Add feedback to sign tool's saved-signature flow to stop rage-clicking

The saved-signature browse/delete/place controls gave too little feedback, so
users re-clicked them repeatedly (the icon-only Use/Remove buttons were the
top rage-click target on /sign, alongside Pause/Resume placement).

- Remove is async (a network round-trip for backend storage) but showed no
  loading state and stayed enabled, so repeated clicks fired several delete
  requests and, once one resolved, the next signature slid into the same card
  slot — looking like nothing happened. It now shows a spinner and disables the
  category controls while the delete is in flight.
- Use is an icon-only control with no confirmation; it now shows a transient
  "Selected" success state so the click clearly registers.
- The saved-signature carousel now wraps around so it is quicker to cycle.
- Pause/Resume placement gained descriptive tooltips and clearer paused-state
  copy explaining what the mode does and how to resume (button or Esc).
- Signing is applied client-side (endpoint: null), so completions never reached
  the shared execute path that emits editor_operation. Emit it on a successful
  apply so abandonment on this flow is measurable.

Generated-By: PostHog Code
Task-Id: 4cd0e9d7-9785-4553-83cf-c6a6ca9c270b

.github/aur/stirling-pdf-desktop/PKGBUILD

@@ -1,6 +1,6 @@
 # Maintainer: Stirling PDF Inc <contact@stirlingpdf.com>
 pkgname=stirling-pdf-desktop
-pkgver=2.14.1
+pkgver=2.14.2
 pkgrel=1
 pkgdesc="Locally hosted, web-based PDF manipulation tool (Tauri desktop app, official Stirling PDF Inc build)"
 arch=('x86_64')

.github/aur/stirling-pdf-server-bin/PKGBUILD

@@ -1,6 +1,6 @@
 # Maintainer: Stirling PDF Inc <contact@stirlingpdf.com>
 pkgname=stirling-pdf-server-bin
-pkgver=2.14.1
+pkgver=2.14.2
 pkgrel=1
 pkgdesc="Locally hosted, web-based PDF manipulation tool (server JAR, prebuilt)"
 arch=('any')

.github/config/.files.yaml

@@ -1,12 +1,12 @@
 build: &build
   - build.gradle
-  - app/(common|core|proprietary)/build.gradle
+  - app/(common|core|proprietary|saas)/build.gradle
   - Taskfile.yml
   - .taskfiles/backend.yml
 
 openapi: &openapi
   - *build
-  - app/(common|core|proprietary)/src/main/java/**
+  - app/(common|core|proprietary|saas)/src/main/java/**
 
 docker-base: &docker-base
   - docker/base/Dockerfile
@@ -23,9 +23,9 @@ docker: &docker
   - *docker-base
 
 project: &project
-  - app/(common|core|proprietary)/src/(main|test)/java/**
+  - app/(common|core|proprietary|saas)/src/(main|test)/java/**
   - *build
-  - "app/(common|core|proprietary)/src/(main|test)/resources/**/!(messages_*.properties|*.md)*"
+  - "app/(common|core|proprietary|saas)/src/(main|test)/resources/**/!(messages_*.properties|*.md)*"
   - exampleYmlFiles/**
   - gradle/**
   - libs/**
@@ -82,7 +82,7 @@ tauri: &tauri
 # tool surfaces it generates models from.
 engine: &engine
   - engine/**
-  - app/(common|core|proprietary)/src/main/java/**
+  - app/(common|core|proprietary|saas)/src/main/java/**
   - .github/workflows/ai-engine.yml
   - Taskfile.yml
   - .taskfiles/engine.yml

.github/labeler-config-srvaroa.yml

@@ -63,6 +63,7 @@ labels:
     files:
       - 'app/core/src/main/resources/static/.*'
       - 'app/proprietary/src/main/resources/static/.*'
+      - 'app/saas/src/main/resources/static/.*'
       - 'frontend/**'
       - 'frontend/.*'
       - 'frontend/**/.*'
@@ -83,6 +84,7 @@ labels:
       - 'app/common/src/main/java/.*.java'
       - 'app/proprietary/src/main/java/.*.java'
       - 'app/core/src/main/java/.*.java'
+      - 'app/saas/src/main/java/.*.java'
 
   - label: 'Back End'
     files:
@@ -90,6 +92,9 @@ labels:
       - 'app/core/src/main/java/stirling/software/SPDF/controller/.*'
       - 'app/core/src/main/resources/settings.yml.template'
       - 'app/core/src/main/resources/application.properties'
+      - 'app/proprietary/src/main/resources/application-proprietary.properties'
+      - 'app/saas/src/main/resources/application-dev.properties'
+      - 'app/saas/src/main/resources/application-saas.properties'
       - 'app/core/src/main/resources/banner.txt'
       - 'app/core/src/main/resources/static/python/png_to_webp.py'
       - 'app/core/src/main/resources/static/python/split_photos.py'
@@ -153,6 +158,7 @@ labels:
       - 'app/common/src/test/.*'
       - 'app/proprietary/src/test/.*'
       - 'app/core/src/test/.*'
+      - 'app/saas/src/test/.*'
       - 'testing/.*'
       - '.github/workflows/scorecards.yml'
       - 'exampleYmlFiles/test_cicd.yml'
@@ -171,3 +177,4 @@ labels:
       - 'app/common/build.gradle'
       - 'app/proprietary/build.gradle'
       - 'app/core/build.gradle'
+      - 'app/saas/build.gradle'

.github/workflows/PR-Auto-Deploy-V2.yml

@@ -256,7 +256,7 @@ jobs:
 
       - name: Build and push V2 image (Docker fork fallback)
         if: env.USE_DEPOT != 'true' && steps.check-image.outputs.exists == 'false'
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           context: .
           file: ./docker/embedded/Dockerfile

.github/workflows/PR-Demo-Comment-with-react.yml

@@ -270,7 +270,7 @@ jobs:
 
       - name: Build and push PR-specific image (Docker fork fallback)
         if: env.USE_DEPOT != 'true'
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           context: .
           file: ./docker/embedded/Dockerfile
@@ -296,7 +296,7 @@ jobs:
 
       - name: Build and push engine image (Docker fork fallback)
         if: env.USE_DEPOT != 'true' && needs.check-comment.outputs.enable_prototypes == 'true'
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           context: ./engine
           file: ./engine/Dockerfile

.github/workflows/backend-build.yml

@@ -47,7 +47,7 @@ jobs:
           distribution: "temurin"
 
       - name: Cache Gradle dependency artifacts
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: |
             ~/.gradle/wrapper

.github/workflows/check-licence.yml

@@ -29,7 +29,7 @@ jobs:
           distribution: "temurin"
 
       - name: Cache Gradle dependency artifacts
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: |
             ~/.gradle/wrapper

.github/workflows/check-openapi.yml

@@ -34,7 +34,7 @@ jobs:
           distribution: "temurin"
 
       - name: Cache Gradle dependency artifacts
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: |
             ~/.gradle/wrapper

.github/workflows/coverage-aggregate.yml

@@ -51,7 +51,7 @@ jobs:
           distribution: "temurin"
 
       - name: Cache Gradle dependency artifacts
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: |
             ~/.gradle/wrapper

.github/workflows/db-migration-test.yml

@@ -37,7 +37,7 @@ jobs:
           distribution: temurin
 
       - name: Cache Gradle dependency artifacts
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: |
             ~/.gradle/wrapper

.github/workflows/deploy-on-v2-commit.yml

@@ -121,7 +121,7 @@ jobs:
 
       - name: Build and push frontend image (Docker fork fallback)
         if: env.USE_DEPOT != 'true' && steps.check-frontend.outputs.exists == 'false'
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           context: .
           file: ./docker/frontend/Dockerfile
@@ -150,7 +150,7 @@ jobs:
 
       - name: Build and push backend image (Docker fork fallback)
         if: env.USE_DEPOT != 'true' && steps.check-backend.outputs.exists == 'false'
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           context: .
           file: ./docker/backend/Dockerfile

.github/workflows/docker-compose-tests.yml

@@ -50,7 +50,7 @@ jobs:
           distribution: "temurin"
 
       - name: Cache Gradle dependency artifacts
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: |
             ~/.gradle/wrapper

.github/workflows/multiOSReleases.yml

@@ -61,7 +61,7 @@ jobs:
           distribution: "temurin"
 
       - name: Cache Gradle dependencies
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: |
             ~/.gradle/caches
@@ -510,6 +510,7 @@ jobs:
       # cargo output unsigned, so checking it produces false negatives.
       - name: Verify Windows Code Signature
         if: ${{ matrix.platform == 'windows-latest' && env.SM_API_KEY != '' && (github.event_name == 'release' || (github.event_name == 'workflow_dispatch' && github.event.inputs.sign != 'false') || github.ref == 'refs/heads/V2-master') }}
+        timeout-minutes: 15
         shell: pwsh
         run: |
           $allSigned = $true
@@ -531,11 +532,26 @@ jobs:
 
           # Extract MSI and verify the inner exe (the file that actually gets installed).
           # This is the critical check - AV flags the installed exe at runtime.
+          # Use lessmsi, not `msiexec /a`: msiexec serializes on the global
+          # _MSIExecute mutex and hangs forever on hosted runners when another
+          # installer is busy. lessmsi reads MSI tables directly - no mutex, no service.
           $msi = $msiFiles[0].FullName
           $extractDir = Join-Path $env:RUNNER_TEMP "msi-verify"
           if (Test-Path $extractDir) { Remove-Item $extractDir -Recurse -Force }
-          $proc = Start-Process msiexec.exe -ArgumentList '/a', $msi, '/qn', "TARGETDIR=$extractDir" -Wait -PassThru -NoNewWindow
-          if ($proc.ExitCode -eq 0) {
+          New-Item -ItemType Directory -Force -Path $extractDir | Out-Null
+
+          choco install lessmsi -y --no-progress --limit-output | Out-Null
+
+          # Bound the extraction and kill on hang (defence in depth over timeout-minutes).
+          $proc = Start-Process lessmsi -ArgumentList 'x', "`"$msi`"", "`"$extractDir\`"" -PassThru -NoNewWindow
+          if (-not $proc.WaitForExit(120000)) {
+            try { $proc.Kill() } catch {}
+            Write-Host "[ERROR] MSI extraction timed out after 120s"
+            $allSigned = $false
+          } elseif ($proc.ExitCode -ne 0) {
+            Write-Host "[ERROR] Failed to extract MSI for verification (exit code: $($proc.ExitCode))"
+            $allSigned = $false
+          } else {
             $innerExe = Get-ChildItem -Path $extractDir -Filter "stirling-pdf.exe" -Recurse -File | Select-Object -First 1
             if ($innerExe) {
               $sig = Get-AuthenticodeSignature -FilePath $innerExe.FullName
@@ -548,9 +564,6 @@ jobs:
               Write-Host "[ERROR] Could not find stirling-pdf.exe inside MSI"
               $allSigned = $false
             }
-          } else {
-            Write-Host "[ERROR] Failed to extract MSI for verification (exit code: $($proc.ExitCode))"
-            $allSigned = $false
           }
 
           if (-not $allSigned) {
@@ -800,7 +813,11 @@ jobs:
         uses: softprops/action-gh-release@b4309332981a82ec1c5618f44dd2e27cc8bfbfda # v3.0.0
         with:
           tag_name: v${{ needs.determine-matrix.outputs.version }}
-          generate_release_notes: true
+          # Don't regenerate/append notes on re-runs, and don't force this into the
+          # "Latest" slot - leave the release body and latest marker as they are.
+          generate_release_notes: false
+          append_body: false
+          make_latest: false
           fail_on_unmatched_files: true
           # Installers + updater payloads + manifest. .sig contents are embedded
           # in latest.json so the .sig files themselves are not uploaded.

.github/workflows/push-docker-base.yml

@@ -85,7 +85,7 @@ jobs:
 
       - name: Build and push base image
         id: build-push-base
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           builder: ${{ steps.buildx.outputs.name }}
           context: docker/base

.github/workflows/push-docker.yml

@@ -66,7 +66,7 @@ jobs:
           distribution: "temurin"
 
       - name: Cache Gradle dependencies
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: |
             ~/.gradle/caches
@@ -145,7 +145,7 @@ jobs:
         id: build-push-latest
         # Empty-tag guard: build-push-action errors when asked to push with no tags.
         if: env.RUN_MAIN_APP == 'true' && steps.meta.outputs.tags != ''
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           builder: ${{ steps.buildx.outputs.name }}
           context: .
@@ -192,7 +192,7 @@ jobs:
 
       - name: Build and push Unified Dockerfile (fat variant)
         id: build-push-fat
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         if: env.RUN_MAIN_APP == 'true' && github.ref != 'refs/heads/main' && github.ref != 'refs/heads/testMain' && steps.meta-fat.outputs.tags != ''
         with:
           builder: ${{ steps.buildx.outputs.name }}
@@ -236,7 +236,7 @@ jobs:
 
       - name: Build and push Unified Dockerfile (ultra-lite variant)
         id: build-push-lite
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         if: env.RUN_MAIN_APP == 'true' && github.ref != 'refs/heads/main' && github.ref != 'refs/heads/testMain' && steps.meta-lite.outputs.tags != ''
         with:
           builder: ${{ steps.buildx.outputs.name }}
@@ -365,7 +365,7 @@ jobs:
       - name: Build and push unoserver image
         id: build-push-unoserver
         if: env.RUN_UNOSERVER == 'true' && steps.unoserverDecision.outputs.mode != 'skip'
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           builder: ${{ steps.buildx.outputs.name }}
           context: .

.github/workflows/test-build-docker.yml

@@ -95,7 +95,7 @@ jobs:
           distribution: "temurin"
 
       - name: Cache Gradle dependency artifacts
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: |
             ~/.gradle/wrapper
@@ -186,7 +186,7 @@ jobs:
       # (multi-platform + gha cache) against the published base image.
       - name: Build ${{ matrix.docker-rev }} (Docker fork fallback)
         if: env.USE_DEPOT != 'true' && inputs.docker-base-changed != 'true'
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           builder: ${{ steps.buildx.outputs.name }}
           context: .
@@ -259,7 +259,7 @@ jobs:
 
       - name: Build docker/unoserver/Dockerfile (Docker fork fallback)
         if: env.USE_DEPOT != 'true'
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           builder: ${{ steps.buildx.outputs.name }}
           context: .

.github/workflows/testdriver.yml

@@ -95,7 +95,7 @@ jobs:
 
       - name: Build and push test image (Docker fork fallback)
         if: env.USE_DEPOT != 'true'
-        uses: docker/build-push-action@bcafcacb16a39f128d818304e6c9c0c18556b85f # v7.1.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           context: .
           file: ./docker/embedded/Dockerfile

.imgbotconfig

@@ -0,0 +1,5 @@
+{
+  "ignoredFiles": [
+    "frontend/editor/src-tauri/icons/icon.png"
+  ]
+}

.taskfiles/frontend.yml

@@ -80,6 +80,12 @@ tasks:
       OPEN: '{{.OPEN | default ""}}'
     env:
       BACKEND_URL: '{{.BACKEND_URL}}'
+      # Dev-only browser-tab label so concurrent worktrees are distinguishable.
+      # Only the worktree folder basename (e.g. "wt1") is exposed — never the
+      # full path, hostname, or user. Consumed at dev-serve time by vite.config
+      # and dropped from production builds.
+      STIRLING_DEV_LABEL:
+        sh: basename "$(git rev-parse --show-toplevel 2>/dev/null || pwd)"
     cmds:
       - npx vite editor --mode {{.MODE}} --port {{.PORT}}{{if .OPEN}} --open{{end}}
 

app/allowed-licenses.json

@@ -208,6 +208,18 @@
       "moduleName": ".*",
       "moduleLicense": "The W3C License"
     },
+    {
+      "moduleName": "com.google.re2j:re2j",
+      "moduleLicense": "Go License"
+    },
+    {
+      "moduleName": "com.hubspot:algebra",
+      "moduleLicense": null
+    },
+    {
+      "moduleName": "com.hubspot.immutables:immutables-exceptions",
+      "moduleLicense": null
+    },
     {
       "moduleName": ".*",
       "moduleLicense": "UnRar License"

app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java

@@ -246,6 +246,14 @@ public class ApplicationProperties {
          * and paused runs are kept regardless of age.
          */
         private int runExpiryMinutes = 30;
+
+        /**
+         * Whether a policy S3 source's custom endpoint may resolve to a loopback, link-local, or
+         * private address. Off by default so a user-supplied endpoint cannot be pointed at internal
+         * services (e.g. the cloud metadata address); enable for a self-hosted MinIO or other
+         * in-network object store.
+         */
+        private boolean allowPrivateS3Endpoints = false;
     }
 
     @Data

app/common/src/main/java/stirling/software/common/service/AutomationRunContext.java

@@ -0,0 +1,56 @@
+package stirling.software.common.service;
+
+/**
+ * Thread-scoped correlation id for one automation run — a single pipeline, policy, or AI-workflow
+ * execution over its input file(s).
+ *
+ * <p>Automations dispatch each tool step as a separate internal loopback POST via {@link
+ * InternalApiClient}. The orchestrator opens a run scope around its dispatch loop; {@code
+ * InternalApiClient} reads {@link #current()} and stamps it on every sub-step request as {@link
+ * #RUN_ID_HEADER}. The SaaS PAYG interceptor uses that header so all sub-steps of ONE run group
+ * into a single charge, while two <em>separate</em> runs that happen to touch identical bytes stay
+ * distinct charges (the old content+time-window grouping merged them).
+ *
+ * <p>Sub-steps dispatch synchronously on the orchestrator's own thread (loopback {@code
+ * RestTemplate}), so this ThreadLocal is visible to {@code InternalApiClient}. The id then crosses
+ * to the receiving request thread via the HTTP header — never via this ThreadLocal.
+ *
+ * <p>No-op when the id is absent (a standalone tool call): the interceptor treats a missing run id
+ * as "its own charge", which is exactly what a one-off call should be.
+ */
+public final class AutomationRunContext {
+
+    /** Header carrying the run id on internal sub-step dispatches. */
+    public static final String RUN_ID_HEADER = "X-Stirling-Run-Id";
+
+    private static final ThreadLocal<String> CURRENT = new ThreadLocal<>();
+
+    private AutomationRunContext() {}
+
+    /**
+     * Opens a run scope on the current thread. Returns an {@link AutoCloseable} that restores the
+     * previously-active id (nesting-safe) — use in try-with-resources around the dispatch loop.
+     */
+    public static Scope open(String runId) {
+        String previous = CURRENT.get();
+        CURRENT.set(runId);
+        return () -> {
+            if (previous == null) {
+                CURRENT.remove();
+            } else {
+                CURRENT.set(previous);
+            }
+        };
+    }
+
+    /** The run id active on this thread, or {@code null} when not inside a run scope. */
+    public static String current() {
+        return CURRENT.get();
+    }
+
+    /** AutoCloseable whose {@link #close()} declares no checked exception. */
+    public interface Scope extends AutoCloseable {
+        @Override
+        void close();
+    }
+}

app/common/src/main/java/stirling/software/common/service/InternalApiClient.java

@@ -8,6 +8,7 @@ import java.nio.file.Files;
 import java.time.Duration;
 import java.util.regex.Pattern;
 
+import org.slf4j.MDC;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.env.Environment;
 import org.springframework.core.io.FileSystemResource;
@@ -60,6 +61,17 @@ public class InternalApiClient {
      */
     public static final String AUTOMATION_HEADER = "X-Stirling-Automation";
 
+    /**
+     * Header carrying the parent policy's name onto each sub-step dispatch, read from MDC key
+     * {@link #POLICY_NAME_MDC_KEY} (set by the policy runner on the worker thread). Lets the audit
+     * layer attribute a tool step to the policy that ran it, instead of showing it as a bare direct
+     * call.
+     */
+    public static final String POLICY_NAME_HEADER = "X-Stirling-Policy-Name";
+
+    /** MDC key the policy runner stamps with the running policy's name; forwarded as a header. */
+    public static final String POLICY_NAME_MDC_KEY = "auditPolicyName";
+
     private final ServletContext servletContext;
     private final UserServiceInterface userService;
     private final TempFileManager tempFileManager;
@@ -111,6 +123,27 @@ public class InternalApiClient {
         // step inside a policy run must bill as AUTOMATION, not AI). Set unconditionally because
         // every caller of this dispatcher is an automation surface by design.
         headers.add(AUTOMATION_HEADER, "true");
+        // Propagate the current automation run id (set by the orchestrator around its dispatch
+        // loop) so the PAYG interceptor groups every sub-step of this one run into a single charge,
+        // and never merges two separate runs that happen to touch identical bytes. Absent → the
+        // receiving call is treated as standalone. See AutomationRunContext.
+        String runId = AutomationRunContext.current();
+        if (runId != null && !runId.isEmpty()) {
+            headers.add(AutomationRunContext.RUN_ID_HEADER, runId);
+        }
+
+        // Forward the parent policy name (set in MDC by the policy runner) so the audited sub-step
+        // ties back to its policy. Single-line, length-capped: it becomes an HTTP header value.
+        String policyName = MDC.get(POLICY_NAME_MDC_KEY);
+        if (policyName != null && !policyName.isBlank()) {
+            String safe = policyName.replaceAll("[\\r\\n]", " ").trim();
+            if (safe.length() > 200) {
+                safe = safe.substring(0, 200);
+            }
+            if (!safe.isEmpty()) {
+                headers.add(POLICY_NAME_HEADER, safe);
+            }
+        }
 
         // A no-file ai/tools call (e.g. create-pdf-from-html-agent) sends only string params, so
         // without this RestTemplate would use urlencoded instead of the multipart the controller

app/common/src/main/java/stirling/software/common/service/TempFileCleanupService.java

@@ -144,8 +144,10 @@ public class TempFileCleanupService {
         int directoriesDeletedCount = 0;
         for (Path directory : registry.getTempDirectories()) {
             try {
-                if (Files.exists(directory)) {
+                if (Files.exists(directory)
+                        && shouldDeleteRegisteredDirectory(directory, maxAgeMillis)) {
                     GeneralUtils.deleteDirectory(directory);
+                    registry.unregisterDirectory(directory);
                     directoriesDeletedCount++;
                     log.debug("Cleaned up temporary directory: {}", directory);
                 }
@@ -275,6 +277,21 @@ public class TempFileCleanupService {
         return totalDeletedCount.get();
     }
 
+    private boolean shouldDeleteRegisteredDirectory(Path directory, long maxAgeMillis) {
+        if (maxAgeMillis <= 0) {
+            return true;
+        }
+
+        try {
+            long currentTime = System.currentTimeMillis();
+            long lastModified = Files.getLastModifiedTime(directory).toMillis();
+            return (currentTime - lastModified) > maxAgeMillis;
+        } catch (IOException e) {
+            log.debug("Could not check directory age, skipping cleanup: {}", directory, e);
+            return false;
+        }
+    }
+
     /** Get the system temp directory path based on configuration or system property. */
     private Path getSystemTempPath() {
         String systemTempDir =

app/common/src/main/java/stirling/software/common/util/TempFileManager.java

@@ -155,6 +155,7 @@ public class TempFileManager {
         if (directory != null && Files.isDirectory(directory)) {
             try {
                 GeneralUtils.deleteDirectory(directory);
+                registry.unregisterDirectory(directory);
                 log.debug("Deleted temp directory: {}", directory.toString());
             } catch (IOException e) {
                 log.warn("Failed to delete temp directory: {}", directory.toString(), e);

app/common/src/main/java/stirling/software/common/util/TempFileRegistry.java

@@ -85,6 +85,18 @@ public class TempFileRegistry {
         return directory;
     }
 
+    /**
+     * Unregister a temporary directory from the registry.
+     *
+     * @param directory The directory to unregister
+     */
+    public void unregisterDirectory(Path directory) {
+        if (directory != null) {
+            tempDirectories.remove(directory);
+            log.debug("Unregistered temp directory: {}", directory.toString());
+        }
+    }
+
     /**
      * Register a third-party temporary file that requires special handling.
      *

app/common/src/test/java/stirling/software/common/service/TempFileCleanupServiceMoreTest.java

@@ -176,11 +176,13 @@ class TempFileCleanupServiceMoreTest {
     class ScheduledCleanup {
 
         @Test
-        @DisplayName("deletes registered temp directories and reports counts")
+        @DisplayName("deletes stale registered temp directories and reports counts")
         void deletesRegisteredDirectories() throws IOException {
             when(tempFileManager.cleanupOldTempFiles(anyLong())).thenReturn(2);
             Path regDir = Files.createDirectories(tempDir.resolve("registeredDir"));
             Files.createFile(regDir.resolve("inside.txt"));
+            Files.setLastModifiedTime(
+                    regDir, FileTime.fromMillis(System.currentTimeMillis() - 2L * 60 * 60 * 1000));
             Set<Path> dirs = new HashSet<>();
             dirs.add(regDir);
             when(registry.getTempDirectories()).thenReturn(dirs);
@@ -193,6 +195,22 @@ class TempFileCleanupServiceMoreTest {
             verify(tempFileManager).cleanupOldTempFiles(anyLong());
         }
 
+        @Test
+        @DisplayName("keeps a fresh registered temp directory")
+        void keepsFreshRegisteredDirectory() throws IOException {
+            when(tempFileManager.cleanupOldTempFiles(anyLong())).thenReturn(0);
+            Path regDir = Files.createDirectories(tempDir.resolve("freshRegisteredDir"));
+            Files.createFile(regDir.resolve("inside.txt"));
+            Set<Path> dirs = new HashSet<>();
+            dirs.add(regDir);
+            when(registry.getTempDirectories()).thenReturn(dirs);
+            lenient().when(registry.contains(any(File.class))).thenReturn(false);
+
+            withIsolatedUserHome(cleanupService::scheduledCleanup);
+
+            assertThat(Files.exists(regDir)).isTrue();
+        }
+
         @Test
         @DisplayName("skips a registered directory that no longer exists")
         void skipsMissingRegisteredDirectory() {

app/core/src/main/java/stirling/software/SPDF/controller/api/pipeline/PipelineProcessor.java

@@ -10,6 +10,7 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.UUID;
 
 import org.springframework.core.io.FileSystemResource;
 import org.springframework.core.io.Resource;
@@ -27,6 +28,7 @@ import stirling.software.SPDF.model.PipelineConfig;
 import stirling.software.SPDF.model.PipelineOperation;
 import stirling.software.SPDF.model.PipelineResult;
 import stirling.software.SPDF.service.ApiDocService;
+import stirling.software.common.service.AutomationRunContext;
 import stirling.software.common.service.InternalApiClient;
 import stirling.software.common.util.TempFileManager;
 import stirling.software.common.util.ZipExtractionUtils;
@@ -71,6 +73,17 @@ public class PipelineProcessor {
 
     PipelineResult runPipelineAgainstFiles(List<Resource> outputFiles, PipelineConfig config)
             throws Exception {
+        // One pipeline execution = one automation run. Scope a run id so every tool sub-step
+        // dispatched via InternalApiClient groups into a single charge on the SaaS billing side
+        // (see AutomationRunContext); pipeline steps run synchronously on this thread.
+        try (AutomationRunContext.Scope ignored =
+                AutomationRunContext.open(UUID.randomUUID().toString())) {
+            return runPipelineAgainstFilesInternal(outputFiles, config);
+        }
+    }
+
+    private PipelineResult runPipelineAgainstFilesInternal(
+            List<Resource> outputFiles, PipelineConfig config) throws Exception {
         PipelineResult result = new PipelineResult();
 
         ByteArrayOutputStream logStream = new ByteArrayOutputStream();

app/core/src/main/java/stirling/software/SPDF/controller/api/security/RedactController.java

@@ -136,11 +136,17 @@ public class RedactController {
                             + "Users can provide text patterns to redact, with options for regex and whole word matching. "
                             + "Input:PDF Output:PDF Type:SISO")
     public ResponseEntity<Resource> redactPdf(@ModelAttribute RedactPdfRequest request) {
-        String[] listOfText = request.getListOfText().split("\n");
+        String rawListOfText = request.getListOfText();
         boolean useRegex = Boolean.TRUE.equals(request.getUseRegex());
         boolean wholeWordSearchBool = Boolean.TRUE.equals(request.getWholeWordSearch());
 
-        if (listOfText.length == 0 || (listOfText.length == 1 && listOfText[0].trim().isEmpty())) {
+        if (rawListOfText == null || rawListOfText.trim().isEmpty()) {
+            throw ExceptionUtils.createIllegalArgumentException(
+                    "error.redaction.no.patterns", "No text patterns provided for redaction");
+        }
+
+        String[] listOfText = rawListOfText.split("\n");
+        if (listOfText.length == 1 && listOfText[0].trim().isEmpty()) {
             throw ExceptionUtils.createIllegalArgumentException(
                     "error.redaction.no.patterns", "No text patterns provided for redaction");
         }

app/core/src/main/resources/static/images/redact-auto.svg

@@ -1,18 +1 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<svg xmlns="http://www.w3.org/2000/svg" version="1.1" xmlns:xlink="http://www.w3.org/1999/xlink" viewBox="0 0 24 24">
-        <symbol id="icon-redact-auto" viewBox="0 0 24 24">  <g id="Layer_2" data-name="Layer 2">
-    <g id="Layer_1-2" data-name="Layer 1">
-      <rect width="24" height="24" style="fill: none"/>
-      <g>
-        <path d="M17.541,15.64258a.91793.91793,0,0,1,.55469-.18555h1.1084a.91586.91586,0,0,1,.55469.18555,1.30889,1.30889,0,0,1,.40429.499,1.57206,1.57206,0,0,1,.15039.68457v5.47754H19.2041V20.21094H18.0957v2.09277H16.9873V16.82617a1.55843,1.55843,0,0,1,.15039-.68457A1.2979,1.2979,0,0,1,17.541,15.64258Zm1.66309,1.10547H18.0957v2.17187h1.1084Z" style="fill: currentColor"/>
-        <path d="M5.68653,22.30351a2.00588,2.00588,0,0,1-2-2v-16A1.92585,1.92585,0,0,1,4.274,2.891a1.92585,1.92585,0,0,1,1.4125-.5875h8l6,6v5.66931h-2V9.30351h-5v-5h-7v16h9.74021v2Z" style="fill: currentColor"/>
-        <rect x="7.69809" y="10.43189" width="4.33778" height="0.79501" style="fill: currentColor"/>
-        <rect x="7.69809" y="12.16889" width="7.31192" height="1.21288" style="fill: currentColor"/>
-        <rect x="7.69809" y="17.14555" width="7.31192" height="1.21288" style="fill: currentColor"/>
-        <rect x="7.69809" y="14.32375" width="7.31192" height="0.57517" style="fill: currentColor"/>
-        <rect x="7.69809" y="15.79848" width="5.25578" height="0.4475" style="fill: currentColor"/>
-      </g>
-    </g>
-  </g>
-    </symbol>
-</svg>
+<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" version="1.1" viewBox="0 0 24 24"><symbol id="icon-redact-auto" viewBox="0 0 24 24"><g id="Layer_2" data-name="Layer 2"><g id="Layer_1-2" data-name="Layer 1"><rect width="24" height="24" style="fill:none"/><g><path d="M17.541,15.64258a.91793.91793,0,0,1,.55469-.18555h1.1084a.91586.91586,0,0,1,.55469.18555,1.30889,1.30889,0,0,1,.40429.499,1.57206,1.57206,0,0,1,.15039.68457v5.47754H19.2041V20.21094H18.0957v2.09277H16.9873V16.82617a1.55843,1.55843,0,0,1,.15039-.68457A1.2979,1.2979,0,0,1,17.541,15.64258Zm1.66309,1.10547H18.0957v2.17187h1.1084Z" style="fill:currentColor"/><path d="M5.68653,22.30351a2.00588,2.00588,0,0,1-2-2v-16A1.92585,1.92585,0,0,1,4.274,2.891a1.92585,1.92585,0,0,1,1.4125-.5875h8l6,6v5.66931h-2V9.30351h-5v-5h-7v16h9.74021v2Z" style="fill:currentColor"/><rect width="4.338" height=".795" x="7.698" y="10.432" style="fill:currentColor"/><rect width="7.312" height="1.213" x="7.698" y="12.169" style="fill:currentColor"/><rect width="7.312" height="1.213" x="7.698" y="17.146" style="fill:currentColor"/><rect width="7.312" height=".575" x="7.698" y="14.324" style="fill:currentColor"/><rect width="5.256" height=".448" x="7.698" y="15.798" style="fill:currentColor"/></g></g></g></symbol></svg>

app/core/src/main/resources/static/images/rename.svg

@@ -1,13 +1 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!-- Uploaded to: SVG Repo, www.svgrepo.com, Generator: SVG Repo Mixer Tools -->
-<svg width="800px" height="800px" viewBox="0 0 512 512" version="1.1" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">
-    <title>rename</title>
-        <symbol id="icon-rename" viewBox="0 0 512 512">
-    <g id="Page-1" stroke="none" stroke-width="1" fill="none" fill-rule="evenodd">
-        <g id="Combined-Shape" fill="currentColor" transform="translate(42.666667, 64.000000)">
-            <path d="M362.666667,1.42108547e-14 L362.666667,21.3333333 L320,21.333 L320,362.666 L362.666667,362.666667 L362.666667,384 L320,383.999 L320,384 L298.666667,384 L298.666,383.999 L256,384 L256,362.666667 L298.666,362.666 L298.666,21.333 L256,21.3333333 L256,1.42108547e-14 L362.666667,1.42108547e-14 Z M426.666667,64 L426.666667,320 L341.333333,320 L341.333333,277.333333 L384,277.333333 L384,106.666667 L341.333333,106.666667 L341.333333,64 L426.666667,64 Z M277.333333,64 L277.333333,320 L3.55271368e-14,320 L3.55271368e-14,64 L277.333333,64 Z M179.2,89.6 L149.333333,89.6 L149.333333,234.666667 C149.333333,248 148.5,256.333333 147.875,264.354167 L147.792993,265.422171 L147.792993,265.422171 L147.714003,266.48894 C147.417695,270.579012 147.2,274.696296 147.2,279.466667 L147.2,279.466667 L177.066667,279.466667 L177.066667,260.266667 C184.941497,273.926888 199.708077,282.130544 215.466667,281.6 C229.540046,281.805757 242.921593,275.508559 251.733333,264.533333 C263.162478,248.989677 269.832496,230.461848 270.933333,211.2 C270.933333,170.666667 249.6,142.933333 217.6,142.933333 C202.507405,142.999748 188.308689,150.099106 179.2,162.133333 L179.2,162.133333 L179.2,89.6 Z M119.466667,162.133333 C107.961824,149.843793 91.4322333,143.546807 74.6666667,145.066667 C57.6785115,144.485924 40.8138255,148.15216 25.6,155.733333 L25.6,155.733333 L34.1333333,177.066667 C45.3979052,171.147831 57.7246848,167.522308 70.4,166.4 C78.5613135,165.511423 86.6853595,168.371259 92.4903835,174.176283 C98.2954074,179.981307 101.155244,188.105353 100.266667,196.266667 L100.266667,196.266667 L100.266667,198.4 L78.9333333,198.4 C65.8181975,197.679203 52.705771,199.864608 40.5333333,204.8 C26.2806563,210.950309 17.6507691,225.621117 19.2,241.066667 C19.0625857,252.057651 23.6679763,262.574827 31.8381493,269.927982 C40.0083223,277.281138 50.9508304,280.757072 61.8666667,279.466667 C77.2795695,280.291768 92.2192911,274.001359 102.4,262.4 L102.4,262.4 L102.4,277.333333 L130.133333,277.333333 C128.292479,266.054406 127.577851,254.620365 128,243.2 L128,243.2 L128,204.8 C129.999138,190.023932 126.995128,175.003882 119.466667,162.133333 Z M98.1333333,213.333333 L98.1333333,238.933333 C92.082572,249.988391 80.836024,257.218314 68.2666667,258.133333 C63.0655139,258.520242 57.9538681,256.621996 54.2659359,252.934064 C50.5780036,249.246132 48.6797582,244.134486 49.0666667,238.933333 C49.0666667,224 59.7333333,215.466667 85.3333333,213.333333 L85.3333333,213.333333 L98.1333333,213.333333 Z M209.066667,166.4 C226.133333,166.4 238.933333,183.466667 238.933333,211.2 C238.933333,238.933333 228.266667,256 211.2,256 C197.298049,255.69869 184.825037,247.383349 179.2,234.666667 L179.2,234.666667 L179.2,187.733333 C185.154203,176.240507 196.263981,168.304951 209.066667,166.4 Z">
-</path>
-        </g>
-    </g>
-    </symbol>
-</svg>
+<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="800" height="800" version="1.1" viewBox="0 0 512 512"><title>rename</title><symbol id="icon-rename" viewBox="0 0 512 512"><g id="Page-1" fill="none" fill-rule="evenodd" stroke="none" stroke-width="1"><g id="Combined-Shape" fill="currentColor"><path d="M362.666667,1.42108547e-14 L362.666667,21.3333333 L320,21.333 L320,362.666 L362.666667,362.666667 L362.666667,384 L320,383.999 L320,384 L298.666667,384 L298.666,383.999 L256,384 L256,362.666667 L298.666,362.666 L298.666,21.333 L256,21.3333333 L256,1.42108547e-14 L362.666667,1.42108547e-14 Z M426.666667,64 L426.666667,320 L341.333333,320 L341.333333,277.333333 L384,277.333333 L384,106.666667 L341.333333,106.666667 L341.333333,64 L426.666667,64 Z M277.333333,64 L277.333333,320 L3.55271368e-14,320 L3.55271368e-14,64 L277.333333,64 Z M179.2,89.6 L149.333333,89.6 L149.333333,234.666667 C149.333333,248 148.5,256.333333 147.875,264.354167 L147.792993,265.422171 L147.792993,265.422171 L147.714003,266.48894 C147.417695,270.579012 147.2,274.696296 147.2,279.466667 L147.2,279.466667 L177.066667,279.466667 L177.066667,260.266667 C184.941497,273.926888 199.708077,282.130544 215.466667,281.6 C229.540046,281.805757 242.921593,275.508559 251.733333,264.533333 C263.162478,248.989677 269.832496,230.461848 270.933333,211.2 C270.933333,170.666667 249.6,142.933333 217.6,142.933333 C202.507405,142.999748 188.308689,150.099106 179.2,162.133333 L179.2,162.133333 L179.2,89.6 Z M119.466667,162.133333 C107.961824,149.843793 91.4322333,143.546807 74.6666667,145.066667 C57.6785115,144.485924 40.8138255,148.15216 25.6,155.733333 L25.6,155.733333 L34.1333333,177.066667 C45.3979052,171.147831 57.7246848,167.522308 70.4,166.4 C78.5613135,165.511423 86.6853595,168.371259 92.4903835,174.176283 C98.2954074,179.981307 101.155244,188.105353 100.266667,196.266667 L100.266667,196.266667 L100.266667,198.4 L78.9333333,198.4 C65.8181975,197.679203 52.705771,199.864608 40.5333333,204.8 C26.2806563,210.950309 17.6507691,225.621117 19.2,241.066667 C19.0625857,252.057651 23.6679763,262.574827 31.8381493,269.927982 C40.0083223,277.281138 50.9508304,280.757072 61.8666667,279.466667 C77.2795695,280.291768 92.2192911,274.001359 102.4,262.4 L102.4,262.4 L102.4,277.333333 L130.133333,277.333333 C128.292479,266.054406 127.577851,254.620365 128,243.2 L128,243.2 L128,204.8 C129.999138,190.023932 126.995128,175.003882 119.466667,162.133333 Z M98.1333333,213.333333 L98.1333333,238.933333 C92.082572,249.988391 80.836024,257.218314 68.2666667,258.133333 C63.0655139,258.520242 57.9538681,256.621996 54.2659359,252.934064 C50.5780036,249.246132 48.6797582,244.134486 49.0666667,238.933333 C49.0666667,224 59.7333333,215.466667 85.3333333,213.333333 L85.3333333,213.333333 L98.1333333,213.333333 Z M209.066667,166.4 C226.133333,166.4 238.933333,183.466667 238.933333,211.2 C238.933333,238.933333 228.266667,256 211.2,256 C197.298049,255.69869 184.825037,247.383349 179.2,234.666667 L179.2,234.666667 L179.2,187.733333 C185.154203,176.240507 196.263981,168.304951 209.066667,166.4 Z" transform="translate(42.666667, 64.000000)"/></g></g></symbol></svg>

app/core/src/test/java/stirling/software/SPDF/controller/api/security/RedactControllerMoreTest.java

@@ -299,6 +299,18 @@ class RedactControllerMoreTest {
             verify(pdfDocumentFactory, never()).load(any(MultipartFile.class));
         }
 
+        @Test
+        @DisplayName("null listOfText throws an illegal-argument error before any load")
+        void nullPatternsThrows() throws Exception {
+            RedactPdfRequest request = new RedactPdfRequest();
+            request.setFileInput(pdfFile(new byte[] {1, 2, 3}));
+            request.setListOfText(null);
+
+            assertThatThrownBy(() -> controller.redactPdf(request))
+                    .isInstanceOf(RuntimeException.class);
+            verify(pdfDocumentFactory, never()).load(any(MultipartFile.class));
+        }
+
         @Test
         @DisplayName("null file input is reported as a failure")
         void nullFileThrows() {

app/proprietary/build.gradle

@@ -66,6 +66,20 @@ dependencies {
 
     implementation "com.google.code.gson:gson:${gsonVersion}"
 
+    // jinjava/jjwt transitively request older Jackson 2 versions; declare the current
+    // version directly so it is selected consistently (root build.gradle pins are the fallback).
+    runtimeOnly "com.fasterxml.jackson.core:jackson-core:${jackson2Version}"
+    runtimeOnly "com.fasterxml.jackson.core:jackson-databind:${jackson2Version}"
+
+    implementation("com.hubspot.jinjava:jinjava:${jinjavaVersion}") {
+        // Compile-time-only annotation artifacts (class-retention annotations, not needed at
+        // runtime) whose declared licences (LGPL / none) fail the licence compatibility check.
+        exclude group: 'com.google.code.findbugs', module: 'annotations'
+        exclude group: 'org.derive4j', module: 'derive4j-annotation'
+        exclude group: 'com.hubspot.immutables', module: 'hubspot-style'
+        exclude group: 'com.hubspot.immutables', module: 'immutable-collection-encodings'
+    }
+
     api 'io.micrometer:micrometer-registry-prometheus'
 
     api "io.jsonwebtoken:jjwt-api:${jwtVersion}"

app/proprietary/src/main/java/stirling/software/proprietary/audit/AuditAspect.java

@@ -142,6 +142,9 @@ public class AuditAspect {
             auditService.addTimingData(
                     auditData, startTime, resp, auditedAnnotation.level(), isHttpRequest);
 
+            // Merge controller-set policy context + the internal-automation marker onto the event.
+            auditService.addAutomationContext(auditData, req);
+
             // Resolve the event type based on annotation and context
             String httpMethod = null;
             String path = null;

app/proprietary/src/main/java/stirling/software/proprietary/audit/AuditContext.java

@@ -0,0 +1,18 @@
+package stirling.software.proprietary.audit;
+
+/**
+ * Request-scoped keys a controller can set to enrich its own audit event with context the generic
+ * aspect can't infer from the HTTP request alone (e.g. the policy a pipeline run belongs to). The
+ * aspect copies these into the audit data in its {@code finally} block, after the controller body
+ * has run. See {@code AuditService#addAutomationContext}.
+ */
+public final class AuditContext {
+
+    /** Request attribute: the name of the policy/pipeline a run executes. */
+    public static final String REQ_ATTR_POLICY_NAME = "stirling.audit.policyName";
+
+    /** Request attribute: the ordered tool endpoint paths a run executes. */
+    public static final String REQ_ATTR_POLICY_STEPS = "stirling.audit.policySteps";
+
+    private AuditContext() {}
+}

app/proprietary/src/main/java/stirling/software/proprietary/audit/ControllerAuditAspect.java

@@ -205,6 +205,10 @@ public class ControllerAuditAspect {
                 // Call auditService but with isHttpRequest=true to skip additional timing
                 auditService.addTimingData(data, start, resp, level, true);
 
+                // Merge controller-set policy context + the internal-automation marker (set after
+                // the body ran, so it must happen here rather than with the pre-proceed HTTP data).
+                auditService.addAutomationContext(data, req);
+
                 // Resolve the event type using the unified method
                 AuditEventType eventType =
                         auditService.resolveEventType(

app/proprietary/src/main/java/stirling/software/proprietary/cluster/s3/S3Clients.java

@@ -133,29 +133,45 @@ public final class S3Clients {
      * storage.s3.allow-private-endpoints=true}.
      */
     static void validateEndpointHost(URI endpoint, boolean allowPrivate) {
+        validateEndpointHost(
+                endpoint,
+                allowPrivate,
+                "storage.s3.endpoint",
+                "set storage.s3.allow-private-endpoints=true to opt in"
+                        + " (e.g. for MinIO or in-cluster S3).");
+    }
+
+    /**
+     * The same private-address guard for S3 endpoints configured outside the {@code storage.s3.*}
+     * block (e.g. per-source policy config), with the setting named in messages supplied by the
+     * caller.
+     */
+    public static void validateEndpointHost(
+            URI endpoint, boolean allowPrivate, String settingName, String optInHint) {
         if (allowPrivate) {
             return;
         }
         String host = endpoint.getHost();
         if (host == null || host.isBlank()) {
-            throw new IllegalStateException("storage.s3.endpoint must include a host: " + endpoint);
+            throw new IllegalStateException(settingName + " must include a host: " + endpoint);
         }
         InetAddress[] addresses;
         try {
             addresses = InetAddress.getAllByName(host);
         } catch (UnknownHostException e) {
             throw new IllegalStateException(
-                    "Unable to resolve storage.s3.endpoint host '" + host + "'", e);
+                    "Unable to resolve " + settingName + " host '" + host + "'", e);
         }
         for (InetAddress address : addresses) {
             if (isPrivateOrLocal(address)) {
                 throw new IllegalStateException(
-                        "storage.s3.endpoint host '"
+                        settingName
+                                + " host '"
                                 + host
                                 + "' resolves to private/link-local address "
                                 + address.getHostAddress()
-                                + "; set storage.s3.allow-private-endpoints=true to opt in"
-                                + " (e.g. for MinIO or in-cluster S3).");
+                                + "; "
+                                + optInHint);
             }
         }
     }

app/proprietary/src/main/java/stirling/software/proprietary/controller/api/CreatePdfAgentController.java

@@ -8,12 +8,14 @@ import java.util.List;
 
 import org.apache.pdfbox.pdmodel.PDDocument;
 import org.springframework.core.io.Resource;
+import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ResponseStatusException;
 
 import io.github.pixee.security.Filenames;
 import io.swagger.v3.oas.annotations.Hidden;
@@ -24,18 +26,24 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.configuration.RuntimePathConfig;
+import stirling.software.common.model.ApplicationProperties;
 import stirling.software.common.service.CustomPDFDocumentFactory;
 import stirling.software.common.util.ProcessExecutor;
 import stirling.software.common.util.TempFile;
 import stirling.software.common.util.TempFileManager;
 import stirling.software.common.util.WebResponseUtils;
+import stirling.software.proprietary.model.api.ai.create.AiDocument;
+import stirling.software.proprietary.service.AiDocumentHtmlRenderer;
+
+import tools.jackson.core.JacksonException;
+import tools.jackson.databind.ObjectMapper;
 
 /**
- * Dispatchable tool that converts an AI-generated HTML string to a PDF via WeasyPrint.
+ * Dispatchable tool that converts an AI-generated document model to a PDF via WeasyPrint.
  *
  * <p>Called by {@link stirling.software.proprietary.service.AiWorkflowService} when the engine
- * emits a {@code CREATE_PDF_FROM_HTML_AGENT} plan step. The HTML comes from a trusted Jinja
- * template so sanitization is intentionally skipped.
+ * emits a {@code CREATE_PDF_FROM_HTML_AGENT} plan step. The engine supplies the document as
+ * structured fields; the HTML is built here from a fixed template.
  */
 @Slf4j
 @Hidden
@@ -48,6 +56,9 @@ public class CreatePdfAgentController {
     private final TempFileManager tempFileManager;
     private final CustomPDFDocumentFactory pdfDocumentFactory;
     private final RuntimePathConfig runtimePathConfig;
+    private final ApplicationProperties applicationProperties;
+    private final ObjectMapper objectMapper;
+    private final AiDocumentHtmlRenderer htmlRenderer;
 
     /**
      * Returns true only when WeasyPrint is definitively unavailable — either the binary could not
@@ -74,32 +85,42 @@ public class CreatePdfAgentController {
             value = "/create-pdf-from-html-agent",
             consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
     @Operation(
-            summary = "Convert AI-generated HTML to a PDF",
+            summary = "Convert an AI-generated document to a PDF",
             description =
-                    "Accepts an HTML document as a plain-text parameter and returns a PDF."
-                            + " This endpoint is dispatched by the AI workflow orchestrator as a"
-                            + " plan step; it is not intended for direct client use.")
-    public ResponseEntity<Resource> createPdfFromHtml(
-            @RequestParam("htmlContent") String htmlContent,
-            @RequestParam("filename") String filename)
+                    "Accepts a structured document as a JSON parameter and returns a PDF. This"
+                            + " endpoint is dispatched by the AI workflow orchestrator as a plan"
+                            + " step; it is not intended for direct client use.")
+    public ResponseEntity<Resource> createPdf(
+            @RequestParam("document") String document, @RequestParam("filename") String filename)
             throws Exception {
 
+        if (!applicationProperties.getAiEngine().isEnabled()) {
+            throw new ResponseStatusException(HttpStatus.NOT_FOUND);
+        }
+
+        AiDocument model;
+        try {
+            model = objectMapper.readValue(document, AiDocument.class);
+        } catch (JacksonException e) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST);
+        }
+
+        String html = htmlRenderer.render(model);
+
         log.info(
-                "[create-pdf-agent] converting HTML to PDF via WeasyPrint — html_bytes={}",
-                htmlContent.length());
+                "[create-pdf-agent] converting document to PDF via WeasyPrint — html_bytes={}",
+                html.length());
 
         try (TempFile htmlFile = tempFileManager.createManagedTempFile(".html");
                 TempFile pdfFile = tempFileManager.createManagedTempFile(".pdf")) {
 
-            Files.writeString(htmlFile.getPath(), htmlContent, StandardCharsets.UTF_8);
+            Files.writeString(htmlFile.getPath(), html, StandardCharsets.UTF_8);
 
             List<String> command = new ArrayList<>();
             command.add(runtimePathConfig.getWeasyPrintPath());
             command.add("-e");
             command.add("utf-8");
             command.add("-v");
-            // SSRF: the HTML is self-contained and the engine validates style colours, so no
-            // external url() reaches WeasyPrint. For full isolation, run it network-isolated.
             command.add(htmlFile.getAbsolutePath());
             command.add(pdfFile.getAbsolutePath());
 
@@ -126,8 +147,8 @@ public class CreatePdfAgentController {
             // avoids materialising the whole document as a byte[] twice (read-all + re-serialise),
             // which matters for large generated documents.
             TempFile tempOut = tempFileManager.createManagedTempFile(".pdf");
-            try (PDDocument document = pdfDocumentFactory.load(pdfFile.getPath())) {
-                document.save(tempOut.getPath().toFile());
+            try (PDDocument pdDocument = pdfDocumentFactory.load(pdfFile.getPath())) {
+                pdDocument.save(tempOut.getPath().toFile());
             } catch (Exception e) {
                 tempOut.close();
                 throw e;

app/proprietary/src/main/java/stirling/software/proprietary/controller/api/FleetUsageController.java

@@ -4,6 +4,7 @@ import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.List;
 
+import org.springframework.context.annotation.Profile;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -38,6 +39,9 @@ import stirling.software.proprietary.security.database.repository.UserRepository
 @PreAuthorize("hasRole('ADMIN')")
 @RequiredArgsConstructor
 @EnterpriseEndpoint
+// Self-hosted only: counts are server-wide. On SaaS this endpoint is owned by the team-scoped
+// SaasFleetUsageController (@Profile("saas")) so one backend can't leak another tenant's usage.
+@Profile("!saas")
 public class FleetUsageController {
 
     private final PersistentAuditEventRepository auditRepository;

app/proprietary/src/main/java/stirling/software/proprietary/integration/crypto/LenientEncryptedStringConverter.java

@@ -0,0 +1,31 @@
+package stirling.software.proprietary.integration.crypto;
+
+import jakarta.persistence.AttributeConverter;
+import jakarta.persistence.Converter;
+
+/**
+ * {@link EncryptedStringConverter} for columns that held plaintext before encryption shipped:
+ * writes are always encrypted, but a stored value that is not valid ciphertext is returned as-is,
+ * so pre-encryption rows keep loading and become encrypted on their next save. The discrimination
+ * is exact for JSON payloads, which can never be mistaken for ciphertext ('{' is not in the Base64
+ * alphabet). The trade-off is that a genuinely corrupted ciphertext surfaces as garbage to the
+ * caller's parser instead of failing here.
+ */
+@Converter
+public class LenientEncryptedStringConverter implements AttributeConverter<String, String> {
+
+    @Override
+    public String convertToDatabaseColumn(String attribute) {
+        return CredentialEncryption.encrypt(attribute);
+    }
+
+    @Override
+    public String convertToEntityAttribute(String dbData) {
+        try {
+            return CredentialEncryption.decrypt(dbData);
+        } catch (IllegalArgumentException | IllegalStateException e) {
+            // Not ciphertext: legacy plaintext from before encryption shipped.
+            return dbData;
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/integration/service/IntegrationConfigService.java

@@ -43,6 +43,10 @@ public class IntegrationConfigService {
     private final OwnershipService ownership;
     private final SecretMasker secretMasker;
     private final ResourceGrantRepository grantRepository;
+    // Bean-discovered extension points: features that understand a type contribute its config
+    // schema and report what still references a config, without this module depending on them.
+    private final List<IntegrationConfigValidator> validators;
+    private final List<IntegrationConfigUsageCheck> usageChecks;
 
     // ---- commands ----
 
@@ -66,13 +70,21 @@ public class IntegrationConfigService {
                         ? DefaultAccessPolicy.EXPLICIT_ONLY
                         : request.defaultAccess());
 
+        // TEAM scope may omit the team id: default to the caller's own team so clients (the
+        // portal) need not know it. assignOwnership still enforces admin-or-leader of that team.
+        Long ownerTeamId = request.ownerTeamId();
+        if (ownerTeamId == null && scope == OwnerScope.TEAM && currentUser.getTeam() != null) {
+            ownerTeamId = currentUser.getTeam().getId();
+        }
         ownership.assignOwnership(
                 cfg,
                 scope,
-                request.ownerTeamId(),
+                ownerTeamId,
                 currentUser,
                 () -> lockedServerExists(cfg.getIntegrationType()));
-        cfg.setConfig(writeJson(secretMasker.sanitize(request.config())));
+        Map<String, Object> config = secretMasker.sanitize(request.config());
+        validateConfig(cfg.getIntegrationType(), config);
+        cfg.setConfig(writeJson(config));
         return repository.save(cfg);
     }
 
@@ -101,8 +113,10 @@ public class IntegrationConfigService {
             cfg.setDefaultAccess(request.defaultAccess());
         }
         if (request.config() != null) {
-            cfg.setConfig(
-                    writeJson(secretMasker.merge(readJson(cfg.getConfig()), request.config())));
+            Map<String, Object> merged =
+                    secretMasker.merge(readJson(cfg.getConfig()), request.config());
+            validateConfig(cfg.getIntegrationType(), merged);
+            cfg.setConfig(writeJson(merged));
         }
         return repository.save(cfg);
     }
@@ -113,6 +127,15 @@ public class IntegrationConfigService {
         if (!ownership.canManage(TYPE, cfg, currentUser)) {
             throw forbidden("You cannot manage this integration");
         }
+        // Refuse to pull a connection out from under whatever still references it.
+        List<String> usages =
+                usageChecks.stream()
+                        .flatMap(check -> check.usagesOf(cfg.getId()).stream())
+                        .toList();
+        if (!usages.isEmpty()) {
+            throw new ResponseStatusException(
+                    HttpStatus.CONFLICT, "Integration is in use by: " + String.join(", ", usages));
+        }
         // Drop grants sharing this config so they do not dangle as dead rows.
         grantRepository.deleteByResourceTypeAndResourceId(TYPE, String.valueOf(cfg.getId()));
         repository.delete(cfg);
@@ -188,6 +211,19 @@ public class IntegrationConfigService {
 
     // ---- integration-specific glue ----
 
+    /** Runs every registered validator for the type; unknown types save free-form. */
+    private void validateConfig(IntegrationType type, Map<String, Object> config) {
+        for (IntegrationConfigValidator validator : validators) {
+            if (validator.type() == type) {
+                try {
+                    validator.validate(config == null ? Map.of() : config);
+                } catch (IllegalArgumentException e) {
+                    throw new ResponseStatusException(HttpStatus.BAD_REQUEST, e.getMessage());
+                }
+            }
+        }
+    }
+
     /** A non-admin can't create a personal config of a type an admin has locked at server scope. */
     private boolean lockedServerExists(IntegrationType type) {
         return repository.findByScope(OwnerScope.SERVER).stream()

app/proprietary/src/main/java/stirling/software/proprietary/integration/service/IntegrationConfigUsageCheck.java

@@ -0,0 +1,15 @@
+package stirling.software.proprietary.integration.service;
+
+import java.util.List;
+
+/**
+ * Reports what still references an integration config, so deletion can be refused instead of
+ * pulling a connection out from under a live consumer. Implementations are beans discovered by
+ * {@link IntegrationConfigService} (e.g. the policy subsystem reporting sources and pipelines that
+ * reference a connection).
+ */
+public interface IntegrationConfigUsageCheck {
+
+    /** Human-readable labels of everything still using the config; empty when unreferenced. */
+    List<String> usagesOf(long configId);
+}

app/proprietary/src/main/java/stirling/software/proprietary/integration/service/IntegrationConfigValidator.java

@@ -0,0 +1,23 @@
+package stirling.software.proprietary.integration.service;
+
+import java.util.Map;
+
+import stirling.software.proprietary.integration.model.IntegrationType;
+
+/**
+ * Validates one integration type's config map at save time. Implementations are beans discovered by
+ * {@link IntegrationConfigService}, so the feature that understands a type (e.g. the policy S3
+ * backend) owns its schema without the integration module depending on it. Types with no registered
+ * validator save free-form.
+ */
+public interface IntegrationConfigValidator {
+
+    /** The type this validator understands. */
+    IntegrationType type();
+
+    /**
+     * Validates the config as it will be stored (secrets already sanitized/merged, so values are
+     * real, never the redaction mask). Throws {@link IllegalArgumentException} on bad config.
+     */
+    void validate(Map<String, Object> config);
+}

app/proprietary/src/main/java/stirling/software/proprietary/model/api/ai/create/AiDocument.java

@@ -0,0 +1,35 @@
+package stirling.software.proprietary.model.api.ai.create;
+
+import java.util.List;
+
+import lombok.Data;
+
+@Data
+public class AiDocument {
+
+    private String title;
+    private String subtitle;
+    private String referenceNumber;
+    private Style style;
+    private List<Section> sections;
+
+    @Data
+    public static class Style {
+        private String primaryColor;
+        private String backgroundColor;
+        private String bodyTextColor;
+    }
+
+    @Data
+    public static class Section {
+        private String type;
+        private String heading;
+        private String body;
+        private List<List<String>> pairs;
+        private List<String> columns;
+        private List<List<String>> rows;
+        private List<String> totalRow;
+        private List<String> items;
+        private List<String> signatories;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/model/api/audit/InfraAuditSummary.java

@@ -17,6 +17,9 @@ public class InfraAuditSummary {
     @Schema(description = "Total events in the returned window", example = "40")
     private int totalEvents;
 
+    @Schema(description = "Policy-run events", example = "3")
+    private int policy;
+
     @Schema(description = "Processing-category events", example = "24")
     private int processing;
 

app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java

@@ -23,6 +23,8 @@ import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestPart;
 import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.context.request.RequestContextHolder;
+import org.springframework.web.context.request.ServletRequestAttributes;
 import org.springframework.web.multipart.MultipartFile;
 import org.springframework.web.server.ResponseStatusException;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@@ -32,6 +34,7 @@ import io.swagger.v3.oas.annotations.Hidden;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 
+import jakarta.servlet.http.HttpServletRequest;
 import jakarta.validation.Valid;
 
 import lombok.RequiredArgsConstructor;
@@ -42,14 +45,18 @@ import stirling.software.common.model.job.JobResponse;
 import stirling.software.common.service.JobOwnershipService;
 import stirling.software.common.util.TempFile;
 import stirling.software.common.util.TempFileManager;
+import stirling.software.proprietary.audit.AuditContext;
 import stirling.software.proprietary.policy.config.PolicyAccessGuard;
 import stirling.software.proprietary.policy.config.PolicyManagementAuthority;
 import stirling.software.proprietary.policy.engine.PolicyRunHandle;
 import stirling.software.proprietary.policy.engine.PolicyRunRegistry;
 import stirling.software.proprietary.policy.engine.PolicyRunner;
 import stirling.software.proprietary.policy.engine.PolicyValidator;
+import stirling.software.proprietary.policy.engine.SweepOutcome;
 import stirling.software.proprietary.policy.ledger.ProcessedLedger;
+import stirling.software.proprietary.policy.model.OutputSpec;
 import stirling.software.proprietary.policy.model.PipelineDefinition;
+import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.model.PolicyInputs;
 import stirling.software.proprietary.policy.model.PolicyRun;
@@ -58,12 +65,15 @@ import stirling.software.proprietary.policy.model.PolicyRunView;
 import stirling.software.proprietary.policy.overview.PoliciesOverviewResponse;
 import stirling.software.proprietary.policy.overview.PolicyOverviewService;
 import stirling.software.proprietary.policy.progress.PolicyProgressListener;
+import stirling.software.proprietary.policy.source.EditorSource;
 import stirling.software.proprietary.policy.source.SourceAccessGuard;
+import stirling.software.proprietary.policy.source.SourceDocCounter;
 import stirling.software.proprietary.policy.source.SourceStore;
 import stirling.software.proprietary.policy.store.PolicyStore;
 import stirling.software.proprietary.policy.trigger.PolicyTrigger;
 import stirling.software.proprietary.policy.trigger.PolicyTriggerManager;
 import stirling.software.proprietary.policy.trigger.TriggerInfo;
+import stirling.software.proprietary.util.SecretMasker;
 
 /**
  * Policy CRUD plus pipeline runs (stored or ad-hoc). Runs are async: returns a run id, poll {@code
@@ -83,6 +93,7 @@ public class PolicyController {
     private final PolicyStore policyStore;
     private final SourceStore sourceStore;
     private final SourceAccessGuard sourceAccessGuard;
+    private final SourceDocCounter docCounter;
     private final PolicyValidator policyValidator;
     private final PolicyAccessGuard policyAccessGuard;
     private final PolicyManagementAuthority policyManagementAuthority;
@@ -107,11 +118,14 @@ public class PolicyController {
             @RequestPart("json") PipelineDefinition definition,
             @Valid @ModelAttribute PolicyRunFiles files)
             throws IOException {
+        stampPolicyAudit(definition);
         requireRunnable(definition);
+        validateAdHocOutput(definition);
         PolicyInputs inputs = toInputs(files);
-        String runId =
-                policyRunner.runAdHoc(definition, inputs, PolicyProgressListener.NOOP).runId();
-        return ResponseEntity.accepted().body(new JobResponse<>(true, runId, null));
+        PolicyRunHandle handle =
+                policyRunner.runAdHoc(definition, inputs, PolicyProgressListener.NOOP);
+        recordEditorDocs(inputs);
+        return ResponseEntity.accepted().body(new JobResponse<>(true, handle.runId(), null));
     }
 
     @PostMapping(value = "/run/stream", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
@@ -125,7 +139,9 @@ public class PolicyController {
             @RequestPart("json") PipelineDefinition definition,
             @Valid @ModelAttribute PolicyRunFiles files)
             throws IOException {
+        stampPolicyAudit(definition);
         requireRunnable(definition);
+        validateAdHocOutput(definition);
         PolicyInputs inputs = toInputs(files);
 
         SseEmitter emitter =
@@ -133,6 +149,7 @@ public class PolicyController {
         emitter.onError(e -> log.warn("Policy run SSE emitter error", e));
 
         PolicyRunHandle handle = policyRunner.runAdHoc(definition, inputs, streamListener(emitter));
+        recordEditorDocs(inputs);
         // whenComplete runs on the worker thread after the run finishes, so the terminal event
         // never races the step events.
         handle.completion()
@@ -202,7 +219,7 @@ public class PolicyController {
                             + " assigned; returns the stored policy with its id.")
     public ResponseEntity<Policy> savePolicy(@RequestBody Policy policy) {
         requirePolicyEditingAllowed();
-        Policy owned = resolveOwnership(policy);
+        Policy owned = withStoredOutputSecrets(resolveOwnership(policy));
         requireAccessibleSources(owned);
         try {
             policyValidator.validate(owned);
@@ -213,7 +230,7 @@ public class PolicyController {
         // Re-sync trigger registrations now so a new/changed folder-watch policy starts being
         // watched immediately instead of after the next reconcile sweep.
         policyTriggerManager.notifyPoliciesChanged();
-        return ResponseEntity.ok(saved);
+        return ResponseEntity.ok(withMaskedOutputSecrets(saved));
     }
 
     @PutMapping("/order")
@@ -282,6 +299,50 @@ public class PolicyController {
                 teamId);
     }
 
+    /** Output secrets never leave the server: reads return the redaction sentinel instead. */
+    private static Policy withMaskedOutputSecrets(Policy policy) {
+        return withOutput(
+                policy,
+                new OutputSpec(
+                        policy.output().type(), SecretMasker.mask(policy.output().options())));
+    }
+
+    /**
+     * An edit that round-trips a masked read sends output secrets back as the sentinel; restore
+     * them from the stored policy so saving without re-typing keeps them (validation then runs
+     * against the real values).
+     */
+    private Policy withStoredOutputSecrets(Policy incoming) {
+        if (incoming.id() == null || incoming.id().isBlank()) {
+            return incoming;
+        }
+        return policyStore
+                .get(incoming.id())
+                .map(
+                        existing ->
+                                withOutput(
+                                        incoming,
+                                        new OutputSpec(
+                                                incoming.output().type(),
+                                                SecretMasker.restoreRedacted(
+                                                        incoming.output().options(),
+                                                        existing.output().options()))))
+                .orElse(incoming);
+    }
+
+    private static Policy withOutput(Policy policy, OutputSpec output) {
+        return new Policy(
+                policy.id(),
+                policy.name(),
+                policy.owner(),
+                policy.enabled(),
+                policy.trigger(),
+                policy.sourceIds(),
+                policy.steps(),
+                output,
+                policy.teamId());
+    }
+
     /**
      * Creating, editing, pausing/resuming, and deleting policies requires the editor role for the
      * caller's team — a team leader on SaaS (see {@link PolicyManagementAuthority}); the global
@@ -306,9 +367,14 @@ public class PolicyController {
     @GetMapping
     @Operation(
             summary = "List policies",
-            description = "Lists the policies belonging to the caller's team.")
+            description =
+                    "Lists the policies belonging to the caller's team. Secret-bearing output"
+                            + " options are returned as a redaction sentinel, never their stored"
+                            + " values.")
     public List<Policy> listPolicies() {
-        return policyAccessGuard.visibleFrom(policyStore);
+        return policyAccessGuard.visibleFrom(policyStore).stream()
+                .map(PolicyController::withMaskedOutputSecrets)
+                .toList();
     }
 
     @GetMapping("/overview")
@@ -337,11 +403,17 @@ public class PolicyController {
     }
 
     @GetMapping("/{policyId}")
-    @Operation(summary = "Get a policy by id")
+    @Operation(
+            summary = "Get a policy by id",
+            description =
+                    "Secret-bearing output options are returned as a redaction sentinel, never"
+                            + " their stored values; an edit that sends the sentinel back keeps"
+                            + " them.")
     public ResponseEntity<Policy> getPolicy(@PathVariable String policyId) {
         return policyStore
                 .get(policyId)
                 .filter(policyAccessGuard::canAccess)
+                .map(PolicyController::withMaskedOutputSecrets)
                 .map(ResponseEntity::ok)
                 .orElseGet(() -> ResponseEntity.notFound().build());
     }
@@ -401,6 +473,7 @@ public class PolicyController {
                                 () ->
                                         new ResponseStatusException(
                                                 HttpStatus.NOT_FOUND, "No policy: " + policyId));
+        stampPolicyAudit(policy.toDefinition());
         PolicyInputs inputs = toInputs(files);
         String runId = policyRunner.runWith(policy, inputs, PolicyProgressListener.NOOP).runId();
         return ResponseEntity.accepted().body(new JobResponse<>(true, runId, null));
@@ -412,9 +485,10 @@ public class PolicyController {
             description =
                     "Pulls the policy's configured sources and runs the pipeline now, regardless of"
                             + " the enabled flag (which only gates automatic triggering). Returns"
-                            + " the ids of the runs started; poll the run-status endpoint for each."
-                            + " Empty when the sources yielded no work to do.")
-    public ResponseEntity<List<String>> trigger(@PathVariable String policyId) {
+                            + " the ids of the runs started (poll the run-status endpoint for each)"
+                            + " plus what the sweep skipped - already-processed, parked-by-failure,"
+                            + " and in-flight counts - so an empty result explains itself.")
+    public ResponseEntity<SweepOutcome> trigger(@PathVariable String policyId) {
         Policy policy =
                 policyStore
                         .get(policyId)
@@ -433,6 +507,60 @@ public class PolicyController {
         }
     }
 
+    /**
+     * Stamp this run's policy name and step endpoints onto the current request so the audit aspect
+     * can label the event as the policy it ran (rather than the generic {@code /run} endpoint) and
+     * record which tools it executed. No-op outside a web request.
+     */
+    private static void stampPolicyAudit(PipelineDefinition definition) {
+        if (definition == null
+                || !(RequestContextHolder.getRequestAttributes()
+                        instanceof ServletRequestAttributes attrs)) {
+            return;
+        }
+        HttpServletRequest request = attrs.getRequest();
+        if (definition.name() != null && !definition.name().isBlank()) {
+            request.setAttribute(AuditContext.REQ_ATTR_POLICY_NAME, definition.name());
+        }
+        List<String> steps =
+                definition.steps().stream()
+                        .map(PipelineStep::operation)
+                        .filter(op -> op != null && !op.isBlank())
+                        .toList();
+        if (!steps.isEmpty()) {
+            request.setAttribute(AuditContext.REQ_ATTR_POLICY_STEPS, steps);
+        }
+    }
+
+    /**
+     * Authorization-check an ad-hoc run's output while the caller's principal is present (this
+     * request thread). The worker thread that later delivers carries no security context, so an S3
+     * output's connection-access check would be skipped there; without this gate a caller could
+     * reference another tenant's connection by id and write to it (confused deputy). Stored
+     * policies are covered by save-time {@link PolicyValidator#validate} instead.
+     */
+    private void validateAdHocOutput(PipelineDefinition definition) {
+        if (definition.output() == null) {
+            return;
+        }
+        try {
+            policyValidator.validateOutput(definition.output());
+        } catch (IllegalArgumentException e) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, e.getMessage());
+        }
+    }
+
+    /**
+     * Ad-hoc runs (AI / one-off pipelines) are still editor activity, so their supplied documents
+     * feed the same virtual editor source as stored editor policies, counted against the caller's
+     * team. A run with no primary documents (generator pipeline) records nothing.
+     */
+    private void recordEditorDocs(PolicyInputs inputs) {
+        docCounter.record(
+                EditorSource.counterKey(sourceAccessGuard.currentTeamId()),
+                inputs.primary().size());
+    }
+
     /**
      * Turn the typed run files into engine {@link PolicyInputs}: the primary documents plus the
      * named supporting-file store, where each asset's {@code key} is the name a step references

app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java

@@ -21,7 +21,9 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.model.job.ResultFile;
+import stirling.software.common.service.AutomationRunContext;
 import stirling.software.common.service.FileStorage;
+import stirling.software.common.service.InternalApiClient;
 import stirling.software.common.service.InternalApiTimeoutException;
 import stirling.software.common.service.JobOwnershipService;
 import stirling.software.common.service.JobQueue;
@@ -148,6 +150,7 @@ public class PolicyEngine {
                         runAsPrincipal(
                                 billingPrincipal,
                                 fileOwner,
+                                definition.name(),
                                 () -> runToCompletion(run, inputs, tracking, completion));
 
         // One admission unit per run; steps run synchronously within it, so this gates heavy work
@@ -199,63 +202,78 @@ public class PolicyEngine {
             PolicyProgressListener listener,
             CompletableFuture<PolicyRun> completion) {
         String runId = run.getRunId();
-        try {
-            run.markRunning();
-            PolicyExecutionResult result =
-                    stepExecutor.execute(run.getDefinition(), inputs, listener);
-            OutputSpec output = run.getDefinition().output();
-            List<ResultFile> outputs =
-                    sinkFor(output)
-                            .deliver(
-                                    new OutputDelivery(runId, run.getPolicyId()),
-                                    result.files(),
-                                    output);
-            taskManager.setMultipleFileResults(runId, outputs);
-            taskManager.setComplete(runId);
-            run.complete(outputs);
-        } catch (PolicyInputRequiredException e) {
-            // Expected path: suspend rather than fail. Persist intermediates as fileIds so the run
-            // can resume after this worker thread is gone.
-            WaitState wait = suspend(e);
-            run.waitForInput(wait);
-            taskManager.addNote(runId, "Waiting for input: " + e.getMessage());
-        } catch (InternalApiTimeoutException e) {
-            String message = toolTimeoutMessage(e);
-            log.error(
-                    "Policy run {} timed out on {}: {}",
-                    runId,
-                    e.getEndpointPath(),
-                    e.getMessage());
-            run.fail(message);
-            taskManager.setError(runId, message);
-        } catch (RestClientResponseException e) {
-            // A downstream tool call returned an error status. When it's a structured entitlement
-            // response (401/402 with a JSON `error` sentinel), surface that code onto the run so
-            // the
-            // client can react — e.g. pop the usage-limit modal — instead of only seeing a generic
-            // failure. We don't interpret the code here (that would couple this module to the saas
-            // billing layer); we just pass it through for the client to map. Other statuses fall
-            // through to the generic failure below.
-            String code = DownstreamEntitlementError.extractCode(e);
-            if (code != null) {
-                log.info("Policy run {} blocked by downstream entitlement gate ({})", runId, code);
-                String message = "Usage limit reached";
-                run.failWithCode(message, code, DownstreamEntitlementError.extractSubscribed(e));
-                taskManager.setError(runId, message);
-            } else {
-                String message = "Policy run failed: " + e.getMessage();
-                log.error("Policy run {} failed (downstream HTTP error)", runId, e);
+        // One policy run = one automation run. Scope the run id on this worker thread (the async
+        // hop already happened) so every tool sub-step dispatched via InternalApiClient groups into
+        // a single charge, and two separate policy runs on the same document stay distinct charges.
+        try (AutomationRunContext.Scope runScope = AutomationRunContext.open(runId)) {
+            try {
+                run.markRunning();
+                PolicyExecutionResult result =
+                        stepExecutor.execute(run.getDefinition(), inputs, listener);
+                OutputSpec output = run.getDefinition().output();
+                List<ResultFile> outputs =
+                        sinkFor(output)
+                                .deliver(
+                                        new OutputDelivery(runId, run.getPolicyId()),
+                                        result.files(),
+                                        output);
+                taskManager.setMultipleFileResults(runId, outputs);
+                taskManager.setComplete(runId);
+                run.complete(outputs);
+            } catch (PolicyInputRequiredException e) {
+                // Expected path: suspend rather than fail. Persist intermediates as fileIds so the
+                // run
+                // can resume after this worker thread is gone.
+                WaitState wait = suspend(e);
+                run.waitForInput(wait);
+                taskManager.addNote(runId, "Waiting for input: " + e.getMessage());
+            } catch (InternalApiTimeoutException e) {
+                String message = toolTimeoutMessage(e);
+                log.error(
+                        "Policy run {} timed out on {}: {}",
+                        runId,
+                        e.getEndpointPath(),
+                        e.getMessage());
                 run.fail(message);
                 taskManager.setError(runId, message);
+            } catch (RestClientResponseException e) {
+                // A downstream tool call returned an error status. When it's a structured
+                // entitlement
+                // response (401/402 with a JSON `error` sentinel), surface that code onto the run
+                // so
+                // the
+                // client can react — e.g. pop the usage-limit modal — instead of only seeing a
+                // generic
+                // failure. We don't interpret the code here (that would couple this module to the
+                // saas
+                // billing layer); we just pass it through for the client to map. Other statuses
+                // fall
+                // through to the generic failure below.
+                String code = DownstreamEntitlementError.extractCode(e);
+                if (code != null) {
+                    log.info(
+                            "Policy run {} blocked by downstream entitlement gate ({})",
+                            runId,
+                            code);
+                    String message = "Usage limit reached";
+                    run.failWithCode(
+                            message, code, DownstreamEntitlementError.extractSubscribed(e));
+                    taskManager.setError(runId, message);
+                } else {
+                    String message = "Policy run failed: " + e.getMessage();
+                    log.error("Policy run {} failed (downstream HTTP error)", runId, e);
+                    run.fail(message);
+                    taskManager.setError(runId, message);
+                }
+            } catch (Exception e) {
+                String message = "Policy run failed: " + e.getMessage();
+                log.error("Policy run {} failed", runId, e);
+                run.fail(message);
+                taskManager.setError(runId, message);
+            } finally {
+                // Always resolve so stream/await callers unblock.
+                completion.complete(run);
             }
-        } catch (Exception e) {
-            String message = "Policy run failed: " + e.getMessage();
-            log.error("Policy run {} failed", runId, e);
-            run.fail(message);
-            taskManager.setError(runId, message);
-        } finally {
-            // Always resolve so stream/await callers unblock.
-            completion.complete(run);
         }
     }
 
@@ -361,18 +379,26 @@ public class PolicyEngine {
      * dispatch attributes (and charges) usage to that user. A null/blank principal runs as-is.
      * Restores the previous MDC value afterward (defensive — worker threads aren't pooled).
      */
-    private static void runAsPrincipal(String billingPrincipal, String fileOwner, Runnable body) {
+    private static void runAsPrincipal(
+            String billingPrincipal, String fileOwner, String policyName, Runnable body) {
         // Billing identity (MDC auditPrincipal) and output-file ownership (JobContext owner) are
         // set
         // independently: usage is charged to billingPrincipal, but stored output files are owned by
         // fileOwner — the user who triggered an org-wide policy — so they can fetch their results.
         // Either may be null (e.g. login disabled, or a trigger-fired run); each is applied only
-        // when present and restored afterward (defensive — worker threads aren't pooled).
+        // when present and restored afterward (defensive — worker threads aren't pooled). The
+        // policy
+        // name rides MDC too so each tool step's loopback dispatch (InternalApiClient) can forward
+        // it as a header, letting the audit tie the step back to its policy.
         String previousPrincipal = MDC.get(AUDIT_PRINCIPAL_MDC_KEY);
+        String previousPolicyName = MDC.get(InternalApiClient.POLICY_NAME_MDC_KEY);
         String previousOwner = JobContext.getOwner();
         if (billingPrincipal != null && !billingPrincipal.isBlank()) {
             MDC.put(AUDIT_PRINCIPAL_MDC_KEY, billingPrincipal);
         }
+        if (policyName != null && !policyName.isBlank()) {
+            MDC.put(InternalApiClient.POLICY_NAME_MDC_KEY, policyName);
+        }
         if (fileOwner != null && !fileOwner.isBlank()) {
             JobContext.setOwner(fileOwner);
         }
@@ -384,6 +410,11 @@ public class PolicyEngine {
             } else {
                 MDC.remove(AUDIT_PRINCIPAL_MDC_KEY);
             }
+            if (previousPolicyName != null) {
+                MDC.put(InternalApiClient.POLICY_NAME_MDC_KEY, previousPolicyName);
+            } else {
+                MDC.remove(InternalApiClient.POLICY_NAME_MDC_KEY);
+            }
             JobContext.setOwner(previousOwner);
         }
     }

app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java

@@ -21,6 +21,7 @@ import stirling.software.proprietary.policy.model.PolicyInputs;
 import stirling.software.proprietary.policy.model.PolicyRun;
 import stirling.software.proprietary.policy.model.PolicyRunStatus;
 import stirling.software.proprietary.policy.progress.PolicyProgressListener;
+import stirling.software.proprietary.policy.source.EditorSource;
 import stirling.software.proprietary.policy.source.Source;
 import stirling.software.proprietary.policy.source.SourceDocCounter;
 import stirling.software.proprietary.policy.source.SourceStore;
@@ -44,7 +45,7 @@ public class PolicyRunner {
     private final ProcessedLedger processedLedger;
 
     /** Full-listing sweep: resolve every source, then reconcile the ledger. */
-    public List<String> run(Policy policy) {
+    public SweepOutcome run(Policy policy) {
         return run(policy, SweepKind.FULL);
     }
 
@@ -52,10 +53,10 @@ public class PolicyRunner {
      * Trigger entry point. Pulls every referenced source; each yielded unit becomes its own run so
      * one failure does not affect the others. No sources means one run with no input (generator
      * pipeline). Missing or disabled sources are skipped so one broken reference does not stop the
-     * rest. Returns the ids of the runs it started (empty when sources yielded no work), so a
-     * manual trigger can report back which runs to follow.
+     * rest. Returns the ids of the runs it started plus what the sweep skipped, so a manual trigger
+     * can report which runs to follow or why nothing ran.
      */
-    public List<String> run(Policy policy, SweepKind sweep) {
+    public SweepOutcome run(Policy policy, SweepKind sweep) {
         long sweepStart = System.currentTimeMillis();
         PolicySweep context = new PolicySweep(policy.id(), sweep, processedLedger);
         List<String> runIds = new ArrayList<>();
@@ -95,13 +96,19 @@ public class PolicyRunner {
                         policy.id());
             }
         }
-        return runIds;
+        return context.outcome(runIds);
     }
 
-    /** Run a stored policy on caller-supplied files (e.g. manual upload), bypassing its sources. */
+    /**
+     * Run a stored policy on caller-supplied files (e.g. an editor upload), bypassing its sources.
+     * The supplied documents are still counted against the virtual {@link EditorSource}, scoped to
+     * the policy's team, so the Sources overview reports the whole team's editor throughput.
+     */
     public PolicyRunHandle runWith(
             Policy policy, PolicyInputs inputs, PolicyProgressListener listener) {
-        return policyEngine.runPolicy(policy, inputs, listener);
+        PolicyRunHandle handle = policyEngine.runPolicy(policy, inputs, listener);
+        docCounter.record(EditorSource.counterKey(policy.teamId()), inputs.primary().size());
+        return handle;
     }
 
     /** Run an ad-hoc pipeline with no stored policy (AI/Automate one-offs). */

app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicySweep.java

@@ -86,4 +86,32 @@ final class PolicySweep implements ResolveContext {
     synchronized Set<String> presentIdentities() {
         return Set.copyOf(present);
     }
+
+    /**
+     * Summarise the sweep from state already in hand (no extra ledger reads): the prefetched rows
+     * were loaded before claiming, and successful claims flipped their entries to PROCESSING, so
+     * what remains DONE or ERROR is exactly what this sweep skipped.
+     */
+    synchronized SweepOutcome outcome(List<String> runIds) {
+        int alreadyProcessed = 0;
+        int parked = 0;
+        int processing = 0;
+        for (String identity : present) {
+            ClaimState state = prefetched.get(identity);
+            if (state == null) {
+                continue;
+            }
+            switch (state.status()) {
+                case DONE -> alreadyProcessed++;
+                case ERROR -> parked++;
+                case PROCESSING, INTERRUPTED -> processing++;
+            }
+        }
+        return new SweepOutcome(
+                runIds,
+                present.size(),
+                alreadyProcessed,
+                parked,
+                Math.max(0, processing - runIds.size()));
+    }
 }

app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyValidator.java

@@ -52,7 +52,19 @@ public class PolicyValidator {
             InputSpec spec = source.toInputSpec();
             inputSourceFor(spec).validate(spec);
         }
-        outputSinkFor(policy.output()).validate(policy.output());
+        validateOutput(policy.output());
+    }
+
+    /**
+     * Validate an output spec against its sink. Must be called on a request thread (caller's
+     * principal present) so an S3 output's connection is authorization-checked against the caller -
+     * ad-hoc runs are never persisted and so never hit {@link #validate(Policy)}, and the worker
+     * thread that later delivers has no principal, so this is their only access gate.
+     *
+     * @throws IllegalArgumentException if the type is unknown or the config is invalid/inaccessible
+     */
+    public void validateOutput(OutputSpec output) {
+        outputSinkFor(output).validate(output);
     }
 
     private PolicyTrigger triggerFor(TriggerConfig config) {

app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/SweepOutcome.java

@@ -0,0 +1,19 @@
+package stirling.software.proprietary.policy.engine;
+
+import java.util.List;
+
+/**
+ * What one policy sweep found and started, so a manual trigger can explain an empty result instead
+ * of a blanket "nothing to do": how many files the sources listed, how many were skipped because
+ * they are already processed at their current version, how many are parked by a failed run (not
+ * retried until they change or history is cleared), and how many are still in flight from an
+ * earlier sweep. Counts are zero for {@link SweepKind#LIGHT} sweeps, which do not take a full
+ * listing.
+ */
+public record SweepOutcome(
+        List<String> runIds, int filesListed, int alreadyProcessed, int parked, int inFlight) {
+
+    public SweepOutcome {
+        runIds = runIds == null ? List.of() : List.copyOf(runIds);
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/input/S3InputSource.java

@@ -0,0 +1,287 @@
+package stirling.software.proprietary.policy.input;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
+import org.springframework.core.io.AbstractResource;
+import org.springframework.core.io.Resource;
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.PolicyInputs;
+import stirling.software.proprietary.policy.s3.S3Config;
+import stirling.software.proprietary.policy.s3.S3ConnectionPool;
+import stirling.software.proprietary.policy.s3.S3ConnectionResolver;
+import stirling.software.proprietary.policy.s3.S3Identities;
+
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+/**
+ * Reads input files from an Amazon S3 (or S3-compatible) bucket; each listed object is its own unit
+ * of work, claimed through the {@link ResolveContext} ledger and tracked in place. Identity and
+ * version gate come from {@link S3Identities}, so the steady-state sweep never downloads content.
+ * Options: "connectionId" references the stored S3 connection (an {@code IntegrationConfig} owning
+ * bucket, region, endpoint, and credentials - resolved by {@link S3ConnectionResolver}); "prefix"
+ * (only keys starting with it are read) and "mode" are per-source, where mode is "consume"
+ * (default: a processed object is deleted once every policy that claimed it has settled
+ * successfully and it is still the version that ran; failures stay in place and are not retried
+ * until they change) or "snapshot" (stateless, every run sees the full set). Keys ending in "/"
+ * (folder placeholders) and keys with a dot-prefixed path segment are never picked up, mirroring
+ * the folder source's hidden-file rule.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+@ConditionalOnBooleanProperty(name = "policies.enabled")
+public class S3InputSource implements InputSource {
+
+    private static final String TYPE = "s3";
+
+    private final S3ConnectionPool connectionPool;
+    private final S3ConnectionResolver connectionResolver;
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    @Override
+    public boolean supports(InputSpec spec) {
+        return spec != null && TYPE.equals(spec.type());
+    }
+
+    /**
+     * Fails fast at save time: an unknown/disabled/unusable connection, bad config shape, a private
+     * endpoint without the operator opt-in, or a bucket the connection cannot list.
+     */
+    @Override
+    public void validate(InputSpec spec) {
+        S3Config config = connectionResolver.resolve(spec.options());
+        try {
+            connectionPool.clientFor(config).listObjectsV2(listRequest(config).maxKeys(1).build());
+        } catch (SdkException e) {
+            throw new IllegalArgumentException(
+                    "cannot access s3://"
+                            + config.bucket()
+                            + "/"
+                            + config.prefix()
+                            + ": "
+                            + e.getMessage(),
+                    e);
+        }
+    }
+
+    @Override
+    public List<ResolvedInput> resolve(InputSpec spec, ResolveContext ctx) throws IOException {
+        S3Config config = connectionResolver.resolve(spec.options());
+        S3Client client = connectionPool.clientFor(config);
+        // A listing failure propagates so the sweep reads it as "could not list" (which vetoes
+        // presence cleanup), never as "verifiably no objects".
+        List<S3Object> objects = listObjects(client, config);
+
+        if (config.snapshot()) {
+            return objects.stream()
+                    .map(
+                            object ->
+                                    ResolvedInput.of(
+                                            PolicyInputs.of(
+                                                    List.of(
+                                                            objectResource(
+                                                                    client, config, object)))))
+                    .toList();
+        }
+
+        ctx.reportPresent(
+                objects.stream()
+                        .map(object -> S3Identities.identity(config.bucket(), object.key()))
+                        .toList());
+
+        List<ResolvedInput> work = new ArrayList<>();
+        for (S3Object object : objects) {
+            String identity = S3Identities.identity(config.bucket(), object.key());
+            String gate = S3Identities.gate(object.eTag(), object.size(), object.lastModified());
+            if (!ctx.claim(identity, gate, null)) {
+                continue;
+            }
+            work.add(
+                    new ResolvedInput(
+                            PolicyInputs.of(List.of(objectResource(client, config, object))),
+                            success ->
+                                    completeConsumed(
+                                            ctx,
+                                            client,
+                                            config,
+                                            object.key(),
+                                            identity,
+                                            gate,
+                                            success)));
+        }
+        return work;
+    }
+
+    /**
+     * Settle at the version this run claimed, then remove the object only when it still carries
+     * that version and every policy that claimed it has settled DONE, mirroring the folder source's
+     * consensus delete. A failed run settles ERROR and never deletes; the DONE row of an object
+     * that could not be deleted still stops reprocessing.
+     */
+    private void completeConsumed(
+            ResolveContext ctx,
+            S3Client client,
+            S3Config config,
+            String key,
+            String identity,
+            String claimGate,
+            boolean success) {
+        ctx.settle(identity, claimGate, null, success);
+        if (!success) {
+            return;
+        }
+        try {
+            HeadObjectResponse head =
+                    client.headObject(
+                            HeadObjectRequest.builder().bucket(config.bucket()).key(key).build());
+            String currentGate =
+                    S3Identities.gate(head.eTag(), head.contentLength(), head.lastModified());
+            if (currentGate.equals(claimGate) && ctx.allSettledDone(identity)) {
+                client.deleteObject(
+                        DeleteObjectRequest.builder().bucket(config.bucket()).key(key).build());
+            }
+        } catch (NoSuchKeyException alreadyGone) {
+            // Removed by the user or a co-watching policy's own consensus delete: nothing to do.
+        } catch (S3Exception e) {
+            if (e.statusCode() == 404) {
+                return;
+            }
+            log.warn("Could not remove consumed S3 object {}: {}", identity, e.getMessage());
+        } catch (SdkException e) {
+            log.warn("Could not remove consumed S3 object {}: {}", identity, e.getMessage());
+        }
+    }
+
+    /** Every ingestible object under the configured prefix, across all listing pages. */
+    private static List<S3Object> listObjects(S3Client client, S3Config config) {
+        List<S3Object> objects = new ArrayList<>();
+        String continuationToken = null;
+        do {
+            ListObjectsV2Request.Builder request = listRequest(config);
+            if (continuationToken != null) {
+                request.continuationToken(continuationToken);
+            }
+            ListObjectsV2Response page = client.listObjectsV2(request.build());
+            for (S3Object object : page.contents()) {
+                if (ingestible(object)) {
+                    objects.add(object);
+                }
+            }
+            continuationToken = page.nextContinuationToken();
+        } while (continuationToken != null);
+        return objects;
+    }
+
+    private static ListObjectsV2Request.Builder listRequest(S3Config config) {
+        ListObjectsV2Request.Builder request =
+                ListObjectsV2Request.builder().bucket(config.bucket());
+        if (!config.prefix().isEmpty()) {
+            request.prefix(config.prefix());
+        }
+        return request;
+    }
+
+    /**
+     * Folder-placeholder keys (ending "/") and keys with a dot-prefixed segment are skipped, so a
+     * hidden convention (e.g. a future output sink's staging prefix) is never re-ingested.
+     */
+    private static boolean ingestible(S3Object object) {
+        String key = object.key();
+        if (key.isEmpty() || key.endsWith("/")) {
+            return false;
+        }
+        for (String segment : key.split("/")) {
+            if (segment.startsWith(".")) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private static Resource objectResource(S3Client client, S3Config config, S3Object object) {
+        return new S3ObjectResource(client, config.bucket(), object);
+    }
+
+    /**
+     * Streams the object on demand, pinned to the ETag observed at listing time so a run never
+     * reads a different version than the sweep claimed (a swapped object fails the read with a
+     * precondition error and the new version is claimed by a later sweep).
+     */
+    private static final class S3ObjectResource extends AbstractResource {
+
+        private final S3Client client;
+        private final String bucket;
+        private final String key;
+        private final String eTag;
+        private final Long size;
+
+        private S3ObjectResource(S3Client client, String bucket, S3Object object) {
+            this.client = client;
+            this.bucket = bucket;
+            this.key = object.key();
+            this.eTag = object.eTag();
+            this.size = object.size();
+        }
+
+        @Override
+        public InputStream getInputStream() throws IOException {
+            GetObjectRequest.Builder request = GetObjectRequest.builder().bucket(bucket).key(key);
+            if (eTag != null && !eTag.isBlank()) {
+                request.ifMatch(eTag);
+            }
+            try {
+                return client.getObject(request.build());
+            } catch (NoSuchKeyException e) {
+                throw new FileNotFoundException(getDescription() + " no longer exists");
+            } catch (SdkException e) {
+                throw new IOException(
+                        "Could not read " + getDescription() + ": " + e.getMessage(), e);
+            }
+        }
+
+        /** Listed just now; readers get a precise error from {@link #getInputStream} instead. */
+        @Override
+        public boolean exists() {
+            return true;
+        }
+
+        @Override
+        public long contentLength() {
+            return size == null ? -1 : size;
+        }
+
+        @Override
+        public String getFilename() {
+            return key.substring(key.lastIndexOf('/') + 1);
+        }
+
+        @Override
+        public String getDescription() {
+            return "S3 object " + S3Identities.identity(bucket, key);
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java

@@ -15,7 +15,6 @@ import java.util.List;
 import java.util.UUID;
 import java.util.stream.Stream;
 
-import org.apache.commons.io.FilenameUtils;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
 import org.springframework.core.io.Resource;
 import org.springframework.http.MediaType;
@@ -82,7 +81,7 @@ public class FolderOutputSink implements PolicyOutputSink {
         List<ResultFile> results = new ArrayList<>();
         for (int i = 0; i < outputs.size(); i++) {
             Resource resource = outputs.get(i);
-            String name = safeName(resource.getFilename(), i);
+            String name = OutputNames.safeName(resource.getFilename(), i);
             Path staged = tmpDir.resolve(UUID.randomUUID().toString());
             String contentHash = stage(resource, staged, delivery.policyId() != null);
             long size = Files.size(staged);
@@ -198,29 +197,14 @@ public class FolderOutputSink implements PolicyOutputSink {
         return Path.of(directory.toString());
     }
 
-    // Strip any directory component / "../" so a crafted output name cannot escape targetDir.
-    private static String safeName(String filename, int index) {
-        if (filename == null || filename.isBlank()) {
-            return "output-" + index;
-        }
-        String name = FilenameUtils.getName(filename);
-        if (name.isBlank() || ".".equals(name) || "..".equals(name)) {
-            return "output-" + index;
-        }
-        return name;
-    }
-
     // Non-colliding path, appending " (n)" before the extension.
     private static Path uniqueTarget(Path dir, String filename) {
         Path candidate = dir.resolve(filename);
         if (!Files.exists(candidate)) {
             return candidate;
         }
-        String base = FilenameUtils.getBaseName(filename);
-        String ext = FilenameUtils.getExtension(filename);
-        String suffix = ext.isEmpty() ? "" : "." + ext;
         for (int n = 1; ; n++) {
-            Path next = dir.resolve(base + " (" + n + ")" + suffix);
+            Path next = dir.resolve(OutputNames.numbered(filename, n));
             if (!Files.exists(next)) {
                 return next;
             }

app/proprietary/src/main/java/stirling/software/proprietary/policy/output/OutputNames.java

@@ -0,0 +1,29 @@
+package stirling.software.proprietary.policy.output;
+
+import org.apache.commons.io.FilenameUtils;
+
+/** Output file naming shared by the sinks: sanitised base names and collision suffixes. */
+final class OutputNames {
+
+    private OutputNames() {}
+
+    /** Strip any directory component / "../" so a crafted output name cannot escape the target. */
+    static String safeName(String filename, int index) {
+        if (filename == null || filename.isBlank()) {
+            return "output-" + index;
+        }
+        String name = FilenameUtils.getName(filename);
+        if (name.isBlank() || ".".equals(name) || "..".equals(name)) {
+            return "output-" + index;
+        }
+        return name;
+    }
+
+    /** The nth alternative for a taken name, appending " (n)" before the extension. */
+    static String numbered(String filename, int n) {
+        String base = FilenameUtils.getBaseName(filename);
+        String ext = FilenameUtils.getExtension(filename);
+        String suffix = ext.isEmpty() ? "" : "." + ext;
+        return base + " (" + n + ")" + suffix;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/output/S3OutputSink.java

@@ -0,0 +1,272 @@
+package stirling.software.proprietary.policy.output;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.DigestOutputStream;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.UUID;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
+import org.springframework.core.io.Resource;
+import org.springframework.http.MediaType;
+import org.springframework.http.MediaTypeFactory;
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.model.job.ResultFile;
+import stirling.software.proprietary.policy.ledger.ProcessedLedger;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.s3.S3Config;
+import stirling.software.proprietary.policy.s3.S3ConnectionPool;
+import stirling.software.proprietary.policy.s3.S3ConnectionResolver;
+import stirling.software.proprietary.policy.s3.S3Identities;
+
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectResponse;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+/**
+ * Uploads a run's outputs to the bucket and key prefix given in the {@link OutputSpec} (same
+ * connection options as the S3 input source; "prefix" is the destination folder). The
+ * record-before-visible obligation is met without a rename step: a single-part PUT's ETag is the
+ * MD5 of its content on plain and SSE-S3 buckets, so the ledger row is recorded at that predicted
+ * gate BEFORE the upload, and the object is claimed under exactly the gate the next listing
+ * returns. Stores where the returned ETag differs (e.g. SSE-KMS) are re-recorded at the actual gate
+ * immediately after the PUT - a narrow race those buckets accept rather than a broken loop. Names
+ * never overwrite: uploads are conditional on the key not existing ({@code If-None-Match: *}),
+ * re-picking "name (n).ext" on collision exactly like the folder sink; stores without
+ * conditional-write support fall back to an existence check per candidate.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+@ConditionalOnBooleanProperty(name = "policies.enabled")
+public class S3OutputSink implements PolicyOutputSink {
+
+    private static final String TYPE = "s3";
+
+    private final S3ConnectionPool connectionPool;
+    private final S3ConnectionResolver connectionResolver;
+    private final ProcessedLedger processedLedger;
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    @Override
+    public boolean supports(OutputSpec spec) {
+        return spec != null && TYPE.equals(spec.type());
+    }
+
+    /**
+     * Connection resolution (including the saving user's right to use it) and endpoint guard only -
+     * no network probe, since write-only credentials (s3:PutObject without s3:ListBucket) are a
+     * legitimate setup for an output bucket and a listing probe would wrongly reject them.
+     */
+    @Override
+    public void validate(OutputSpec spec) {
+        connectionPool.clientFor(connectionResolver.resolve(spec.options()));
+    }
+
+    @Override
+    public List<ResultFile> deliver(
+            OutputDelivery delivery, List<Resource> outputs, OutputSpec spec) throws IOException {
+        S3Config config = connectionResolver.resolve(spec.options());
+        S3Client client = connectionPool.clientFor(config);
+
+        List<ResultFile> results = new ArrayList<>();
+        for (int i = 0; i < outputs.size(); i++) {
+            Resource resource = outputs.get(i);
+            String name = OutputNames.safeName(resource.getFilename(), i);
+            Path staged = Files.createTempFile("s3-output-", ".tmp");
+            try {
+                String predictedGate = stage(resource, staged, delivery.policyId() != null);
+                long size = Files.size(staged);
+                String key = upload(delivery, client, config, name, staged, predictedGate);
+                String contentType =
+                        MediaTypeFactory.getMediaType(name)
+                                .orElse(MediaType.APPLICATION_OCTET_STREAM)
+                                .toString();
+                results.add(
+                        ResultFile.builder()
+                                .fileId(UUID.randomUUID().toString())
+                                .fileName(S3Identities.identity(config.bucket(), key))
+                                .contentType(contentType)
+                                .fileSize(size)
+                                .build());
+                log.debug(
+                        "Wrote policy run {} output to {}",
+                        delivery.runId(),
+                        S3Identities.identity(config.bucket(), key));
+            } finally {
+                try {
+                    Files.deleteIfExists(staged);
+                } catch (IOException e) {
+                    log.warn("Could not remove S3 staging file {}: {}", staged, e.getMessage());
+                }
+            }
+        }
+        return results;
+    }
+
+    /**
+     * Spool the output to a local staging file (S3 needs a known content length, and the body must
+     * be re-readable across collision retries). For a recorded delivery the MD5 - the predicted
+     * single-part ETag - is digested in the same pass; ad-hoc runs record nothing and skip it.
+     */
+    private static String stage(Resource resource, Path staged, boolean recorded)
+            throws IOException {
+        if (!recorded) {
+            try (InputStream is = resource.getInputStream();
+                    OutputStream out = Files.newOutputStream(staged)) {
+                is.transferTo(out);
+            }
+            return null;
+        }
+        MessageDigest digest = newMd5();
+        try (InputStream is = resource.getInputStream();
+                DigestOutputStream out =
+                        new DigestOutputStream(Files.newOutputStream(staged), digest)) {
+            is.transferTo(out);
+        }
+        return HexFormat.of().formatHex(digest.digest());
+    }
+
+    /**
+     * The S3 shape of the folder sink's record-then-rename loop. The ledger row must exist before
+     * the object is visible, so it is recorded at the predicted gate before the PUT; losing the
+     * chosen key to a concurrent writer (the conditional PUT fails) forgets the just-recorded row -
+     * whatever object actually owns that key must stay claimable at any version - then re-picks. A
+     * PUT that never made the object visible also forgets its row.
+     */
+    private String upload(
+            OutputDelivery delivery,
+            S3Client client,
+            S3Config config,
+            String name,
+            Path staged,
+            String predictedGate)
+            throws IOException {
+        String keyPrefix = keyPrefix(config);
+        boolean conditionalPuts = true;
+        for (int attempt = 0; ; attempt++) {
+            String key = keyPrefix + (attempt == 0 ? name : OutputNames.numbered(name, attempt));
+            String identity = S3Identities.identity(config.bucket(), key);
+            if (!conditionalPuts && exists(client, config.bucket(), key)) {
+                continue;
+            }
+            if (delivery.policyId() != null) {
+                processedLedger.recordOutput(delivery.policyId(), identity, predictedGate, null);
+            }
+            PutObjectRequest.Builder put =
+                    PutObjectRequest.builder().bucket(config.bucket()).key(key);
+            if (conditionalPuts) {
+                put.ifNoneMatch("*");
+            }
+            try {
+                PutObjectResponse response =
+                        client.putObject(put.build(), RequestBody.fromFile(staged));
+                reRecordIfGateDiffers(delivery, identity, predictedGate, response);
+                return key;
+            } catch (S3Exception e) {
+                forgetRecorded(delivery, identity, predictedGate);
+                if (conditionalPuts && e.statusCode() == 412) {
+                    // Known edge: if our own PUT succeeded server-side but the response was lost
+                    // and the SDK retried, that retry 412s here too - we then upload under the
+                    // next name, leaving the first object row-less (claimable, single duplicate).
+                    // Requires a response-lost network flake at exactly this moment; accepted.
+                    log.debug("Output key {} taken concurrently; re-picking", identity);
+                    continue;
+                }
+                if (conditionalPuts && e.statusCode() == 501) {
+                    // Store without conditional-write support: retry this candidate with a plain
+                    // existence check instead.
+                    log.debug(
+                            "Conditional PUT unsupported by {}; falling back to existence checks",
+                            config.bucket());
+                    conditionalPuts = false;
+                    attempt--;
+                    continue;
+                }
+                throw new IOException("Could not upload " + identity + ": " + e.getMessage(), e);
+            } catch (SdkException e) {
+                forgetRecorded(delivery, identity, predictedGate);
+                throw new IOException("Could not upload " + identity + ": " + e.getMessage(), e);
+            }
+        }
+    }
+
+    /**
+     * On buckets where a PUT's ETag is not the content MD5 (e.g. SSE-KMS), re-record at the gate
+     * listings will actually return. The row is briefly at the wrong gate while the object is
+     * already visible - the narrow race such stores trade for a working self-output skip.
+     */
+    private void reRecordIfGateDiffers(
+            OutputDelivery delivery,
+            String identity,
+            String predictedGate,
+            PutObjectResponse response) {
+        if (delivery.policyId() == null) {
+            return;
+        }
+        String actualGate = S3Identities.gate(response.eTag(), null, null);
+        if (!actualGate.equals(predictedGate)) {
+            log.debug(
+                    "PUT ETag for {} differs from content MD5 (encrypted bucket?); re-recording",
+                    identity);
+            processedLedger.recordOutput(delivery.policyId(), identity, actualGate, null);
+        }
+    }
+
+    private void forgetRecorded(OutputDelivery delivery, String identity, String predictedGate) {
+        if (delivery.policyId() != null) {
+            processedLedger.forgetOutput(delivery.policyId(), identity, predictedGate);
+        }
+    }
+
+    private static boolean exists(S3Client client, String bucket, String key) {
+        try {
+            client.headObject(HeadObjectRequest.builder().bucket(bucket).key(key).build());
+            return true;
+        } catch (NoSuchKeyException e) {
+            return false;
+        } catch (S3Exception e) {
+            if (e.statusCode() == 404) {
+                return false;
+            }
+            throw e;
+        }
+    }
+
+    /** The configured prefix as a key-path prefix: "processed" and "processed/" mean the same. */
+    private static String keyPrefix(S3Config config) {
+        String prefix = config.prefix();
+        if (prefix.isEmpty() || prefix.endsWith("/")) {
+            return prefix;
+        }
+        return prefix + "/";
+    }
+
+    private static MessageDigest newMd5() {
+        try {
+            return MessageDigest.getInstance("MD5");
+        } catch (NoSuchAlgorithmException e) {
+            throw new IllegalStateException("MD5 unavailable", e);
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigration.java

@@ -0,0 +1,212 @@
+package stirling.software.proprietary.policy.s3;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.proprietary.access.model.DefaultAccessPolicy;
+import stirling.software.proprietary.access.model.OwnerScope;
+import stirling.software.proprietary.integration.model.IntegrationConfig;
+import stirling.software.proprietary.integration.model.IntegrationType;
+import stirling.software.proprietary.integration.repository.IntegrationConfigRepository;
+import stirling.software.proprietary.model.Team;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.source.Source;
+import stirling.software.proprietary.policy.source.SourceStore;
+import stirling.software.proprietary.policy.store.PolicyStore;
+import stirling.software.proprietary.security.repository.TeamRepository;
+
+import tools.jackson.databind.ObjectMapper;
+
+/**
+ * One-time, idempotent extraction of legacy embedded S3 credentials into stored connections:
+ * sources and policy outputs written before connections shipped carry bucket/credentials in their
+ * own options; this rewrites each to reference a (deduplicated) S3 {@link IntegrationConfig} and
+ * keeps only per-use options (prefix, mode). MUST be programmatic - the option JSON is encrypted at
+ * the application layer, so no SQL migration can read it.
+ *
+ * <p>Idempotent by construction: rewritten rows no longer embed credentials, so re-runs find
+ * nothing to do. Connections are deduplicated against both this run's extractions and existing S3
+ * connections; a concurrent multi-node boot can at worst create a redundant connection row, never
+ * corrupt a source. Ownership follows the owning row: team-scoped when the source/policy has a
+ * team, server-scoped otherwise (single-operator self-hosted).
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@ConditionalOnBooleanProperty(name = "policies.enabled")
+public class EmbeddedS3CredentialMigration {
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+    private static final List<String> CONNECTION_OPTIONS =
+            List.of("bucket", "region", "endpoint", "accessKeyId", "secretAccessKey");
+    // Field separator for the dedup key: a unit-separator control char that cannot appear in a
+    // bucket/region/endpoint/credential, so distinct field sets can never collide.
+    private static final char DELIMITER = '\u001f';
+
+    private final SourceStore sourceStore;
+    private final PolicyStore policyStore;
+    private final IntegrationConfigRepository connections;
+    private final TeamRepository teamRepository;
+
+    @EventListener(ApplicationReadyEvent.class)
+    @Transactional
+    public void migrate() {
+        Map<String, IntegrationConfig> byCredentialKey = indexExistingConnections();
+        int migrated = 0;
+        for (Source source : sourceStore.all()) {
+            if (!"s3".equals(source.type()) || !embedsCredentials(source.options())) {
+                continue;
+            }
+            IntegrationConfig connection =
+                    connectionFor(source.options(), source.teamId(), byCredentialKey);
+            sourceStore.save(withOptions(source, referencing(connection, source.options(), true)));
+            migrated++;
+        }
+        for (Policy policy : policyStore.all()) {
+            OutputSpec output = policy.output();
+            if (!"s3".equals(output.type()) || !embedsCredentials(output.options())) {
+                continue;
+            }
+            IntegrationConfig connection =
+                    connectionFor(output.options(), policy.teamId(), byCredentialKey);
+            policyStore.save(
+                    withOutput(
+                            policy,
+                            new OutputSpec(
+                                    output.type(),
+                                    referencing(connection, output.options(), false))));
+            migrated++;
+        }
+        if (migrated > 0) {
+            log.info("Extracted embedded S3 credentials from {} row(s) into connections", migrated);
+        }
+    }
+
+    private static boolean embedsCredentials(Map<String, Object> options) {
+        return options.get("accessKeyId") != null;
+    }
+
+    /** Reuses an existing connection with identical coordinates+credentials, else creates one. */
+    private IntegrationConfig connectionFor(
+            Map<String, Object> options, Long teamId, Map<String, IntegrationConfig> byKey) {
+        String key = credentialKey(options);
+        IntegrationConfig existing = byKey.get(key);
+        if (existing != null) {
+            return existing;
+        }
+        IntegrationConfig connection = new IntegrationConfig();
+        connection.setIntegrationType(IntegrationType.S3);
+        connection.setName(connectionName(options, byKey));
+        connection.setEnabled(true);
+        connection.setLocked(false);
+        connection.setDefaultAccess(DefaultAccessPolicy.EXPLICIT_ONLY);
+        Team team = teamId == null ? null : teamRepository.findById(teamId).orElse(null);
+        if (team != null) {
+            connection.setScope(OwnerScope.TEAM);
+            connection.setOwnerTeam(team);
+        } else {
+            // No team (teamless self-hosted, or a source whose team was since deleted): server
+            // scope, i.e. admin-owned. An orphaned-team source's non-admin editor would then need
+            // an admin to re-share the connection - acceptable for the narrow orphaned case.
+            connection.setScope(OwnerScope.SERVER);
+        }
+        Map<String, Object> config = new LinkedHashMap<>();
+        for (String option : CONNECTION_OPTIONS) {
+            Object value = options.get(option);
+            if (value != null && !value.toString().isBlank()) {
+                config.put(option, value);
+            }
+        }
+        connection.setConfig(OBJECT_MAPPER.writeValueAsString(config));
+        IntegrationConfig saved = connections.save(connection);
+        byKey.put(key, saved);
+        return saved;
+    }
+
+    /** The rewritten options: the connection reference plus per-use settings only. */
+    private static Map<String, Object> referencing(
+            IntegrationConfig connection, Map<String, Object> legacy, boolean keepMode) {
+        Map<String, Object> options = new LinkedHashMap<>();
+        options.put(S3ConnectionResolver.CONNECTION_ID_OPTION, connection.getId());
+        Object prefix = legacy.get("prefix");
+        if (prefix != null && !prefix.toString().isBlank()) {
+            options.put("prefix", prefix);
+        }
+        Object mode = legacy.get("mode");
+        if (keepMode && mode != null && !mode.toString().isBlank()) {
+            options.put("mode", mode);
+        }
+        return options;
+    }
+
+    private Map<String, IntegrationConfig> indexExistingConnections() {
+        Map<String, IntegrationConfig> byKey = new LinkedHashMap<>();
+        for (IntegrationConfig connection : connections.findAll()) {
+            if (connection.getIntegrationType() != IntegrationType.S3) {
+                continue;
+            }
+            try {
+                Map<String, Object> config =
+                        OBJECT_MAPPER.readValue(connection.getConfig(), Map.class);
+                byKey.putIfAbsent(credentialKey(config), connection);
+            } catch (Exception e) {
+                log.debug(
+                        "Skipping unreadable S3 connection {} while indexing: {}",
+                        connection.getId(),
+                        e.getMessage());
+            }
+        }
+        return byKey;
+    }
+
+    private static String credentialKey(Map<String, Object> options) {
+        StringBuilder key = new StringBuilder();
+        for (String option : CONNECTION_OPTIONS) {
+            Object value = options.get(option);
+            key.append(value == null ? "" : value.toString().trim()).append(DELIMITER);
+        }
+        return key.toString();
+    }
+
+    private static String connectionName(
+            Map<String, Object> options, Map<String, IntegrationConfig> byKey) {
+        String base = "S3: " + options.getOrDefault("bucket", "bucket");
+        long sameName = byKey.values().stream().filter(c -> c.getName().startsWith(base)).count();
+        return sameName == 0 ? base : base + " (" + (sameName + 1) + ")";
+    }
+
+    private static Source withOptions(Source source, Map<String, Object> options) {
+        return new Source(
+                source.id(),
+                source.name(),
+                source.type(),
+                options,
+                source.enabled(),
+                source.owner(),
+                source.teamId());
+    }
+
+    private static Policy withOutput(Policy policy, OutputSpec output) {
+        return new Policy(
+                policy.id(),
+                policy.name(),
+                policy.owner(),
+                policy.enabled(),
+                policy.trigger(),
+                policy.sourceIds(),
+                policy.steps(),
+                output,
+                policy.teamId());
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/PolicyS3ConnectionUsageCheck.java

@@ -0,0 +1,56 @@
+package stirling.software.proprietary.policy.s3;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+
+import stirling.software.proprietary.integration.service.IntegrationConfigUsageCheck;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.source.Source;
+import stirling.software.proprietary.policy.source.SourceStore;
+import stirling.software.proprietary.policy.store.PolicyStore;
+
+/**
+ * Reports the policy sources and pipeline outputs referencing an S3 connection, so the connection
+ * cannot be deleted out from under them (mirrors {@code SourceController}'s referenced-source
+ * delete guard). Scans in memory - fine at admin-dashboard scale, always consistent with the live
+ * stores.
+ */
+@Component
+@RequiredArgsConstructor
+@ConditionalOnBooleanProperty(name = "policies.enabled")
+public class PolicyS3ConnectionUsageCheck implements IntegrationConfigUsageCheck {
+
+    private final SourceStore sourceStore;
+    private final PolicyStore policyStore;
+
+    @Override
+    public List<String> usagesOf(long configId) {
+        List<String> usages = new ArrayList<>();
+        for (Source source : sourceStore.all()) {
+            if (references(source.options(), configId)) {
+                usages.add("source '" + source.name() + "'");
+            }
+        }
+        for (Policy policy : policyStore.all()) {
+            if (references(policy.output().options(), configId)) {
+                usages.add("pipeline '" + policy.name() + "'");
+            }
+        }
+        return usages;
+    }
+
+    private static boolean references(Map<String, Object> options, long configId) {
+        try {
+            Long reference = S3ConnectionResolver.connectionId(options);
+            return reference != null && reference == configId;
+        } catch (IllegalArgumentException unparseable) {
+            return false;
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3Config.java

@@ -0,0 +1,105 @@
+package stirling.software.proprietary.policy.s3;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+
+/**
+ * The fully resolved connection settings the S3 input source and output sink run with - normally
+ * produced by {@link S3ConnectionResolver} merging a stored connection (bucket, region, endpoint,
+ * credentials) with per-use options (prefix, mode), or parsed directly from legacy options that
+ * still embed credentials. Credentials are required: there is deliberately no fallback to the
+ * server's own AWS credential chain, so user-supplied config can never borrow the host's identity.
+ * {@code snapshot} is input-only and ignored by the sink.
+ */
+public record S3Config(
+        String bucket,
+        String region,
+        String prefix,
+        String endpoint,
+        String accessKeyId,
+        String secretAccessKey,
+        boolean snapshot) {
+
+    private static final String BUCKET_OPTION = "bucket";
+    private static final String REGION_OPTION = "region";
+    private static final String PREFIX_OPTION = "prefix";
+    private static final String ENDPOINT_OPTION = "endpoint";
+    private static final String ACCESS_KEY_ID_OPTION = "accessKeyId";
+    private static final String SECRET_ACCESS_KEY_OPTION = "secretAccessKey";
+    private static final String MODE_OPTION = "mode";
+    private static final String MODE_CONSUME = "consume";
+    private static final String MODE_SNAPSHOT = "snapshot";
+
+    public static S3Config from(Map<String, Object> options) {
+        String bucket = trimmed(options.get(BUCKET_OPTION));
+        if (bucket == null) {
+            throw new IllegalArgumentException("s3 config requires a 'bucket' option");
+        }
+        String region = trimmed(options.get(REGION_OPTION));
+        String prefix = trimmed(options.get(PREFIX_OPTION));
+        if (prefix != null && prefix.startsWith("/")) {
+            prefix = prefix.substring(1);
+        }
+        String endpoint = validEndpoint(trimmed(options.get(ENDPOINT_OPTION)));
+        String accessKeyId = trimmed(options.get(ACCESS_KEY_ID_OPTION));
+        String secretAccessKey = trimmed(options.get(SECRET_ACCESS_KEY_OPTION));
+        if (accessKeyId == null || secretAccessKey == null) {
+            throw new IllegalArgumentException(
+                    "s3 config requires an 'accessKeyId' and 'secretAccessKey'");
+        }
+        String mode = trimmed(options.get(MODE_OPTION));
+        if (mode != null && !MODE_CONSUME.equals(mode) && !MODE_SNAPSHOT.equals(mode)) {
+            throw new IllegalArgumentException("s3 config 'mode' must be 'consume' or 'snapshot'");
+        }
+        return new S3Config(
+                bucket,
+                region == null ? "us-east-1" : region,
+                prefix == null ? "" : prefix,
+                endpoint,
+                accessKeyId,
+                secretAccessKey,
+                MODE_SNAPSHOT.equals(mode));
+    }
+
+    private static String validEndpoint(String endpoint) {
+        if (endpoint == null) {
+            return null;
+        }
+        URI uri;
+        try {
+            uri = new URI(endpoint);
+        } catch (URISyntaxException e) {
+            throw new IllegalArgumentException("s3 config 'endpoint' is not a valid URL", e);
+        }
+        if (!"http".equals(uri.getScheme()) && !"https".equals(uri.getScheme())) {
+            throw new IllegalArgumentException(
+                    "s3 config 'endpoint' must be an http(s) URL, e.g. https://s3.example.com");
+        }
+        return endpoint;
+    }
+
+    private static String trimmed(Object value) {
+        if (value == null) {
+            return null;
+        }
+        String text = value.toString().trim();
+        return text.isEmpty() ? null : text;
+    }
+
+    /** Never prints the credentials, so an accidental log line cannot leak them. */
+    @Override
+    public String toString() {
+        return "S3Config[bucket="
+                + bucket
+                + ", region="
+                + region
+                + ", prefix="
+                + prefix
+                + ", endpoint="
+                + endpoint
+                + ", snapshot="
+                + snapshot
+                + "]";
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3ConnectionPool.java

@@ -0,0 +1,110 @@
+package stirling.software.proprietary.policy.s3;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
+import org.springframework.stereotype.Service;
+
+import jakarta.annotation.PreDestroy;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.cluster.s3.S3Clients;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3ClientBuilder;
+import software.amazon.awssdk.services.s3.S3Configuration;
+
+/**
+ * Long-lived {@link S3Client}s for policy S3 sources and sinks, one per distinct {@link S3Config},
+ * closed at shutdown. An edited spec simply maps to a new entry, and a stale entry costs nothing
+ * (the URL-connection HTTP client holds no pooled sockets or threads). Clients sign exclusively
+ * with the spec's own credentials - there is deliberately no fallback to the server's AWS
+ * credential chain, so user-supplied config can never borrow the host's identity. Endpoints are
+ * guarded against private addresses before a client is ever built, since they come from portal
+ * users rather than the operator.
+ */
+@Service
+@ConditionalOnBooleanProperty(name = "policies.enabled")
+public class S3ConnectionPool {
+
+    private final ApplicationProperties applicationProperties;
+    private final Function<S3Config, S3Client> clientFactory;
+    private final Map<S3Config, S3Client> clients = new ConcurrentHashMap<>();
+
+    @Autowired
+    public S3ConnectionPool(ApplicationProperties applicationProperties) {
+        this(applicationProperties, S3ConnectionPool::buildClient);
+    }
+
+    /** Factory-injecting constructor for tests. */
+    public S3ConnectionPool(
+            ApplicationProperties applicationProperties,
+            Function<S3Config, S3Client> clientFactory) {
+        this.applicationProperties = applicationProperties;
+        this.clientFactory = clientFactory;
+    }
+
+    public S3Client clientFor(S3Config config) {
+        return clients.computeIfAbsent(
+                config,
+                c -> {
+                    requirePermittedEndpoint(c);
+                    return clientFactory.apply(c);
+                });
+    }
+
+    /**
+     * A user-supplied endpoint must not reach loopback, link-local, or private addresses unless the
+     * operator has opted in via {@code policies.allowPrivateS3Endpoints}.
+     */
+    private void requirePermittedEndpoint(S3Config config) {
+        if (config.endpoint() == null) {
+            return;
+        }
+        try {
+            S3Clients.validateEndpointHost(
+                    URI.create(config.endpoint()),
+                    applicationProperties.getPolicies().isAllowPrivateS3Endpoints(),
+                    "S3 source endpoint",
+                    "set policies.allowPrivateS3Endpoints=true to opt in (e.g. for a local"
+                            + " MinIO).");
+        } catch (IllegalStateException e) {
+            throw new IllegalArgumentException(e.getMessage(), e);
+        }
+    }
+
+    private static S3Client buildClient(S3Config config) {
+        S3ClientBuilder builder =
+                S3Client.builder()
+                        .httpClient(UrlConnectionHttpClient.create())
+                        .region(Region.of(config.region()))
+                        // Path-style addressing whenever a custom endpoint is set: S3-compatible
+                        // stores rarely support virtual-hosted bucket DNS.
+                        .serviceConfiguration(
+                                S3Configuration.builder()
+                                        .pathStyleAccessEnabled(config.endpoint() != null)
+                                        .build())
+                        .credentialsProvider(
+                                StaticCredentialsProvider.create(
+                                        AwsBasicCredentials.create(
+                                                config.accessKeyId(), config.secretAccessKey())));
+        if (config.endpoint() != null) {
+            builder.endpointOverride(URI.create(config.endpoint()));
+        }
+        return builder.build();
+    }
+
+    @PreDestroy
+    void closeClients() {
+        clients.values().forEach(S3Client::close);
+        clients.clear();
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3ConnectionResolver.java

@@ -0,0 +1,151 @@
+package stirling.software.proprietary.policy.s3;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
+import org.springframework.security.core.Authentication;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.security.core.userdetails.UserDetails;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.proprietary.access.model.ResourceType;
+import stirling.software.proprietary.access.service.OwnershipService;
+import stirling.software.proprietary.integration.model.IntegrationConfig;
+import stirling.software.proprietary.integration.model.IntegrationType;
+import stirling.software.proprietary.integration.repository.IntegrationConfigRepository;
+import stirling.software.proprietary.security.model.User;
+import stirling.software.proprietary.security.service.UserService;
+
+import tools.jackson.core.type.TypeReference;
+import tools.jackson.databind.ObjectMapper;
+
+/**
+ * Turns a source's or output's options into a full {@link S3Config} by dereferencing its {@code
+ * connectionId} to a stored S3 {@link IntegrationConfig} (the connection owns bucket, region,
+ * endpoint, and credentials; the options own per-use settings such as prefix and mode). Options
+ * with no {@code connectionId} fall back to legacy embedded credentials, so rows written before
+ * connections shipped keep working until {@link EmbeddedS3CredentialMigration} rewrites them.
+ *
+ * <p>When an authenticated caller is present (save-time validation), they must be allowed to use
+ * the connection. Background sweeps and deliveries run with no caller and skip that check: the
+ * referencing source or policy was access-checked when it was saved.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+@Transactional(readOnly = true)
+@ConditionalOnBooleanProperty(name = "policies.enabled")
+public class S3ConnectionResolver {
+
+    static final String CONNECTION_ID_OPTION = "connectionId";
+    private static final String PREFIX_OPTION = "prefix";
+    private static final String MODE_OPTION = "mode";
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    private final IntegrationConfigRepository connections;
+    private final OwnershipService ownership;
+    private final UserService userService;
+
+    public S3Config resolve(Map<String, Object> options) {
+        Long connectionId = connectionId(options);
+        if (connectionId == null) {
+            // Legacy embedded credentials, pending migration.
+            return S3Config.from(options);
+        }
+        IntegrationConfig connection =
+                connections
+                        .findById(connectionId)
+                        .filter(cfg -> cfg.getIntegrationType() == IntegrationType.S3)
+                        .filter(this::usableByCurrentUser)
+                        // Existence and access collapse into one error: a caller must not be able
+                        // to tell "no such connection" from "someone else's connection" and
+                        // enumerate ids. The id/name are never echoed.
+                        .orElseThrow(
+                                () ->
+                                        new IllegalArgumentException(
+                                                "unknown or inaccessible s3 connection"));
+        if (!connection.isEnabled()) {
+            throw new IllegalArgumentException("s3 connection is disabled");
+        }
+        Map<String, Object> merged = new LinkedHashMap<>(connectionConfig(connection));
+        copyPerUseOption(options, merged, PREFIX_OPTION);
+        copyPerUseOption(options, merged, MODE_OPTION);
+        return S3Config.from(merged);
+    }
+
+    /** The {@code connectionId} option as a long, or null when the options are legacy-embedded. */
+    static Long connectionId(Map<String, Object> options) {
+        Object reference = options.get(CONNECTION_ID_OPTION);
+        if (reference == null || (reference instanceof String s && s.isBlank())) {
+            return null;
+        }
+        if (reference instanceof Number number) {
+            return number.longValue();
+        }
+        try {
+            return Long.valueOf(reference.toString().trim());
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException(
+                    "s3 'connectionId' is not a valid connection reference: " + reference);
+        }
+    }
+
+    /**
+     * Whether the current caller may use this connection. With no principal - a background sweep or
+     * delivery on a worker thread that carries no {@code SecurityContext} - access is treated as
+     * already established: stored policies are validated with the caller present at save time, and
+     * ad-hoc runs are validated on the request thread before dispatch (see {@code
+     * PolicyValidator#validateOutput}). A missing principal must therefore never be the ONLY thing
+     * standing between a caller and a connection, or the check becomes a confused deputy.
+     */
+    private boolean usableByCurrentUser(IntegrationConfig connection) {
+        User user = currentUser();
+        return user == null || ownership.canUse(ResourceType.INTEGRATION_CONFIG, connection, user);
+    }
+
+    // Mirrors ResourceAccessSecurity's principal resolution; null when unauthenticated.
+    private User currentUser() {
+        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
+        if (auth == null || !auth.isAuthenticated()) {
+            return null;
+        }
+        Object principal = auth.getPrincipal();
+        if (principal instanceof User user) {
+            return user;
+        }
+        if (principal instanceof UserDetails userDetails) {
+            return userService.findByUsername(userDetails.getUsername()).orElse(null);
+        }
+        if (principal instanceof String username && !"anonymousUser".equals(username)) {
+            return userService.findByUsername(username).orElse(null);
+        }
+        return null;
+    }
+
+    private static Map<String, Object> connectionConfig(IntegrationConfig connection) {
+        String json = connection.getConfig();
+        if (json == null || json.isBlank()) {
+            return Map.of();
+        }
+        try {
+            return OBJECT_MAPPER.readValue(
+                    json, new TypeReference<LinkedHashMap<String, Object>>() {});
+        } catch (Exception e) {
+            throw new IllegalArgumentException(
+                    "s3 connection '" + connection.getName() + "' has unreadable config", e);
+        }
+    }
+
+    private static void copyPerUseOption(
+            Map<String, Object> options, Map<String, Object> merged, String key) {
+        Object value = options.get(key);
+        if (value != null && !value.toString().isBlank()) {
+            merged.put(key, value);
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3Identities.java

@@ -0,0 +1,28 @@
+package stirling.software.proprietary.policy.s3;
+
+import java.time.Instant;
+
+/**
+ * The S3 backend's identity and version scheme, shared by {@code S3InputSource} and {@code
+ * S3OutputSink} so outputs are recorded under exactly the identity and gate the next listing
+ * derives. Identity is {@code s3://bucket/key}; the gate is the ETag every listing returns for free
+ * (multipart ETags are not content hashes, so any ETag change simply reads as a new version).
+ */
+public final class S3Identities {
+
+    private S3Identities() {}
+
+    public static String identity(String bucket, String key) {
+        return "s3://" + bucket + "/" + key;
+    }
+
+    /** ETag stripped of its quotes; falls back to size:lastModified for stores that omit it. */
+    public static String gate(String eTag, Long size, Instant lastModified) {
+        if (eTag != null && !eTag.isBlank()) {
+            return eTag.replace("\"", "");
+        }
+        return (size == null ? -1 : size)
+                + ":"
+                + (lastModified == null ? 0 : lastModified.toEpochMilli());
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/S3IntegrationValidator.java

@@ -0,0 +1,51 @@
+package stirling.software.proprietary.policy.s3;
+
+import java.net.URI;
+import java.util.Map;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.cluster.s3.S3Clients;
+import stirling.software.proprietary.integration.model.IntegrationType;
+import stirling.software.proprietary.integration.service.IntegrationConfigValidator;
+
+/**
+ * The S3 connection schema, enforced when an S3 {@link IntegrationType} config is saved: bucket and
+ * credentials required, endpoint an http(s) URL that must not reach private addresses without the
+ * operator opt-in - the same rules {@link S3ConnectionPool} enforces before signing, moved to save
+ * time so a bad connection fails in the form rather than in a sweep.
+ */
+@Component
+@RequiredArgsConstructor
+@ConditionalOnBooleanProperty(name = "policies.enabled")
+public class S3IntegrationValidator implements IntegrationConfigValidator {
+
+    private final ApplicationProperties applicationProperties;
+
+    @Override
+    public IntegrationType type() {
+        return IntegrationType.S3;
+    }
+
+    @Override
+    public void validate(Map<String, Object> config) {
+        S3Config parsed = S3Config.from(config);
+        if (parsed.endpoint() == null) {
+            return;
+        }
+        try {
+            S3Clients.validateEndpointHost(
+                    URI.create(parsed.endpoint()),
+                    applicationProperties.getPolicies().isAllowPrivateS3Endpoints(),
+                    "S3 connection endpoint",
+                    "set policies.allowPrivateS3Endpoints=true to opt in (e.g. for a local"
+                            + " MinIO).");
+        } catch (IllegalStateException e) {
+            throw new IllegalArgumentException(e.getMessage(), e);
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/EditorSource.java

@@ -0,0 +1,27 @@
+package stirling.software.proprietary.policy.source;
+
+/**
+ * The Editor as a virtual, always-present source. Unlike a persisted {@link Source} it is neither
+ * stored nor configurable: it stands for the documents a team processes by running policies from
+ * the in-app editor (the {@code POST /api/v1/policies/{id}/run} path). Its throughput is tracked
+ * through {@link SourceDocCounter} under a synthetic, team-scoped key, so each team sees only its
+ * own editor activity and the client is only ever handed the opaque {@link #ID}, never a team.
+ */
+public final class EditorSource {
+
+    /** The single, stable id and type the client sees for the editor row. */
+    public static final String ID = "editor";
+
+    public static final String TYPE = "editor";
+
+    private EditorSource() {}
+
+    /**
+     * The per-team {@link SourceDocCounter} key. A {@code null} team (login disabled / self-hosted
+     * single user) shares one global bucket; otherwise counts are partitioned by team so a team's
+     * total aggregates every member's editor runs and no other team's.
+     */
+    public static String counterKey(Long teamId) {
+        return teamId == null ? ID : ID + ":" + teamId;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceAccessGuard.java

@@ -34,6 +34,13 @@ public class SourceAccessGuard {
 
     /** Team a new source is stamped with: the creator's team. {@code null} when login disabled. */
     public Long teamForNewSource() {
+        return currentTeamId();
+    }
+
+    /**
+     * The current user's team (what scopes their sources), or {@code null} when login is disabled.
+     */
+    public Long currentTeamId() {
         return enforced() ? policyManagementAuthority.currentUserTeamId() : null;
     }
 

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java

@@ -1,6 +1,7 @@
 package stirling.software.proprietary.policy.source;
 
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
 import org.springframework.http.HttpStatus;
@@ -29,6 +30,7 @@ import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.store.PolicyStore;
 import stirling.software.proprietary.policy.trigger.PolicyTriggerManager;
+import stirling.software.proprietary.util.SecretMasker;
 
 /**
  * CRUD for persisted, reusable input connections plus the Sources overview for the admin portal. A
@@ -65,11 +67,16 @@ public class SourceController {
     }
 
     @GetMapping("/{sourceId}")
-    @Operation(summary = "Get a source by id")
+    @Operation(
+            summary = "Get a source by id",
+            description =
+                    "Secret-bearing options are returned as a redaction sentinel, never their"
+                            + " stored values; an edit that sends the sentinel back keeps them.")
     public ResponseEntity<Source> get(@PathVariable String sourceId) {
         return sourceStore
                 .get(sourceId)
                 .filter(sourceAccessGuard::canAccess)
+                .map(SourceController::withMaskedSecrets)
                 .map(ResponseEntity::ok)
                 .orElseGet(() -> ResponseEntity.notFound().build());
     }
@@ -81,6 +88,10 @@ public class SourceController {
                     "The trailing 30-day per-day document series (oldest first) for the source's"
                             + " sparkline.")
     public ResponseEntity<List<Long>> documentCounts(@PathVariable String sourceId) {
+        // The editor is virtual: its series is tracked per team, not against a persisted source.
+        if (EditorSource.ID.equals(sourceId)) {
+            return ResponseEntity.ok(overviewService.editorDailySeries());
+        }
         return sourceStore
                 .get(sourceId)
                 .filter(sourceAccessGuard::canAccess)
@@ -97,7 +108,8 @@ public class SourceController {
                             + " matching source type.")
     public ResponseEntity<Source> save(@RequestBody Source source) {
         requireSourceEditingAllowed();
-        Source owned = resolveOwnership(source);
+        requireNotEditor(source.id(), source.type());
+        Source owned = withStoredSecrets(resolveOwnership(source));
         try {
             validateConfig(owned);
         } catch (IllegalArgumentException e) {
@@ -107,7 +119,7 @@ public class SourceController {
         // An edited folder source can change which directory needs watching, so re-sync trigger
         // registrations now instead of waiting for the next reconcile.
         policyTriggerManager.notifyPoliciesChanged();
-        return ResponseEntity.ok(saved);
+        return ResponseEntity.ok(withMaskedSecrets(saved));
     }
 
     @DeleteMapping("/{sourceId}")
@@ -118,6 +130,7 @@ public class SourceController {
                             + " so the connection can't be pulled out from under a live policy.")
     public ResponseEntity<Void> delete(@PathVariable String sourceId) {
         requireSourceEditingAllowed();
+        requireNotEditor(sourceId, null);
         Source source = sourceStore.get(sourceId).filter(sourceAccessGuard::canAccess).orElse(null);
         if (source == null) {
             return ResponseEntity.notFound().build();
@@ -169,6 +182,42 @@ public class SourceController {
                 teamId);
     }
 
+    private static Source withOptions(Source source, Map<String, Object> options) {
+        return new Source(
+                source.id(),
+                source.name(),
+                source.type(),
+                options,
+                source.enabled(),
+                source.owner(),
+                source.teamId());
+    }
+
+    /** Secrets never leave the server: reads return the redaction sentinel in their place. */
+    private static Source withMaskedSecrets(Source source) {
+        return withOptions(source, SecretMasker.mask(source.options()));
+    }
+
+    /**
+     * An edit that round-trips a masked read sends secrets back as the sentinel; restore them from
+     * the stored source so saving without re-typing keeps them (validation then runs against the
+     * real values).
+     */
+    private Source withStoredSecrets(Source incoming) {
+        if (incoming.id() == null || incoming.id().isBlank()) {
+            return incoming;
+        }
+        return sourceStore
+                .get(incoming.id())
+                .map(
+                        existing ->
+                                withOptions(
+                                        incoming,
+                                        SecretMasker.restoreRedacted(
+                                                incoming.options(), existing.options())))
+                .orElse(incoming);
+    }
+
     /** Validate the config against the bean that handles the source's type, as the engine will. */
     private void validateConfig(Source source) {
         InputSpec spec = source.toInputSpec();
@@ -195,6 +244,18 @@ public class SourceController {
         }
     }
 
+    /**
+     * The editor is a built-in, virtual source: it is always present and cannot be created, edited,
+     * or deleted like a persisted connection. Reject any attempt to touch it by id or type.
+     */
+    private static void requireNotEditor(String id, String type) {
+        if (EditorSource.ID.equals(id) || EditorSource.TYPE.equals(type)) {
+            throw new ResponseStatusException(
+                    HttpStatus.BAD_REQUEST,
+                    "The editor is a built-in source and cannot be created, edited, or deleted");
+        }
+    }
+
     /** Names of the caller's visible policies that reference the given source. */
     private List<String> referencingPolicyNames(String sourceId) {
         return policyAccessGuard.visibleFrom(policyStore).stream()

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceEntity.java

@@ -3,6 +3,7 @@ package stirling.software.proprietary.policy.source;
 import java.io.Serializable;
 
 import jakarta.persistence.Column;
+import jakarta.persistence.Convert;
 import jakarta.persistence.Entity;
 import jakarta.persistence.Id;
 import jakarta.persistence.Table;
@@ -11,6 +12,8 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
 
+import stirling.software.proprietary.integration.crypto.LenientEncryptedStringConverter;
+
 /**
  * JPA row for a {@link Source}. The whole source lives as JSON in {@code sourceJson} (authoritative
  * on read); the scalar columns are denormalized copies for querying. {@code owner} and {@code
@@ -45,6 +48,9 @@ public class SourceEntity implements Serializable {
     @Column(name = "enabled")
     private boolean enabled;
 
+    // Encrypted at rest: source options carry user-supplied credentials (e.g. an S3 secret
+    // access key). Lenient so rows written before encryption shipped still load.
+    @Convert(converter = LenientEncryptedStringConverter.class)
     @Column(name = "source_json", columnDefinition = "text")
     private String sourceJson;
 }

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java

@@ -14,6 +14,7 @@ import lombok.RequiredArgsConstructor;
 import stirling.software.proprietary.policy.config.PolicyAccessGuard;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.store.PolicyStore;
+import stirling.software.proprietary.util.SecretMasker;
 
 /**
  * Builds the Sources overview: every persisted source the caller's team owns, shown exactly once,
@@ -40,7 +41,7 @@ public class SourceOverviewService {
         Map<String, DocStats> docStats =
                 docCounter.statsFor(sources.stream().map(Source::id).toList());
 
-        List<SourceView> views =
+        List<SourceView> persisted =
                 sources.stream()
                         .map(
                                 source ->
@@ -55,7 +56,13 @@ public class SourceOverviewService {
                                         .thenComparing(SourceView::name))
                         .toList();
 
-        return new SourcesResponse(buildKpis(views), views);
+        // The editor is a built-in source: always present and pinned first. The KPI strip counts
+        // only the connections a team configures, so the editor is left out of the KPIs.
+        List<SourceView> views = new ArrayList<>();
+        views.add(editorView(policies));
+        views.addAll(persisted);
+
+        return new SourcesResponse(buildKpis(persisted), views);
     }
 
     /**
@@ -66,6 +73,49 @@ public class SourceOverviewService {
         return docCounter.dailySeriesFor(sourceId);
     }
 
+    /** The 30-day daily editor document series (oldest first) for the caller's team. */
+    public List<Long> editorDailySeries() {
+        return docCounter.dailySeriesFor(
+                EditorSource.counterKey(sourceAccessGuard.currentTeamId()));
+    }
+
+    /**
+     * The always-present editor row. It has no stored config; its documents are those the team has
+     * processed by running policies from the editor, and it is "used by" every policy that targets
+     * the editor as its source.
+     */
+    private SourceView editorView(List<Policy> policies) {
+        String key = EditorSource.counterKey(sourceAccessGuard.currentTeamId());
+        DocStats docs = docCounter.statsFor(List.of(key)).getOrDefault(key, DocStats.ZERO);
+        List<SourceView.PolicyRef> refs =
+                policies.stream()
+                        .filter(SourceOverviewService::runsFromEditor)
+                        .map(policy -> new SourceView.PolicyRef(policy.id(), policy.name()))
+                        .toList();
+        return new SourceView(
+                EditorSource.ID,
+                "Editor",
+                EditorSource.TYPE,
+                "active",
+                refs.size(),
+                refs,
+                List.of(),
+                docs.total(),
+                docs.last24h(),
+                docs.last30d());
+    }
+
+    /**
+     * Whether a policy runs from the editor. Editor membership is carried in the policy's output
+     * metadata ({@code output.options.sources}) - a client-side list the editor writes when a
+     * policy targets it - rather than as a persisted {@code sourceId}, because the editor is
+     * virtual and has no stored source to reference.
+     */
+    private static boolean runsFromEditor(Policy policy) {
+        Object sources = policy.output().options().get("sources");
+        return sources instanceof List<?> list && list.contains(EditorSource.ID);
+    }
+
     /** Policies referencing each source id, across the caller's visible policies. */
     private static Map<String, List<Policy>> referencesBySource(List<Policy> policies) {
         Map<String, List<Policy>> bySource = new HashMap<>();
@@ -104,13 +154,18 @@ public class SourceOverviewService {
         return referenceCount == 0 ? "unused" : "active";
     }
 
-    /** Generic key/value view of the source's config - works for any source type. */
+    /**
+     * Generic key/value view of the source's config - works for any source type. Secret-bearing
+     * options (e.g. an S3 secret access key) are redacted, not omitted, so the overview still shows
+     * that a credential is configured.
+     */
     private static List<SourceView.DetailRow> configRows(Source source) {
-        return source.options().entrySet().stream()
+        Map<String, Object> masked = SecretMasker.mask(source.options());
+        return source.options().keySet().stream()
                 .map(
-                        entry ->
+                        key ->
                                 new SourceView.DetailRow(
-                                        humanize(entry.getKey()), String.valueOf(entry.getValue())))
+                                        humanize(key), String.valueOf(masked.get(key))))
                 .toList();
     }
 

app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyEntity.java

@@ -3,6 +3,7 @@ package stirling.software.proprietary.policy.store;
 import java.io.Serializable;
 
 import jakarta.persistence.Column;
+import jakarta.persistence.Convert;
 import jakarta.persistence.Entity;
 import jakarta.persistence.Id;
 import jakarta.persistence.Table;
@@ -11,6 +12,8 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
 
+import stirling.software.proprietary.integration.crypto.LenientEncryptedStringConverter;
+
 /**
  * JPA row for a {@link stirling.software.proprietary.policy.model.Policy}. The whole policy lives
  * as JSON in {@code policyJson} (authoritative on read); the scalar columns are denormalized copies
@@ -55,6 +58,9 @@ public class PolicyEntity implements Serializable {
     @Column(name = "sort_order")
     private Integer sortOrder;
 
+    // Encrypted at rest: output options carry user-supplied credentials (e.g. an S3 secret
+    // access key). Lenient so rows written before encryption shipped still load.
+    @Convert(converter = LenientEncryptedStringConverter.class)
     @Column(name = "policy_json", columnDefinition = "text")
     private String policyJson;
 }

app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ScheduleTrigger.java

@@ -99,11 +99,17 @@ public class ScheduleTrigger implements PolicyTrigger {
             // Baseline a newly-seen policy to now so it does not fire immediately.
             Instant last = lastFiredByPolicy.computeIfAbsent(policy.id(), id -> now);
             ZonedDateTime next = config.schedule().nextAfter(last.atZone(config.zone()));
-            if (!next.toInstant().isAfter(now)) {
-                lastFiredByPolicy.put(policy.id(), now);
-                log.info("Scheduled policy {} ({}) is due", policy.id(), policy.name());
-                policyRunner.run(policy);
+            if (next.toInstant().isAfter(now)) {
+                continue;
             }
+            ZonedDateTime later = config.schedule().nextAfter(next);
+            while (!later.toInstant().isAfter(now)) {
+                next = later;
+                later = config.schedule().nextAfter(later);
+            }
+            lastFiredByPolicy.put(policy.id(), next.toInstant());
+            log.info("Scheduled policy {} ({}) is due", policy.id(), policy.name());
+            policyRunner.run(policy);
         }
     }
 

app/proprietary/src/main/java/stirling/software/proprietary/repository/PersistentAuditEventRepository.java

@@ -184,6 +184,20 @@ public interface PersistentAuditEventRepository extends JpaRepository<Persistent
     Page<PersistentAuditEvent> findByPrincipalIn(
             @Param("principals") List<String> principals, Pageable pageable);
 
+    // Noise-excluding scans for the portal: keep the page window full of meaningful events so the
+    // list doesn't shrink as read/polling noise (UI_DATA/HTTP_REQUEST) grows in the recent window.
+    @Query("SELECT e FROM PersistentAuditEvent e WHERE e.type NOT IN :excludedTypes")
+    Page<PersistentAuditEvent> findByTypeNotIn(
+            @Param("excludedTypes") List<String> excludedTypes, Pageable pageable);
+
+    @Query(
+            "SELECT e FROM PersistentAuditEvent e WHERE e.type NOT IN :excludedTypes AND e.principal"
+                    + " IN :principals")
+    Page<PersistentAuditEvent> findByTypeNotInAndPrincipalIn(
+            @Param("excludedTypes") List<String> excludedTypes,
+            @Param("principals") List<String> principals,
+            Pageable pageable);
+
     @Query(
             "SELECT e FROM PersistentAuditEvent e WHERE e.type IN :types AND e.timestamp BETWEEN :startDate AND :endDate")
     Page<PersistentAuditEvent> findByTypeInAndTimestampBetween(
@@ -276,4 +290,25 @@ public interface PersistentAuditEventRepository extends JpaRepository<Persistent
             @Param("source") String source,
             @Param("excludeType") String excludeType,
             @Param("since") Instant since);
+
+    // Team-scoped (SaaS) variants: same free-UI counts, constrained to a team's member principals.
+    @Query(
+            "SELECT COUNT(e) FROM PersistentAuditEvent e "
+                    + "WHERE e.type IN :types AND e.source = :source "
+                    + "AND e.principal IN :principals AND e.timestamp > :since")
+    long countByTypeInAndSourceAndPrincipalInAndTimestampAfter(
+            @Param("types") List<String> types,
+            @Param("source") String source,
+            @Param("principals") List<String> principals,
+            @Param("since") Instant since);
+
+    @Query(
+            "SELECT COUNT(DISTINCT e.principal) FROM PersistentAuditEvent e "
+                    + "WHERE e.source = :source AND e.type <> :excludeType "
+                    + "AND e.principal IN :principals AND e.timestamp > :since")
+    long countDistinctPrincipalsBySourceExcludingTypeAndPrincipalInAfter(
+            @Param("source") String source,
+            @Param("excludeType") String excludeType,
+            @Param("principals") List<String> principals,
+            @Param("since") Instant since);
 }

app/proprietary/src/main/java/stirling/software/proprietary/security/model/User.java

@@ -103,7 +103,6 @@ public class User implements UserDetails, Serializable {
 
     @ElementCollection
     @MapKeyColumn(name = "setting_key")
-    @Lob
     @Column(name = "setting_value", columnDefinition = "text")
     @CollectionTable(name = "user_settings", joinColumns = @JoinColumn(name = "user_id"))
     @JsonIgnore

app/proprietary/src/main/java/stirling/software/proprietary/service/AiDocumentHtmlRenderer.java

@@ -0,0 +1,135 @@
+package stirling.software.proprietary.service;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.stereotype.Component;
+
+import com.hubspot.jinjava.Jinjava;
+import com.hubspot.jinjava.JinjavaConfig;
+
+import stirling.software.proprietary.model.api.ai.create.AiDocument;
+
+/** Renders an {@link AiDocument} to HTML using a Jinja template loaded from the classpath. */
+@Component
+public class AiDocumentHtmlRenderer {
+
+    private static final String TEMPLATE_PATH = "templates/ai/create/document.html.jinja2";
+
+    private static final Pattern SAFE_COLOR = Pattern.compile("^#[0-9a-fA-F]{6}$");
+
+    private final Jinjava jinjava;
+    private final String template;
+
+    public AiDocumentHtmlRenderer() {
+        JinjavaConfig config =
+                JinjavaConfig.newBuilder().withNestedInterpretationEnabled(false).build();
+        this.jinjava = new Jinjava(config);
+        this.template = loadTemplate();
+    }
+
+    public String render(AiDocument doc) {
+        return jinjava.render(template, buildContext(doc));
+    }
+
+    private static Map<String, Object> buildContext(AiDocument doc) {
+        Map<String, Object> context = new LinkedHashMap<>();
+        context.put("title", doc.getTitle());
+        context.put("subtitle", doc.getSubtitle());
+        context.put("reference_number", doc.getReferenceNumber());
+
+        AiDocument.Style style = doc.getStyle();
+        if (style != null) {
+            context.put("style_primary", safeColor(style.getPrimaryColor()));
+            context.put("style_background", safeColor(style.getBackgroundColor()));
+            context.put("style_body", safeColor(style.getBodyTextColor()));
+        }
+
+        List<Map<String, Object>> sections = new ArrayList<>();
+        if (doc.getSections() != null) {
+            for (AiDocument.Section section : doc.getSections()) {
+                if (section != null && section.getType() != null) {
+                    sections.add(buildSection(section));
+                }
+            }
+        }
+        context.put("sections", sections);
+        return context;
+    }
+
+    private static Map<String, Object> buildSection(AiDocument.Section section) {
+        Map<String, Object> node = new LinkedHashMap<>();
+        node.put("type", section.getType());
+        node.put("heading", section.getHeading());
+        switch (section.getType()) {
+            case "text" -> node.put("paragraphs", paragraphs(section.getBody()));
+            case "key_value" -> node.put("pairs", pairs(section.getPairs()));
+            case "line_items" -> {
+                node.put("columns", orEmpty(section.getColumns()));
+                node.put("rows", orEmptyRows(section.getRows()));
+                node.put("total_row", emptyToNull(section.getTotalRow()));
+            }
+            case "bullet_list" -> node.put("items", orEmpty(section.getItems()));
+            case "signature" -> node.put("signatories", orEmpty(section.getSignatories()));
+            default -> {}
+        }
+        return node;
+    }
+
+    private static List<String> paragraphs(String body) {
+        String text = body == null ? "" : body;
+        List<String> out = new ArrayList<>();
+        for (String paragraph : text.split("\n\n")) {
+            out.add(paragraph.replace("\n", " "));
+        }
+        return out;
+    }
+
+    private static List<Map<String, String>> pairs(List<List<String>> pairs) {
+        List<Map<String, String>> out = new ArrayList<>();
+        if (pairs != null) {
+            for (List<String> pair : pairs) {
+                Map<String, String> node = new LinkedHashMap<>();
+                node.put("label", pair.isEmpty() ? "" : pair.get(0));
+                node.put("value", pair.size() < 2 ? "" : pair.get(1));
+                out.add(node);
+            }
+        }
+        return out;
+    }
+
+    private static List<String> orEmpty(List<String> values) {
+        return values == null ? List.of() : values;
+    }
+
+    private static List<List<String>> orEmptyRows(List<List<String>> rows) {
+        return rows == null ? List.of() : rows;
+    }
+
+    private static List<String> emptyToNull(List<String> values) {
+        return values == null || values.isEmpty() ? null : values;
+    }
+
+    private static String safeColor(String value) {
+        if (value == null) {
+            return null;
+        }
+        String trimmed = value.trim();
+        return SAFE_COLOR.matcher(trimmed).matches() ? trimmed : null;
+    }
+
+    private static String loadTemplate() {
+        try {
+            return new ClassPathResource(TEMPLATE_PATH).getContentAsString(StandardCharsets.UTF_8);
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/service/AiWorkflowService.java

@@ -7,6 +7,7 @@ import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.stream.Collectors;
 
 import org.apache.commons.io.FilenameUtils;
@@ -27,6 +28,7 @@ import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.service.AutomationRunContext;
 import stirling.software.common.service.CustomPDFDocumentFactory;
 import stirling.software.common.service.FileStorage;
 import stirling.software.common.service.InternalApiTimeoutException;
@@ -156,33 +158,41 @@ public class AiWorkflowService {
             throws IOException {
         validateRequest(request);
 
-        // Key by opaque file id, not filename. Filenames aren't guaranteed unique across an
-        // upload (users can rotate the same 'scan.pdf' twice), and the engine identifies files
-        // by id in every response shape that asks Java to look a file up again.
-        Map<String, MultipartFile> filesById = new LinkedHashMap<>();
-        List<AiFile> files = new ArrayList<>();
-        for (AiWorkflowFileInput fileInput : request.getFileInputs()) {
-            MultipartFile multipartFile = fileInput.getFileInput();
-            AiFile aiFile =
-                    new AiFile(
-                            fileIdStrategy.idFor(multipartFile),
-                            multipartFile.getOriginalFilename());
-            filesById.put(aiFile.getId(), multipartFile);
-            files.add(aiFile);
-        }
+        // One AI orchestration = one automation run. Scope a run id (on whichever thread runs
+        // orchestrate — request thread for sync, stream-executor for streaming) so every tool
+        // sub-step it dispatches via PolicyExecutor → InternalApiClient groups into one charge.
+        try (AutomationRunContext.Scope ignored =
+                AutomationRunContext.open(UUID.randomUUID().toString())) {
 
-        WorkflowTurnRequest initialRequest = new WorkflowTurnRequest();
-        initialRequest.setUserMessage(request.getUserMessage().trim());
-        initialRequest.setFiles(files);
-        initialRequest.setConversationHistory(new ArrayList<>(request.getConversationHistory()));
-        initialRequest.setEnabledEndpoints(endpointResolver.getEnabledEndpointUrls());
-        listener.onProgress(AiWorkflowProgressEvent.of(AiWorkflowPhase.ANALYZING));
+            // Key by opaque file id, not filename. Filenames aren't guaranteed unique across an
+            // upload (users can rotate the same 'scan.pdf' twice), and the engine identifies files
+            // by id in every response shape that asks Java to look a file up again.
+            Map<String, MultipartFile> filesById = new LinkedHashMap<>();
+            List<AiFile> files = new ArrayList<>();
+            for (AiWorkflowFileInput fileInput : request.getFileInputs()) {
+                MultipartFile multipartFile = fileInput.getFileInput();
+                AiFile aiFile =
+                        new AiFile(
+                                fileIdStrategy.idFor(multipartFile),
+                                multipartFile.getOriginalFilename());
+                filesById.put(aiFile.getId(), multipartFile);
+                files.add(aiFile);
+            }
 
-        WorkflowState state = new WorkflowState.Pending(initialRequest);
-        while (state instanceof WorkflowState.Pending pending) {
-            state = advance(pending.request(), filesById, listener);
+            WorkflowTurnRequest initialRequest = new WorkflowTurnRequest();
+            initialRequest.setUserMessage(request.getUserMessage().trim());
+            initialRequest.setFiles(files);
+            initialRequest.setConversationHistory(
+                    new ArrayList<>(request.getConversationHistory()));
+            initialRequest.setEnabledEndpoints(endpointResolver.getEnabledEndpointUrls());
+            listener.onProgress(AiWorkflowProgressEvent.of(AiWorkflowPhase.ANALYZING));
+
+            WorkflowState state = new WorkflowState.Pending(initialRequest);
+            while (state instanceof WorkflowState.Pending pending) {
+                state = advance(pending.request(), filesById, listener);
+            }
+            return ((WorkflowState.Terminal) state).response();
         }
-        return ((WorkflowState.Terminal) state).response();
     }
 
     private WorkflowState advance(

app/proprietary/src/main/java/stirling/software/proprietary/service/AuditService.java

@@ -37,9 +37,11 @@ import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.model.api.PDFFile;
 import stirling.software.common.service.CustomPDFDocumentFactory;
+import stirling.software.common.service.InternalApiClient;
 import stirling.software.common.util.RegexPatternUtils;
 import stirling.software.common.util.RequestUriUtils;
 import stirling.software.proprietary.accountlink.BillableOperationClassifier;
+import stirling.software.proprietary.audit.AuditContext;
 import stirling.software.proprietary.audit.AuditEventType;
 import stirling.software.proprietary.audit.AuditLevel;
 import stirling.software.proprietary.audit.Audited;
@@ -469,6 +471,73 @@ public class AuditService {
         }
     }
 
+    /**
+     * Merge controller-supplied context into the audit data, read in the aspect's {@code finally}
+     * after the controller body has run. A policy run stamps its name and step endpoints as request
+     * attributes ({@link AuditContext}); the internal loopback dispatch that executes each pipeline
+     * step carries the automation marker header ({@link InternalApiClient#AUTOMATION_HEADER}).
+     * Surfacing these lets the portal label a run as its policy instead of the raw {@code /run}
+     * endpoint, and flag its sub-steps as automation rather than direct user actions.
+     *
+     * @param data The existing audit data map
+     * @param req The current request, or null when not in a web context
+     */
+    public void addAutomationContext(Map<String, Object> data, HttpServletRequest req) {
+        if (req == null) {
+            return;
+        }
+        Object policyName = req.getAttribute(AuditContext.REQ_ATTR_POLICY_NAME);
+        if (policyName == null) {
+            // A pipeline step's loopback dispatch carries its parent policy name as a header, so
+            // the
+            // step audit ties back to the policy that ran it (not a bare direct call).
+            String header = req.getHeader(InternalApiClient.POLICY_NAME_HEADER);
+            if (header != null && !header.isBlank()) {
+                policyName = header;
+            }
+        }
+        if (policyName != null) {
+            // Both sources are caller-controlled (the header is spoofable and, unlike the sender in
+            // InternalApiClient, uncapped); strip newlines and cap before persisting to audit JSON.
+            String safe = capLabel(String.valueOf(policyName));
+            if (!safe.isEmpty()) {
+                data.put("policyName", safe);
+            }
+        }
+        Object steps = req.getAttribute(AuditContext.REQ_ATTR_POLICY_STEPS);
+        if (steps instanceof List<?> list && !list.isEmpty()) {
+            // Caller-supplied and unbounded; cap count and each entry so a crafted run can't
+            // bloat the audit JSON the portal cache loads and parses in bulk.
+            List<String> safeSteps =
+                    list.stream()
+                            .limit(MAX_POLICY_STEPS)
+                            .map(s -> capLabel(String.valueOf(s)))
+                            .filter(s -> !s.isEmpty())
+                            .toList();
+            if (!safeSteps.isEmpty()) {
+                data.put("policySteps", safeSteps);
+            }
+        }
+        if ("true".equalsIgnoreCase(req.getHeader(InternalApiClient.AUTOMATION_HEADER))) {
+            data.put("automation", Boolean.TRUE);
+        }
+    }
+
+    /** Max characters kept for a persisted policy label; mirrors the InternalApiClient send cap. */
+    private static final int MAX_LABEL_LEN = 200;
+
+    /** Max step endpoints kept on a run's audit event; the portal only shows the first few. */
+    private static final int MAX_POLICY_STEPS = 50;
+
+    /** Single-line, length-capped label safe to persist to audit JSON and render in the portal. */
+    private static String capLabel(String value) {
+        if (value == null) {
+            return "";
+        }
+        String safe = value.replaceAll("[\\r\\n]", " ").trim();
+        return safe.length() > MAX_LABEL_LEN ? safe.substring(0, MAX_LABEL_LEN) : safe;
+    }
+
     /**
      * Add method arguments to the audit data
      *

app/proprietary/src/main/java/stirling/software/proprietary/service/PortalAuditReadService.java

@@ -9,6 +9,7 @@ import org.springframework.stereotype.Service;
 
 import lombok.RequiredArgsConstructor;
 
+import stirling.software.proprietary.audit.AuditEventType;
 import stirling.software.proprietary.audit.PortalAuditEventRow;
 import stirling.software.proprietary.model.security.PersistentAuditEvent;
 import stirling.software.proprietary.repository.PersistentAuditEventRepository;
@@ -24,12 +25,21 @@ public class PortalAuditReadService {
     /** Newest rows to scan; each surface filters this down to what it shows. */
     private static final int SCAN_LIMIT = 400;
 
+    /**
+     * Read/polling noise excluded at the query level so the scan window stays full of meaningful
+     * events. Otherwise a busy scope's recent rows fill with these and the visible list shrinks as
+     * traffic grows - the "audit getting smaller over time" a user would see. No portal surface
+     * shows these types anyway (the infra tab and documents feed both drop them).
+     */
+    private static final List<String> NOISE_TYPES =
+            List.of(AuditEventType.UI_DATA.name(), AuditEventType.HTTP_REQUEST.name());
+
     private final PersistentAuditEventRepository auditRepository;
 
     /** Recent whole-server events (admins). */
     @Cacheable(value = CACHE_NAME, key = "'server'")
     public List<PortalAuditEventRow> serverEvents() {
-        return toRows(auditRepository.findAll(recentPage()).getContent());
+        return toRows(auditRepository.findByTypeNotIn(NOISE_TYPES, recentPage()).getContent());
     }
 
     /** Recent events by the given principals (team scope). Empty principals yield an empty list. */
@@ -38,7 +48,10 @@ public class PortalAuditReadService {
         if (principals.isEmpty()) {
             return List.of();
         }
-        return toRows(auditRepository.findByPrincipalIn(principals, recentPage()).getContent());
+        return toRows(
+                auditRepository
+                        .findByTypeNotInAndPrincipalIn(NOISE_TYPES, principals, recentPage())
+                        .getContent());
     }
 
     private static PageRequest recentPage() {

app/proprietary/src/main/java/stirling/software/proprietary/service/PortalDocumentsService.java

@@ -61,8 +61,17 @@ public class PortalDocumentsService {
                 continue;
             }
             String path = asString(data.get("path"));
-            String source = sourceLabel(asString(data.get("__origin")));
-            String product = "API integration".equals(source) ? "API" : "Editor";
+            // Pipeline steps run over an internal loopback (API-key auth), so origin alone reads as
+            // "API". The automation marker distinguishes a policy-run step from real API traffic.
+            boolean automation = isAutomation(data);
+            String policyName = asString(data.get("policyName"));
+            String source =
+                    automation
+                            ? (policyName != null && !policyName.isBlank()
+                                    ? "Policy: " + policyName
+                                    : "Policy automation")
+                            : sourceLabel(asString(data.get("__origin")));
+            String product = automation ? "Automation" : productLabel(source);
             String action = prettyTool(path);
             boolean failed = isFailure(data);
             Instant ts = event.timestamp();
@@ -174,6 +183,15 @@ public class PortalDocumentsService {
         return "Web upload";
     }
 
+    private static String productLabel(String source) {
+        return "API integration".equals(source) ? "API" : "Editor";
+    }
+
+    private static boolean isAutomation(Map<String, Object> data) {
+        Object v = data.get("automation");
+        return Boolean.TRUE.equals(v) || "true".equalsIgnoreCase(String.valueOf(v));
+    }
+
     private static String docType(String contentType, String name) {
         String ct = contentType == null ? "" : contentType.toLowerCase(Locale.ROOT);
         if (ct.contains("pdf") || name.toLowerCase(Locale.ROOT).endsWith(".pdf")) {

app/proprietary/src/main/java/stirling/software/proprietary/service/PortalInfraAuditService.java

@@ -54,6 +54,7 @@ public class PortalInfraAuditService {
                         .limit(RETURN_LIMIT)
                         .toList();
 
+        int policy = (int) events.stream().filter(e -> "policy".equals(e.getCategory())).count();
         int processing =
                 (int) events.stream().filter(e -> "processing".equals(e.getCategory())).count();
         int elevation =
@@ -63,6 +64,7 @@ public class PortalInfraAuditService {
         InfraAuditSummary summary =
                 InfraAuditSummary.builder()
                         .totalEvents(events.size())
+                        .policy(policy)
                         .processing(processing)
                         .elevation(elevation)
                         .config(config)
@@ -84,20 +86,32 @@ public class PortalInfraAuditService {
     private InfraAuditEventDto toDto(PortalAuditEventRow event) {
         Map<String, Object> data = parseData(event);
         String path = asString(data.get("path"));
-        String category = categoryFor(event.type(), path);
+        String policyName = asString(data.get("policyName"));
+        boolean automation = isAutomation(data);
+        // Classify a dispatch by its real run-path URI, not a policyName: the latter can be spoofed
+        // via the X-Stirling-Policy-Name header to make a direct call pose as a policy row.
+        boolean policyDispatch = isPolicyRunPath(path) && !automation;
+        // A dispatch is its own "policy" category so the UI badges it as a policy run, not a
+        // generic processing op; its internal steps keep their real tool category.
+        String category = policyDispatch ? "policy" : categoryFor(event.type(), path);
 
         return InfraAuditEventDto.builder()
                 .id(String.valueOf(event.id()))
                 .timestamp(event.timestamp() == null ? "" : TS_FORMAT.format(event.timestamp()))
                 .category(category)
-                .action(actionFor(event.type(), path))
+                .action(actionFor(event.type(), path, policyName, automation))
                 .actor(event.principal())
-                .target(targetFor(category, path, data))
+                .target(targetFor(category, path, data, policyDispatch))
                 .status(statusFor(event.type(), category, data))
                 .latencyMs(asLong(data.get("latencyMs")))
                 .build();
     }
 
+    private static boolean isAutomation(Map<String, Object> data) {
+        Object v = data.get("automation");
+        return Boolean.TRUE.equals(v) || "true".equalsIgnoreCase(String.valueOf(v));
+    }
+
     private Map<String, Object> parseData(PortalAuditEventRow event) {
         if (event.data() == null || event.data().isEmpty()) {
             return Map.of();
@@ -140,7 +154,37 @@ public class PortalInfraAuditService {
                 || p.contains("redact");
     }
 
-    private static String actionFor(String type, String path) {
+    /**
+     * Label for a row. A genuine policy dispatch (a {@code /policies/.../run} request) shows the
+     * policy it ran; an internal pipeline step (automation marker) is flagged so it isn't read as a
+     * direct action. The name is only shown as the action on a real run URI, so a spoofed
+     * X-Stirling-Policy-Name header on a direct tool call can't overwrite its true action.
+     */
+    private static String actionFor(
+            String type, String path, String policyName, boolean automation) {
+        if (!automation && isPolicyRunPath(path)) {
+            // A run with no name (ad-hoc pipeline) still reads better than the "run" endpoint.
+            return policyName != null ? policyName : "Policy run";
+        }
+        String base = baseActionFor(type, path);
+        if (automation) {
+            return policyName != null
+                    ? base + " (policy: " + policyName + ")"
+                    : base + " (automation)";
+        }
+        return base;
+    }
+
+    /**
+     * The pipeline-run endpoints: {@code /policies/run}, {@code /run/stream}, {@code /{id}/run}.
+     */
+    private static boolean isPolicyRunPath(String path) {
+        return path != null
+                && path.contains("/policies/")
+                && (path.endsWith("/run") || path.endsWith("/run/stream"));
+    }
+
+    private static String baseActionFor(String type, String path) {
         AuditEventType t = AuditEventType.fromString(type);
         if (t == null) {
             return prettyTool(path);
@@ -191,7 +235,32 @@ public class PortalInfraAuditService {
         return sb.isEmpty() ? "PDF operation" : sb.toString();
     }
 
-    private static String targetFor(String category, String path, Map<String, Object> data) {
+    /** "Auto Redact, Compress PDF" from the run's step endpoints; first three, then "+N more". */
+    private static String prettyStepList(Object steps) {
+        if (!(steps instanceof List<?> list) || list.isEmpty()) {
+            return null;
+        }
+        int shown = Math.min(3, list.size());
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i < shown; i++) {
+            if (i > 0) {
+                sb.append(", ");
+            }
+            sb.append(prettyTool(asString(list.get(i))));
+        }
+        if (list.size() > shown) {
+            sb.append(" +").append(list.size() - shown).append(" more");
+        }
+        return sb.toString();
+    }
+
+    private static String targetFor(
+            String category, String path, Map<String, Object> data, boolean policyDispatch) {
+        if (policyDispatch) {
+            // The run touches no single file at this level; show the tools the policy runs instead.
+            String steps = prettyStepList(data.get("policySteps"));
+            return steps != null ? steps : "Pipeline";
+        }
         if ("auth".equals(category)) {
             // Auth events don't act on a resource; the session is the closest thing.
             return "Web session";

app/proprietary/src/main/java/stirling/software/proprietary/util/SecretMasker.java

@@ -1,5 +1,6 @@
 package stirling.software.proprietary.util;
 
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.regex.Pattern;
@@ -13,10 +14,15 @@ import stirling.software.common.util.RegexPatternUtils;
 @Slf4j
 public final class SecretMasker {
 
+    /** The placeholder masked values are replaced with; reads as "a secret is set". */
+    public static final String REDACTED = "********";
+
     private static final Pattern SENSITIVE =
             RegexPatternUtils.getInstance()
                     .getPattern(
-                            "(?i)\\b(password|token|secret|api[_-]?key|authorization|auth|jwt|cred|cert)\\b");
+                            // secret[_-]?access[_-]?key precedes plain secret so camelCase keys
+                            // like secretAccessKey (no word boundary after "secret") still match.
+                            "(?i)\\b(password|token|secret[_-]?access[_-]?key|secret|api[_-]?key|authorization|auth|jwt|cred|cert)\\b");
 
     private SecretMasker() {}
 
@@ -47,8 +53,28 @@ public final class SecretMasker {
 
     private static Object deepMaskValue(String key, Object value) {
         if (key != null && SENSITIVE.matcher(key).find()) {
-            return "***REDACTED***";
+            return REDACTED;
         }
         return deepMask(value);
     }
+
+    /**
+     * Restore top-level values the caller sent back as the {@link #REDACTED} sentinel from the
+     * stored map, so a masked read can round-trip through an edit without re-typing secrets. A
+     * sentinel with no stored counterpart is left as-is (it fails whatever validates it, rather
+     * than silently passing an unset secret).
+     */
+    public static Map<String, Object> restoreRedacted(
+            Map<String, Object> incoming, Map<String, Object> stored) {
+        if (incoming == null || stored == null) {
+            return incoming;
+        }
+        Map<String, Object> merged = new LinkedHashMap<>(incoming);
+        merged.replaceAll(
+                (key, value) ->
+                        REDACTED.equals(value) && stored.containsKey(key)
+                                ? stored.get(key)
+                                : value);
+        return merged;
+    }
 }

app/proprietary/src/main/resources/templates/ai/create/document.html.jinja2

@@ -1,3 +1,4 @@
+{%- autoescape true -%}
 <!DOCTYPE html>
 <html lang="en">
 <head>
@@ -175,18 +176,18 @@
     color: var(--color-label);
   }
 </style>
-{%- if doc.style %}
+{%- if style_primary or style_background or style_body %}
 <style>
   :root {
-    {%- if doc.style.primary_color %}
-    --color-primary: {{ doc.style.primary_color }};
+    {%- if style_primary %}
+    --color-primary: {{ style_primary }};
     {%- endif %}
-    {%- if doc.style.background_color %}
-    --color-bg: {{ doc.style.background_color }};
+    {%- if style_background %}
+    --color-bg: {{ style_background }};
     {%- endif %}
-    {%- if doc.style.body_text_color %}
-    --color-body: {{ doc.style.body_text_color }};
-    --color-label: {{ doc.style.body_text_color }};
+    {%- if style_body %}
+    --color-body: {{ style_body }};
+    --color-label: {{ style_body }};
     {%- endif %}
   }
 </style>
@@ -195,16 +196,16 @@
 <body>
 
 <div class="doc-header">
-  <div class="doc-title">{{ doc.title }}</div>
-  {%- if doc.subtitle %}
-  <div class="doc-subtitle">{{ doc.subtitle }}</div>
+  <div class="doc-title">{{ title }}</div>
+  {%- if subtitle %}
+  <div class="doc-subtitle">{{ subtitle }}</div>
   {%- endif %}
-  {%- if doc.reference_number %}
-  <div class="doc-reference">{{ doc.reference_number }}</div>
+  {%- if reference_number %}
+  <div class="doc-reference">{{ reference_number }}</div>
   {%- endif %}
 </div>
 
-{%- for section in doc.sections %}
+{%- for section in sections %}
 
 {%- if section.type == "text" %}
 <section>
@@ -212,8 +213,8 @@
   <h2>{{ section.heading }}</h2>
   {%- endif %}
   <div class="text-body">
-    {%- for para in section.body.split('\n\n') %}
-    <p>{{ para | replace('\n', ' ') }}</p>
+    {%- for para in section.paragraphs %}
+    <p>{{ para }}</p>
     {%- endfor %}
   </div>
 </section>
@@ -225,10 +226,10 @@
   {%- endif %}
   <table class="kv-table">
     <tbody>
-      {%- for label, value in section.pairs %}
+      {%- for pair in section.pairs %}
       <tr>
-        <td class="kv-label">{{ label }}</td>
-        <td class="kv-value">{{ value }}</td>
+        <td class="kv-label">{{ pair.label }}</td>
+        <td class="kv-value">{{ pair.value }}</td>
       </tr>
       {%- endfor %}
     </tbody>
@@ -299,3 +300,4 @@
 
 </body>
 </html>
+{%- endautoescape %}

app/proprietary/src/test/java/stirling/software/proprietary/access/service/ResourceAccessServiceTest.java

@@ -2,6 +2,7 @@ package stirling.software.proprietary.access.service;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.when;
 
 import java.lang.reflect.Field;
@@ -258,6 +259,24 @@ class ResourceAccessServiceTest {
                 .isFalse();
     }
 
+    @Test
+    void teamLeadDefaultOnTeamResourceDeniesForeignTeamsLead() {
+        // Team-owned resource: the default admits only the owning team's leads. A lead of
+        // some other team must be denied even though an unscoped "is any team leader"
+        // check would admit them (lenient stub: the scoped path must never consult it).
+        stubGrants();
+        User foreignLead = user(6);
+        lenient().when(teamLeadLookup.isAnyTeamLeader(foreignLead)).thenReturn(true);
+        assertThat(
+                        service.canUseResource(
+                                TYPE,
+                                RID,
+                                PrincipalRef.team(7L),
+                                DefaultAccessPolicy.ADMINS_AND_TEAM_LEADS,
+                                foreignLead))
+                .isFalse();
+    }
+
     // ---- portal convenience (default policy ADMINS_AND_TEAM_LEADS) ----
 
     @Test

app/proprietary/src/test/java/stirling/software/proprietary/integration/crypto/LenientEncryptedStringConverterTest.java

@@ -0,0 +1,45 @@
+package stirling.software.proprietary.integration.crypto;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import javax.crypto.KeyGenerator;
+import javax.crypto.SecretKey;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class LenientEncryptedStringConverterTest {
+
+    private final LenientEncryptedStringConverter converter = new LenientEncryptedStringConverter();
+
+    @BeforeAll
+    static void initKey() throws Exception {
+        KeyGenerator generator = KeyGenerator.getInstance("AES");
+        generator.init(256);
+        SecretKey key = generator.generateKey();
+        CredentialEncryption.initialiseForTesting(key);
+    }
+
+    @Test
+    void roundTripsThroughCiphertext() {
+        String json = "{\"bucket\":\"inbox\",\"secretAccessKey\":\"shh\"}";
+
+        String stored = converter.convertToDatabaseColumn(json);
+
+        assertThat(stored).isNotEqualTo(json).doesNotContain("shh");
+        assertThat(converter.convertToEntityAttribute(stored)).isEqualTo(json);
+    }
+
+    @Test
+    void legacyPlaintextRowsPassThroughOnRead() {
+        String legacy = "{\"bucket\":\"inbox\",\"mode\":\"consume\"}";
+
+        assertThat(converter.convertToEntityAttribute(legacy)).isEqualTo(legacy);
+    }
+
+    @Test
+    void nullsPassThrough() {
+        assertThat(converter.convertToDatabaseColumn(null)).isNull();
+        assertThat(converter.convertToEntityAttribute(null)).isNull();
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/integration/service/IntegrationConfigServiceTest.java

@@ -13,9 +13,9 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.springframework.http.HttpStatus;
@@ -52,7 +52,58 @@ class IntegrationConfigServiceTest {
     @Mock
     private stirling.software.proprietary.access.repository.ResourceGrantRepository grantRepository;
 
-    @InjectMocks private IntegrationConfigService service;
+    @Mock private IntegrationConfigValidator validator;
+    @Mock private IntegrationConfigUsageCheck usageCheck;
+
+    private IntegrationConfigService service;
+
+    @BeforeEach
+    void setUp() {
+        service =
+                new IntegrationConfigService(
+                        repository,
+                        ownership,
+                        secretMasker,
+                        grantRepository,
+                        List.of(validator),
+                        List.of(usageCheck));
+    }
+
+    @Test
+    void createRejectsAConfigItsTypeValidatorRefuses() {
+        when(secretMasker.sanitize(any())).thenReturn(Map.of());
+        when(validator.type()).thenReturn(IntegrationType.API);
+        org.mockito.Mockito.doThrow(new IllegalArgumentException("api config needs a 'url'"))
+                .when(validator)
+                .validate(any());
+
+        assertThatThrownBy(
+                        () ->
+                                service.create(
+                                        request(IntegrationType.API, OwnerScope.USER, null),
+                                        user(7)))
+                .isInstanceOf(ResponseStatusException.class)
+                .satisfies(
+                        e ->
+                                assertThat(((ResponseStatusException) e).getStatusCode())
+                                        .isEqualTo(HttpStatus.BAD_REQUEST));
+    }
+
+    @Test
+    void deleteRefusedWhileAnythingStillReferencesTheConfig() {
+        IntegrationConfig cfg = config(9L);
+        when(repository.findById(9L)).thenReturn(Optional.of(cfg));
+        when(ownership.canManage(any(), eq(cfg), any())).thenReturn(true);
+        when(usageCheck.usagesOf(9L)).thenReturn(List.of("source 'Claims intake'"));
+
+        assertThatThrownBy(() -> service.delete(9L, user(7)))
+                .isInstanceOf(ResponseStatusException.class)
+                .satisfies(
+                        e ->
+                                assertThat(((ResponseStatusException) e).getStatusCode())
+                                        .isEqualTo(HttpStatus.CONFLICT));
+        verify(repository, org.mockito.Mockito.never()).delete(any(IntegrationConfig.class));
+    }
 
     @Test
     void createDelegatesOwnershipAndSanitizesConfig() {

app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java

@@ -4,11 +4,13 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
@@ -17,6 +19,7 @@ import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.springframework.http.HttpStatus;
@@ -34,16 +37,21 @@ import stirling.software.proprietary.policy.engine.PolicyRunHandle;
 import stirling.software.proprietary.policy.engine.PolicyRunRegistry;
 import stirling.software.proprietary.policy.engine.PolicyRunner;
 import stirling.software.proprietary.policy.engine.PolicyValidator;
+import stirling.software.proprietary.policy.engine.SweepOutcome;
 import stirling.software.proprietary.policy.ledger.ProcessedLedger;
+import stirling.software.proprietary.policy.model.OutputSpec;
 import stirling.software.proprietary.policy.model.PipelineDefinition;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.model.PolicyRun;
 import stirling.software.proprietary.policy.model.PolicyRunView;
 import stirling.software.proprietary.policy.progress.PolicyProgressListener;
+import stirling.software.proprietary.policy.source.EditorSource;
 import stirling.software.proprietary.policy.source.SourceAccessGuard;
+import stirling.software.proprietary.policy.source.SourceDocCounter;
 import stirling.software.proprietary.policy.source.SourceStore;
 import stirling.software.proprietary.policy.trigger.PolicyTriggerManager;
+import stirling.software.proprietary.util.SecretMasker;
 
 @ExtendWith(MockitoExtension.class)
 @DisplayName("PolicyController")
@@ -54,6 +62,7 @@ class PolicyControllerTest {
     @Mock private stirling.software.proprietary.policy.store.PolicyStore policyStore;
     @Mock private SourceStore sourceStore;
     @Mock private SourceAccessGuard sourceAccessGuard;
+    @Mock private SourceDocCounter docCounter;
     @Mock private PolicyValidator policyValidator;
     @Mock private PolicyAccessGuard policyAccessGuard;
     @Mock private PolicyManagementAuthority policyManagementAuthority;
@@ -87,6 +96,7 @@ class PolicyControllerTest {
                         policyStore,
                         sourceStore,
                         sourceAccessGuard,
+                        docCounter,
                         policyValidator,
                         policyAccessGuard,
                         policyManagementAuthority,
@@ -128,6 +138,17 @@ class PolicyControllerTest {
         return new Policy(id, "name", "owner", true, null, List.of(), List.of(), null, teamId);
     }
 
+    private static Policy s3OutputPolicy(String id, String secret) {
+        OutputSpec output =
+                new OutputSpec(
+                        "s3",
+                        Map.of(
+                                "bucket", "outbox",
+                                "accessKeyId", "AKIAEXAMPLE",
+                                "secretAccessKey", secret));
+        return new Policy(id, "name", "owner", true, null, List.of(), List.of(), output, 1L);
+    }
+
     private static PolicyRunHandle handle(String runId) {
         PolicyRun run = new PolicyRun(runId, null, definitionWithStep());
         return new PolicyRunHandle(runId, CompletableFuture.completedFuture(run));
@@ -150,6 +171,18 @@ class PolicyControllerTest {
             assertThat(response.getBody().getJobId()).isEqualTo("run-1");
         }
 
+        @Test
+        @DisplayName("feeds the editor source, scoped to the caller's team")
+        void adHocRunFeedsTheEditorSource() throws Exception {
+            when(policyRunner.runAdHoc(any(), any(), eq(PolicyProgressListener.NOOP)))
+                    .thenReturn(handle("run-1"));
+            when(sourceAccessGuard.currentTeamId()).thenReturn(3L);
+
+            controller.run(definitionWithStep(), new PolicyRunFiles());
+
+            verify(docCounter).record(EditorSource.counterKey(3L), 0L);
+        }
+
         @Test
         @DisplayName("rejects a pipeline with no steps")
         void rejectsEmptyPipeline() {
@@ -162,6 +195,29 @@ class PolicyControllerTest {
                                     assertThat(((ResponseStatusException) e).getStatusCode())
                                             .isEqualTo(HttpStatus.BAD_REQUEST));
         }
+
+        @Test
+        @DisplayName("rejects an ad-hoc output the caller cannot use, on the request thread")
+        void rejectsUnauthorizedAdHocOutput() {
+            // The confused-deputy guard: an S3 output referencing a connection the caller may not
+            // use is validated here (principal present) and refused before any worker dispatch.
+            PipelineDefinition definition =
+                    new PipelineDefinition(
+                            "pipe",
+                            List.of(new PipelineStep("/api/v1/misc/compress-pdf", null)),
+                            new OutputSpec("s3", Map.of("connectionId", 999)));
+            doThrow(new IllegalArgumentException("unknown or inaccessible s3 connection"))
+                    .when(policyValidator)
+                    .validateOutput(any());
+
+            assertThatThrownBy(() -> controller.run(definition, new PolicyRunFiles()))
+                    .isInstanceOf(ResponseStatusException.class)
+                    .satisfies(
+                            e ->
+                                    assertThat(((ResponseStatusException) e).getStatusCode())
+                                            .isEqualTo(HttpStatus.BAD_REQUEST));
+            verify(policyRunner, never()).runAdHoc(any(), any(), any());
+        }
     }
 
     @Nested
@@ -263,6 +319,27 @@ class PolicyControllerTest {
             verify(policyTriggerManager).notifyPoliciesChanged();
         }
 
+        @Test
+        @DisplayName("saving the sentinel back keeps the stored output secret")
+        void saveRestoresOutputSecrets() {
+            applicationProperties.getSecurity().setEnableLogin(false);
+            Policy existing = s3OutputPolicy("p1", "shh");
+            when(policyStore.get("p1")).thenReturn(Optional.of(existing));
+            when(policyAccessGuard.canAccess(existing)).thenReturn(true);
+            when(policyStore.save(any())).thenAnswer(i -> i.getArgument(0));
+
+            ResponseEntity<Policy> response =
+                    controller.savePolicy(s3OutputPolicy("p1", SecretMasker.REDACTED));
+
+            ArgumentCaptor<Policy> stored = ArgumentCaptor.forClass(Policy.class);
+            verify(policyStore).save(stored.capture());
+            assertThat(stored.getValue().output().options().get("secretAccessKey"))
+                    .isEqualTo("shh");
+            // The save response is masked again; only the store sees the real value.
+            assertThat(response.getBody().output().options().get("secretAccessKey"))
+                    .isEqualTo(SecretMasker.REDACTED);
+        }
+
         @Test
         @DisplayName("forbidden when login enabled and caller cannot edit")
         void forbidden() {
@@ -363,6 +440,20 @@ class PolicyControllerTest {
             assertThat(response.getBody().id()).isEqualTo("a");
         }
 
+        @Test
+        @DisplayName("getPolicy returns output secrets as the redaction sentinel")
+        void getMasksOutputSecrets() {
+            Policy p = s3OutputPolicy("a", "shh");
+            when(policyStore.get("a")).thenReturn(Optional.of(p));
+            when(policyAccessGuard.canAccess(p)).thenReturn(true);
+
+            Policy read = controller.getPolicy("a").getBody();
+
+            assertThat(read.output().options().get("secretAccessKey"))
+                    .isEqualTo(SecretMasker.REDACTED);
+            assertThat(read.output().options().get("bucket")).isEqualTo("outbox");
+        }
+
         @Test
         @DisplayName("getPolicy returns 404 when not accessible")
         void getNotAccessible() {
@@ -536,17 +627,18 @@ class PolicyControllerTest {
         }
 
         @Test
-        @DisplayName("trigger runs an accessible policy against its sources and returns run ids")
+        @DisplayName("trigger runs an accessible policy against its sources and returns the sweep")
         void triggersRun() {
             Policy p = policy("a", 1L);
             when(policyStore.get("a")).thenReturn(Optional.of(p));
             when(policyAccessGuard.canAccess(p)).thenReturn(true);
-            when(policyRunner.run(p)).thenReturn(List.of("run-a", "run-b"));
+            SweepOutcome outcome = new SweepOutcome(List.of("run-a", "run-b"), 3, 1, 0, 0);
+            when(policyRunner.run(p)).thenReturn(outcome);
 
-            ResponseEntity<List<String>> response = controller.trigger("a");
+            ResponseEntity<SweepOutcome> response = controller.trigger("a");
 
             assertThat(response.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED);
-            assertThat(response.getBody()).containsExactly("run-a", "run-b");
+            assertThat(response.getBody()).isEqualTo(outcome);
         }
 
         @Test

app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyExecutorTest.java

@@ -166,8 +166,8 @@ class PolicyExecutorTest {
                                 new PipelineStep(
                                         createPdf,
                                         Map.of(
-                                                "htmlContent",
-                                                "<p>hi</p>",
+                                                "document",
+                                                "{\"title\":\"PO\",\"sections\":[]}",
                                                 "filename",
                                                 "purchase-order.pdf"))),
                         PolicyInputs.of(List.of()),

app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java

@@ -1,5 +1,6 @@
 package stirling.software.proprietary.policy.engine;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -26,10 +27,12 @@ import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.core.io.ByteArrayResource;
 
 import stirling.software.proprietary.policy.input.InputSource;
 import stirling.software.proprietary.policy.input.ResolveContext;
 import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
 import stirling.software.proprietary.policy.ledger.ProcessedLedger;
 import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.OutputSpec;
@@ -39,6 +42,7 @@ import stirling.software.proprietary.policy.model.PolicyInputs;
 import stirling.software.proprietary.policy.model.PolicyRun;
 import stirling.software.proprietary.policy.model.PolicyRunStatus;
 import stirling.software.proprietary.policy.progress.PolicyProgressListener;
+import stirling.software.proprietary.policy.source.EditorSource;
 import stirling.software.proprietary.policy.source.InProcessSourceDocCounter;
 import stirling.software.proprietary.policy.source.InProcessSourceStore;
 import stirling.software.proprietary.policy.source.Source;
@@ -56,6 +60,7 @@ class PolicyRunnerTest {
     @Mock private ProcessedLedger processedLedger;
 
     private final SourceStore sourceStore = new InProcessSourceStore();
+    private final InProcessSourceDocCounter docCounter = new InProcessSourceDocCounter();
     private PolicyRunner runner;
 
     @BeforeEach
@@ -65,7 +70,7 @@ class PolicyRunnerTest {
                         policyEngine,
                         List.of(folderSource),
                         sourceStore,
-                        new InProcessSourceDocCounter(),
+                        docCounter,
                         processedLedger);
     }
 
@@ -85,6 +90,42 @@ class PolicyRunnerTest {
         verify(processedLedger).deleteUnseen(eq("p1"), anyLong());
     }
 
+    @Test
+    void reportsWhatTheSweepSkippedSoAnEmptyTriggerExplainsItself() throws Exception {
+        InProcessProcessedLedger ledger = new InProcessProcessedLedger();
+        PolicyRunner reporting =
+                new PolicyRunner(
+                        policyEngine,
+                        List.of(folderSource),
+                        sourceStore,
+                        new InProcessSourceDocCounter(),
+                        ledger);
+        InputSpec spec = InputSpec.folder("/in");
+        Policy policy = policy(List.of(spec));
+        // One file already processed at its current version, one parked by a failed run.
+        ledger.claim("p1", "/in/done.pdf", "g1", null);
+        ledger.settle("p1", "/in/done.pdf", "g1", null, true);
+        ledger.claim("p1", "/in/failed.pdf", "g2", null);
+        ledger.settle("p1", "/in/failed.pdf", "g2", null, false);
+        when(folderSource.supports(spec)).thenReturn(true);
+        when(folderSource.resolve(eq(spec), any()))
+                .thenAnswer(
+                        invocation -> {
+                            ResolveContext ctx = invocation.getArgument(1);
+                            ctx.reportPresent(List.of("/in/done.pdf", "/in/failed.pdf"));
+                            // Both are at their settled versions, so neither claims.
+                            return List.of();
+                        });
+
+        SweepOutcome outcome = reporting.run(policy);
+
+        assertTrue(outcome.runIds().isEmpty());
+        assertEquals(2, outcome.filesListed());
+        assertEquals(1, outcome.alreadyProcessed());
+        assertEquals(1, outcome.parked());
+        assertEquals(0, outcome.inFlight());
+    }
+
     @Test
     void pullsEverySourceAndRunsOnePerUnitOfWork() throws Exception {
         InputSpec spec = InputSpec.folder("/in");
@@ -257,6 +298,33 @@ class PolicyRunnerTest {
         verifyNoInteractions(folderSource);
     }
 
+    @Test
+    void runWithRecordsSuppliedDocsAgainstTheEditorSourceForThePolicyTeam() {
+        Policy policy =
+                new Policy(
+                        "p1",
+                        "p",
+                        "owner",
+                        true,
+                        null,
+                        List.of(),
+                        List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
+                        OutputSpec.inline(),
+                        7L);
+        PolicyInputs inputs =
+                PolicyInputs.of(
+                        List.of(
+                                new ByteArrayResource("a".getBytes()),
+                                new ByteArrayResource("b".getBytes())));
+        when(policyEngine.runPolicy(policy, inputs, PolicyProgressListener.NOOP))
+                .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>()));
+
+        runner.runWith(policy, inputs, PolicyProgressListener.NOOP);
+
+        String key = EditorSource.counterKey(7L);
+        assertEquals(2, docCounter.statsFor(List.of(key)).get(key).total());
+    }
+
     /** Persists each spec as a source and returns a policy referencing them by id. */
     private Policy policy(List<InputSpec> sources) {
         List<String> sourceIds =

app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyValidatorTest.java

@@ -82,6 +82,28 @@ class PolicyValidatorTest {
         assertTrue(ex.getMessage().contains("schedule"));
     }
 
+    @Test
+    void validateOutputDelegatesToTheSink() {
+        when(outputSink.supports(any())).thenReturn(true);
+        OutputSpec output = new OutputSpec("s3", Map.of("connectionId", 1));
+
+        validator.validateOutput(output);
+
+        verify(outputSink).validate(output);
+    }
+
+    @Test
+    void validateOutputSurfacesAnInaccessibleConnection() {
+        when(outputSink.supports(any())).thenReturn(true);
+        doThrow(new IllegalArgumentException("unknown or inaccessible s3 connection"))
+                .when(outputSink)
+                .validate(any());
+
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> validator.validateOutput(new OutputSpec("s3", Map.of("connectionId", 1))));
+    }
+
     @Test
     void rejectsAnUnknownTriggerType() {
         when(trigger.type()).thenReturn("schedule");

app/proprietary/src/test/java/stirling/software/proprietary/policy/input/S3InputSourceMinioTest.java

@@ -0,0 +1,236 @@
+package stirling.software.proprietary.policy.input;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.MinIOContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.s3.S3ConnectionPool;
+import stirling.software.proprietary.policy.s3.S3TestConnections;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+
+/**
+ * End-to-end {@link S3InputSource} test against a real S3 API (MinIO), through the production
+ * client factory: listing, claiming, streaming, consensus delete, and save-time validation.
+ */
+@Testcontainers(disabledWithoutDocker = true)
+class S3InputSourceMinioTest {
+
+    private static final String POLICY = "p1";
+    private static final String ACCESS_KEY = "minioadmin";
+    private static final String SECRET_KEY = "minioadmin";
+
+    @Container
+    static MinIOContainer minio =
+            new MinIOContainer("minio/minio:latest")
+                    .withUserName(ACCESS_KEY)
+                    .withPassword(SECRET_KEY);
+
+    private static S3Client adminClient;
+    private static int bucketCounter;
+
+    private String bucket;
+    private S3InputSource source;
+    private InProcessProcessedLedger ledger;
+    private RecordingContext ctx;
+
+    @BeforeEach
+    void setUp() {
+        if (adminClient == null) {
+            adminClient =
+                    S3Client.builder()
+                            .endpointOverride(java.net.URI.create(minio.getS3URL()))
+                            .httpClient(UrlConnectionHttpClient.create())
+                            .region(Region.US_EAST_1)
+                            .credentialsProvider(
+                                    StaticCredentialsProvider.create(
+                                            AwsBasicCredentials.create(ACCESS_KEY, SECRET_KEY)))
+                            .serviceConfiguration(
+                                    S3Configuration.builder().pathStyleAccessEnabled(true).build())
+                            .build();
+        }
+        bucket = "policy-inbox-" + ++bucketCounter;
+        adminClient.createBucket(CreateBucketRequest.builder().bucket(bucket).build());
+
+        // The MinIO endpoint resolves to loopback, so the operator opt-in must be on.
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowPrivateS3Endpoints(true);
+        source =
+                new S3InputSource(
+                        new S3ConnectionPool(properties), S3TestConnections.legacyResolver());
+        ledger = new InProcessProcessedLedger();
+        ctx = new RecordingContext();
+    }
+
+    @Test
+    void consumeListsStreamsAndDeletesByConsensus() throws IOException {
+        put("incoming/doc.pdf", "pdf bytes");
+        put("incoming/other.txt", "text");
+
+        List<ResolvedInput> work = source.resolve(spec(Map.of("prefix", "incoming/")), ctx);
+
+        assertThat(work).hasSize(2);
+        assertThat(ctx.present)
+                .containsExactlyInAnyOrder(
+                        "s3://" + bucket + "/incoming/doc.pdf",
+                        "s3://" + bucket + "/incoming/other.txt");
+        assertThat(read(work.get(0))).isIn("pdf bytes", "text");
+        // In flight: nothing to claim on a second sweep.
+        assertThat(source.resolve(spec(Map.of("prefix", "incoming/")), ctx)).isEmpty();
+
+        work.forEach(unit -> unit.onComplete().accept(true));
+        assertThat(exists("incoming/doc.pdf")).isFalse();
+        assertThat(exists("incoming/other.txt")).isFalse();
+    }
+
+    @Test
+    void aFailedObjectStaysInTheBucket() throws IOException {
+        put("doc.pdf", "data");
+
+        source.resolve(spec(Map.of()), ctx).get(0).onComplete().accept(false);
+
+        assertThat(exists("doc.pdf")).isTrue();
+        assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty();
+    }
+
+    @Test
+    void anObjectOverwrittenMidRunSurvivesTheDeleteAndRunsAgain() throws IOException {
+        put("doc.pdf", "v1");
+
+        List<ResolvedInput> work = source.resolve(spec(Map.of()), ctx);
+        put("doc.pdf", "v2 with a different etag");
+        work.get(0).onComplete().accept(true);
+
+        assertThat(exists("doc.pdf")).isTrue();
+        assertThat(source.resolve(spec(Map.of()), ctx)).hasSize(1);
+    }
+
+    @Test
+    void prefixLimitsWhatIsRead() throws IOException {
+        put("incoming/doc.pdf", "data");
+        put("archive/old.pdf", "data");
+
+        List<ResolvedInput> work = source.resolve(spec(Map.of("prefix", "incoming/")), ctx);
+
+        assertThat(work).hasSize(1);
+        assertThat(ctx.present).containsExactly("s3://" + bucket + "/incoming/doc.pdf");
+    }
+
+    @Test
+    void validateAcceptsAReachableBucketAndRejectsBadCredentials() {
+        source.validate(spec(Map.of()));
+
+        Map<String, Object> wrongSecret = new HashMap<>(baseOptions());
+        wrongSecret.put("secretAccessKey", "not-the-secret");
+        assertThatThrownBy(() -> source.validate(new InputSpec("s3", wrongSecret)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("cannot access");
+
+        Map<String, Object> missingBucket = new HashMap<>(baseOptions());
+        missingBucket.put("bucket", "no-such-bucket-here");
+        assertThatThrownBy(() -> source.validate(new InputSpec("s3", missingBucket)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("cannot access");
+    }
+
+    @Test
+    void aPrivateEndpointIsRejectedWithoutTheOperatorOptIn() {
+        S3InputSource guarded =
+                new S3InputSource(
+                        new S3ConnectionPool(new ApplicationProperties()),
+                        S3TestConnections.legacyResolver());
+
+        assertThatThrownBy(() -> guarded.validate(spec(Map.of())))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("policies.allowPrivateS3Endpoints");
+    }
+
+    private Map<String, Object> baseOptions() {
+        return Map.of(
+                "bucket", bucket,
+                "endpoint", minio.getS3URL(),
+                "accessKeyId", ACCESS_KEY,
+                "secretAccessKey", SECRET_KEY);
+    }
+
+    private InputSpec spec(Map<String, Object> extra) {
+        Map<String, Object> options = new HashMap<>(baseOptions());
+        options.putAll(extra);
+        return new InputSpec("s3", options);
+    }
+
+    private void put(String key, String content) {
+        adminClient.putObject(
+                PutObjectRequest.builder().bucket(bucket).key(key).build(),
+                RequestBody.fromString(content, StandardCharsets.UTF_8));
+    }
+
+    private boolean exists(String key) {
+        try {
+            adminClient.headObject(HeadObjectRequest.builder().bucket(bucket).key(key).build());
+            return true;
+        } catch (NoSuchKeyException e) {
+            return false;
+        }
+    }
+
+    private static String read(ResolvedInput unit) throws IOException {
+        try (InputStream stream = unit.inputs().primary().get(0).getInputStream()) {
+            return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
+        }
+    }
+
+    private class RecordingContext implements ResolveContext {
+
+        private final List<String> present = new ArrayList<>();
+
+        @Override
+        public boolean claim(String identity, String gate, Supplier<String> contentHash) {
+            return ledger.claim(POLICY, identity, gate, contentHash);
+        }
+
+        @Override
+        public void settle(
+                String identity, String finalGate, String finalContentHash, boolean success) {
+            ledger.settle(POLICY, identity, finalGate, finalContentHash, success);
+        }
+
+        @Override
+        public boolean allSettledDone(String identity) {
+            return ledger.allSettledDone(identity);
+        }
+
+        @Override
+        public void reportPresent(Collection<String> identities) {
+            present.addAll(identities);
+        }
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/input/S3InputSourceTest.java

@@ -0,0 +1,329 @@
+package stirling.software.proprietary.policy.input;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.s3.S3ConnectionPool;
+import stirling.software.proprietary.policy.s3.S3TestConnections;
+
+import software.amazon.awssdk.core.ResponseInputStream;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.http.AbortableInputStream;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+/**
+ * Tests for {@link S3InputSource}: consume mode tracks objects in place through the ledger and
+ * removes them by consensus, snapshot stays stateless, and discovery skips folder placeholders and
+ * dot-prefixed keys.
+ */
+@ExtendWith(MockitoExtension.class)
+class S3InputSourceTest {
+
+    private static final String POLICY = "p1";
+    private static final String BUCKET = "inbox-bucket";
+
+    @Mock private S3Client s3Client;
+
+    private S3InputSource source;
+    private InProcessProcessedLedger ledger;
+    private RecordingContext ctx;
+
+    @BeforeEach
+    void setUp() {
+        source =
+                new S3InputSource(
+                        new S3ConnectionPool(new ApplicationProperties(), config -> s3Client),
+                        S3TestConnections.legacyResolver());
+        ledger = new InProcessProcessedLedger();
+        ctx = new RecordingContext();
+    }
+
+    @Test
+    void consumeRemovesTheObjectOnceProcessed() throws IOException {
+        listingReturns(object("doc.pdf", "\"etag-1\""));
+        headReturns("doc.pdf", "\"etag-1\"");
+
+        List<ResolvedInput> work = source.resolve(spec(), ctx);
+
+        assertEquals(1, work.size());
+        assertEquals(1, work.get(0).inputs().primary().size());
+        // In flight: a second sweep does not pick it up again.
+        assertTrue(source.resolve(spec(), ctx).isEmpty());
+
+        work.get(0).onComplete().accept(true);
+        verify(s3Client).deleteObject(any(DeleteObjectRequest.class));
+        assertTrue(source.resolve(spec(), ctx).isEmpty());
+    }
+
+    @Test
+    void anObjectReplacedMidRunSurvivesTheDelete() throws IOException {
+        listingReturns(object("doc.pdf", "\"etag-1\""));
+        // The object is overwritten while the run is executing.
+        headReturns("doc.pdf", "\"etag-2\"");
+
+        List<ResolvedInput> work = source.resolve(spec(), ctx);
+        work.get(0).onComplete().accept(true);
+
+        // The delete is version-guarded: the replacement is not the object that ran, so it stays
+        // and is claimed as fresh work instead of being marked processed.
+        verify(s3Client, never()).deleteObject(any(DeleteObjectRequest.class));
+        listingReturns(object("doc.pdf", "\"etag-2\""));
+        assertEquals(1, source.resolve(spec(), ctx).size());
+    }
+
+    @Test
+    void aSharedObjectIsRemovedOnlyOnceEveryPolicyHasProcessedIt() throws IOException {
+        listingReturns(object("doc.pdf", "\"etag-1\""));
+        headReturns("doc.pdf", "\"etag-1\"");
+        RecordingContext other = new RecordingContext("p2");
+
+        List<ResolvedInput> mine = source.resolve(spec(), ctx);
+        List<ResolvedInput> theirs = source.resolve(spec(), other);
+        assertEquals(1, mine.size());
+        assertEquals(1, theirs.size());
+
+        mine.get(0).onComplete().accept(true);
+        // The other policy's claim is still in flight, so the first finisher must not delete.
+        verify(s3Client, never()).deleteObject(any(DeleteObjectRequest.class));
+
+        theirs.get(0).onComplete().accept(true);
+        verify(s3Client).deleteObject(any(DeleteObjectRequest.class));
+    }
+
+    @Test
+    void aFailedObjectStaysAndIsNotRetriedUntilItChanges() throws IOException {
+        listingReturns(object("doc.pdf", "\"etag-1\""));
+
+        source.resolve(spec(), ctx).get(0).onComplete().accept(false);
+
+        verify(s3Client, never()).deleteObject(any(DeleteObjectRequest.class));
+        assertTrue(source.resolve(spec(), ctx).isEmpty());
+
+        // A new upload carries a new ETag, which reads as a new version and retries.
+        listingReturns(object("doc.pdf", "\"etag-2\""));
+        assertEquals(1, source.resolve(spec(), ctx).size());
+    }
+
+    @Test
+    void snapshotReadsStatelesslyEverySweep() throws IOException {
+        listingReturns(object("doc.pdf", "\"etag-1\""));
+        InputSpec spec = new InputSpec("s3", options(Map.of("mode", "snapshot")));
+
+        List<ResolvedInput> first = source.resolve(spec, ctx);
+        first.get(0).onComplete().accept(true);
+        List<ResolvedInput> second = source.resolve(spec, ctx);
+
+        assertEquals(1, first.size());
+        assertEquals(1, second.size());
+        verify(s3Client, never()).deleteObject(any(DeleteObjectRequest.class));
+        assertTrue(ctx.present.isEmpty());
+    }
+
+    @Test
+    void folderPlaceholdersAndDotPrefixedKeysAreSkipped() throws IOException {
+        listingReturns(
+                object("doc.pdf", "\"etag-1\""),
+                object("incoming/", "\"etag-2\""),
+                object(".stirling/tmp/staged.pdf", "\"etag-3\""),
+                object("incoming/.hidden.pdf", "\"etag-4\""));
+
+        List<ResolvedInput> work = source.resolve(spec(), ctx);
+
+        assertEquals(1, work.size());
+        assertEquals(List.of("s3://" + BUCKET + "/doc.pdf"), ctx.present);
+    }
+
+    @Test
+    void listingPagesAreAllRead() throws IOException {
+        ListObjectsV2Response firstPage =
+                ListObjectsV2Response.builder()
+                        .contents(object("a.pdf", "\"etag-a\""))
+                        .nextContinuationToken("next")
+                        .build();
+        ListObjectsV2Response secondPage =
+                ListObjectsV2Response.builder().contents(object("b.pdf", "\"etag-b\"")).build();
+        when(s3Client.listObjectsV2(any(ListObjectsV2Request.class)))
+                .thenReturn(firstPage, secondPage);
+
+        assertEquals(2, source.resolve(spec(), ctx).size());
+    }
+
+    @Test
+    void aListingFailurePropagatesSoTheSweepVetoesCleanup() {
+        when(s3Client.listObjectsV2(any(ListObjectsV2Request.class)))
+                .thenThrow(SdkClientException.create("connection refused"));
+
+        assertThrows(SdkClientException.class, () -> source.resolve(spec(), ctx));
+    }
+
+    @Test
+    void resourceStreamsTheObjectAndNamesItByKeyBasename() throws IOException {
+        listingReturns(object("incoming/doc.pdf", "\"etag-1\""));
+        byte[] payload = "data".getBytes(StandardCharsets.UTF_8);
+        when(s3Client.getObject(any(GetObjectRequest.class)))
+                .thenReturn(
+                        new ResponseInputStream<>(
+                                GetObjectResponse.builder().build(),
+                                AbortableInputStream.create(new ByteArrayInputStream(payload))));
+
+        var resource = source.resolve(spec(), ctx).get(0).inputs().primary().get(0);
+
+        assertEquals("doc.pdf", resource.getFilename());
+        // Content length comes from the listing, not a download.
+        assertEquals(4, resource.contentLength());
+        try (var stream = resource.getInputStream()) {
+            assertEquals("data", new String(stream.readAllBytes(), StandardCharsets.UTF_8));
+        }
+    }
+
+    @Test
+    void aMissingETagFallsBackToSizeAndLastModified() throws IOException {
+        Instant modified = Instant.parse("2026-01-01T00:00:00Z");
+        listingReturns(S3Object.builder().key("doc.pdf").size(4L).lastModified(modified).build());
+
+        assertEquals(1, source.resolve(spec(), ctx).size());
+        // The same gate on the next sweep reads as already claimed.
+        listingReturns(S3Object.builder().key("doc.pdf").size(4L).lastModified(modified).build());
+        assertTrue(source.resolve(spec(), ctx).isEmpty());
+    }
+
+    @Test
+    void validateRejectsBadConfig() {
+        // No bucket.
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> source.validate(new InputSpec("s3", Map.of())));
+        // Credentials are required, never the server's own identity - together and individually.
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> source.validate(new InputSpec("s3", Map.of("bucket", BUCKET))));
+        assertThrows(
+                IllegalArgumentException.class,
+                () ->
+                        source.validate(
+                                new InputSpec(
+                                        "s3", Map.of("bucket", BUCKET, "accessKeyId", "AKIA"))));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> source.validate(new InputSpec("s3", options(Map.of("mode", "sideways")))));
+        assertThrows(
+                IllegalArgumentException.class,
+                () ->
+                        source.validate(
+                                new InputSpec(
+                                        "s3", options(Map.of("endpoint", "ftp://example.com")))));
+    }
+
+    @Test
+    void validateRejectsAnUnreachableBucket() {
+        when(s3Client.listObjectsV2(any(ListObjectsV2Request.class)))
+                .thenThrow(SdkClientException.create("connection refused"));
+
+        assertThrows(IllegalArgumentException.class, () -> source.validate(spec()));
+    }
+
+    private static InputSpec spec() {
+        return new InputSpec("s3", options(Map.of()));
+    }
+
+    /** The required options (bucket + credentials) plus any extras under test. */
+    private static Map<String, Object> options(Map<String, Object> extra) {
+        Map<String, Object> options = new HashMap<>(extra);
+        options.put("bucket", BUCKET);
+        options.put("accessKeyId", "AKIAEXAMPLE");
+        options.put("secretAccessKey", "shh");
+        return options;
+    }
+
+    private static S3Object object(String key, String eTag) {
+        return S3Object.builder()
+                .key(key)
+                .eTag(eTag)
+                .size(4L)
+                .lastModified(Instant.parse("2026-01-01T00:00:00Z"))
+                .build();
+    }
+
+    private void listingReturns(S3Object... objects) {
+        when(s3Client.listObjectsV2(any(ListObjectsV2Request.class)))
+                .thenReturn(ListObjectsV2Response.builder().contents(objects).build());
+    }
+
+    private void headReturns(String key, String eTag) {
+        when(s3Client.headObject(any(HeadObjectRequest.class)))
+                .thenReturn(
+                        HeadObjectResponse.builder()
+                                .eTag(eTag)
+                                .contentLength(4L)
+                                .lastModified(Instant.parse("2026-01-01T00:00:00Z"))
+                                .build());
+    }
+
+    private class RecordingContext implements ResolveContext {
+
+        private final String policyId;
+        private final List<String> present = new ArrayList<>();
+
+        private RecordingContext() {
+            this(POLICY);
+        }
+
+        private RecordingContext(String policyId) {
+            this.policyId = policyId;
+        }
+
+        @Override
+        public boolean claim(String identity, String gate, Supplier<String> contentHash) {
+            return ledger.claim(policyId, identity, gate, contentHash);
+        }
+
+        @Override
+        public void settle(
+                String identity, String finalGate, String finalContentHash, boolean success) {
+            ledger.settle(policyId, identity, finalGate, finalContentHash, success);
+        }
+
+        @Override
+        public boolean allSettledDone(String identity) {
+            return ledger.allSettledDone(identity);
+        }
+
+        @Override
+        public void reportPresent(Collection<String> identities) {
+            present.addAll(identities);
+        }
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/output/S3OutputSinkMinioTest.java

@@ -0,0 +1,212 @@
+package stirling.software.proprietary.policy.output;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.Resource;
+import org.testcontainers.containers.MinIOContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.model.job.ResultFile;
+import stirling.software.proprietary.policy.input.ResolveContext;
+import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.input.S3InputSource;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.s3.S3ConnectionPool;
+import stirling.software.proprietary.policy.s3.S3TestConnections;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.ResponseInputStream;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+
+/**
+ * End-to-end {@link S3OutputSink} test against a real S3 API (MinIO): uploads, collision renaming,
+ * and - composed with {@link S3InputSource} - the loop-safety guarantee that a policy writing into
+ * a bucket it also watches never re-ingests its own outputs, while a second policy still can.
+ */
+@Testcontainers(disabledWithoutDocker = true)
+class S3OutputSinkMinioTest {
+
+    private static final String POLICY = "p1";
+    private static final String ACCESS_KEY = "minioadmin";
+    private static final String SECRET_KEY = "minioadmin";
+
+    @Container
+    static MinIOContainer minio =
+            new MinIOContainer("minio/minio:latest")
+                    .withUserName(ACCESS_KEY)
+                    .withPassword(SECRET_KEY);
+
+    private static S3Client adminClient;
+    private static int bucketCounter;
+
+    private String bucket;
+    private S3OutputSink sink;
+    private S3InputSource source;
+    private InProcessProcessedLedger ledger;
+
+    @BeforeEach
+    void setUp() {
+        if (adminClient == null) {
+            adminClient =
+                    S3Client.builder()
+                            .endpointOverride(URI.create(minio.getS3URL()))
+                            .httpClient(UrlConnectionHttpClient.create())
+                            .region(Region.US_EAST_1)
+                            .credentialsProvider(
+                                    StaticCredentialsProvider.create(
+                                            AwsBasicCredentials.create(ACCESS_KEY, SECRET_KEY)))
+                            .serviceConfiguration(
+                                    S3Configuration.builder().pathStyleAccessEnabled(true).build())
+                            .build();
+        }
+        bucket = "policy-outbox-" + ++bucketCounter;
+        adminClient.createBucket(CreateBucketRequest.builder().bucket(bucket).build());
+
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowPrivateS3Endpoints(true);
+        S3ConnectionPool pool = new S3ConnectionPool(properties);
+        ledger = new InProcessProcessedLedger();
+        sink = new S3OutputSink(pool, S3TestConnections.legacyResolver(), ledger);
+        source = new S3InputSource(pool, S3TestConnections.legacyResolver());
+    }
+
+    @Test
+    void uploadsOutputsUnderThePrefix() throws IOException {
+        List<ResultFile> results =
+                sink.deliver(
+                        new OutputDelivery("run-1", POLICY),
+                        List.of(output("doc.pdf", "pdf bytes")),
+                        outputSpec("processed/"));
+
+        assertThat(results).hasSize(1);
+        assertThat(results.get(0).getFileName()).isEqualTo("s3://" + bucket + "/processed/doc.pdf");
+        assertThat(objectContent("processed/doc.pdf")).isEqualTo("pdf bytes");
+    }
+
+    @Test
+    void anExistingKeyIsNeverOverwritten() throws IOException {
+        adminClient.putObject(
+                PutObjectRequest.builder().bucket(bucket).key("doc.pdf").build(),
+                RequestBody.fromString("theirs", StandardCharsets.UTF_8));
+
+        List<ResultFile> results =
+                sink.deliver(
+                        new OutputDelivery("run-1", POLICY),
+                        List.of(output("doc.pdf", "ours")),
+                        outputSpec(""));
+
+        assertThat(results.get(0).getFileName()).isEqualTo("s3://" + bucket + "/doc (1).pdf");
+        assertThat(objectContent("doc.pdf")).isEqualTo("theirs");
+        assertThat(objectContent("doc (1).pdf")).isEqualTo("ours");
+    }
+
+    @Test
+    void aPolicyWritingIntoItsWatchedBucketSkipsItsOwnOutputsButAnotherPolicyChains()
+            throws IOException {
+        sink.deliver(
+                new OutputDelivery("run-1", POLICY),
+                List.of(output("result.pdf", "produced")),
+                outputSpec(""));
+
+        // The producing policy's sweep sees its own output at the recorded gate and skips it.
+        assertThat(source.resolve(inputSpec(), new RecordingContext(POLICY))).isEmpty();
+
+        // A different policy watching the same bucket has no row and processes it - chaining.
+        List<ResolvedInput> chained = source.resolve(inputSpec(), new RecordingContext("p2"));
+        assertThat(chained).hasSize(1);
+        try (InputStream stream = chained.get(0).inputs().primary().get(0).getInputStream()) {
+            assertThat(new String(stream.readAllBytes(), StandardCharsets.UTF_8))
+                    .isEqualTo("produced");
+        }
+    }
+
+    private OutputSpec outputSpec(String prefix) {
+        return new OutputSpec(
+                "s3",
+                Map.of(
+                        "bucket", bucket,
+                        "prefix", prefix,
+                        "endpoint", minio.getS3URL(),
+                        "accessKeyId", ACCESS_KEY,
+                        "secretAccessKey", SECRET_KEY));
+    }
+
+    private InputSpec inputSpec() {
+        return new InputSpec(
+                "s3",
+                Map.of(
+                        "bucket", bucket,
+                        "endpoint", minio.getS3URL(),
+                        "accessKeyId", ACCESS_KEY,
+                        "secretAccessKey", SECRET_KEY));
+    }
+
+    private String objectContent(String key) throws IOException {
+        try (ResponseInputStream<GetObjectResponse> stream =
+                adminClient.getObject(GetObjectRequest.builder().bucket(bucket).key(key).build())) {
+            return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
+        }
+    }
+
+    private static Resource output(String name, String content) {
+        return new ByteArrayResource(content.getBytes(StandardCharsets.UTF_8)) {
+            @Override
+            public String getFilename() {
+                return name;
+            }
+        };
+    }
+
+    private class RecordingContext implements ResolveContext {
+
+        private final String policyId;
+
+        private RecordingContext(String policyId) {
+            this.policyId = policyId;
+        }
+
+        @Override
+        public boolean claim(String identity, String gate, Supplier<String> contentHash) {
+            return ledger.claim(policyId, identity, gate, contentHash);
+        }
+
+        @Override
+        public void settle(
+                String identity, String finalGate, String finalContentHash, boolean success) {
+            ledger.settle(policyId, identity, finalGate, finalContentHash, success);
+        }
+
+        @Override
+        public boolean allSettledDone(String identity) {
+            return ledger.allSettledDone(identity);
+        }
+
+        @Override
+        public void reportPresent(Collection<String> identities) {}
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/output/S3OutputSinkTest.java

@@ -0,0 +1,269 @@
+package stirling.software.proprietary.policy.output;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.Resource;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.model.job.ResultFile;
+import stirling.software.proprietary.policy.ledger.ClaimState;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
+import stirling.software.proprietary.policy.ledger.ProcessedFileStatus;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.s3.S3ConnectionPool;
+import stirling.software.proprietary.policy.s3.S3TestConnections;
+
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectResponse;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+/**
+ * Tests for {@link S3OutputSink}: the ledger row exists before the object is visible, collisions
+ * re-pick names, ad-hoc runs record nothing, and encrypted-bucket ETags are re-recorded.
+ */
+@ExtendWith(MockitoExtension.class)
+class S3OutputSinkTest {
+
+    private static final String POLICY = "p1";
+    private static final String BUCKET = "outbox-bucket";
+    private static final OutputDelivery DELIVERY = new OutputDelivery("run-1", POLICY);
+    private static final OutputDelivery AD_HOC = new OutputDelivery("run-2", null);
+
+    @Mock private S3Client s3Client;
+
+    private S3OutputSink sink;
+    private InProcessProcessedLedger ledger;
+    private final List<PutObjectRequest> puts = new ArrayList<>();
+
+    @BeforeEach
+    void setUp() {
+        ledger = new InProcessProcessedLedger();
+        sink =
+                new S3OutputSink(
+                        new S3ConnectionPool(new ApplicationProperties(), config -> s3Client),
+                        S3TestConnections.legacyResolver(),
+                        ledger);
+    }
+
+    @Test
+    void recordsTheRowBeforeTheObjectBecomesVisible() throws IOException {
+        // The row for the exact key must already be settled DONE at the moment the PUT runs -
+        // record-before-visible, asserted from inside the upload itself.
+        List<ClaimState> stateAtPutTime = new ArrayList<>();
+        when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class)))
+                .thenAnswer(
+                        invocation -> {
+                            PutObjectRequest request = invocation.getArgument(0);
+                            puts.add(request);
+                            stateAtPutTime.add(stateFor(identity(request.key())));
+                            return PutObjectResponse.builder().eTag(quotedMd5("data")).build();
+                        });
+
+        List<ResultFile> results =
+                sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec());
+
+        assertEquals(1, results.size());
+        assertEquals("s3://" + BUCKET + "/processed/doc.pdf", results.get(0).getFileName());
+        assertEquals(4, results.get(0).getFileSize());
+        assertNotNull(stateAtPutTime.get(0));
+        assertEquals(ProcessedFileStatus.DONE, stateAtPutTime.get(0).status());
+        assertEquals(md5("data"), stateAtPutTime.get(0).gate());
+        assertTrue(puts.get(0).ifNoneMatch() != null);
+    }
+
+    @Test
+    void aTakenKeyIsForgottenAndRePicked() throws IOException {
+        when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class)))
+                .thenAnswer(
+                        invocation -> {
+                            PutObjectRequest request = invocation.getArgument(0);
+                            puts.add(request);
+                            if (puts.size() == 1) {
+                                throw s3Error(412, "PreconditionFailed");
+                            }
+                            return PutObjectResponse.builder().eTag(quotedMd5("data")).build();
+                        });
+
+        List<ResultFile> results =
+                sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec());
+
+        assertEquals("s3://" + BUCKET + "/processed/doc (1).pdf", results.get(0).getFileName());
+        // The lost candidate's row is gone; only the delivered key is recorded.
+        assertNull(stateFor(identity("processed/doc.pdf")));
+        assertNotNull(stateFor(identity("processed/doc (1).pdf")));
+    }
+
+    @Test
+    void anEncryptedBucketETagIsReRecordedAtTheActualGate() throws IOException {
+        when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class)))
+                .thenReturn(PutObjectResponse.builder().eTag("\"kms-opaque-etag\"").build());
+
+        sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec());
+
+        assertEquals("kms-opaque-etag", stateFor(identity("processed/doc.pdf")).gate());
+    }
+
+    @Test
+    void anAdHocDeliveryRecordsNothing() throws IOException {
+        when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class)))
+                .thenReturn(PutObjectResponse.builder().eTag(quotedMd5("data")).build());
+
+        sink.deliver(AD_HOC, List.of(output("doc.pdf", "data")), spec());
+
+        assertNull(stateFor(identity("processed/doc.pdf")));
+    }
+
+    @Test
+    void aFailedUploadForgetsItsRowAndThrows() {
+        when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class)))
+                .thenThrow(SdkClientException.create("connection refused"));
+
+        assertThrows(
+                IOException.class,
+                () -> sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec()));
+
+        assertNull(stateFor(identity("processed/doc.pdf")));
+    }
+
+    @Test
+    void aStoreWithoutConditionalPutsFallsBackToExistenceChecks() throws IOException {
+        when(s3Client.headObject(any(HeadObjectRequest.class))).thenThrow(s3Error(404, "NotFound"));
+        when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class)))
+                .thenAnswer(
+                        invocation -> {
+                            PutObjectRequest request = invocation.getArgument(0);
+                            puts.add(request);
+                            if (request.ifNoneMatch() != null) {
+                                throw s3Error(501, "NotImplemented");
+                            }
+                            return PutObjectResponse.builder().eTag(quotedMd5("data")).build();
+                        });
+
+        List<ResultFile> results =
+                sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec());
+
+        // Same key, second attempt unconditional.
+        assertEquals("s3://" + BUCKET + "/processed/doc.pdf", results.get(0).getFileName());
+        assertEquals(2, puts.size());
+        assertNull(puts.get(1).ifNoneMatch());
+        assertNotNull(stateFor(identity("processed/doc.pdf")));
+    }
+
+    @Test
+    void aBarePrefixGetsItsSlash() throws IOException {
+        when(s3Client.putObject(any(PutObjectRequest.class), any(RequestBody.class)))
+                .thenAnswer(
+                        invocation -> {
+                            puts.add(invocation.getArgument(0));
+                            return PutObjectResponse.builder().eTag(quotedMd5("data")).build();
+                        });
+
+        sink.deliver(DELIVERY, List.of(output("doc.pdf", "data")), spec("processed"));
+
+        assertEquals("processed/doc.pdf", puts.get(0).key());
+    }
+
+    @Test
+    void validateRejectsBadConfigShape() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> sink.validate(new OutputSpec("s3", Map.of())));
+        // Credentials are required, never the server's own identity.
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> sink.validate(new OutputSpec("s3", Map.of("bucket", BUCKET))));
+        assertThrows(
+                IllegalArgumentException.class,
+                () ->
+                        sink.validate(
+                                new OutputSpec(
+                                        "s3", Map.of("bucket", BUCKET, "accessKeyId", "AKIA"))));
+    }
+
+    @Test
+    void supportsOnlyS3Specs() {
+        assertTrue(sink.supports(spec()));
+        assertFalse(sink.supports(OutputSpec.inline()));
+        assertFalse(sink.supports(null));
+    }
+
+    private static OutputSpec spec() {
+        return spec("processed/");
+    }
+
+    private static OutputSpec spec(String prefix) {
+        return new OutputSpec(
+                "s3",
+                Map.of(
+                        "bucket",
+                        BUCKET,
+                        "prefix",
+                        prefix,
+                        "accessKeyId",
+                        "AKIAEXAMPLE",
+                        "secretAccessKey",
+                        "shh"));
+    }
+
+    private static String identity(String key) {
+        return "s3://" + BUCKET + "/" + key;
+    }
+
+    private ClaimState stateFor(String identity) {
+        return ledger.statesFor(POLICY, List.of(identity)).get(identity);
+    }
+
+    private static Resource output(String name, String content) {
+        return new ByteArrayResource(content.getBytes(StandardCharsets.UTF_8)) {
+            @Override
+            public String getFilename() {
+                return name;
+            }
+        };
+    }
+
+    private static String md5(String content) {
+        try {
+            return HexFormat.of()
+                    .formatHex(
+                            MessageDigest.getInstance("MD5")
+                                    .digest(content.getBytes(StandardCharsets.UTF_8)));
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+    private static String quotedMd5(String content) {
+        return "\"" + md5(content) + "\"";
+    }
+
+    private static AwsServiceException s3Error(int status, String code) {
+        return S3Exception.builder().statusCode(status).message(code).build();
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigrationTest.java

@@ -0,0 +1,212 @@
+package stirling.software.proprietary.policy.s3;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import stirling.software.proprietary.access.model.OwnerScope;
+import stirling.software.proprietary.integration.model.IntegrationConfig;
+import stirling.software.proprietary.integration.repository.IntegrationConfigRepository;
+import stirling.software.proprietary.model.Team;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineStep;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.source.InProcessSourceStore;
+import stirling.software.proprietary.policy.source.Source;
+import stirling.software.proprietary.policy.store.InProcessPolicyStore;
+import stirling.software.proprietary.security.repository.TeamRepository;
+
+/**
+ * Tests for {@link EmbeddedS3CredentialMigration}: legacy embedded credentials become deduplicated
+ * team-scoped connections, rewritten rows keep only per-use options, and re-runs are no-ops.
+ */
+@ExtendWith(MockitoExtension.class)
+class EmbeddedS3CredentialMigrationTest {
+
+    @Mock private IntegrationConfigRepository connections;
+    @Mock private TeamRepository teamRepository;
+
+    private final InProcessSourceStore sourceStore = new InProcessSourceStore();
+    private final InProcessPolicyStore policyStore = new InProcessPolicyStore();
+    private EmbeddedS3CredentialMigration migration;
+
+    @BeforeEach
+    void setUp() {
+        migration =
+                new EmbeddedS3CredentialMigration(
+                        sourceStore, policyStore, connections, teamRepository);
+        AtomicLong ids = new AtomicLong(100);
+        // Lenient: the nothing-to-migrate cases never create a connection.
+        lenient().when(connections.findAll()).thenReturn(List.of());
+        lenient()
+                .when(connections.save(any()))
+                .thenAnswer(
+                        invocation -> {
+                            IntegrationConfig saved = invocation.getArgument(0);
+                            if (saved.getId() == null) {
+                                saved.setId(ids.incrementAndGet());
+                            }
+                            return saved;
+                        });
+    }
+
+    @Test
+    void extractsSharedCredentialsIntoOneTeamScopedConnection() {
+        Team team = new Team();
+        team.setId(7L);
+        when(teamRepository.findById(7L)).thenReturn(Optional.of(team));
+        Source source =
+                sourceStore.save(
+                        new Source(
+                                null,
+                                "Claims intake",
+                                "s3",
+                                Map.of(
+                                        "bucket", "inbox",
+                                        "prefix", "incoming/",
+                                        "mode", "snapshot",
+                                        "accessKeyId", "AKIAEXAMPLE",
+                                        "secretAccessKey", "shh"),
+                                true,
+                                "alice",
+                                7L));
+        Policy policy =
+                policyStore.save(
+                        new Policy(
+                                null,
+                                "Rotate",
+                                "alice",
+                                true,
+                                null,
+                                List.of(source.id()),
+                                List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
+                                new OutputSpec(
+                                        "s3",
+                                        Map.of(
+                                                "bucket", "inbox",
+                                                "prefix", "processed/",
+                                                "accessKeyId", "AKIAEXAMPLE",
+                                                "secretAccessKey", "shh")),
+                                7L));
+
+        migration.migrate();
+
+        // Same bucket + credentials on both rows: exactly one connection extracted.
+        verify(connections, times(1)).save(any());
+        Map<String, Object> sourceOptions = sourceStore.get(source.id()).orElseThrow().options();
+        assertEquals(101L, sourceOptions.get("connectionId"));
+        assertEquals("incoming/", sourceOptions.get("prefix"));
+        assertEquals("snapshot", sourceOptions.get("mode"));
+        assertNull(sourceOptions.get("accessKeyId"));
+        assertNull(sourceOptions.get("secretAccessKey"));
+        assertNull(sourceOptions.get("bucket"));
+
+        Map<String, Object> outputOptions =
+                policyStore.get(policy.id()).orElseThrow().output().options();
+        assertEquals(101L, outputOptions.get("connectionId"));
+        assertEquals("processed/", outputOptions.get("prefix"));
+        assertNull(outputOptions.get("secretAccessKey"));
+    }
+
+    @Test
+    void connectionOwnershipFollowsTheSourceTeam() {
+        Team team = new Team();
+        team.setId(7L);
+        when(teamRepository.findById(7L)).thenReturn(Optional.of(team));
+        sourceStore.save(s3Source("teamed", 7L));
+
+        migration.migrate();
+
+        verify(connections)
+                .save(
+                        org.mockito.ArgumentMatchers.argThat(
+                                connection ->
+                                        connection.getScope() == OwnerScope.TEAM
+                                                && connection.getOwnerTeam() == team));
+    }
+
+    @Test
+    void teamlessRowsBecomeServerScopedConnections() {
+        sourceStore.save(s3Source("solo", null));
+
+        migration.migrate();
+
+        verify(connections)
+                .save(
+                        org.mockito.ArgumentMatchers.argThat(
+                                connection -> connection.getScope() == OwnerScope.SERVER));
+    }
+
+    @Test
+    void aSecondRunFindsNothingToDo() {
+        sourceStore.save(s3Source("once", null));
+
+        migration.migrate();
+        migration.migrate();
+
+        // One connection from the first run; the rewritten source no longer embeds credentials.
+        verify(connections, times(1)).save(any());
+    }
+
+    @Test
+    void nonS3AndAlreadyMigratedRowsAreUntouched() {
+        Source folder =
+                sourceStore.save(
+                        new Source(
+                                null,
+                                "Folder",
+                                "folder",
+                                Map.of("directory", "/in"),
+                                true,
+                                "alice",
+                                null));
+        Source migrated =
+                sourceStore.save(
+                        new Source(
+                                null,
+                                "Done already",
+                                "s3",
+                                Map.of("connectionId", 55L, "prefix", "in/"),
+                                true,
+                                "alice",
+                                null));
+
+        migration.migrate();
+
+        verify(connections, times(0)).save(any());
+        assertEquals(
+                Map.of("directory", "/in"), sourceStore.get(folder.id()).orElseThrow().options());
+        assertEquals(
+                Map.of("connectionId", 55L, "prefix", "in/"),
+                sourceStore.get(migrated.id()).orElseThrow().options());
+    }
+
+    private static Source s3Source(String name, Long teamId) {
+        return new Source(
+                null,
+                name,
+                "s3",
+                Map.of(
+                        "bucket", "inbox",
+                        "accessKeyId", "AKIAEXAMPLE",
+                        "secretAccessKey", "shh"),
+                true,
+                "alice",
+                teamId);
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/PolicyS3ConnectionUsageCheckTest.java

@@ -0,0 +1,52 @@
+package stirling.software.proprietary.policy.s3;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineStep;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.source.InProcessSourceStore;
+import stirling.software.proprietary.policy.source.Source;
+import stirling.software.proprietary.policy.store.InProcessPolicyStore;
+
+/** Tests for {@link PolicyS3ConnectionUsageCheck}'s reference scan across sources and outputs. */
+class PolicyS3ConnectionUsageCheckTest {
+
+    private final InProcessSourceStore sourceStore = new InProcessSourceStore();
+    private final InProcessPolicyStore policyStore = new InProcessPolicyStore();
+    private final PolicyS3ConnectionUsageCheck check =
+            new PolicyS3ConnectionUsageCheck(sourceStore, policyStore);
+
+    @Test
+    void reportsSourcesAndOutputsReferencingTheConnection() {
+        sourceStore.save(
+                new Source(
+                        null,
+                        "Claims intake",
+                        "s3",
+                        Map.of("connectionId", 5L, "prefix", "in/"),
+                        true,
+                        "alice",
+                        null));
+        policyStore.save(
+                new Policy(
+                        null,
+                        "Rotate",
+                        "alice",
+                        true,
+                        null,
+                        List.of(),
+                        List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
+                        new OutputSpec("s3", Map.of("connectionId", "5")),
+                        null));
+
+        assertThat(check.usagesOf(5))
+                .containsExactlyInAnyOrder("source 'Claims intake'", "pipeline 'Rotate'");
+        assertThat(check.usagesOf(6)).isEmpty();
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/S3ConnectionResolverTest.java

@@ -0,0 +1,149 @@
+package stirling.software.proprietary.policy.s3;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import java.util.Optional;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
+import org.springframework.security.core.context.SecurityContextHolder;
+
+import stirling.software.proprietary.access.service.OwnershipService;
+import stirling.software.proprietary.integration.model.IntegrationConfig;
+import stirling.software.proprietary.integration.model.IntegrationType;
+import stirling.software.proprietary.integration.repository.IntegrationConfigRepository;
+import stirling.software.proprietary.security.model.User;
+import stirling.software.proprietary.security.service.UserService;
+
+/**
+ * Tests for {@link S3ConnectionResolver}: connection dereferencing with per-use overrides, the
+ * legacy embedded fallback, and the save-time access check that background sweeps skip.
+ */
+@ExtendWith(MockitoExtension.class)
+class S3ConnectionResolverTest {
+
+    @Mock private IntegrationConfigRepository connections;
+    @Mock private OwnershipService ownership;
+    @Mock private UserService userService;
+
+    @AfterEach
+    void clearSecurityContext() {
+        SecurityContextHolder.clearContext();
+    }
+
+    @Test
+    void resolvesAConnectionAndMergesPerUseOptions() {
+        when(connections.findById(9L)).thenReturn(Optional.of(s3Connection(9L, true)));
+
+        S3Config config =
+                resolver()
+                        .resolve(
+                                Map.of(
+                                        "connectionId", 9L,
+                                        "prefix", "incoming/",
+                                        "mode", "snapshot"));
+
+        assertEquals("inbox", config.bucket());
+        assertEquals("AKIAEXAMPLE", config.accessKeyId());
+        assertEquals("incoming/", config.prefix());
+        assertTrue(config.snapshot());
+    }
+
+    @Test
+    void acceptsAStringConnectionReference() {
+        when(connections.findById(9L)).thenReturn(Optional.of(s3Connection(9L, true)));
+
+        assertEquals("inbox", resolver().resolve(Map.of("connectionId", "9")).bucket());
+    }
+
+    @Test
+    void fallsBackToLegacyEmbeddedCredentials() {
+        S3Config config =
+                resolver()
+                        .resolve(
+                                Map.of(
+                                        "bucket", "legacy",
+                                        "accessKeyId", "AKIAEXAMPLE",
+                                        "secretAccessKey", "shh"));
+
+        assertEquals("legacy", config.bucket());
+    }
+
+    @Test
+    void rejectsUnknownDisabledOrWrongTypeConnections() {
+        when(connections.findById(1L)).thenReturn(Optional.empty());
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> resolver().resolve(Map.of("connectionId", 1L)));
+
+        when(connections.findById(2L)).thenReturn(Optional.of(s3Connection(2L, false)));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> resolver().resolve(Map.of("connectionId", 2L)));
+
+        IntegrationConfig mcp = s3Connection(3L, true);
+        mcp.setIntegrationType(IntegrationType.MCP);
+        when(connections.findById(3L)).thenReturn(Optional.of(mcp));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> resolver().resolve(Map.of("connectionId", 3L)));
+    }
+
+    @Test
+    void anAuthenticatedSaverMustBeAllowedToUseTheConnection() {
+        when(connections.findById(9L)).thenReturn(Optional.of(s3Connection(9L, true)));
+        User saver = new User();
+        saver.setUsername("alice");
+        SecurityContextHolder.getContext()
+                .setAuthentication(
+                        new UsernamePasswordAuthenticationToken(saver, null, java.util.List.of()));
+        when(ownership.canUse(any(), any(IntegrationConfig.class), eq(saver))).thenReturn(false);
+
+        // Denied reads the same as unknown and never echoes the connection name, so ids can't be
+        // enumerated by probing.
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> resolver().resolve(Map.of("connectionId", 9L)));
+        try {
+            resolver().resolve(Map.of("connectionId", 9L));
+        } catch (IllegalArgumentException e) {
+            org.junit.jupiter.api.Assertions.assertFalse(
+                    e.getMessage().contains("Claims bucket"),
+                    "access-denied error must not leak the connection name");
+        }
+    }
+
+    @Test
+    void backgroundSweepsWithNoUserSkipTheAccessCheck() {
+        when(connections.findById(9L)).thenReturn(Optional.of(s3Connection(9L, true)));
+
+        // No authentication in the context: resolution succeeds without consulting ownership.
+        assertEquals("inbox", resolver().resolve(Map.of("connectionId", 9L)).bucket());
+    }
+
+    private S3ConnectionResolver resolver() {
+        return new S3ConnectionResolver(connections, ownership, userService);
+    }
+
+    private static IntegrationConfig s3Connection(long id, boolean enabled) {
+        IntegrationConfig connection = new IntegrationConfig();
+        connection.setId(id);
+        connection.setIntegrationType(IntegrationType.S3);
+        connection.setName("Claims bucket");
+        connection.setEnabled(enabled);
+        connection.setConfig(
+                "{\"bucket\":\"inbox\",\"accessKeyId\":\"AKIAEXAMPLE\","
+                        + "\"secretAccessKey\":\"shh\"}");
+        return connection;
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/S3IntegrationValidatorTest.java

@@ -0,0 +1,71 @@
+package stirling.software.proprietary.policy.s3;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.integration.model.IntegrationType;
+
+/**
+ * Tests for {@link S3IntegrationValidator}: the S3 connection schema fails at save time - missing
+ * credentials, bad endpoints, and private endpoints without the operator opt-in.
+ */
+class S3IntegrationValidatorTest {
+
+    @Test
+    void acceptsACompleteConnection() {
+        assertThatCode(
+                        () ->
+                                validator(false)
+                                        .validate(
+                                                Map.of(
+                                                        "bucket", "inbox",
+                                                        "accessKeyId", "AKIAEXAMPLE",
+                                                        "secretAccessKey", "shh")))
+                .doesNotThrowAnyException();
+    }
+
+    @Test
+    void rejectsMissingCredentialsOrBucket() {
+        assertThatThrownBy(() -> validator(false).validate(Map.of("bucket", "inbox")))
+                .isInstanceOf(IllegalArgumentException.class);
+        assertThatThrownBy(
+                        () ->
+                                validator(false)
+                                        .validate(
+                                                Map.of(
+                                                        "accessKeyId", "AKIAEXAMPLE",
+                                                        "secretAccessKey", "shh")))
+                .isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test
+    void rejectsAPrivateEndpointWithoutTheOperatorOptIn() {
+        Map<String, Object> config =
+                Map.of(
+                        "bucket", "inbox",
+                        "accessKeyId", "AKIAEXAMPLE",
+                        "secretAccessKey", "shh",
+                        "endpoint", "http://localhost:9000");
+
+        assertThatThrownBy(() -> validator(false).validate(config))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("allowPrivateS3Endpoints");
+        assertThatCode(() -> validator(true).validate(config)).doesNotThrowAnyException();
+    }
+
+    @Test
+    void itOnlyClaimsTheS3Type() {
+        org.junit.jupiter.api.Assertions.assertEquals(IntegrationType.S3, validator(false).type());
+    }
+
+    private static S3IntegrationValidator validator(boolean allowPrivateEndpoints) {
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowPrivateS3Endpoints(allowPrivateEndpoints);
+        return new S3IntegrationValidator(properties);
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/S3TestConnections.java

@@ -0,0 +1,24 @@
+package stirling.software.proprietary.policy.s3;
+
+import static org.mockito.Mockito.mock;
+
+import stirling.software.proprietary.access.service.OwnershipService;
+import stirling.software.proprietary.integration.repository.IntegrationConfigRepository;
+import stirling.software.proprietary.security.service.UserService;
+
+/** Test fixtures for S3 connection plumbing shared across the policy S3 tests. */
+public final class S3TestConnections {
+
+    private S3TestConnections() {}
+
+    /**
+     * A resolver for tests whose options embed credentials directly (the legacy pass-through path),
+     * so its collaborators are never touched.
+     */
+    public static S3ConnectionResolver legacyResolver() {
+        return new S3ConnectionResolver(
+                mock(IntegrationConfigRepository.class),
+                mock(OwnershipService.class),
+                mock(UserService.class));
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java

@@ -28,6 +28,7 @@ import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.store.InProcessPolicyStore;
 import stirling.software.proprietary.policy.store.PolicyStore;
 import stirling.software.proprietary.policy.trigger.PolicyTriggerManager;
+import stirling.software.proprietary.util.SecretMasker;
 
 /**
  * Tests for {@link SourceController}'s delete guard: a source still referenced by a policy is
@@ -109,6 +110,107 @@ class SourceControllerTest {
         assertEquals(404, controller.delete("nope").getStatusCode().value());
     }
 
+    @Test
+    void documentCountsForTheEditorReturnsTheTeamSeries() {
+        ResponseEntity<List<Long>> response = controller.documentCounts(EditorSource.ID);
+
+        assertEquals(200, response.getStatusCode().value());
+        assertEquals(30, response.getBody().size());
+    }
+
+    @Test
+    void theEditorIsBuiltInAndCannotBeDeleted() {
+        ResponseStatusException ex =
+                assertThrows(
+                        ResponseStatusException.class, () -> controller.delete(EditorSource.ID));
+
+        assertEquals(400, ex.getStatusCode().value());
+    }
+
+    @Test
+    void theEditorIsBuiltInAndCannotBeSaved() {
+        Source editor = new Source(null, "Editor", "editor", Map.of(), true, null, null);
+
+        ResponseStatusException ex =
+                assertThrows(ResponseStatusException.class, () -> controller.save(editor));
+
+        assertEquals(400, ex.getStatusCode().value());
+    }
+
+    @Test
+    void readsReturnSecretsAsTheRedactionSentinel() {
+        Source saved = sourceStore.save(s3Source("shh"));
+
+        Source read = controller.get(saved.id()).getBody();
+
+        assertEquals(SecretMasker.REDACTED, read.options().get("secretAccessKey"));
+        assertEquals("AKIAEXAMPLE", read.options().get("accessKeyId"));
+        // The store itself keeps the real value.
+        assertEquals(
+                "shh", sourceStore.get(saved.id()).orElseThrow().options().get("secretAccessKey"));
+    }
+
+    @Test
+    void savingTheSentinelBackKeepsTheStoredSecret() {
+        Source saved = sourceStore.save(s3Source("shh"));
+
+        Source edited =
+                new Source(
+                        saved.id(),
+                        "Renamed",
+                        saved.type(),
+                        Map.of(
+                                "bucket", "inbox",
+                                "accessKeyId", "AKIAEXAMPLE",
+                                "secretAccessKey", SecretMasker.REDACTED),
+                        true,
+                        saved.owner(),
+                        saved.teamId());
+        Source response = controller.save(edited).getBody();
+
+        assertEquals(
+                "shh", sourceStore.get(saved.id()).orElseThrow().options().get("secretAccessKey"));
+        // The save response is masked too; only the store sees the real value.
+        assertEquals(SecretMasker.REDACTED, response.options().get("secretAccessKey"));
+    }
+
+    @Test
+    void savingANewSecretReplacesTheStoredOne() {
+        Source saved = sourceStore.save(s3Source("old-secret"));
+
+        Source edited =
+                new Source(
+                        saved.id(),
+                        saved.name(),
+                        saved.type(),
+                        Map.of(
+                                "bucket", "inbox",
+                                "accessKeyId", "AKIAEXAMPLE",
+                                "secretAccessKey", "new-secret"),
+                        true,
+                        saved.owner(),
+                        saved.teamId());
+        controller.save(edited);
+
+        assertEquals(
+                "new-secret",
+                sourceStore.get(saved.id()).orElseThrow().options().get("secretAccessKey"));
+    }
+
+    private static Source s3Source(String secret) {
+        return new Source(
+                null,
+                "Bucket intake",
+                "s3",
+                Map.of(
+                        "bucket", "inbox",
+                        "accessKeyId", "AKIAEXAMPLE",
+                        "secretAccessKey", secret),
+                true,
+                "owner",
+                null);
+    }
+
     private static Source folderSource() {
         return new Source(
                 null, "Claims intake", "folder", Map.of("directory", "/in"), true, "owner", null);

app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceOverviewServiceTest.java

@@ -56,9 +56,10 @@ class SourceOverviewServiceTest {
 
         SourcesResponse response = service.overview();
 
-        assertEquals(3, response.sources().size());
-        // Sorted most-referenced first, so the shared source A leads.
-        assertEquals(a.id(), response.sources().get(0).id());
+        assertEquals(4, response.sources().size());
+        // The built-in editor is pinned first; persisted sources follow, most-referenced leading.
+        assertEquals(EditorSource.ID, response.sources().get(0).id());
+        assertEquals(a.id(), response.sources().get(1).id());
 
         SourceView av = find(response, a.id());
         assertEquals(2, av.referenceCount());
@@ -123,13 +124,57 @@ class SourceOverviewServiceTest {
 
         SourcesResponse response = scoped.overview();
 
-        assertEquals(1, response.sources().size());
-        SourceView view = response.sources().get(0);
-        assertEquals(ours.id(), view.id());
+        assertEquals(2, response.sources().size());
+        assertEquals(EditorSource.ID, response.sources().get(0).id());
+        SourceView view = find(response, ours.id());
         assertEquals(1, view.referenceCount());
         assertEquals(List.of(1L, 1L, 0L), response.kpis().stream().map(SourceKpi::value).toList());
     }
 
+    @Test
+    void theEditorSourceIsAlwaysPresentEvenWithNoConnections() {
+        SourcesResponse response = service.overview();
+
+        assertEquals(1, response.sources().size());
+        SourceView editor = response.sources().get(0);
+        assertEquals(EditorSource.ID, editor.id());
+        assertEquals("editor", editor.type());
+        assertEquals("active", editor.status());
+        assertEquals(0, editor.referenceCount());
+        // KPIs describe configured connections, so the built-in editor is left out of them.
+        assertEquals(List.of(0L, 0L, 0L), response.kpis().stream().map(SourceKpi::value).toList());
+    }
+
+    @Test
+    void theEditorSourceIsUsedByEveryPolicyThatRunsFromIt() {
+        editorPolicy("Redact on upload");
+        editorPolicy("Classify on upload");
+        // A folder-sourced policy does not target the editor, so it must not inflate the count.
+        policyReferencing("Folder sweep", source("Folder", "/f").id());
+
+        SourceView editor = find(service.overview(), EditorSource.ID);
+
+        assertEquals(2, editor.referenceCount());
+        assertTrue(
+                editor.referencingPolicies().stream()
+                        .map(SourceView.PolicyRef::name)
+                        .toList()
+                        .containsAll(List.of("Redact on upload", "Classify on upload")));
+    }
+
+    @Test
+    void theEditorSourceReportsTheTeamsRecordedDocumentThroughput() {
+        // Login disabled, so the team is null and the editor shares the global counter bucket.
+        docCounter.record(EditorSource.counterKey(null), 4);
+        docCounter.record(EditorSource.counterKey(null), 6);
+
+        SourceView editor = find(service.overview(), EditorSource.ID);
+
+        assertEquals(10, editor.docsTotal());
+        assertEquals(10, editor.docs24h());
+        assertEquals(10, editor.docs30d());
+    }
+
     @Test
     void documentCountsReflectRecordedDocs() {
         Source a = source("A", "/a");
@@ -177,6 +222,22 @@ class SourceOverviewServiceTest {
                         OutputSpec.inline()));
     }
 
+    /**
+     * A policy that targets the editor: membership rides in its output metadata, not a sourceId.
+     */
+    private void editorPolicy(String name) {
+        policyStore.save(
+                new Policy(
+                        null,
+                        name,
+                        "owner",
+                        true,
+                        null,
+                        List.of(),
+                        List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
+                        new OutputSpec("inline", Map.of("sources", List.of("editor")))));
+    }
+
     private void teamPolicy(String name, Long teamId, String... sourceIds) {
         policyStore.save(
                 new Policy(