Add S3 policy source (#6948)

# Description of Changes
* Adds an Amazon S3 Source & Output
* Removes folder source from SaaS
* Some miscellaneous UX fixes around pipelines

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/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/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/policy/controller/PolicyController.java

@@ -48,7 +48,9 @@ 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.Policy;
 import stirling.software.proprietary.policy.model.PolicyInputs;
@@ -64,6 +66,7 @@ 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
@@ -202,7 +205,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 +216,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 +285,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 +353,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 +389,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());
     }
@@ -412,9 +470,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)

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

@@ -44,7 +44,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 +52,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,7 +95,7 @@ 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. */

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/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,286 @@
+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.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 (see {@link S3Config}): "bucket" (required), "region" (default us-east-1), "prefix" (only
+ * keys starting with it are read), "endpoint" (S3-compatible stores such as MinIO; path-style
+ * addressing is used automatically), "accessKeyId" and "secretAccessKey" (required; requests are
+ * never signed with the server's own AWS identity), and "mode" which 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;
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    @Override
+    public boolean supports(InputSpec spec) {
+        return spec != null && TYPE.equals(spec.type());
+    }
+
+    /**
+     * Fails fast at save time: bad config shape, a private endpoint without the operator opt-in, or
+     * a bucket the supplied credentials cannot list.
+     */
+    @Override
+    public void validate(InputSpec spec) {
+        S3Config config = S3Config.from(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 = S3Config.from(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,270 @@
+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.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 ProcessedLedger processedLedger;
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    @Override
+    public boolean supports(OutputSpec spec) {
+        return spec != null && TYPE.equals(spec.type());
+    }
+
+    /**
+     * Config shape 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(S3Config.from(spec.options()));
+    }
+
+    @Override
+    public List<ResultFile> deliver(
+            OutputDelivery delivery, List<Resource> outputs, OutputSpec spec) throws IOException {
+        S3Config config = S3Config.from(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/S3Config.java

@@ -0,0 +1,103 @@
+package stirling.software.proprietary.policy.s3;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+
+/**
+ * Connection settings shared by the S3 input source and output sink, parsed from a spec's options
+ * map. 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/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/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());
     }
@@ -97,7 +104,7 @@ public class SourceController {
                             + " matching source type.")
     public ResponseEntity<Source> save(@RequestBody Source source) {
         requireSourceEditingAllowed();
-        Source owned = resolveOwnership(source);
+        Source owned = withStoredSecrets(resolveOwnership(source));
         try {
             validateConfig(owned);
         } catch (IllegalArgumentException e) {
@@ -107,7 +114,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}")
@@ -169,6 +176,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();

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,
@@ -104,13 +105,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/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/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/policy/controller/PolicyControllerTest.java

@@ -9,6 +9,7 @@ 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 +18,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,7 +36,9 @@ 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;
@@ -44,6 +48,7 @@ import stirling.software.proprietary.policy.progress.PolicyProgressListener;
 import stirling.software.proprietary.policy.source.SourceAccessGuard;
 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")
@@ -128,6 +133,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));
@@ -263,6 +279,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 +400,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 +587,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/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;
@@ -30,6 +31,7 @@ import org.mockito.junit.jupiter.MockitoExtension;
 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;
@@ -85,6 +87,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");

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

@@ -0,0 +1,231 @@
+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 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));
+        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()));
+
+        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,327 @@
+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 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));
+        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,211 @@
+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 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, ledger);
+        source = new S3InputSource(pool);
+    }
+
+    @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,267 @@
+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 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),
+                        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/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,80 @@ class SourceControllerTest {
         assertEquals(404, controller.delete("nope").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/trigger/ScheduleTriggerTest.java

@@ -70,6 +70,41 @@ class ScheduleTriggerTest {
         verify(policyRunner, times(1)).run(eq(policy));
     }
 
+    @Test
+    void anIntervalMatchingTheSweepPeriodFiresEverySweepDespiteJitter() {
+        Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES));
+        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+
+        Instant t0 = Instant.parse("2026-06-05T10:00:00Z");
+        trigger.sweep(t0); // baseline
+        // The sweep that fires runs a few ms late (scheduler jitter)...
+        trigger.sweep(t0.plusSeconds(60).plusMillis(5));
+        verify(policyRunner, times(1)).run(eq(policy));
+
+        // ...and the next sweep lands exactly on the 60s grid. Anchoring lastFired to the due
+        // time (not the jittered observation) means this must still fire, not alias to skip.
+        trigger.sweep(t0.plusSeconds(120));
+        verify(policyRunner, times(2)).run(eq(policy));
+    }
+
+    @Test
+    void aGapFiresOnceNotOncePerMissedInterval() {
+        Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES));
+        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+
+        Instant t0 = Instant.parse("2026-06-05T10:00:00Z");
+        trigger.sweep(t0); // baseline
+        // Ten minutes of downtime: nine missed due points collapse into one firing.
+        trigger.sweep(t0.plusSeconds(600));
+        verify(policyRunner, times(1)).run(eq(policy));
+
+        // Not due again until a full interval after the latest due point.
+        trigger.sweep(t0.plusSeconds(630));
+        verify(policyRunner, times(1)).run(eq(policy));
+        trigger.sweep(t0.plusSeconds(660));
+        verify(policyRunner, times(2)).run(eq(policy));
+    }
+
     @Test
     void doesNotFireBeforeTheNextScheduledTime() {
         Policy policy = scheduled("p1", new Schedule.Daily(LocalTime.of(3, 0))); // 03:00 UTC daily

app/proprietary/src/test/java/stirling/software/proprietary/util/SecretMaskerTest.java

@@ -14,8 +14,8 @@ import org.junit.jupiter.api.Test;
  * Unit tests for {@link SecretMasker}.
  *
  * <p>Assumptions: - Key matching is case-insensitive via the pattern in SENSITIVE. - If the key
- * matches a sensitive pattern, the value is replaced with "***REDACTED***". - Nested maps and lists
- * are searched recursively. - Null maps and null values are ignored or returned as null. -
+ * matches a sensitive pattern, the value is replaced with SecretMasker.REDACTED. - Nested maps and
+ * lists are searched recursively. - Null maps and null values are ignored or returned as null. -
  * Non-sensitive keys/values remain unchanged.
  */
 class SecretMaskerTest {
@@ -40,7 +40,7 @@ class SecretMaskerTest {
 
             Map<String, Object> result = SecretMasker.mask(input);
 
-            assertEquals("***REDACTED***", result.get("password"));
+            assertEquals(SecretMasker.REDACTED, result.get("password"));
             assertEquals("john", result.get("username"));
         }
 
@@ -55,11 +55,54 @@ class SecretMaskerTest {
 
             Map<String, Object> result = SecretMasker.mask(input);
 
-            assertEquals("***REDACTED***", result.get("Api-Key"));
-            assertEquals("***REDACTED***", result.get("TOKEN"));
+            assertEquals(SecretMasker.REDACTED, result.get("Api-Key"));
+            assertEquals(SecretMasker.REDACTED, result.get("TOKEN"));
             assertEquals("keepme", result.get("normal"));
         }
 
+        @Test
+        @DisplayName("restoreRedacted swaps sentinels for stored values, leaves the rest")
+        void restoreRedactedRoundTripsAnEdit() {
+            Map<String, Object> stored =
+                    Map.of("secretAccessKey", "shh", "accessKeyId", "AKIAEXAMPLE");
+            Map<String, Object> incoming =
+                    Map.of(
+                            "secretAccessKey", SecretMasker.REDACTED,
+                            "accessKeyId", "AKIA-NEW",
+                            "bucket", "inbox");
+
+            Map<String, Object> merged = SecretMasker.restoreRedacted(incoming, stored);
+
+            assertEquals("shh", merged.get("secretAccessKey"));
+            assertEquals("AKIA-NEW", merged.get("accessKeyId"));
+            assertEquals("inbox", merged.get("bucket"));
+        }
+
+        @Test
+        @DisplayName("restoreRedacted leaves a sentinel with no stored counterpart in place")
+        void restoreRedactedWithoutStoredValueStaysSentinel() {
+            Map<String, Object> merged =
+                    SecretMasker.restoreRedacted(
+                            Map.of("secretAccessKey", SecretMasker.REDACTED), Map.of());
+
+            assertEquals(SecretMasker.REDACTED, merged.get("secretAccessKey"));
+        }
+
+        @Test
+        @DisplayName("should mask camelCase secretAccessKey despite no word boundary")
+        void shouldMaskCamelCaseSecretAccessKey() {
+            Map<String, Object> input =
+                    Map.of(
+                            "secretAccessKey", "shh",
+                            "accessKeyId", "AKIAEXAMPLE");
+
+            Map<String, Object> result = SecretMasker.mask(input);
+
+            assertEquals(SecretMasker.REDACTED, result.get("secretAccessKey"));
+            // Access key ids are username-like, not secrets.
+            assertEquals("AKIAEXAMPLE", result.get("accessKeyId"));
+        }
+
         @Test
         @DisplayName("should mask nested map sensitive keys")
         void shouldMaskNestedMapSensitiveKeys() {
@@ -77,9 +120,9 @@ class SecretMaskerTest {
             Map<String, Object> result = SecretMasker.mask(input);
 
             Map<String, Object> outer = (Map<String, Object>) result.get("outer");
-            assertEquals("***REDACTED***", outer.get("jwt"));
+            assertEquals(SecretMasker.REDACTED, outer.get("jwt"));
             Map<String, Object> inner = (Map<String, Object>) outer.get("inner");
-            assertEquals("***REDACTED***", inner.get("secret"));
+            assertEquals(SecretMasker.REDACTED, inner.get("secret"));
             assertEquals("ok", inner.get("other"));
         }
 
@@ -98,7 +141,7 @@ class SecretMaskerTest {
 
             List<?> list = (List<?>) result.get("list");
             Map<String, Object> first = (Map<String, Object>) list.get(0);
-            assertEquals("***REDACTED***", first.get("token"));
+            assertEquals(SecretMasker.REDACTED, first.get("token"));
             Map<String, Object> second = (Map<String, Object>) list.get(1);
             assertEquals("john", second.get("username"));
             assertEquals("stringValue", list.get(2));
@@ -170,7 +213,8 @@ class SecretMaskerTest {
             Map<String, Object> outer = (Map<String, Object>) result.get("outer");
             assertTrue(outer.containsKey(null), "Null key should be preserved");
             assertEquals("plainText", outer.get(null), "Value for null key must not be masked");
-            assertEquals("***REDACTED***", outer.get("password"), "Sensitive keys must be masked");
+            assertEquals(
+                    SecretMasker.REDACTED, outer.get("password"), "Sensitive keys must be masked");
         }
     }
 }

frontend/editor/public/locales/en-US/translation.toml

@@ -7246,6 +7246,11 @@ operations_one = "Operation ({{count}})"
 operations_other = "Operations ({{count}})"
 output = "Output"
 removeStep = "Remove operation"
+s3Configure = "Configure"
+s3Done = "Done"
+s3ModalTitle = "Amazon S3 output"
+s3NotConfigured = "Not configured"
+s3PrefixHelp = "Outputs are uploaded under this key prefix."
 save = "Save changes"
 scheduleEvery = "Run every"
 sources = "Sources"
@@ -7265,6 +7270,7 @@ confirm = "Delete"
 title = "Delete pipeline?"
 
 [portal.pipelines.detail]
+clearHistory = "Clear history"
 delete = "Delete pipeline"
 run = "Run now"
 
@@ -7282,12 +7288,19 @@ total = "Pipelines"
 [portal.pipelines.output]
 folder = "Write to folder"
 inline = "Return files"
+s3 = "Write to Amazon S3"
 
 [portal.pipelines.run]
+allProcessed_one = "Nothing to run: the source's {{count}} document has already been processed."
+allProcessed_other = "Nothing to run: all {{count}} documents in the sources have already been processed."
 completed_one = "Run completed."
 completed_other = "All {{count}} runs completed."
 empty = "Nothing to run: the sources had no documents to process."
 failed = "Run failed: {{error}}"
+historyCleared = "History cleared. The next run reprocesses everything currently in the sources."
+inFlight = "Nothing new to run: documents are still being processed from an earlier run."
+parked_one = "Nothing to run: {{count}} document failed previously and is parked. Fix the cause, then clear history to retry it."
+parked_other = "Nothing to run: {{count}} documents failed previously and are parked. Fix the cause, then clear history to retry them."
 running = "Run started; still in progress."
 timeout = "Run is taking longer than expected; it may still finish in the background."
 
@@ -7974,6 +7987,42 @@ label = "Folder depth"
 all = "Include subfolders"
 top = "Top level only"
 
+[portal.sources.types.s3]
+description = "Pull documents from an Amazon S3 or S3-compatible bucket."
+label = "Amazon S3"
+
+[portal.sources.types.s3.fields.accessKeyId]
+label = "Access key ID"
+
+[portal.sources.types.s3.fields.bucket]
+label = "Bucket"
+placeholder = "my-company-inbox"
+
+[portal.sources.types.s3.fields.endpoint]
+helperText = "Leave blank for Amazon S3. Set to use an S3-compatible service such as MinIO."
+label = "Custom endpoint"
+placeholder = "https://s3.example.com"
+
+[portal.sources.types.s3.fields.mode]
+helperText = "Consume removes each object from the bucket once every policy has processed it."
+label = "Read mode"
+
+[portal.sources.types.s3.fields.mode.options]
+consume = "Consume: process each object once"
+snapshot = "Snapshot: re-read the bucket every run"
+
+[portal.sources.types.s3.fields.prefix]
+helperText = "Only objects whose keys start with this prefix are processed."
+label = "Key prefix"
+placeholder = "incoming/"
+
+[portal.sources.types.s3.fields.region]
+label = "Region"
+placeholder = "us-east-1"
+
+[portal.sources.types.s3.fields.secretAccessKey]
+label = "Secret access key"
+
 [portal.sources.types.unknown]
 label = "Source"
 

frontend/editor/src/portal-saas/components/pipelines/outputModes.test.ts

@@ -0,0 +1,9 @@
+import { describe, expect, it } from "vitest";
+// Resolves to the SaaS override (src/portal-saas) via the @portal cascade.
+import { availableOutputModes } from "@portal/components/pipelines/outputModes";
+
+describe("availableOutputModes (SaaS)", () => {
+  it("offers only s3: no server filesystem, and inline results would expire unseen", () => {
+    expect(availableOutputModes()).toEqual(["s3"]);
+  });
+});

frontend/editor/src/portal-saas/components/pipelines/outputModes.ts

@@ -0,0 +1,12 @@
+import type { PipelineOutputMode } from "@portal/api/pipelines";
+
+/**
+ * Hosted deployments never write to the server's filesystem (the backend's
+ * FolderAccessGuard denies it outright), so folder outputs are not offered in
+ * the pipeline builder. Inline is not offered either: inline results live in
+ * transient job storage with no portal download surface, so for an unattended
+ * pipeline they would simply expire unseen.
+ */
+export function availableOutputModes(): PipelineOutputMode[] {
+  return ["s3"];
+}

frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.test.ts

@@ -0,0 +1,13 @@
+import { describe, expect, it } from "vitest";
+// Resolves to the SaaS override (src/portal-saas) via the @portal cascade.
+import { creatableSourceTypes } from "@portal/components/sources/creatableSourceTypes";
+
+describe("creatableSourceTypes (SaaS)", () => {
+  it("never offers folder sources: hosted deployments do not read the server filesystem", () => {
+    expect(creatableSourceTypes().map((t) => t.type)).not.toContain("folder");
+  });
+
+  it("still offers the cloud source types", () => {
+    expect(creatableSourceTypes().map((t) => t.type)).toContain("s3");
+  });
+});

frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.ts

@@ -0,0 +1,13 @@
+import {
+  CREATABLE_SOURCE_TYPES,
+  type CreatableSourceType,
+} from "@portal/components/sources/sourceTypes";
+
+/**
+ * Hosted deployments never read the server's filesystem (the backend's
+ * FolderAccessGuard denies it outright), so folder connections are not offered
+ * in the connect wizard.
+ */
+export function creatableSourceTypes(): CreatableSourceType[] {
+  return CREATABLE_SOURCE_TYPES.filter((type) => type.type !== "folder");
+}

frontend/editor/src/portal/api/pipelines.ts

@@ -29,6 +29,9 @@ export interface OutputSpec {
   options: Record<string, unknown>;
 }
 
+/** The output destinations the pipeline builder can offer. */
+export type PipelineOutputMode = "inline" | "folder" | "s3";
+
 /**
  * The stored policy record: the create/update body (`id` blank on create) and what
  * the backend returns from GET/POST. Mirrors Policy.java exactly; `owner`/`teamId`
@@ -150,12 +153,26 @@ export async function fetchTriggers(): Promise<TriggerInfo[]> {
 }
 
 /**
- * POST /api/v1/policies/{id}/trigger: run the pipeline now against its configured
- * sources, regardless of the enabled flag. Returns the ids of the runs started
- * (empty when the sources yielded no work); poll {@link fetchRun} for each.
+ * What a manual trigger found and started. Mirrors the backend `SweepOutcome`:
+ * when `runIds` is empty, the counts say why - no files listed, all already
+ * processed at their current version, parked by a failed run, or still in
+ * flight from an earlier sweep.
  */
-export async function triggerPipeline(id: string): Promise<string[]> {
-  return apiClient.local.json<string[]>(
+export interface TriggerOutcome {
+  runIds: string[];
+  filesListed: number;
+  alreadyProcessed: number;
+  parked: number;
+  inFlight: number;
+}
+
+/**
+ * POST /api/v1/policies/{id}/trigger: run the pipeline now against its configured
+ * sources, regardless of the enabled flag. Returns the runs started plus what the
+ * sweep skipped; poll {@link fetchRun} for each run id.
+ */
+export async function triggerPipeline(id: string): Promise<TriggerOutcome> {
+  return apiClient.local.json<TriggerOutcome>(
     `/api/v1/policies/${encodeURIComponent(id)}/trigger`,
     { method: "POST" },
   );

frontend/editor/src/portal/components/pipelines/outputModes.ts

@@ -0,0 +1,11 @@
+import type { PipelineOutputMode } from "@portal/api/pipelines";
+
+/**
+ * The output destinations the pipeline builder offers. An extension point:
+ * deployments where a destination cannot work shadow this module and filter
+ * the list (e.g. hosted deployments never write to the server's filesystem,
+ * so folder outputs are not offered there).
+ */
+export function availableOutputModes(): PipelineOutputMode[] {
+  return ["inline", "folder", "s3"];
+}

frontend/editor/src/portal/components/sources/ConnectWizard.test.tsx

@@ -74,6 +74,53 @@ describe("ConnectWizard", () => {
     });
   });
 
+  it("creates an s3 source with a masked secret in review", async () => {
+    createSource.mockResolvedValue({ id: "src-2" });
+
+    renderWithMantine(
+      <ConnectWizard open onClose={vi.fn()} onCreated={vi.fn()} />,
+    );
+
+    // Step 0: pick the S3 type card.
+    fireEvent.click(screen.getByText("portal.sources.types.s3.label"));
+    fireEvent.click(screen.getByText("portal.sources.wizard.continue"));
+
+    // Step 1: name, bucket, credentials. The secret renders as a password
+    // input, so it is not part of the textbox roles.
+    const inputs = screen.getAllByRole("textbox") as HTMLInputElement[];
+    fireEvent.change(inputs[0], { target: { value: "Claims bucket" } });
+    fireEvent.change(inputs[1], { target: { value: "claims-inbox" } });
+    fireEvent.change(inputs[4], { target: { value: "AKIAEXAMPLE" } });
+    const secret = document.querySelector(
+      'input[type="password"]',
+    ) as HTMLInputElement;
+    fireEvent.change(secret, { target: { value: "shh-secret" } });
+    fireEvent.click(screen.getByText("portal.sources.wizard.continue"));
+
+    // Step 2: the secret is masked in review, never echoed.
+    expect(screen.queryByText("shh-secret")).not.toBeInTheDocument();
+    expect(screen.getByText("********")).toBeInTheDocument();
+
+    fireEvent.click(screen.getByText("portal.sources.actions.connectSource"));
+    await waitFor(() => {
+      expect(createSource).toHaveBeenCalledTimes(1);
+    });
+    expect(createSource).toHaveBeenCalledWith({
+      name: "Claims bucket",
+      type: "s3",
+      options: {
+        bucket: "claims-inbox",
+        region: "us-east-1",
+        prefix: "",
+        accessKeyId: "AKIAEXAMPLE",
+        secretAccessKey: "shh-secret",
+        endpoint: "",
+        mode: "consume",
+      },
+      enabled: true,
+    });
+  });
+
   it("edits an existing source: prefilled, skips type, submits with its id", async () => {
     createSource.mockResolvedValue({ id: "s1" });
     const onCreated = vi.fn();

frontend/editor/src/portal/components/sources/ConnectWizard.tsx

@@ -11,15 +11,16 @@ import {
 } from "@app/ui";
 import { errorMessage } from "@portal/api/http";
 import { createSource, type Source } from "@portal/api/sources";
+import { creatableSourceTypes } from "@portal/components/sources/creatableSourceTypes";
 import {
-  CREATABLE_SOURCE_TYPES,
   defaultOptions,
   sourceTypeMeta,
   type CreatableSourceType,
 } from "@portal/components/sources/sourceTypes";
 import "@portal/views/Sources.css";
 
-const DEFAULT_TYPE = CREATABLE_SOURCE_TYPES[0];
+const OFFERED_TYPES = creatableSourceTypes();
+const DEFAULT_TYPE = OFFERED_TYPES[0];
 
 /** Wizard steps. Editing skips type selection (the type is fixed once created). */
 type StepId = "type" | "configure" | "review";
@@ -35,9 +36,9 @@ interface ConnectWizardProps {
   source?: Source;
 }
 
-/** The creatable-type metadata for a source's stored type, falling back to folder. */
+/** The creatable-type metadata for a source's stored type, falling back to the first offered. */
 function typeFor(type: string | undefined): CreatableSourceType {
-  return CREATABLE_SOURCE_TYPES.find((t) => t.type === type) ?? DEFAULT_TYPE;
+  return OFFERED_TYPES.find((t) => t.type === type) ?? DEFAULT_TYPE;
 }
 
 /** Source options coerced to strings for the form, defaulted from the type's fields. */
@@ -199,7 +200,7 @@ export function ConnectWizard({
 
       {stepId === "type" && (
         <div className="portal-sources__type-grid">
-          {CREATABLE_SOURCE_TYPES.map((ct) => (
+          {OFFERED_TYPES.map((ct) => (
             <Button
               key={ct.type}
               variant="tertiary"
@@ -251,6 +252,7 @@ export function ConnectWizard({
                 />
               ) : (
                 <Input
+                  type={field.control === "password" ? "password" : undefined}
                   value={options[field.key] ?? ""}
                   placeholder={
                     field.placeholderKey ? t(field.placeholderKey) : undefined
@@ -280,7 +282,11 @@ export function ConnectWizard({
               <StatTile
                 key={field.key}
                 label={t(field.labelKey)}
-                value={options[field.key] || "—"}
+                value={
+                  field.control === "password" && options[field.key]
+                    ? "********"
+                    : options[field.key] || "—"
+                }
               />
             ))}
           </div>

frontend/editor/src/portal/components/sources/creatableSourceTypes.ts

@@ -0,0 +1,14 @@
+import {
+  CREATABLE_SOURCE_TYPES,
+  type CreatableSourceType,
+} from "@portal/components/sources/sourceTypes";
+
+/**
+ * The source types the connect wizard offers. An extension point: deployments
+ * where a type cannot work shadow this module and filter the list (e.g. hosted
+ * deployments never read the server's filesystem, so folder sources are not
+ * offered there).
+ */
+export function creatableSourceTypes(): CreatableSourceType[] {
+  return CREATABLE_SOURCE_TYPES;
+}

frontend/editor/src/portal/components/sources/sourceTypes.ts

@@ -26,6 +26,11 @@ const SOURCE_TYPE_META: Record<string, SourceTypeMeta> = {
     icon: "✏",
     accent: "success",
   },
+  s3: {
+    labelKey: "portal.sources.types.s3.label",
+    icon: "☁",
+    accent: "brand",
+  },
 };
 
 const UNKNOWN_TYPE_META: SourceTypeMeta = {
@@ -42,7 +47,7 @@ export function sourceTypeMeta(type: string): SourceTypeMeta {
 export interface SourceFieldDef {
   key: string;
   labelKey: string;
-  control: "text" | "select";
+  control: "text" | "password" | "select";
   required?: boolean;
   placeholderKey?: string;
   helperTextKey?: string;
@@ -130,6 +135,70 @@ export const CREATABLE_SOURCE_TYPES: CreatableSourceType[] = [
       },
     ],
   },
+  {
+    type: "s3",
+    labelKey: "portal.sources.types.s3.label",
+    descriptionKey: "portal.sources.types.s3.description",
+    fields: [
+      {
+        key: "bucket",
+        labelKey: "portal.sources.types.s3.fields.bucket.label",
+        control: "text",
+        required: true,
+        placeholderKey: "portal.sources.types.s3.fields.bucket.placeholder",
+      },
+      {
+        key: "region",
+        labelKey: "portal.sources.types.s3.fields.region.label",
+        control: "text",
+        defaultValue: "us-east-1",
+        placeholderKey: "portal.sources.types.s3.fields.region.placeholder",
+      },
+      {
+        key: "prefix",
+        labelKey: "portal.sources.types.s3.fields.prefix.label",
+        control: "text",
+        placeholderKey: "portal.sources.types.s3.fields.prefix.placeholder",
+        helperTextKey: "portal.sources.types.s3.fields.prefix.helperText",
+      },
+      {
+        key: "accessKeyId",
+        labelKey: "portal.sources.types.s3.fields.accessKeyId.label",
+        control: "text",
+        required: true,
+      },
+      {
+        key: "secretAccessKey",
+        labelKey: "portal.sources.types.s3.fields.secretAccessKey.label",
+        control: "password",
+        required: true,
+      },
+      {
+        key: "endpoint",
+        labelKey: "portal.sources.types.s3.fields.endpoint.label",
+        control: "text",
+        placeholderKey: "portal.sources.types.s3.fields.endpoint.placeholder",
+        helperTextKey: "portal.sources.types.s3.fields.endpoint.helperText",
+      },
+      {
+        key: "mode",
+        labelKey: "portal.sources.types.s3.fields.mode.label",
+        control: "select",
+        defaultValue: "consume",
+        helperTextKey: "portal.sources.types.s3.fields.mode.helperText",
+        options: [
+          {
+            value: "consume",
+            labelKey: "portal.sources.types.s3.fields.mode.options.consume",
+          },
+          {
+            value: "snapshot",
+            labelKey: "portal.sources.types.s3.fields.mode.options.snapshot",
+          },
+        ],
+      },
+    ],
+  },
 ];
 
 /** Default option values for a type's create form. */

frontend/editor/src/portal/mocks/handlers/pipelines.ts

@@ -165,7 +165,13 @@ export const pipelinesHandlers = [
   http.post("/api/v1/policies/:id/trigger", async ({ params }) => {
     if (!store.some((p) => p.id === params.id)) return undefined;
     await delay(120);
-    return HttpResponse.json([`run_${Date.now().toString(36)}`]);
+    return HttpResponse.json({
+      runIds: [`run_${Date.now().toString(36)}`],
+      filesListed: 1,
+      alreadyProcessed: 0,
+      parked: 0,
+      inFlight: 0,
+    });
   }),
 
   // Raw policy by id. Only our pipeline ids are served here; everything else falls

frontend/editor/src/portal/views/PipelineBuilder.css

@@ -392,3 +392,27 @@
   min-width: 1.5rem;
   min-height: 1.5rem;
 }
+
+.portal-builder__s3-output {
+  display: flex;
+  align-items: center;
+  gap: 0.625rem;
+}
+
+.portal-builder__s3-summary {
+  font-size: 0.8125rem;
+  color: var(--color-text-1);
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+
+.portal-builder__s3-summary.is-unset {
+  color: var(--color-text-4);
+}
+
+.portal-builder__s3-fields {
+  display: flex;
+  flex-direction: column;
+  gap: 0.75rem;
+}

frontend/editor/src/portal/views/PipelineBuilder.test.tsx

@@ -7,7 +7,7 @@ import {
 } from "@testing-library/react";
 import { MantineProvider } from "@mantine/core";
 import { MemoryRouter, Route, Routes } from "react-router-dom";
-import type { Policy } from "@portal/api/pipelines";
+import type { Policy, TriggerOutcome } from "@portal/api/pipelines";
 import type { ToolRegistryCatalog } from "@app/contexts/ToolRegistryContext";
 import type { ToolRegistryEntry } from "@app/data/toolsTaxonomy";
 import { PipelineBuilder } from "@portal/views/PipelineBuilder";
@@ -45,6 +45,11 @@ vi.mock("@portal/api/sources", () => ({
   fetchSources: () => fetchSources(),
 }));
 
+const clearProcessedHistory = vi.fn();
+vi.mock("@portal/api/policies", () => ({
+  clearProcessedHistory: (id: string) => clearProcessedHistory(id),
+}));
+
 // One editable tool, Compress, so the picker and step settings have something to render.
 vi.mock("@app/contexts/ToolRegistryContext", () => {
   const compress = {
@@ -100,6 +105,17 @@ const POLICY: Policy = {
   output: { type: "inline", options: {} },
 };
 
+function outcome(overrides: Partial<TriggerOutcome>): TriggerOutcome {
+  return {
+    runIds: [],
+    filesListed: 0,
+    alreadyProcessed: 0,
+    parked: 0,
+    inFlight: 0,
+    ...overrides,
+  };
+}
+
 function renderBuilder(initial: string) {
   return render(
     <MemoryRouter initialEntries={[initial]}>
@@ -129,8 +145,10 @@ describe("PipelineBuilder", () => {
     fetchSources.mockResolvedValue({ kpis: [], sources: [] });
     savePipeline.mockResolvedValue({});
     deletePipeline.mockResolvedValue(undefined);
-    triggerPipeline.mockResolvedValue(["run-1"]);
+    triggerPipeline.mockResolvedValue(outcome({ runIds: ["run-1"] }));
     fetchRun.mockResolvedValue({ status: "COMPLETED" });
+    clearProcessedHistory.mockReset();
+    clearProcessedHistory.mockResolvedValue(undefined);
   });
 
   it("builds a new pipeline: name it, add a tool, and save", async () => {
@@ -159,6 +177,60 @@ describe("PipelineBuilder", () => {
     expect(await screen.findByText("pipelines list")).toBeInTheDocument();
   });
 
+  it("saves an s3 output with its connection options", async () => {
+    renderBuilder("/processor/pipelines/new");
+
+    fireEvent.change(await screen.findByRole("textbox"), {
+      target: { value: "Bucket to bucket" },
+    });
+    fireEvent.click(screen.getByLabelText("portal.pipelines.output.s3"));
+
+    // With s3 selected but no bucket, saving is blocked and the summary reads
+    // unconfigured; the connection fields live behind the Configure modal.
+    expect(
+      screen.getByText("portal.pipelines.composer.create").closest("button"),
+    ).toBeDisabled();
+    expect(
+      screen.getByText("portal.pipelines.composer.s3NotConfigured"),
+    ).toBeInTheDocument();
+    fireEvent.click(screen.getByText("portal.pipelines.composer.s3Configure"));
+
+    // Textboxes: name, then the modal's bucket, region, prefix, access key id,
+    // endpoint; the secret renders as a password input outside the textbox role.
+    const inputs = screen.getAllByRole("textbox") as HTMLInputElement[];
+    fireEvent.change(inputs[1], { target: { value: "claims-processed" } });
+    fireEvent.change(inputs[3], { target: { value: "processed/" } });
+    fireEvent.change(inputs[4], { target: { value: "AKIAEXAMPLE" } });
+    const secret = document.querySelector(
+      'input[type="password"]',
+    ) as HTMLInputElement;
+    fireEvent.change(secret, { target: { value: "shh-secret" } });
+    fireEvent.click(screen.getByText("portal.pipelines.composer.s3Done"));
+
+    // The summary now shows the configured destination.
+    expect(
+      screen.getByText("s3://claims-processed/processed/"),
+    ).toBeInTheDocument();
+    fireEvent.click(screen.getByText("portal.pipelines.composer.create"));
+
+    await waitFor(() => expect(savePipeline).toHaveBeenCalledTimes(1));
+    expect(savePipeline).toHaveBeenCalledWith(
+      expect.objectContaining({
+        output: {
+          type: "s3",
+          options: {
+            bucket: "claims-processed",
+            region: "us-east-1",
+            prefix: "processed/",
+            endpoint: "",
+            accessKeyId: "AKIAEXAMPLE",
+            secretAccessKey: "shh-secret",
+          },
+        },
+      }),
+    );
+  });
+
   it("runs an existing pipeline and reports success", async () => {
     renderBuilder("/processor/pipelines/plc-1");
 
@@ -170,6 +242,45 @@ describe("PipelineBuilder", () => {
     ).toBeInTheDocument();
   });
 
+  it("explains an empty trigger when files are parked by a failed run", async () => {
+    triggerPipeline.mockResolvedValue(outcome({ filesListed: 2, parked: 2 }));
+    renderBuilder("/processor/pipelines/plc-1");
+
+    fireEvent.click(await screen.findByText("portal.pipelines.detail.run"));
+
+    expect(
+      await screen.findByText("portal.pipelines.run.parked"),
+    ).toBeInTheDocument();
+  });
+
+  it("explains an empty trigger when everything is already processed", async () => {
+    triggerPipeline.mockResolvedValue(
+      outcome({ filesListed: 3, alreadyProcessed: 3 }),
+    );
+    renderBuilder("/processor/pipelines/plc-1");
+
+    fireEvent.click(await screen.findByText("portal.pipelines.detail.run"));
+
+    expect(
+      await screen.findByText("portal.pipelines.run.allProcessed"),
+    ).toBeInTheDocument();
+  });
+
+  it("clears processed history from the header and confirms", async () => {
+    renderBuilder("/processor/pipelines/plc-1");
+
+    fireEvent.click(
+      await screen.findByText("portal.pipelines.detail.clearHistory"),
+    );
+
+    await waitFor(() =>
+      expect(clearProcessedHistory).toHaveBeenCalledWith("plc-1"),
+    );
+    expect(
+      await screen.findByText("portal.pipelines.run.historyCleared"),
+    ).toBeInTheDocument();
+  });
+
   it("blocks saving a step that needs an uploaded file", async () => {
     renderBuilder("/processor/pipelines/new");
 

frontend/editor/src/portal/views/PipelineBuilder.tsx

@@ -5,6 +5,7 @@ import ArrowBackRoundedIcon from "@mui/icons-material/ArrowBackRounded";
 import KeyboardArrowUpRoundedIcon from "@mui/icons-material/KeyboardArrowUpRounded";
 import KeyboardArrowDownRoundedIcon from "@mui/icons-material/KeyboardArrowDownRounded";
 import DeleteOutlineRoundedIcon from "@mui/icons-material/DeleteOutlineRounded";
+import HistoryRoundedIcon from "@mui/icons-material/HistoryRounded";
 import AddRoundedIcon from "@mui/icons-material/AddRounded";
 import PlayArrowRoundedIcon from "@mui/icons-material/PlayArrowRounded";
 import {
@@ -42,9 +43,13 @@ import {
   type OutputSpec,
   type Policy,
   type PolicyRunView,
+  type PipelineOutputMode,
   type TriggerConfig,
   type TriggerInfo,
+  type TriggerOutcome,
 } from "@portal/api/pipelines";
+import { clearProcessedHistory } from "@portal/api/policies";
+import { availableOutputModes } from "@portal/components/pipelines/outputModes";
 import { fetchSources, type SourceView } from "@portal/api/sources";
 import { useAsync } from "@portal/hooks/useAsync";
 import { VIEW_PATHS, toPortalPath } from "@portal/contexts/ViewContext";
@@ -53,7 +58,29 @@ import { PipelineStepSettings } from "@portal/components/pipelines/PipelineStepS
 import { ToolPicker } from "@portal/components/pipelines/ToolPicker";
 import "@portal/views/PipelineBuilder.css";
 
-type OutputMode = "inline" | "folder";
+type OutputMode = PipelineOutputMode;
+
+/** New pipelines (and specs of unoffered types) start on the first offered destination. */
+const DEFAULT_OUTPUT_MODE = availableOutputModes()[0];
+
+/** The s3 output's connection fields, mirrored from the OutputSpec options. */
+interface S3OutputOptions {
+  bucket: string;
+  region: string;
+  prefix: string;
+  endpoint: string;
+  accessKeyId: string;
+  secretAccessKey: string;
+}
+
+const EMPTY_S3_OUTPUT: S3OutputOptions = {
+  bucket: "",
+  region: "us-east-1",
+  prefix: "",
+  endpoint: "",
+  accessKeyId: "",
+  secretAccessKey: "",
+};
 type ScheduleUnit = "MINUTES" | "HOURS" | "DAYS";
 
 const SCHEDULE_UNITS: ScheduleUnit[] = ["MINUTES", "HOURS", "DAYS"];
@@ -95,14 +122,32 @@ function parseTrigger(trigger: TriggerConfig | null): {
 function parseOutput(output: OutputSpec | undefined): {
   mode: OutputMode;
   directory: string;
+  s3: S3OutputOptions;
 } {
   if (output?.type === "folder") {
     return {
       mode: "folder",
       directory: String(output.options?.directory ?? ""),
+      s3: EMPTY_S3_OUTPUT,
     };
   }
-  return { mode: "inline", directory: "" };
+  if (output?.type === "s3") {
+    const option = (key: keyof S3OutputOptions, fallback = "") =>
+      String(output.options?.[key] ?? fallback);
+    return {
+      mode: "s3",
+      directory: "",
+      s3: {
+        bucket: option("bucket"),
+        region: option("region", "us-east-1"),
+        prefix: option("prefix"),
+        endpoint: option("endpoint"),
+        accessKeyId: option("accessKeyId"),
+        secretAccessKey: option("secretAccessKey"),
+      },
+    };
+  }
+  return { mode: DEFAULT_OUTPUT_MODE, directory: "", s3: EMPTY_S3_OUTPUT };
 }
 
 /**
@@ -149,12 +194,15 @@ export function PipelineBuilder() {
   const [triggerType, setTriggerType] = useState<string>(MANUAL);
   const [scheduleCount, setScheduleCount] = useState("1");
   const [scheduleUnit, setScheduleUnit] = useState<ScheduleUnit>("HOURS");
-  const [outputMode, setOutputMode] = useState<OutputMode>("inline");
+  const [outputMode, setOutputMode] = useState<OutputMode>(DEFAULT_OUTPUT_MODE);
   const [outputDirectory, setOutputDirectory] = useState("");
+  const [outputS3, setOutputS3] = useState<S3OutputOptions>(EMPTY_S3_OUTPUT);
+  const [s3ConfigOpen, setS3ConfigOpen] = useState(false);
   const [submitting, setSubmitting] = useState(false);
   const [error, setError] = useState<string | null>(null);
   const [seeded, setSeeded] = useState(false);
   const [running, setRunning] = useState(false);
+  const [clearingHistory, setClearingHistory] = useState(false);
   const [runResult, setRunResult] = useState<RunResult | null>(null);
   const [pendingDelete, setPendingDelete] = useState(false);
   const [deleting, setDeleting] = useState(false);
@@ -186,6 +234,7 @@ export function PipelineBuilder() {
     setScheduleUnit(trigger.unit);
     setOutputMode(output.mode);
     setOutputDirectory(output.directory);
+    setOutputS3(output.s3);
     setSeeded(true);
   }, [isEdit, policyState.data, allTools, seeded]);
 
@@ -222,6 +271,10 @@ export function PipelineBuilder() {
     if (selected && !triggerAvailable(selected)) setTriggerType(MANUAL);
   }, [triggerType, triggers, triggerAvailable]);
 
+  function setS3Field(key: keyof S3OutputOptions, value: string) {
+    setOutputS3((current) => ({ ...current, [key]: value }));
+  }
+
   function toggleSource(sourceId: string, checked: boolean) {
     setSourceIds((ids) =>
       checked
@@ -286,6 +339,7 @@ export function PipelineBuilder() {
     scheduleUnit,
     outputMode,
     outputDirectory,
+    outputS3,
   });
   const baseline = useRef<string | null>(null);
   useEffect(() => {
@@ -295,7 +349,13 @@ export function PipelineBuilder() {
 
   const scheduleCountValid =
     triggerType !== "schedule" || Number(scheduleCount) > 0;
-  const outputValid = outputMode !== "folder" || outputDirectory.trim() !== "";
+  const s3OutputValid =
+    outputMode !== "s3" ||
+    (outputS3.bucket.trim() !== "" &&
+      outputS3.accessKeyId.trim() !== "" &&
+      outputS3.secretAccessKey.trim() !== "");
+  const outputValid =
+    (outputMode !== "folder" || outputDirectory.trim() !== "") && s3OutputValid;
   const canSave =
     name.trim() !== "" &&
     scheduleCountValid &&
@@ -357,7 +417,9 @@ export function PipelineBuilder() {
     const output: OutputSpec =
       outputMode === "folder"
         ? { type: "folder", options: { directory: outputDirectory.trim() } }
-        : { type: "inline", options: {} };
+        : outputMode === "s3"
+          ? { type: "s3", options: { ...outputS3 } }
+          : { type: "inline", options: {} };
     const policy: Policy = {
       id: policyState.data?.id ?? undefined,
       name: name.trim(),
@@ -387,15 +449,37 @@ export function PipelineBuilder() {
     return null;
   }
 
+  /** Explain an empty trigger: parked files outrank blander reasons. */
+  function emptySweepResult(outcome: TriggerOutcome): RunResult {
+    if (outcome.parked > 0) {
+      return {
+        tone: "warning",
+        text: t("portal.pipelines.run.parked", { count: outcome.parked }),
+      };
+    }
+    if (outcome.inFlight > 0) {
+      return { tone: "info", text: t("portal.pipelines.run.inFlight") };
+    }
+    if (outcome.alreadyProcessed > 0) {
+      return {
+        tone: "info",
+        text: t("portal.pipelines.run.allProcessed", {
+          count: outcome.alreadyProcessed,
+        }),
+      };
+    }
+    return { tone: "info", text: t("portal.pipelines.run.empty") };
+  }
+
   async function handleRun() {
     if (running || !id) return;
     setRunning(true);
     setRunResult(null);
     try {
-      const runIds = await triggerPipeline(id);
+      const outcome = await triggerPipeline(id);
+      const runIds = outcome.runIds;
       if (runIds.length === 0) {
-        if (mounted.current)
-          setRunResult({ tone: "info", text: t("portal.pipelines.run.empty") });
+        if (mounted.current) setRunResult(emptySweepResult(outcome));
         return;
       }
       const finals = await Promise.all(runIds.map((runId) => awaitRun(runId)));
@@ -428,6 +512,30 @@ export function PipelineBuilder() {
     }
   }
 
+  /**
+   * Forget which source files this pipeline has processed, so the next sweep
+   * reprocesses everything currently in its sources (the standard retry for a
+   * parked-by-failure file). Does not touch the files themselves.
+   */
+  async function handleClearHistory() {
+    if (clearingHistory || !id) return;
+    setClearingHistory(true);
+    setRunResult(null);
+    try {
+      await clearProcessedHistory(id);
+      if (mounted.current)
+        setRunResult({
+          tone: "success",
+          text: t("portal.pipelines.run.historyCleared"),
+        });
+    } catch (e) {
+      if (mounted.current)
+        setRunResult({ tone: "danger", text: errorMessage(e) });
+    } finally {
+      if (mounted.current) setClearingHistory(false);
+    }
+  }
+
   async function confirmDelete() {
     if (!id || deleting) return;
     setDeleting(true);
@@ -494,6 +602,17 @@ export function PipelineBuilder() {
               >
                 {t("portal.pipelines.detail.run")}
               </Button>
+              <Button
+                variant="secondary"
+                size="sm"
+                loading={clearingHistory}
+                onClick={handleClearHistory}
+                leftSection={
+                  <HistoryRoundedIcon style={{ fontSize: "1.125rem" }} />
+                }
+              >
+                {t("portal.pipelines.detail.clearHistory")}
+              </Button>
               <Button
                 variant="secondary"
                 size="sm"
@@ -630,10 +749,10 @@ export function PipelineBuilder() {
               name="pipeline-output"
               value={outputMode}
               onChange={setOutputMode}
-              options={[
-                { value: "inline", label: t("portal.pipelines.output.inline") },
-                { value: "folder", label: t("portal.pipelines.output.folder") },
-              ]}
+              options={availableOutputModes().map((mode) => ({
+                value: mode,
+                label: t(`portal.pipelines.output.${mode}`),
+              }))}
             />
             {outputMode === "folder" && (
               <FormField
@@ -648,6 +767,27 @@ export function PipelineBuilder() {
                 />
               </FormField>
             )}
+            {outputMode === "s3" && (
+              <div className="portal-builder__s3-output">
+                <span
+                  className={
+                    "portal-builder__s3-summary" +
+                    (outputS3.bucket ? "" : " is-unset")
+                  }
+                >
+                  {outputS3.bucket
+                    ? `s3://${outputS3.bucket}/${outputS3.prefix}`
+                    : t("portal.pipelines.composer.s3NotConfigured")}
+                </span>
+                <Button
+                  variant="secondary"
+                  size="sm"
+                  onClick={() => setS3ConfigOpen(true)}
+                >
+                  {t("portal.pipelines.composer.s3Configure")}
+                </Button>
+              </div>
+            )}
           </div>
         </div>
       </section>
@@ -860,6 +1000,78 @@ export function PipelineBuilder() {
       >
         <p>{t("portal.pipelines.builder.unsavedBody")}</p>
       </Modal>
+
+      <Modal
+        open={s3ConfigOpen}
+        onClose={() => setS3ConfigOpen(false)}
+        title={t("portal.pipelines.composer.s3ModalTitle")}
+        footer={
+          <div className="portal-pipelines__composer-footer">
+            <Button size="sm" onClick={() => setS3ConfigOpen(false)}>
+              {t("portal.pipelines.composer.s3Done")}
+            </Button>
+          </div>
+        }
+      >
+        <div className="portal-builder__s3-fields">
+          <FormField
+            label={t("portal.sources.types.s3.fields.bucket.label")}
+            required
+          >
+            <Input
+              value={outputS3.bucket}
+              placeholder="my-company-inbox"
+              onChange={(e) => setS3Field("bucket", e.target.value)}
+            />
+          </FormField>
+          <FormField label={t("portal.sources.types.s3.fields.region.label")}>
+            <Input
+              value={outputS3.region}
+              placeholder="us-east-1"
+              onChange={(e) => setS3Field("region", e.target.value)}
+            />
+          </FormField>
+          <FormField
+            label={t("portal.sources.types.s3.fields.prefix.label")}
+            helperText={t("portal.pipelines.composer.s3PrefixHelp")}
+          >
+            <Input
+              value={outputS3.prefix}
+              placeholder="processed/"
+              onChange={(e) => setS3Field("prefix", e.target.value)}
+            />
+          </FormField>
+          <FormField
+            label={t("portal.sources.types.s3.fields.accessKeyId.label")}
+            required
+          >
+            <Input
+              value={outputS3.accessKeyId}
+              onChange={(e) => setS3Field("accessKeyId", e.target.value)}
+            />
+          </FormField>
+          <FormField
+            label={t("portal.sources.types.s3.fields.secretAccessKey.label")}
+            required
+          >
+            <Input
+              type="password"
+              value={outputS3.secretAccessKey}
+              onChange={(e) => setS3Field("secretAccessKey", e.target.value)}
+            />
+          </FormField>
+          <FormField
+            label={t("portal.sources.types.s3.fields.endpoint.label")}
+            helperText={t("portal.sources.types.s3.fields.endpoint.helperText")}
+          >
+            <Input
+              value={outputS3.endpoint}
+              placeholder="https://s3.example.com"
+              onChange={(e) => setS3Field("endpoint", e.target.value)}
+            />
+          </FormField>
+        </div>
+      </Modal>
     </div>
   );
 }