Improve logic for tracking which files have already been processed in policies (#6903)

# Description of Changes
Replaces the `.stirling/done` folder and its friends with a ledger in
the DB which tracks which documents have been processed. This should
scale dramatically better since it's just a few bytes being written for
each PDF processed, rather than each PDF being duplicated and held in
the folder forever. It's designed to work with the current folder
source, but also with S3 buckets and other sources in mind - each source
will define its own strategy for ensuring it knows whether the documents
have had policies run on them or not, and they all get written to the
same ledger.

AGENTS.md

@@ -453,6 +453,7 @@ The frontend is organized with a clear separation of concerns:
 
 - **CRITICAL**: Always update translations in `en-US` only - all other languages (including `en-GB`) are handled separately
 - Translation files are located in `frontend/editor/public/locales/`
+- After changing any translation file, run `task pre-commit:fix`
 
 ## Important Notes
 

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

@@ -48,6 +48,7 @@ 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.ledger.ProcessedLedger;
 import stirling.software.proprietary.policy.model.PipelineDefinition;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.model.PolicyInputs;
@@ -87,6 +88,7 @@ public class PolicyController {
     private final PolicyManagementAuthority policyManagementAuthority;
     private final PolicyTriggerManager policyTriggerManager;
     private final PolicyOverviewService policyOverviewService;
+    private final ProcessedLedger processedLedger;
     private final List<PolicyTrigger> policyTriggers;
     private final ApplicationProperties applicationProperties;
     private final TempFileManager tempFileManager;
@@ -352,6 +354,7 @@ public class PolicyController {
         boolean accessible =
                 policyStore.get(policyId).filter(policyAccessGuard::canAccess).isPresent();
         if (accessible && policyStore.delete(policyId)) {
+            processedLedger.clearPolicy(policyId);
             // Cancel any now-orphaned folder watch promptly rather than leaving the WatchKey open
             // until the next reconcile sweep.
             policyTriggerManager.notifyPoliciesChanged();
@@ -360,6 +363,25 @@ public class PolicyController {
         return ResponseEntity.notFound().build();
     }
 
+    @DeleteMapping("/{policyId}/processed-history")
+    @Operation(
+            summary = "Clear a policy's processed-file history",
+            description =
+                    "Forgets which source files this policy has already processed, so its next"
+                            + " sweep reprocesses everything currently in its sources. Does not"
+                            + " touch the files themselves.")
+    public ResponseEntity<Void> clearProcessedHistory(@PathVariable String policyId) {
+        requirePolicyEditingAllowed();
+        // Scope to the caller's team: a policy in another team reads as not-found.
+        boolean accessible =
+                policyStore.get(policyId).filter(policyAccessGuard::canAccess).isPresent();
+        if (!accessible) {
+            return ResponseEntity.notFound().build();
+        }
+        processedLedger.clearPolicy(policyId);
+        return ResponseEntity.noContent().build();
+    }
+
     @PostMapping(value = "/{policyId}/run", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
     @Operation(
             summary = "Run a stored policy",

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

@@ -35,6 +35,7 @@ import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.model.PolicyInputs;
 import stirling.software.proprietary.policy.model.PolicyRun;
 import stirling.software.proprietary.policy.model.WaitState;
+import stirling.software.proprietary.policy.output.OutputDelivery;
 import stirling.software.proprietary.policy.output.PolicyOutputSink;
 import stirling.software.proprietary.policy.progress.PolicyProgressListener;
 import stirling.software.proprietary.service.DownstreamEntitlementError;
@@ -203,7 +204,12 @@ public class PolicyEngine {
             PolicyExecutionResult result =
                     stepExecutor.execute(run.getDefinition(), inputs, listener);
             OutputSpec output = run.getDefinition().output();
-            List<ResultFile> outputs = sinkFor(output).deliver(runId, result.files(), output);
+            List<ResultFile> outputs =
+                    sinkFor(output)
+                            .deliver(
+                                    new OutputDelivery(runId, run.getPolicyId()),
+                                    result.files(),
+                                    output);
             taskManager.setMultipleFileResults(runId, outputs);
             taskManager.setComplete(runId);
             run.complete(outputs);

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

@@ -13,6 +13,7 @@ import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.proprietary.policy.input.InputSource;
 import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.ledger.ProcessedLedger;
 import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.PipelineDefinition;
 import stirling.software.proprietary.policy.model.Policy;
@@ -27,7 +28,8 @@ import stirling.software.proprietary.policy.source.SourceStore;
 /**
  * Turns a policy's referenced sources into runs: each {@code sourceId} is resolved live to its
  * persisted {@link Source}, then to an {@link InputSpec}. Triggers decide <em>when</em> and call
- * {@link #run(Policy)}; the controller uses the supplied-input and ad-hoc entry points.
+ * {@link #run(Policy)}; the controller uses the supplied-input and ad-hoc entry points. A {@link
+ * SweepKind#FULL} sweep also reconciles the processed-file ledger against what is present.
  */
 @Slf4j
 @Service
@@ -39,6 +41,12 @@ public class PolicyRunner {
     private final List<InputSource> inputSources;
     private final SourceStore sourceStore;
     private final SourceDocCounter docCounter;
+    private final ProcessedLedger processedLedger;
+
+    /** Full-listing sweep: resolve every source, then reconcile the ledger. */
+    public List<String> run(Policy policy) {
+        return run(policy, SweepKind.FULL);
+    }
 
     /**
      * Trigger entry point. Pulls every referenced source; each yielded unit becomes its own run so
@@ -47,15 +55,21 @@ public class PolicyRunner {
      * 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.
      */
-    public List<String> run(Policy policy) {
+    public List<String> run(Policy policy, SweepKind sweep) {
+        long sweepStart = System.currentTimeMillis();
+        PolicySweep context = new PolicySweep(policy.id(), sweep, processedLedger);
+        List<String> runIds = new ArrayList<>();
         List<String> sourceIds = policy.sourceIds();
         if (sourceIds.isEmpty()) {
-            return List.of(startRun(policy, PolicyInputs.of(List.of()), unused -> {}));
+            // Generator pipeline: one run with no input. Still fall through to the cleanup
+            // below so rows recorded for its folder outputs are pruned like anything else,
+            // instead of accumulating until the policy is deleted.
+            runIds.add(startRun(policy, PolicyInputs.of(List.of()), unused -> {}));
         }
-        List<String> runIds = new ArrayList<>();
         for (String sourceId : sourceIds) {
             Source source = sourceStore.get(sourceId).orElse(null);
             if (source == null) {
+                // No veto: a deleted source's rows should age out via the cleanup below.
                 log.warn("Policy {} references missing source {}; skipping", policy.id(), sourceId);
                 continue;
             }
@@ -65,9 +79,21 @@ public class PolicyRunner {
                         sourceId,
                         source.name(),
                         policy.id());
+                // Veto: a paused source's files cannot be stamped, so they must not be pruned.
+                context.vetoCleanup();
                 continue;
             }
-            runIds.addAll(pullAndRun(policy, sourceId, source.toInputSpec()));
+            runIds.addAll(pullAndRun(policy, sourceId, source.toInputSpec(), context));
+        }
+        if (context.cleanupAllowed()) {
+            processedLedger.markSeen(policy.id(), context.presentIdentities());
+            int removed = processedLedger.deleteUnseen(policy.id(), sweepStart);
+            if (removed > 0) {
+                log.debug(
+                        "Pruned {} ledger row(s) for files no longer present (policy {})",
+                        removed,
+                        policy.id());
+            }
         }
         return runIds;
     }
@@ -86,26 +112,33 @@ public class PolicyRunner {
 
     /**
      * Resolves the source and starts a run per unit; records how many documents the source fed and
-     * returns the ids of the runs started.
+     * returns the ids of the runs started. Any source that could not be listed completely vetoes
+     * this sweep's ledger cleanup.
      */
-    private List<String> pullAndRun(Policy policy, String sourceId, InputSpec spec) {
+    private List<String> pullAndRun(
+            Policy policy, String sourceId, InputSpec spec, PolicySweep context) {
         InputSource source = sourceFor(spec);
         if (source == null) {
             log.warn(
                     "No input source for type '{}' (policy {}); skipping",
                     spec.type(),
                     policy.id());
+            context.vetoCleanup();
             return List.of();
         }
+        if (!source.listsExhaustively()) {
+            context.vetoCleanup();
+        }
         List<ResolvedInput> work;
         try {
-            work = source.resolve(spec);
+            work = source.resolve(spec, context);
         } catch (IOException | RuntimeException e) {
             log.warn(
                     "Failed to resolve source '{}' for policy {}: {}",
                     spec.type(),
                     policy.id(),
                     e.getMessage());
+            context.vetoCleanup();
             return List.of();
         }
         List<String> runIds = new ArrayList<>();

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

@@ -0,0 +1,89 @@
+package stirling.software.proprietary.policy.engine;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
+
+import stirling.software.proprietary.policy.input.ResolveContext;
+import stirling.software.proprietary.policy.ledger.ClaimState;
+import stirling.software.proprietary.policy.ledger.ProcessedFileStatus;
+import stirling.software.proprietary.policy.ledger.ProcessedLedger;
+
+/**
+ * The {@link ResolveContext} for one policy sweep: scopes ledger calls to the policy, gathers the
+ * present-identity union across sources, prefetches claim state in bulk so per-file claims skip
+ * their row lookup, and vetoes presence cleanup when any source could not be listed completely
+ * (pruning would wrongly forget its files).
+ */
+final class PolicySweep implements ResolveContext {
+
+    private final String policyId;
+    private final SweepKind kind;
+    private final ProcessedLedger ledger;
+    private final Set<String> present = new HashSet<>();
+    // Claim states loaded in bulk at reportPresent; a claim outside the prefetch falls back to a
+    // single lookup. A stale entry cannot double-claim (the ledger re-checks every transition),
+    // it can only defer a file to the next sweep.
+    private final Map<String, ClaimState> prefetched = new HashMap<>();
+    private final Set<String> prefetchedIdentities = new HashSet<>();
+    private boolean cleanupVetoed;
+
+    PolicySweep(String policyId, SweepKind kind, ProcessedLedger ledger) {
+        this.policyId = policyId;
+        this.kind = kind;
+        this.ledger = ledger;
+    }
+
+    @Override
+    public synchronized boolean claim(String identity, String gate, Supplier<String> contentHash) {
+        ClaimState observed =
+                prefetchedIdentities.contains(identity)
+                        ? prefetched.get(identity)
+                        : ledger.statesFor(policyId, List.of(identity)).get(identity);
+        boolean claimed = ledger.claim(policyId, identity, gate, contentHash, observed);
+        if (claimed) {
+            // A nested source surfacing the same file later in this sweep sees it in flight
+            // without another lookup.
+            prefetchedIdentities.add(identity);
+            prefetched.put(identity, new ClaimState(ProcessedFileStatus.PROCESSING, gate, null));
+        }
+        return claimed;
+    }
+
+    @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) {
+        // Deliberately not policy-scoped: consume deletion needs every claimant's consensus.
+        return ledger.allSettledDone(identity);
+    }
+
+    @Override
+    public synchronized void reportPresent(Collection<String> identities) {
+        if (kind == SweepKind.FULL) {
+            present.addAll(identities);
+        }
+        prefetched.putAll(ledger.statesFor(policyId, identities));
+        prefetchedIdentities.addAll(identities);
+    }
+
+    synchronized void vetoCleanup() {
+        cleanupVetoed = true;
+    }
+
+    synchronized boolean cleanupAllowed() {
+        return kind == SweepKind.FULL && !cleanupVetoed;
+    }
+
+    synchronized Set<String> presentIdentities() {
+        return Set.copyOf(present);
+    }
+}

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

@@ -0,0 +1,10 @@
+package stirling.software.proprietary.policy.engine;
+
+/**
+ * How thorough a policy sweep is: {@link #FULL} (complete listing; also stamps presence and prunes
+ * the ledger) or {@link #LIGHT} (event-driven; claims only, cost proportional to what changed).
+ */
+public enum SweepKind {
+    FULL,
+    LIGHT
+}

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

@@ -1,12 +1,17 @@
 package stirling.software.proprietary.policy.input;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.FileVisitResult;
 import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
-import java.nio.file.StandardCopyOption;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
@@ -19,17 +24,20 @@ import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.util.FileReadinessChecker;
 import stirling.software.proprietary.policy.config.FolderAccessGuard;
+import stirling.software.proprietary.policy.ledger.FolderIdentities;
 import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.PolicyInputs;
 
 /**
- * Reads input files from a directory; each ready file is its own unit of work so one failure does
- * not affect the others.
- *
- * <p>Mode option: "consume" (default) claims each file by moving it into {@code
- * .stirling/processing} then routes it to {@code .stirling/done} or {@code .stirling/error}, so
- * each file runs once; "snapshot" reads without moving, so every run sees the full set. Readiness
- * is checked first so files mid-write are skipped.
+ * Reads input files from a directory; each ready file is its own unit of work, claimed through the
+ * {@link ResolveContext} ledger rather than moved aside, so nothing accumulates in a work
+ * directory. Options: "mode" is "consume" (default: a processed file is removed 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); "recursive" descends into subdirectories; "identity" is "stat" (default, any size/mtime
+ * change is a new version) or "hash" (content-verified, so a touch does not reprocess). Hidden
+ * files and directories, including the legacy {@code .stirling} work dir, are never picked up, and
+ * files mid-write are skipped by the readiness check.
  */
 @Slf4j
 @Service
@@ -38,11 +46,6 @@ import stirling.software.proprietary.policy.model.PolicyInputs;
 public class FolderInputSource implements InputSource {
 
     private static final String TYPE = FolderAccessGuard.FOLDER_TYPE;
-    // Bookkeeping lives under one hidden dir so the watched folder stays tidy.
-    private static final String WORK_SUBDIR = ".stirling";
-    private static final String PROCESSING_SUBDIR = "processing";
-    private static final String DONE_SUBDIR = "done";
-    private static final String ERROR_SUBDIR = "error";
 
     private final FileReadinessChecker readinessChecker;
     private final FolderAccessGuard accessGuard;
@@ -68,70 +71,195 @@ public class FolderInputSource implements InputSource {
     }
 
     @Override
-    public List<ResolvedInput> resolve(InputSpec spec) throws IOException {
+    public List<ResolvedInput> resolve(InputSpec spec, ResolveContext ctx) throws IOException {
         FolderConfig config = FolderConfig.from(spec.options());
         Path inputDir = accessGuard.requirePermitted(config.directory());
         if (!Files.isDirectory(inputDir)) {
-            log.debug("Folder input dir does not exist: {}", inputDir);
-            return List.of();
+            // Fail rather than return empty: an unmounted drive must read as "could not list",
+            // which vetoes the sweep's presence cleanup, not as "verifiably no files", which
+            // would wipe the policy's history and reprocess everything on remount.
+            throw new NoSuchFileException(
+                    inputDir.toString(), null, "input directory does not exist");
+        }
+        Path canonicalDir = FolderIdentities.canonicalDir(inputDir);
+        List<Path> present = listFiles(inputDir, config.recursive());
+
+        if (config.snapshot()) {
+            List<ResolvedInput> work = new ArrayList<>();
+            for (Path file : present) {
+                if (readinessChecker.isReady(file)) {
+                    work.add(ResolvedInput.of(PolicyInputs.of(List.of(fileResource(file)))));
+                }
+            }
+            return work;
         }
 
-        List<Path> ready = new ArrayList<>();
-        try (Stream<Path> entries = Files.list(inputDir)) {
-            entries.filter(Files::isRegularFile)
-                    .filter(readinessChecker::isReady)
-                    .forEach(ready::add);
-        }
+        ctx.reportPresent(
+                present.stream()
+                        .map(file -> FolderIdentities.identity(canonicalDir, inputDir, file))
+                        .toList());
 
         List<ResolvedInput> work = new ArrayList<>();
-        for (Path file : ready) {
-            if (config.snapshot()) {
-                work.add(ResolvedInput.of(PolicyInputs.of(List.of(fileResource(file)))));
-            } else {
-                Path claimed = claim(inputDir, file);
-                if (claimed == null) {
-                    continue; // another sweep/process grabbed it
-                }
-                work.add(
-                        new ResolvedInput(
-                                PolicyInputs.of(List.of(fileResource(claimed))),
-                                success -> route(inputDir, claimed, success)));
+        for (Path file : present) {
+            if (!readinessChecker.isReady(file)) {
+                continue;
             }
+            String identity = FolderIdentities.identity(canonicalDir, inputDir, file);
+            MemoizedContentHash contentHash =
+                    config.hashIdentity() ? new MemoizedContentHash(file) : null;
+            String gate;
+            boolean claimed;
+            try {
+                gate = FolderIdentities.statGate(file);
+                claimed = ctx.claim(identity, gate, contentHash);
+            } catch (IOException | UncheckedIOException e) {
+                log.debug("Could not read {} for its version: {}", file, e.getMessage());
+                continue; // vanished or unreadable mid-sweep; the next sweep sees the truth
+            }
+            if (!claimed) {
+                continue;
+            }
+            work.add(
+                    new ResolvedInput(
+                            PolicyInputs.of(List.of(fileResource(file))),
+                            success ->
+                                    completeConsumed(
+                                            ctx, identity, file, gate, contentHash, success)));
         }
         return work;
     }
 
-    // Atomic move into processing/: only one sweep can win the claim, the rest see the file gone.
-    private Path claim(Path inputDir, Path file) {
+    /**
+     * Settle at the version this run claimed - never a re-read, so a file replaced mid-run reads as
+     * a new unclaimed version next sweep instead of being marked processed. Then remove the input
+     * only when it is still the processed version (a mid-run replacement must survive) and every
+     * policy that claimed it has settled DONE, so co-watching policies all read the original and
+     * one failure parks the file for everyone. A failed run settles ERROR and never deletes; the
+     * DONE row of a file that could not be deleted still stops reprocessing.
+     */
+    private static void completeConsumed(
+            ResolveContext ctx,
+            String identity,
+            Path file,
+            String claimGate,
+            MemoizedContentHash contentHash,
+            boolean success) {
+        ctx.settle(identity, claimGate, claimedHash(file, claimGate, contentHash), success);
+        if (!success) {
+            return;
+        }
         try {
-            Path processingDir = workDir(inputDir, PROCESSING_SUBDIR);
-            Files.createDirectories(processingDir);
-            Path claimed = uniqueTarget(processingDir, file.getFileName().toString());
-            Files.move(file, claimed, StandardCopyOption.ATOMIC_MOVE);
-            return claimed;
+            if (FolderIdentities.statGate(file).equals(claimGate) && ctx.allSettledDone(identity)) {
+                Files.deleteIfExists(file);
+            }
+        } catch (NoSuchFileException alreadyGone) {
+            // Removed by the user or a co-watching policy's own consensus delete: nothing to do.
         } catch (IOException e) {
-            log.debug("Could not claim {}: {}", file, e.getMessage());
+            log.warn("Could not remove consumed input {}: {}", file, e.getMessage());
+        }
+    }
+
+    /**
+     * The claimed version's content hash: the value computed during the claim when the ledger
+     * consulted the verifier, else computed now while the file is still at the claimed gate (so the
+     * hash describes what actually ran), else null. Always null in stat mode.
+     */
+    private static String claimedHash(Path file, String claimGate, MemoizedContentHash hash) {
+        if (hash == null) {
             return null;
         }
+        String computed = hash.valueIfComputed();
+        if (computed != null) {
+            return computed;
+        }
+        try {
+            if (FolderIdentities.statGate(file).equals(claimGate)) {
+                return hash.get();
+            }
+        } catch (IOException | UncheckedIOException e) {
+            log.debug("Could not hash {} at settle: {}", file, e.getMessage());
+        }
+        return null;
     }
 
-    private void route(Path inputDir, Path claimed, boolean success) {
-        String subdir = success ? DONE_SUBDIR : ERROR_SUBDIR;
-        try {
-            Path destDir = workDir(inputDir, subdir);
-            Files.createDirectories(destDir);
-            Files.move(
-                    claimed,
-                    uniqueTarget(destDir, claimed.getFileName().toString()),
-                    StandardCopyOption.ATOMIC_MOVE);
-        } catch (IOException e) {
-            log.warn(
-                    "Could not move processed input {} to {}: {}", claimed, subdir, e.getMessage());
+    /** Lazy verification tier: invoked at most once by the ledger, retained for the settle. */
+    private static final class MemoizedContentHash implements Supplier<String> {
+
+        private final Path file;
+        private volatile String value;
+
+        private MemoizedContentHash(Path file) {
+            this.file = file;
+        }
+
+        @Override
+        public String get() {
+            if (value == null) {
+                try {
+                    value = FolderIdentities.contentHash(file);
+                } catch (IOException e) {
+                    throw new UncheckedIOException(e);
+                }
+            }
+            return value;
+        }
+
+        String valueIfComputed() {
+            return value;
         }
     }
 
-    private static Path workDir(Path inputDir, String subdir) {
-        return inputDir.resolve(WORK_SUBDIR).resolve(subdir);
+    /** Every non-hidden regular file in the source, readable or not. */
+    private static List<Path> listFiles(Path inputDir, boolean recursive) throws IOException {
+        List<Path> files = new ArrayList<>();
+        if (!recursive) {
+            try (Stream<Path> entries = Files.list(inputDir)) {
+                entries.filter(Files::isRegularFile)
+                        .filter(file -> !hidden(file))
+                        .forEach(files::add);
+            }
+            return files;
+        }
+        // Hidden subtrees are pruned wholesale; symlinked directories are not followed.
+        Files.walkFileTree(
+                inputDir,
+                new SimpleFileVisitor<>() {
+                    @Override
+                    public FileVisitResult preVisitDirectory(
+                            Path dir, BasicFileAttributes attributes) {
+                        if (!dir.equals(inputDir) && hidden(dir)) {
+                            return FileVisitResult.SKIP_SUBTREE;
+                        }
+                        return FileVisitResult.CONTINUE;
+                    }
+
+                    @Override
+                    public FileVisitResult visitFile(Path file, BasicFileAttributes attributes) {
+                        if (attributes.isRegularFile() && !hidden(file)) {
+                            files.add(file);
+                        }
+                        return FileVisitResult.CONTINUE;
+                    }
+
+                    @Override
+                    public FileVisitResult visitFileFailed(Path file, IOException e) {
+                        log.debug("Skipping unreadable entry {}: {}", file, e.getMessage());
+                        return FileVisitResult.CONTINUE;
+                    }
+                });
+        return files;
+    }
+
+    private static boolean hidden(Path path) {
+        Path name = path.getFileName();
+        if (name != null && name.toString().startsWith(".")) {
+            return true;
+        }
+        try {
+            return Files.isHidden(path);
+        } catch (IOException e) {
+            return false;
+        }
     }
 
     private static Resource fileResource(Path path) {
@@ -144,27 +272,15 @@ public class FolderInputSource implements InputSource {
         };
     }
 
-    private static Path uniqueTarget(Path dir, String filename) {
-        Path candidate = dir.resolve(filename);
-        if (!Files.exists(candidate)) {
-            return candidate;
-        }
-        int dot = filename.lastIndexOf('.');
-        String base = dot < 0 ? filename : filename.substring(0, dot);
-        String ext = dot < 0 ? "" : filename.substring(dot);
-        for (int n = 1; ; n++) {
-            Path next = dir.resolve(base + " (" + n + ")" + ext);
-            if (!Files.exists(next)) {
-                return next;
-            }
-        }
-    }
-
-    record FolderConfig(Path directory, boolean snapshot) {
+    record FolderConfig(Path directory, boolean snapshot, boolean recursive, boolean hashIdentity) {
 
         private static final String DIRECTORY_OPTION = "directory";
         private static final String MODE_OPTION = "mode";
         private static final String MODE_SNAPSHOT = "snapshot";
+        private static final String RECURSIVE_OPTION = "recursive";
+        private static final String IDENTITY_OPTION = "identity";
+        private static final String IDENTITY_STAT = "stat";
+        private static final String IDENTITY_HASH = "hash";
 
         static FolderConfig from(Map<String, Object> options) {
             Object directory = options.get(DIRECTORY_OPTION);
@@ -173,7 +289,17 @@ public class FolderInputSource implements InputSource {
             }
             Object mode = options.get(MODE_OPTION);
             boolean snapshot = mode != null && MODE_SNAPSHOT.equals(mode.toString());
-            return new FolderConfig(Path.of(directory.toString()), snapshot);
+            Object recursive = options.get(RECURSIVE_OPTION);
+            boolean recurse = recursive != null && Boolean.parseBoolean(recursive.toString());
+            Object identity = options.get(IDENTITY_OPTION);
+            boolean hash = identity != null && IDENTITY_HASH.equals(identity.toString());
+            if (identity != null
+                    && !IDENTITY_STAT.equals(identity.toString())
+                    && !IDENTITY_HASH.equals(identity.toString())) {
+                throw new IllegalArgumentException(
+                        "folder input 'identity' must be 'stat' or 'hash'");
+            }
+            return new FolderConfig(Path.of(directory.toString()), snapshot, recurse, hash);
         }
     }
 }

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

@@ -24,9 +24,21 @@ public interface InputSource {
 
     /**
      * Resolve the spec into zero or more units of work, each carrying one run's files and a
-     * completion hook. Empty list means nothing to run right now.
+     * completion hook. Empty list means nothing to run right now. Discovery is read-only - files
+     * stay where the user put them; "already processed" is tracked through {@code ctx} (claim on
+     * pickup, settle on completion, report what is present so stale ledger rows can be pruned).
      */
-    List<ResolvedInput> resolve(InputSpec spec) throws IOException;
+    List<ResolvedInput> resolve(InputSpec spec, ResolveContext ctx) throws IOException;
+
+    /**
+     * Whether {@link #resolve} observes everything in the source (a complete listing) rather than
+     * e.g. only what events surfaced. Presence cleanup of the ledger is skipped for the whole
+     * policy unless every enabled source says true - wrongly pruning history would reprocess a
+     * whole folder, while keeping a few stale rows costs nothing.
+     */
+    default boolean listsExhaustively() {
+        return true;
+    }
 
     /**
      * Filesystem dirs this source draws from, for the folder-watch trigger. Advisory: resolving is

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

@@ -0,0 +1,36 @@
+package stirling.software.proprietary.policy.input;
+
+import java.util.Collection;
+import java.util.function.Supplier;
+
+/**
+ * A source's policy-scoped window onto the processed-file ledger for one sweep. Thread-safe and
+ * valid for the lifetime of the work units the source issued ({@link #settle} fires from async run
+ * completions).
+ */
+public interface ResolveContext {
+
+    /**
+     * Atomically claim a file at its current version; true means this sweep runs it. A null {@code
+     * contentHash} makes any gate change a new version; a non-null supplier is invoked at most
+     * once, only on a gate mismatch, and a matching hash refreshes the stored gate instead of
+     * reprocessing. Supplier exceptions propagate.
+     */
+    boolean claim(String identity, String gate, Supplier<String> contentHash);
+
+    /** Record a claimed file's outcome at its final version ({@code finalContentHash} nullable). */
+    void settle(String identity, String finalGate, String finalContentHash, boolean success);
+
+    /**
+     * Whether every policy holding a ledger row for this identity has settled it DONE. Cross-policy
+     * by design: consume-mode deletion is a consensus of all claimants, so a shared input is
+     * removed only once nobody still needs it (in-flight, failed, and interrupted rows all veto).
+     */
+    boolean allSettledDone(String identity);
+
+    /**
+     * Report every identity present right now, readable or not; feeds presence cleanup of rows
+     * whose file is gone.
+     */
+    void reportPresent(Collection<String> identities);
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ClaimState.java

@@ -0,0 +1,9 @@
+package stirling.software.proprietary.policy.ledger;
+
+/**
+ * A row's claim-relevant state as read by {@link ProcessedLedger#statesFor}: what a sweep observed
+ * before deciding a claim. May be stale by the time the claim runs; every ledger transition
+ * re-checks the observed state in its WHERE clause, so staleness defers a claim to a later sweep
+ * rather than double-running one.
+ */
+public record ClaimState(ProcessedFileStatus status, String gate, String contentHash) {}

app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/FolderIdentities.java

@@ -0,0 +1,41 @@
+package stirling.software.proprietary.policy.ledger;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.BasicFileAttributes;
+
+import stirling.software.proprietary.billing.ContentHasher;
+
+/**
+ * The folder backend's identity and version scheme, shared by {@code FolderInputSource} and {@code
+ * FolderOutputSink} so outputs are recorded under exactly the identity the next scan derives.
+ * Directories are canonicalised with {@code toRealPath()} so symlinked aliases agree.
+ */
+public final class FolderIdentities {
+
+    private FolderIdentities() {}
+
+    /** Canonical form of a configured directory; resolves symlinks, so the dir must exist. */
+    public static Path canonicalDir(Path dir) throws IOException {
+        return dir.toRealPath();
+    }
+
+    /** Identity of {@code file} under {@code dir}: its path re-rooted onto the canonical dir. */
+    public static String identity(Path canonicalDir, Path dir, Path file) {
+        return canonicalDir.resolve(dir.relativize(file)).normalize().toString();
+    }
+
+    /** The cheap version gate: a change to content length or mtime means "look closer". */
+    public static String statGate(Path file) throws IOException {
+        BasicFileAttributes attributes = Files.readAttributes(file, BasicFileAttributes.class);
+        return attributes.size() + ":" + attributes.lastModifiedTime().toMillis();
+    }
+
+    /**
+     * The strong version token: distinguishes a real change from a touch, at the cost of a read.
+     */
+    public static String contentHash(Path file) throws IOException {
+        return ContentHasher.sha256(file);
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/IdentityHasher.java

@@ -0,0 +1,19 @@
+package stirling.software.proprietary.policy.ledger;
+
+import java.nio.charset.StandardCharsets;
+
+import stirling.software.proprietary.billing.ContentHasher;
+
+/**
+ * Fixed-width key form of a source-owned identity, so any identity length fits the ledger's primary
+ * key. Backend-agnostic: every source type's identities are keyed through here, which is why this
+ * does not live with the folder backend's {@link FolderIdentities}.
+ */
+public final class IdentityHasher {
+
+    private IdentityHasher() {}
+
+    public static String identityHash(String identity) {
+        return ContentHasher.sha256(identity.getBytes(StandardCharsets.UTF_8));
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/InProcessProcessedLedger.java

@@ -0,0 +1,229 @@
+package stirling.software.proprietary.policy.ledger;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * In-memory {@link ProcessedLedger} for tests and DB-less wiring; kept semantically identical to
+ * {@code JpaProcessedLedger} by the shared contract test.
+ */
+public class InProcessProcessedLedger implements ProcessedLedger {
+
+    private final Map<String, Map<String, Row>> rowsByPolicy = new HashMap<>();
+    private final Supplier<Long> nowMillis;
+
+    public InProcessProcessedLedger() {
+        this(System::currentTimeMillis);
+    }
+
+    public InProcessProcessedLedger(Supplier<Long> nowMillis) {
+        this.nowMillis = nowMillis;
+    }
+
+    @Override
+    public synchronized Map<String, ClaimState> statesFor(
+            String policyId, Collection<String> identities) {
+        Map<String, Row> rows = rowsByPolicy.getOrDefault(policyId, Map.of());
+        Map<String, ClaimState> states = new HashMap<>();
+        for (String identity : identities) {
+            Row row = rows.get(identity);
+            if (row != null) {
+                states.put(identity, new ClaimState(row.status, row.gate, row.contentHash));
+            }
+        }
+        return states;
+    }
+
+    // Single-lock store: the live row is never staler than any observed snapshot, so decide
+    // against it directly; the conditional updates of the JPA ledger yield the same outcomes.
+    @Override
+    public synchronized boolean claim(
+            String policyId,
+            String identity,
+            String gate,
+            Supplier<String> contentHash,
+            ClaimState observed) {
+        Map<String, Row> rows = rowsByPolicy.computeIfAbsent(policyId, key -> new HashMap<>());
+        long now = nowMillis.get();
+        Row row = rows.get(identity);
+        if (row == null) {
+            String hash = contentHash == null ? null : contentHash.get();
+            rows.put(identity, new Row(gate, hash, ProcessedFileStatus.PROCESSING, 1, now));
+            return true;
+        }
+        if (row.status == ProcessedFileStatus.PROCESSING) {
+            return false;
+        }
+        if (gate.equals(row.gate)) {
+            if (row.status == ProcessedFileStatus.INTERRUPTED && row.attempts < MAX_ATTEMPTS) {
+                row.status = ProcessedFileStatus.PROCESSING;
+                row.attempts++;
+                row.lastSeen = now;
+                return true;
+            }
+            return false;
+        }
+        if (contentHash == null) {
+            row.gate = gate;
+            row.contentHash = null;
+            row.status = ProcessedFileStatus.PROCESSING;
+            row.attempts = 1;
+            row.lastSeen = now;
+            return true;
+        }
+        String hash = contentHash.get();
+        if (Objects.equals(hash, row.contentHash)) {
+            if (row.status == ProcessedFileStatus.INTERRUPTED && row.attempts < MAX_ATTEMPTS) {
+                row.gate = gate;
+                row.status = ProcessedFileStatus.PROCESSING;
+                row.attempts++;
+                row.lastSeen = now;
+                return true;
+            }
+            if (row.status != ProcessedFileStatus.INTERRUPTED) {
+                row.gate = gate;
+                row.lastSeen = now;
+            }
+            return false;
+        }
+        row.gate = gate;
+        row.contentHash = hash;
+        row.status = ProcessedFileStatus.PROCESSING;
+        row.attempts = 1;
+        row.lastSeen = now;
+        return true;
+    }
+
+    @Override
+    public synchronized void settle(
+            String policyId,
+            String identity,
+            String finalGate,
+            String finalContentHash,
+            boolean success) {
+        upsertSettled(
+                policyId,
+                identity,
+                finalGate,
+                finalContentHash,
+                success ? ProcessedFileStatus.DONE : ProcessedFileStatus.ERROR);
+    }
+
+    @Override
+    public synchronized void recordOutput(
+            String policyId, String identity, String gate, String contentHash) {
+        upsertSettled(policyId, identity, gate, contentHash, ProcessedFileStatus.DONE);
+    }
+
+    @Override
+    public synchronized void forgetOutput(String policyId, String identity, String gate) {
+        Map<String, Row> rows = rowsByPolicy.get(policyId);
+        if (rows == null) {
+            return;
+        }
+        Row row = rows.get(identity);
+        if (row != null && row.status == ProcessedFileStatus.DONE && gate.equals(row.gate)) {
+            rows.remove(identity);
+        }
+    }
+
+    private void upsertSettled(
+            String policyId,
+            String identity,
+            String gate,
+            String contentHash,
+            ProcessedFileStatus status) {
+        Map<String, Row> rows = rowsByPolicy.computeIfAbsent(policyId, key -> new HashMap<>());
+        long now = nowMillis.get();
+        Row row = rows.get(identity);
+        if (row == null) {
+            rows.put(identity, new Row(gate, contentHash, status, 1, now));
+            return;
+        }
+        row.gate = gate;
+        row.contentHash = contentHash;
+        row.status = status;
+        row.lastSeen = now;
+    }
+
+    @Override
+    public synchronized boolean allSettledDone(String identity) {
+        for (Map<String, Row> rows : rowsByPolicy.values()) {
+            Row row = rows.get(identity);
+            if (row != null && row.status != ProcessedFileStatus.DONE) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public synchronized void markSeen(String policyId, Collection<String> identities) {
+        Map<String, Row> rows = rowsByPolicy.get(policyId);
+        if (rows == null) {
+            return;
+        }
+        long now = nowMillis.get();
+        for (String identity : identities) {
+            Row row = rows.get(identity);
+            if (row != null) {
+                row.lastSeen = now;
+            }
+        }
+    }
+
+    @Override
+    public synchronized int deleteUnseen(String policyId, long seenSinceMillis) {
+        Map<String, Row> rows = rowsByPolicy.get(policyId);
+        if (rows == null) {
+            return 0;
+        }
+        int before = rows.size();
+        rows.values()
+                .removeIf(
+                        row ->
+                                row.lastSeen < seenSinceMillis
+                                        && row.status != ProcessedFileStatus.PROCESSING);
+        return before - rows.size();
+    }
+
+    @Override
+    public synchronized void clearPolicy(String policyId) {
+        rowsByPolicy.remove(policyId);
+    }
+
+    @Override
+    public synchronized void recoverInterrupted() {
+        for (Map<String, Row> rows : rowsByPolicy.values()) {
+            for (Row row : rows.values()) {
+                if (row.status == ProcessedFileStatus.PROCESSING) {
+                    row.status = ProcessedFileStatus.INTERRUPTED;
+                }
+            }
+        }
+    }
+
+    private static final class Row {
+        private String gate;
+        private String contentHash;
+        private ProcessedFileStatus status;
+        private int attempts;
+        private long lastSeen;
+
+        private Row(
+                String gate,
+                String contentHash,
+                ProcessedFileStatus status,
+                int attempts,
+                long lastSeen) {
+            this.gate = gate;
+            this.contentHash = contentHash;
+            this.status = status;
+            this.attempts = attempts;
+            this.lastSeen = lastSeen;
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/JpaProcessedLedger.java

@@ -0,0 +1,212 @@
+package stirling.software.proprietary.policy.ledger;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.stereotype.Service;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Durable {@link ProcessedLedger}; the runtime bean. A fresh claim is a flushed insert so a
+ * concurrent winner surfaces as a constraint violation; every other transition is a conditional
+ * update that re-checks the observed state, so a lost race reports 0 rows and the caller skips.
+ * Boot recovery assumes the single node the folder-watch trigger assumes: runs live in memory, so
+ * after a restart every PROCESSING row is stale.
+ */
+@Slf4j
+@Service
+@ConditionalOnBooleanProperty(name = "policies.enabled")
+public class JpaProcessedLedger implements ProcessedLedger {
+
+    private static final int STAMP_CHUNK = 500;
+
+    private final ProcessedFileRepository repository;
+    private final Supplier<Long> nowMillis;
+
+    @Autowired
+    public JpaProcessedLedger(ProcessedFileRepository repository) {
+        this(repository, System::currentTimeMillis);
+    }
+
+    // Clock seam so tests can pin "now"; the runtime bean uses the wall clock above.
+    JpaProcessedLedger(ProcessedFileRepository repository, Supplier<Long> nowMillis) {
+        this.repository = repository;
+        this.nowMillis = nowMillis;
+    }
+
+    @Override
+    public Map<String, ClaimState> statesFor(String policyId, Collection<String> identities) {
+        if (identities.isEmpty()) {
+            return Map.of();
+        }
+        Map<String, String> identityByHash = new HashMap<>();
+        for (String identity : identities) {
+            identityByHash.put(IdentityHasher.identityHash(identity), identity);
+        }
+        Map<String, ClaimState> states = new HashMap<>();
+        List<String> hashes = List.copyOf(identityByHash.keySet());
+        for (int from = 0; from < hashes.size(); from += STAMP_CHUNK) {
+            List<ProcessedFileEntity> rows =
+                    repository.findByPolicyIdAndIdentityHashIn(
+                            policyId,
+                            hashes.subList(from, Math.min(from + STAMP_CHUNK, hashes.size())));
+            for (ProcessedFileEntity row : rows) {
+                states.put(
+                        identityByHash.get(row.getIdentityHash()),
+                        new ClaimState(row.getStatus(), row.getSignature(), row.getContentHash()));
+            }
+        }
+        return states;
+    }
+
+    @Override
+    public boolean claim(
+            String policyId,
+            String identity,
+            String gate,
+            Supplier<String> contentHash,
+            ClaimState observed) {
+        String identityHash = IdentityHasher.identityHash(identity);
+        long now = nowMillis.get();
+        if (observed == null) {
+            try {
+                repository.saveAndFlush(
+                        new ProcessedFileEntity(
+                                policyId,
+                                identityHash,
+                                identity,
+                                gate,
+                                contentHash == null ? null : contentHash.get(),
+                                ProcessedFileStatus.PROCESSING,
+                                now));
+                return true;
+            } catch (DataIntegrityViolationException concurrentClaim) {
+                return false;
+            }
+        }
+        if (observed.status() == ProcessedFileStatus.PROCESSING) {
+            return false;
+        }
+        if (gate.equals(observed.gate())) {
+            if (observed.status() == ProcessedFileStatus.INTERRUPTED) {
+                return repository.retryInterruptedAtGate(
+                                policyId, identityHash, gate, MAX_ATTEMPTS, now)
+                        > 0;
+            }
+            return false;
+        }
+        if (contentHash == null) {
+            return repository.reclaimAtNewGate(policyId, identityHash, gate, now) > 0;
+        }
+        String hash = contentHash.get();
+        if (hash.equals(observed.contentHash())) {
+            if (observed.status() == ProcessedFileStatus.INTERRUPTED) {
+                return repository.retryInterruptedSameContent(
+                                policyId, identityHash, gate, hash, MAX_ATTEMPTS, now)
+                        > 0;
+            }
+            repository.refreshGate(policyId, identityHash, gate, hash, now);
+            return false;
+        }
+        return repository.reclaimAtNewContent(policyId, identityHash, gate, hash, now) > 0;
+    }
+
+    @Override
+    public void settle(
+            String policyId,
+            String identity,
+            String finalGate,
+            String finalContentHash,
+            boolean success) {
+        upsertSettled(
+                policyId,
+                identity,
+                finalGate,
+                finalContentHash,
+                success ? ProcessedFileStatus.DONE : ProcessedFileStatus.ERROR);
+    }
+
+    @Override
+    public void recordOutput(String policyId, String identity, String gate, String contentHash) {
+        upsertSettled(policyId, identity, gate, contentHash, ProcessedFileStatus.DONE);
+    }
+
+    @Override
+    public void forgetOutput(String policyId, String identity, String gate) {
+        repository.deleteDoneAt(policyId, IdentityHasher.identityHash(identity), gate);
+    }
+
+    /**
+     * Settle-or-insert: the row may have been presence-cleaned mid-run, and an output row may be
+     * brand new.
+     */
+    private void upsertSettled(
+            String policyId,
+            String identity,
+            String gate,
+            String contentHash,
+            ProcessedFileStatus status) {
+        String identityHash = IdentityHasher.identityHash(identity);
+        long now = nowMillis.get();
+        if (repository.settle(policyId, identityHash, gate, contentHash, status, now) > 0) {
+            return;
+        }
+        try {
+            ProcessedFileEntity row =
+                    new ProcessedFileEntity(
+                            policyId, identityHash, identity, gate, contentHash, status, now);
+            repository.saveAndFlush(row);
+        } catch (DataIntegrityViolationException concurrentInsert) {
+            repository.settle(policyId, identityHash, gate, contentHash, status, now);
+        }
+    }
+
+    @Override
+    public boolean allSettledDone(String identity) {
+        return !repository.existsByIdentityHashAndStatusNot(
+                IdentityHasher.identityHash(identity), ProcessedFileStatus.DONE);
+    }
+
+    @Override
+    public void markSeen(String policyId, Collection<String> identities) {
+        if (identities.isEmpty()) {
+            return;
+        }
+        List<String> hashes = identities.stream().map(IdentityHasher::identityHash).toList();
+        long now = nowMillis.get();
+        for (int from = 0; from < hashes.size(); from += STAMP_CHUNK) {
+            repository.stampSeen(
+                    policyId,
+                    hashes.subList(from, Math.min(from + STAMP_CHUNK, hashes.size())),
+                    now);
+        }
+    }
+
+    @Override
+    public int deleteUnseen(String policyId, long seenSinceMillis) {
+        return repository.deleteUnseen(policyId, seenSinceMillis);
+    }
+
+    @Override
+    public void clearPolicy(String policyId) {
+        repository.deleteByPolicy(policyId);
+    }
+
+    @Override
+    @EventListener(ApplicationReadyEvent.class)
+    public void recoverInterrupted() {
+        int recovered = repository.markAllProcessingInterrupted(nowMillis.get());
+        if (recovered > 0) {
+            log.info("Recovered {} policy input file(s) interrupted by shutdown", recovered);
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileEntity.java

@@ -0,0 +1,97 @@
+package stirling.software.proprietary.policy.ledger;
+
+import java.io.Serializable;
+
+import org.springframework.data.domain.Persistable;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.EnumType;
+import jakarta.persistence.Enumerated;
+import jakarta.persistence.Id;
+import jakarta.persistence.IdClass;
+import jakarta.persistence.Table;
+import jakarta.persistence.Transient;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+/**
+ * One processed-file ledger row: the version a policy last settled a file at, and where it is in
+ * the claim lifecycle. Keyed by SHA-256 of the source-owned identity so any identity length fits a
+ * fixed-width index. {@code isNew} is always true: the entity is only saved for fresh inserts
+ * (everything else is a conditional update), so a lost insert race surfaces as a constraint
+ * violation rather than a silent merge.
+ */
+@Entity
+@Table(name = "policy_processed_files")
+@IdClass(ProcessedFileId.class)
+@NoArgsConstructor
+@Getter
+@Setter
+public class ProcessedFileEntity implements Serializable, Persistable<ProcessedFileId> {
+
+    private static final long serialVersionUID = 1L;
+
+    @Id
+    @Column(name = "policy_id")
+    private String policyId;
+
+    @Id
+    @Column(name = "identity_hash", length = 64)
+    private String identityHash;
+
+    @Column(name = "identity", length = 4096)
+    private String identity;
+
+    @Column(name = "signature")
+    private String signature;
+
+    @Column(name = "content_hash", length = 64)
+    private String contentHash;
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "status", length = 16)
+    private ProcessedFileStatus status;
+
+    @Column(name = "attempts")
+    private int attempts;
+
+    @Column(name = "last_seen")
+    private long lastSeen;
+
+    @Column(name = "updated_at")
+    private long updatedAt;
+
+    public ProcessedFileEntity(
+            String policyId,
+            String identityHash,
+            String identity,
+            String signature,
+            String contentHash,
+            ProcessedFileStatus status,
+            long nowMillis) {
+        this.policyId = policyId;
+        this.identityHash = identityHash;
+        this.identity = identity;
+        this.signature = signature;
+        this.contentHash = contentHash;
+        this.status = status;
+        this.attempts = 1;
+        this.lastSeen = nowMillis;
+        this.updatedAt = nowMillis;
+    }
+
+    @Override
+    @Transient
+    public ProcessedFileId getId() {
+        return new ProcessedFileId(policyId, identityHash);
+    }
+
+    @Override
+    @Transient
+    public boolean isNew() {
+        return true;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileId.java

@@ -0,0 +1,37 @@
+package stirling.software.proprietary.policy.ledger;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/** Composite key for {@link ProcessedFileEntity}: one row per policy per file identity. */
+public class ProcessedFileId implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private String policyId;
+    private String identityHash;
+
+    public ProcessedFileId() {}
+
+    public ProcessedFileId(String policyId, String identityHash) {
+        this.policyId = policyId;
+        this.identityHash = identityHash;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof ProcessedFileId other)) {
+            return false;
+        }
+        return Objects.equals(policyId, other.policyId)
+                && Objects.equals(identityHash, other.identityHash);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(policyId, identityHash);
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileRepository.java

@@ -0,0 +1,195 @@
+package stirling.software.proprietary.policy.ledger;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * Conditional updates for the processed-file ledger: each claim variant re-checks in its WHERE
+ * clause the state it was decided against, so a racing claim loses cleanly with 0 rows updated.
+ * Transactional per call so the ledger can run them without an enclosing transaction.
+ */
+@Repository
+public interface ProcessedFileRepository
+        extends JpaRepository<ProcessedFileEntity, ProcessedFileId> {
+
+    /**
+     * Re-claim a settled row at a new gate without content verification; clears the stored hash,
+     * which described content this claim never checked.
+     */
+    @Modifying
+    @Transactional
+    @Query(
+            "update ProcessedFileEntity e set e.status ="
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING,"
+                    + " e.signature = :gate, e.contentHash = null, e.attempts = 1,"
+                    + " e.lastSeen = :now, e.updatedAt = :now"
+                    + " where e.policyId = :policyId and e.identityHash = :identityHash"
+                    + " and e.status <>"
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING"
+                    + " and e.signature <> :gate")
+    int reclaimAtNewGate(
+            @Param("policyId") String policyId,
+            @Param("identityHash") String identityHash,
+            @Param("gate") String gate,
+            @Param("now") long now);
+
+    /** Re-claim a settled row whose content verifiably changed (or was never hashed). */
+    @Modifying
+    @Transactional
+    @Query(
+            "update ProcessedFileEntity e set e.status ="
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING,"
+                    + " e.signature = :gate, e.contentHash = :contentHash, e.attempts = 1,"
+                    + " e.lastSeen = :now, e.updatedAt = :now"
+                    + " where e.policyId = :policyId and e.identityHash = :identityHash"
+                    + " and e.status <>"
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING"
+                    + " and (e.contentHash is null or e.contentHash <> :contentHash)")
+    int reclaimAtNewContent(
+            @Param("policyId") String policyId,
+            @Param("identityHash") String identityHash,
+            @Param("gate") String gate,
+            @Param("contentHash") String contentHash,
+            @Param("now") long now);
+
+    /** The gate moved but the content did not: track the new gate without changing status. */
+    @Modifying
+    @Transactional
+    @Query(
+            "update ProcessedFileEntity e set e.signature = :gate, e.lastSeen = :now,"
+                    + " e.updatedAt = :now"
+                    + " where e.policyId = :policyId and e.identityHash = :identityHash"
+                    + " and e.status <>"
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING"
+                    + " and e.contentHash = :contentHash and e.signature <> :gate")
+    int refreshGate(
+            @Param("policyId") String policyId,
+            @Param("identityHash") String identityHash,
+            @Param("gate") String gate,
+            @Param("contentHash") String contentHash,
+            @Param("now") long now);
+
+    /** Bounded retry of an INTERRUPTED row at the same gate. */
+    @Modifying
+    @Transactional
+    @Query(
+            "update ProcessedFileEntity e set e.status ="
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING,"
+                    + " e.attempts = e.attempts + 1, e.lastSeen = :now, e.updatedAt = :now"
+                    + " where e.policyId = :policyId and e.identityHash = :identityHash"
+                    + " and e.status ="
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.INTERRUPTED"
+                    + " and e.signature = :gate and e.attempts < :maxAttempts")
+    int retryInterruptedAtGate(
+            @Param("policyId") String policyId,
+            @Param("identityHash") String identityHash,
+            @Param("gate") String gate,
+            @Param("maxAttempts") int maxAttempts,
+            @Param("now") long now);
+
+    /** Bounded retry of an INTERRUPTED row whose gate moved but whose content is unchanged. */
+    @Modifying
+    @Transactional
+    @Query(
+            "update ProcessedFileEntity e set e.status ="
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING,"
+                    + " e.signature = :gate, e.attempts = e.attempts + 1, e.lastSeen = :now,"
+                    + " e.updatedAt = :now"
+                    + " where e.policyId = :policyId and e.identityHash = :identityHash"
+                    + " and e.status ="
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.INTERRUPTED"
+                    + " and e.contentHash = :contentHash and e.attempts < :maxAttempts")
+    int retryInterruptedSameContent(
+            @Param("policyId") String policyId,
+            @Param("identityHash") String identityHash,
+            @Param("gate") String gate,
+            @Param("contentHash") String contentHash,
+            @Param("maxAttempts") int maxAttempts,
+            @Param("now") long now);
+
+    /**
+     * Unconditional settle (only the claiming run settles a row); returns 0 when the row was
+     * removed mid-run so the caller re-inserts.
+     */
+    @Modifying
+    @Transactional
+    @Query(
+            "update ProcessedFileEntity e set e.status = :status, e.signature = :gate,"
+                    + " e.contentHash = :contentHash, e.lastSeen = :now, e.updatedAt = :now"
+                    + " where e.policyId = :policyId and e.identityHash = :identityHash")
+    int settle(
+            @Param("policyId") String policyId,
+            @Param("identityHash") String identityHash,
+            @Param("gate") String gate,
+            @Param("contentHash") String contentHash,
+            @Param("status") ProcessedFileStatus status,
+            @Param("now") long now);
+
+    /** Whether any policy's row at this identity is in a state other than {@code status}. */
+    boolean existsByIdentityHashAndStatusNot(String identityHash, ProcessedFileStatus status);
+
+    /** One policy's rows across a chunk of identity hashes, for a sweep's claim snapshot. */
+    List<ProcessedFileEntity> findByPolicyIdAndIdentityHashIn(
+            String policyId, Collection<String> identityHashes);
+
+    /**
+     * Remove an output record whose rename never landed, only while still settled exactly as
+     * recorded; a row a claim has since taken over is left alone.
+     */
+    @Modifying
+    @Transactional
+    @Query(
+            "delete from ProcessedFileEntity e where e.policyId = :policyId"
+                    + " and e.identityHash = :identityHash and e.signature = :gate"
+                    + " and e.status ="
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.DONE")
+    int deleteDoneAt(
+            @Param("policyId") String policyId,
+            @Param("identityHash") String identityHash,
+            @Param("gate") String gate);
+
+    /** Stamp presence for the given identities; chunked by the caller for very large folders. */
+    @Modifying
+    @Transactional
+    @Query(
+            "update ProcessedFileEntity e set e.lastSeen = :now"
+                    + " where e.policyId = :policyId and e.identityHash in :identityHashes")
+    int stampSeen(
+            @Param("policyId") String policyId,
+            @Param("identityHashes") Collection<String> identityHashes,
+            @Param("now") long now);
+
+    /**
+     * Presence cleanup: remove rows not stamped since the sweep began, keeping in-flight claims.
+     */
+    @Modifying
+    @Transactional
+    @Query(
+            "delete from ProcessedFileEntity e where e.policyId = :policyId"
+                    + " and e.lastSeen < :cutoff and e.status <>"
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING")
+    int deleteUnseen(@Param("policyId") String policyId, @Param("cutoff") long cutoff);
+
+    @Modifying
+    @Transactional
+    @Query("delete from ProcessedFileEntity e where e.policyId = :policyId")
+    int deleteByPolicy(@Param("policyId") String policyId);
+
+    /** Boot recovery: after a restart every PROCESSING row is stale (single node). */
+    @Modifying
+    @Transactional
+    @Query(
+            "update ProcessedFileEntity e set e.status ="
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.INTERRUPTED,"
+                    + " e.updatedAt = :now"
+                    + " where e.status ="
+                    + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING")
+    int markAllProcessingInterrupted(@Param("now") long now);
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileStatus.java

@@ -0,0 +1,17 @@
+package stirling.software.proprietary.policy.ledger;
+
+/** Lifecycle of one {@code (policy, file)} ledger row. */
+public enum ProcessedFileStatus {
+
+    /** Claimed; a run is in flight. */
+    PROCESSING,
+
+    /** Run completed at this version. */
+    DONE,
+
+    /** Run failed; skipped until the file changes (clear-history is the manual retry). */
+    ERROR,
+
+    /** Was PROCESSING when the JVM died; retried a bounded number of times. */
+    INTERRUPTED
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedLedger.java

@@ -0,0 +1,100 @@
+package stirling.software.proprietary.policy.ledger;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+/**
+ * Remembers which files a policy has processed, one row per {@code (policy, identity)}, so sources
+ * track files in place. Identities are opaque source-owned strings; versions are two-tier: a cheap
+ * gate compared every sweep plus an optional content hash consulted only when the gate moves.
+ * Presence reconciliation ({@link #markSeen} + {@link #deleteUnseen}) keeps the table bounded.
+ */
+public interface ProcessedLedger {
+
+    /**
+     * Claims of one version before an {@link ProcessedFileStatus#INTERRUPTED} row stops retrying.
+     */
+    int MAX_ATTEMPTS = 3;
+
+    /**
+     * One-query snapshot of the rows for these identities, keyed by identity; identities with no
+     * row are absent. Feeds the {@code observed} parameter of {@link #claim(String, String, String,
+     * Supplier, ClaimState)} so a sweep decides its claims without a per-file lookup.
+     */
+    Map<String, ClaimState> statesFor(String policyId, Collection<String> identities);
+
+    /**
+     * Atomically claim a file at its current version, deciding against {@code observed} (this row's
+     * entry from {@link #statesFor}; null means no row was seen); true means this caller runs it. A
+     * stale {@code observed} cannot double-claim - every transition re-checks the observed state,
+     * so a lost race skips until a later sweep. A null {@code contentHash} makes any gate change a
+     * new version; a non-null supplier is invoked at most once, only on a gate mismatch, and a
+     * matching hash refreshes the stored gate instead of reprocessing. Supplier exceptions
+     * propagate.
+     */
+    boolean claim(
+            String policyId,
+            String identity,
+            String gate,
+            Supplier<String> contentHash,
+            ClaimState observed);
+
+    /** Snapshot-then-claim convenience for a single file; sweeps batch via {@link #statesFor}. */
+    default boolean claim(
+            String policyId, String identity, String gate, Supplier<String> contentHash) {
+        return claim(
+                policyId,
+                identity,
+                gate,
+                contentHash,
+                statesFor(policyId, List.of(identity)).get(identity));
+    }
+
+    /** Record a claimed file's outcome at its final version ({@code finalContentHash} nullable). */
+    void settle(
+            String policyId,
+            String identity,
+            String finalGate,
+            String finalContentHash,
+            boolean success);
+
+    /**
+     * Record a produced file as {@link ProcessedFileStatus#DONE} so the policy skips its own
+     * outputs. Must be called before the file is visible at this identity; other policies have no
+     * row and still process it.
+     */
+    void recordOutput(String policyId, String identity, String gate, String contentHash);
+
+    /**
+     * Remove an output record whose file never became visible (its rename lost the name race to a
+     * concurrent writer), so whatever file actually owns that identity is claimable at any version.
+     * A no-op unless the row is still settled exactly as recorded, so a claim that took the row
+     * over in the meantime is left alone.
+     */
+    void forgetOutput(String policyId, String identity, String gate);
+
+    /**
+     * Whether every row at this identity - across all policies, by design - is {@link
+     * ProcessedFileStatus#DONE}. Consume-mode deletion gates on this so a shared input is removed
+     * only once every claimant has processed it; in-flight, failed, and interrupted rows all veto,
+     * parking the file. Vacuously true when no rows exist.
+     */
+    boolean allSettledDone(String identity);
+
+    /** Stamp presence for every identity a full-listing sweep observed. */
+    void markSeen(String policyId, Collection<String> identities);
+
+    /**
+     * Remove rows not seen since {@code seenSinceMillis}, keeping in-flight claims. Only call after
+     * every enabled source listed completely; returns the number of rows removed.
+     */
+    int deleteUnseen(String policyId, long seenSinceMillis);
+
+    /** Forget everything for a policy. */
+    void clearPolicy(String policyId);
+
+    /** Boot recovery: flip stale in-flight claims to {@link ProcessedFileStatus#INTERRUPTED}. */
+    void recoverInterrupted();
+}

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

@@ -2,11 +2,18 @@ package stirling.software.proprietary.policy.output;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.file.FileAlreadyExistsException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.security.DigestOutputStream;
+import java.security.MessageDigest;
+import java.time.Duration;
+import java.time.Instant;
 import java.util.ArrayList;
 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;
@@ -19,14 +26,18 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.model.job.ResultFile;
+import stirling.software.proprietary.billing.ContentHasher;
 import stirling.software.proprietary.policy.config.FolderAccessGuard;
+import stirling.software.proprietary.policy.ledger.FolderIdentities;
+import stirling.software.proprietary.policy.ledger.ProcessedLedger;
 import stirling.software.proprietary.policy.model.OutputSpec;
 
 /**
- * Writes a run's outputs to the {@code directory} given in the {@link OutputSpec}. Files are
- * streamed (not buffered) and uniquely named to avoid clobbering. Returned {@link ResultFile}s
- * carry a synthetic id since the deliverable is the file on disk, not a {@code FileStorage} entry,
- * so folder outputs are not downloadable via {@code /files/{id}}.
+ * Writes a run's outputs to the {@code directory} given in the {@link OutputSpec}. Each output is
+ * staged under a hidden {@code .stirling/tmp} dir, recorded in the processed-file ledger, then
+ * atomically renamed into place, so the producing policy's row exists before the file is
+ * discoverable and half-written outputs are never visible. Returned {@link ResultFile}s carry a
+ * synthetic id since the deliverable is the file on disk, not a {@code FileStorage} entry.
  */
 @Slf4j
 @Service
@@ -37,7 +48,11 @@ public class FolderOutputSink implements PolicyOutputSink {
     static final String TYPE = FolderAccessGuard.FOLDER_TYPE;
     static final String DIRECTORY_OPTION = "directory";
 
+    // Staging entries are renamed away within one delivery; anything older is a crash leftover.
+    private static final Duration STALE_TMP_AGE = Duration.ofDays(1);
+
     private final FolderAccessGuard accessGuard;
+    private final ProcessedLedger processedLedger;
 
     @Override
     public String type() {
@@ -55,20 +70,25 @@ public class FolderOutputSink implements PolicyOutputSink {
     }
 
     @Override
-    public List<ResultFile> deliver(String runId, List<Resource> outputs, OutputSpec spec)
-            throws IOException {
+    public List<ResultFile> deliver(
+            OutputDelivery delivery, List<Resource> outputs, OutputSpec spec) throws IOException {
         Path targetDir = accessGuard.requirePermitted(directoryOf(spec));
         Files.createDirectories(targetDir);
+        Path canonicalDir = FolderIdentities.canonicalDir(targetDir);
+        Path tmpDir = canonicalDir.resolve(".stirling").resolve("tmp");
+        Files.createDirectories(tmpDir);
+        sweepStaleTmp(tmpDir);
 
         List<ResultFile> results = new ArrayList<>();
         for (int i = 0; i < outputs.size(); i++) {
             Resource resource = outputs.get(i);
             String name = safeName(resource.getFilename(), i);
-            Path target = uniqueTarget(targetDir, name);
-            try (InputStream is = resource.getInputStream()) {
-                Files.copy(is, target);
-            }
-            long size = Files.size(target);
+            Path staged = tmpDir.resolve(UUID.randomUUID().toString());
+            String contentHash = stage(resource, staged, delivery.policyId() != null);
+            long size = Files.size(staged);
+            // Size and mtime survive the rename.
+            String gate = FolderIdentities.statGate(staged);
+            Path target = moveIntoPlace(delivery, canonicalDir, name, staged, gate, contentHash);
             String contentType =
                     MediaTypeFactory.getMediaType(name)
                             .orElse(MediaType.APPLICATION_OCTET_STREAM)
@@ -80,11 +100,95 @@ public class FolderOutputSink implements PolicyOutputSink {
                             .contentType(contentType)
                             .fileSize(size)
                             .build());
-            log.debug("Wrote policy run {} output to {}", runId, target);
+            log.debug("Wrote policy run {} output to {}", delivery.runId(), target);
         }
         return results;
     }
 
+    /**
+     * Stream the output to its staging path. For a recorded delivery (stored policy) the content
+     * hash is digested in the same pass, so the ledger gets both version tiers without re-reading a
+     * possibly huge output; ad-hoc runs record nothing and skip the digest entirely.
+     */
+    private static String stage(Resource resource, Path staged, boolean hashed) throws IOException {
+        if (!hashed) {
+            try (InputStream is = resource.getInputStream()) {
+                Files.copy(is, staged);
+            }
+            return null;
+        }
+        MessageDigest digest = ContentHasher.newSha256();
+        try (InputStream is = resource.getInputStream();
+                DigestOutputStream out =
+                        new DigestOutputStream(Files.newOutputStream(staged), digest)) {
+            is.transferTo(out);
+        }
+        return ContentHasher.toHex(digest.digest());
+    }
+
+    /**
+     * The ledger row must exist before the file is visible at its final path, or a sweep could
+     * claim the producing policy's own output in the gap. Losing the chosen name to a concurrent
+     * writer forgets the just-recorded row - whatever file actually owns that name must stay
+     * claimable at any version - then re-picks.
+     */
+    private Path moveIntoPlace(
+            OutputDelivery delivery,
+            Path dir,
+            String name,
+            Path staged,
+            String gate,
+            String contentHash)
+            throws IOException {
+        while (true) {
+            Path target = uniqueTarget(dir, name);
+            if (delivery.policyId() != null) {
+                processedLedger.recordOutput(
+                        delivery.policyId(), target.toString(), gate, contentHash);
+            }
+            try {
+                Files.move(staged, target, StandardCopyOption.ATOMIC_MOVE);
+                return target;
+            } catch (FileAlreadyExistsException raced) {
+                if (delivery.policyId() != null) {
+                    processedLedger.forgetOutput(delivery.policyId(), target.toString(), gate);
+                }
+                log.debug("Output name {} taken concurrently; re-picking", target);
+            }
+        }
+    }
+
+    /** Best-effort removal of staging leftovers from crashed deliveries. */
+    private static void sweepStaleTmp(Path tmpDir) {
+        Instant cutoff = Instant.now().minus(STALE_TMP_AGE);
+        try (Stream<Path> entries = Files.list(tmpDir)) {
+            entries.filter(Files::isRegularFile)
+                    .filter(
+                            entry -> {
+                                try {
+                                    return Files.getLastModifiedTime(entry)
+                                            .toInstant()
+                                            .isBefore(cutoff);
+                                } catch (IOException e) {
+                                    return false;
+                                }
+                            })
+                    .forEach(
+                            entry -> {
+                                try {
+                                    Files.deleteIfExists(entry);
+                                } catch (IOException e) {
+                                    log.debug(
+                                            "Could not remove stale staging file {}: {}",
+                                            entry,
+                                            e.getMessage());
+                                }
+                            });
+        } catch (IOException e) {
+            log.debug("Could not sweep staging dir {}: {}", tmpDir, e.getMessage());
+        }
+    }
+
     private static Path directoryOf(OutputSpec spec) {
         Object directory = spec.options().get(DIRECTORY_OPTION);
         if (directory == null || directory.toString().isBlank()) {

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

@@ -41,8 +41,8 @@ public class InlineOutputSink implements PolicyOutputSink {
     }
 
     @Override
-    public List<ResultFile> deliver(String runId, List<Resource> outputs, OutputSpec spec)
-            throws IOException {
+    public List<ResultFile> deliver(
+            OutputDelivery delivery, List<Resource> outputs, OutputSpec spec) throws IOException {
         List<ResultFile> results = new ArrayList<>();
         for (int i = 0; i < outputs.size(); i++) {
             Resource resource = outputs.get(i);

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

@@ -0,0 +1,8 @@
+package stirling.software.proprietary.policy.output;
+
+/**
+ * Context for one run's output delivery. {@code policyId} is null for ad-hoc pipelines; when
+ * present, sinks record outputs in the processed-file ledger so the producing policy does not
+ * re-ingest them.
+ */
+public record OutputDelivery(String runId, String policyId) {}

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

@@ -25,6 +25,6 @@ public interface PolicyOutputSink {
     default void validate(OutputSpec spec) {}
 
     /** Persist/deliver the output files and return their descriptors. */
-    List<ResultFile> deliver(String runId, List<Resource> outputs, OutputSpec spec)
+    List<ResultFile> deliver(OutputDelivery delivery, List<Resource> outputs, OutputSpec spec)
             throws IOException;
 }

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

@@ -29,6 +29,7 @@ import lombok.extern.slf4j.Slf4j;
 import stirling.software.common.model.ApplicationProperties;
 import stirling.software.proprietary.policy.config.FolderAccessGuard;
 import stirling.software.proprietary.policy.engine.PolicyRunner;
+import stirling.software.proprietary.policy.engine.SweepKind;
 import stirling.software.proprietary.policy.input.InputSource;
 import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.Policy;
@@ -202,7 +203,8 @@ public class FolderWatchTrigger implements PolicyTrigger {
             }
             if (dirs.stream().anyMatch(changedDirs::contains)) {
                 log.debug("Folder-watch policy {} ({}) saw activity", policy.id(), policy.name());
-                policyRunner.run(policy);
+                // Light: the periodic reconcile does the full sweep.
+                policyRunner.run(policy, SweepKind.LIGHT);
             }
         }
     }

app/proprietary/src/main/java/stirling/software/proprietary/security/configuration/DatabaseConfig.java

@@ -34,6 +34,7 @@ import stirling.software.common.model.exception.UnsupportedProviderException;
             "stirling.software.proprietary.workflow.repository",
             "stirling.software.proprietary.policy.store",
             "stirling.software.proprietary.policy.source",
+            "stirling.software.proprietary.policy.ledger",
             "stirling.software.proprietary.accountlink",
             "stirling.software.proprietary.access.repository",
             "stirling.software.proprietary.integration.repository",
@@ -46,6 +47,7 @@ import stirling.software.common.model.exception.UnsupportedProviderException;
     "stirling.software.proprietary.workflow.model",
     "stirling.software.proprietary.policy.store",
     "stirling.software.proprietary.policy.source",
+    "stirling.software.proprietary.policy.ledger",
     "stirling.software.proprietary.accountlink",
     "stirling.software.proprietary.access.model",
     "stirling.software.proprietary.integration.model",

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

@@ -34,6 +34,7 @@ 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.ledger.ProcessedLedger;
 import stirling.software.proprietary.policy.model.PipelineDefinition;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
@@ -62,6 +63,8 @@ class PolicyControllerTest {
     private stirling.software.proprietary.policy.overview.PolicyOverviewService
             policyOverviewService;
 
+    @Mock private ProcessedLedger processedLedger;
+
     @Mock private TempFileManager tempFileManager;
     @Mock private JobOwnershipService jobOwnershipService;
 
@@ -89,6 +92,7 @@ class PolicyControllerTest {
                         policyManagementAuthority,
                         policyTriggerManager,
                         policyOverviewService,
+                        processedLedger,
                         policyTriggers,
                         applicationProperties,
                         tempFileManager,
@@ -398,6 +402,7 @@ class PolicyControllerTest {
             ResponseEntity<Void> response = controller.deletePolicy("a");
 
             assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NO_CONTENT);
+            verify(processedLedger).clearPolicy("a");
             verify(policyTriggerManager).notifyPoliciesChanged();
         }
 
@@ -431,6 +436,53 @@ class PolicyControllerTest {
         }
     }
 
+    @Nested
+    @DisplayName("clearProcessedHistory")
+    class ClearProcessedHistory {
+
+        @Test
+        @DisplayName("clears an accessible policy's history")
+        void clears() {
+            applicationProperties.getSecurity().setEnableLogin(false);
+            Policy p = policy("a", 1L);
+            when(policyStore.get("a")).thenReturn(Optional.of(p));
+            when(policyAccessGuard.canAccess(p)).thenReturn(true);
+
+            ResponseEntity<Void> response = controller.clearProcessedHistory("a");
+
+            assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NO_CONTENT);
+            verify(processedLedger).clearPolicy("a");
+        }
+
+        @Test
+        @DisplayName("returns 404 when policy is not accessible")
+        void notAccessible() {
+            applicationProperties.getSecurity().setEnableLogin(false);
+            Policy p = policy("a", 1L);
+            when(policyStore.get("a")).thenReturn(Optional.of(p));
+            when(policyAccessGuard.canAccess(p)).thenReturn(false);
+
+            ResponseEntity<Void> response = controller.clearProcessedHistory("a");
+
+            assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
+            verify(processedLedger, never()).clearPolicy(any());
+        }
+
+        @Test
+        @DisplayName("forbidden when login enabled and caller cannot edit")
+        void forbidden() {
+            applicationProperties.getSecurity().setEnableLogin(true);
+            when(policyManagementAuthority.canEditPolicies()).thenReturn(false);
+
+            assertThatThrownBy(() -> controller.clearProcessedHistory("a"))
+                    .isInstanceOf(ResponseStatusException.class)
+                    .satisfies(
+                            e ->
+                                    assertThat(((ResponseStatusException) e).getStatusCode())
+                                            .isEqualTo(HttpStatus.FORBIDDEN));
+        }
+    }
+
     @Nested
     @DisplayName("runStoredPolicy")
     class RunStoredPolicy {

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

@@ -4,15 +4,19 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -24,7 +28,9 @@ import org.mockito.Mock;
 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.ProcessedLedger;
 import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.OutputSpec;
 import stirling.software.proprietary.policy.model.PipelineStep;
@@ -39,15 +45,15 @@ import stirling.software.proprietary.policy.source.Source;
 import stirling.software.proprietary.policy.source.SourceStore;
 
 /**
- * Tests for {@link PolicyRunner}: the one place that turns a policy's sources into runs. Verifies
- * it pulls every source, runs one job per unit of work, feeds each unit's completion hook the run
- * outcome, and that a generator (no sources) still runs once.
+ * Tests for {@link PolicyRunner}: the one place that turns a policy's sources into runs, and the
+ * orchestrator of ledger hygiene (presence stamping + cleanup on complete FULL sweeps).
  */
 @ExtendWith(MockitoExtension.class)
 class PolicyRunnerTest {
 
     @Mock private PolicyEngine policyEngine;
     @Mock private InputSource folderSource;
+    @Mock private ProcessedLedger processedLedger;
 
     private final SourceStore sourceStore = new InProcessSourceStore();
     private PolicyRunner runner;
@@ -59,7 +65,8 @@ class PolicyRunnerTest {
                         policyEngine,
                         List.of(folderSource),
                         sourceStore,
-                        new InProcessSourceDocCounter());
+                        new InProcessSourceDocCounter(),
+                        processedLedger);
     }
 
     @Test
@@ -73,6 +80,9 @@ class PolicyRunnerTest {
         ArgumentCaptor<PolicyInputs> inputs = ArgumentCaptor.forClass(PolicyInputs.class);
         verify(policyEngine).runPolicy(eq(policy), inputs.capture(), any());
         assertTrue(inputs.getValue().primary().isEmpty());
+        // Ledger hygiene still runs: rows recorded for a generator policy's folder outputs
+        // are pruned by its own sweeps rather than accumulating until the policy is deleted.
+        verify(processedLedger).deleteUnseen(eq("p1"), anyLong());
     }
 
     @Test
@@ -80,7 +90,7 @@ class PolicyRunnerTest {
         InputSpec spec = InputSpec.folder("/in");
         Policy policy = policy(List.of(spec));
         when(folderSource.supports(spec)).thenReturn(true);
-        when(folderSource.resolve(spec))
+        when(folderSource.resolve(eq(spec), any()))
                 .thenReturn(
                         List.of(
                                 ResolvedInput.of(PolicyInputs.of(List.of())),
@@ -100,7 +110,7 @@ class PolicyRunnerTest {
         AtomicBoolean outcome = new AtomicBoolean(false);
         ResolvedInput unit = new ResolvedInput(PolicyInputs.of(List.of()), outcome::set);
         when(folderSource.supports(spec)).thenReturn(true);
-        when(folderSource.resolve(spec)).thenReturn(List.of(unit));
+        when(folderSource.resolve(eq(spec), any())).thenReturn(List.of(unit));
         CompletableFuture<PolicyRun> completion = new CompletableFuture<>();
         when(policyEngine.runPolicy(any(), any(), any()))
                 .thenReturn(new PolicyRunHandle("r", completion));
@@ -121,7 +131,7 @@ class PolicyRunnerTest {
         AtomicBoolean outcome = new AtomicBoolean(true);
         ResolvedInput unit = new ResolvedInput(PolicyInputs.of(List.of()), outcome::set);
         when(folderSource.supports(spec)).thenReturn(true);
-        when(folderSource.resolve(spec)).thenReturn(List.of(unit));
+        when(folderSource.resolve(eq(spec), any())).thenReturn(List.of(unit));
         CompletableFuture<PolicyRun> completion = new CompletableFuture<>();
         when(policyEngine.runPolicy(any(), any(), any()))
                 .thenReturn(new PolicyRunHandle("r", completion));
@@ -143,6 +153,98 @@ class PolicyRunnerTest {
         verifyNoInteractions(policyEngine);
     }
 
+    @Test
+    void aFullSweepStampsPresenceAndPrunesUnseenRows() throws Exception {
+        InputSpec spec = InputSpec.folder("/in");
+        Policy policy = policy(List.of(spec));
+        when(folderSource.supports(spec)).thenReturn(true);
+        when(folderSource.listsExhaustively()).thenReturn(true);
+        when(folderSource.resolve(eq(spec), any()))
+                .thenAnswer(
+                        invocation -> {
+                            ResolveContext ctx = invocation.getArgument(1);
+                            ctx.reportPresent(List.of("/in/a.pdf", "/in/b.pdf"));
+                            return List.of();
+                        });
+
+        runner.run(policy);
+
+        // Presence reporting also bulk-prefetches claim state: one lookup for the whole listing.
+        verify(processedLedger).statesFor(eq("p1"), eq(List.of("/in/a.pdf", "/in/b.pdf")));
+        verify(processedLedger).markSeen("p1", Set.of("/in/a.pdf", "/in/b.pdf"));
+        verify(processedLedger).deleteUnseen(eq("p1"), anyLong());
+    }
+
+    @Test
+    void aLightSweepClaimsButSkipsLedgerHygiene() throws Exception {
+        InputSpec spec = InputSpec.folder("/in");
+        Policy policy = policy(List.of(spec));
+        when(folderSource.supports(spec)).thenReturn(true);
+        when(folderSource.resolve(eq(spec), any()))
+                .thenReturn(List.of(ResolvedInput.of(PolicyInputs.of(List.of()))));
+        when(policyEngine.runPolicy(any(), any(), any()))
+                .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>()));
+
+        runner.run(policy, SweepKind.LIGHT);
+
+        verify(policyEngine).runPolicy(eq(policy), any(), any());
+        verify(processedLedger, never()).markSeen(any(), any());
+        verify(processedLedger, never()).deleteUnseen(any(), anyLong());
+    }
+
+    @Test
+    void aSourceThatFailsToResolveVetoesCleanupButOthersStillRun() throws Exception {
+        InputSpec broken = InputSpec.folder("/broken");
+        InputSpec healthy = InputSpec.folder("/healthy");
+        Policy policy = policy(List.of(broken, healthy));
+        when(folderSource.supports(any())).thenReturn(true);
+        when(folderSource.listsExhaustively()).thenReturn(true);
+        when(folderSource.resolve(eq(broken), any())).thenThrow(new IOException("mount gone"));
+        when(folderSource.resolve(eq(healthy), any()))
+                .thenReturn(List.of(ResolvedInput.of(PolicyInputs.of(List.of()))));
+        when(policyEngine.runPolicy(any(), any(), any()))
+                .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>()));
+
+        runner.run(policy);
+
+        verify(policyEngine).runPolicy(eq(policy), any(), any()); // healthy source still ran
+        verify(processedLedger, never()).deleteUnseen(any(), anyLong()); // history preserved
+    }
+
+    @Test
+    void aDisabledSourceVetoesCleanup() {
+        InputSpec spec = InputSpec.folder("/in");
+        String pausedId = sourceStore.save(disabledSourceFrom(spec)).id();
+        Policy policy = policyReferencing(List.of(pausedId));
+
+        runner.run(policy);
+
+        verify(processedLedger, never()).deleteUnseen(any(), anyLong());
+    }
+
+    @Test
+    void aNonExhaustiveSourceVetoesCleanup() throws Exception {
+        InputSpec spec = InputSpec.folder("/in");
+        Policy policy = policy(List.of(spec));
+        when(folderSource.supports(spec)).thenReturn(true);
+        when(folderSource.listsExhaustively()).thenReturn(false);
+        when(folderSource.resolve(eq(spec), any())).thenReturn(List.of());
+
+        runner.run(policy);
+
+        verify(processedLedger, never()).deleteUnseen(any(), anyLong());
+    }
+
+    @Test
+    void aMissingSourceDoesNotVetoCleanup() {
+        // A deleted source's rows age out precisely because cleanup still runs.
+        Policy policy = policyReferencing(List.of("ghost-source-id"));
+
+        runner.run(policy);
+
+        verify(processedLedger).deleteUnseen(eq("p1"), anyLong());
+    }
+
     @Test
     void runWithSuppliedInputsBypassesSources() {
         Policy policy = policy(List.of(InputSpec.folder("/in")));
@@ -159,6 +261,10 @@ class PolicyRunnerTest {
     private Policy policy(List<InputSpec> sources) {
         List<String> sourceIds =
                 sources.stream().map(spec -> sourceStore.save(sourceFrom(spec)).id()).toList();
+        return policyReferencing(sourceIds);
+    }
+
+    private static Policy policyReferencing(List<String> sourceIds) {
         return new Policy(
                 "p1",
                 "p",
@@ -173,4 +279,8 @@ class PolicyRunnerTest {
     private static Source sourceFrom(InputSpec spec) {
         return new Source(null, "src", spec.type(), spec.options(), true, "owner", null);
     }
+
+    private static Source disabledSourceFrom(InputSpec spec) {
+        return new Source(null, "src", spec.type(), spec.options(), false, "owner", null);
+    }
 }

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

@@ -1,17 +1,23 @@
 package stirling.software.proprietary.policy.input;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
 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.lenient;
+import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
+import java.time.Instant;
+import java.util.ArrayList;
+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;
@@ -24,18 +30,26 @@ import org.springframework.core.env.StandardEnvironment;
 import stirling.software.common.model.ApplicationProperties;
 import stirling.software.common.util.FileReadinessChecker;
 import stirling.software.proprietary.policy.config.FolderAccessGuard;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
 import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.source.InProcessSourceStore;
 
-/** Tests for {@link FolderInputSource}: consume (claim + route) and snapshot (read-only) modes. */
+/**
+ * Tests for {@link FolderInputSource}: consume mode tracks files in place through the ledger,
+ * snapshot stays stateless, and discovery skips hidden entries and honours the recursive option.
+ */
 @ExtendWith(MockitoExtension.class)
 class FolderInputSourceTest {
 
+    private static final String POLICY = "p1";
+
     @Mock private FileReadinessChecker readinessChecker;
 
     @TempDir Path tempDir;
 
     private FolderInputSource source;
+    private InProcessProcessedLedger ledger;
+    private RecordingContext ctx;
 
     @BeforeEach
     void setUp() {
@@ -45,73 +59,279 @@ class FolderInputSourceTest {
                 new FolderAccessGuard(
                         properties, new StandardEnvironment(), new InProcessSourceStore());
         source = new FolderInputSource(readinessChecker, guard);
+        ledger = new InProcessProcessedLedger();
+        ctx = new RecordingContext();
         // Lenient: the missing-dir / nonexistent-dir cases return before any readiness check.
         lenient().when(readinessChecker.isReady(any())).thenReturn(true);
     }
 
     @Test
-    void consumeClaimsFilesAndRoutesToDoneOnSuccess() throws IOException {
+    void consumeRemovesTheFileOnceProcessed() throws IOException {
         Path inputDir = Files.createDirectories(tempDir.resolve("in"));
-        Files.writeString(inputDir.resolve("doc.pdf"), "data");
+        Path file = inputDir.resolve("doc.pdf");
+        Files.writeString(file, "data");
 
-        List<ResolvedInput> work = source.resolve(InputSpec.folder(inputDir.toString()));
+        List<ResolvedInput> work = source.resolve(InputSpec.folder(inputDir.toString()), ctx);
 
         assertEquals(1, work.size());
         assertEquals(1, work.get(0).inputs().primary().size());
-        // Claimed out of the input dir.
-        assertFalse(Files.exists(inputDir.resolve("doc.pdf")));
-        assertTrue(
-                Files.exists(
-                        inputDir.resolve(".stirling").resolve("processing").resolve("doc.pdf")));
+        // In flight: still on disk, but a second sweep does not pick it up again.
+        assertTrue(Files.exists(file));
+        assertTrue(Files.notExists(inputDir.resolve(".stirling")));
+        assertTrue(source.resolve(InputSpec.folder(inputDir.toString()), ctx).isEmpty());
 
         work.get(0).onComplete().accept(true);
-        assertTrue(Files.exists(inputDir.resolve(".stirling").resolve("done").resolve("doc.pdf")));
-        assertFalse(
-                Files.exists(
-                        inputDir.resolve(".stirling").resolve("processing").resolve("doc.pdf")));
+        assertTrue(Files.notExists(file));
+        assertTrue(source.resolve(InputSpec.folder(inputDir.toString()), ctx).isEmpty());
     }
 
     @Test
-    void consumeRoutesToErrorOnFailure() throws IOException {
+    void aFileReplacedMidRunSurvivesTheDeleteAndRunsAgain() throws IOException {
         Path inputDir = Files.createDirectories(tempDir.resolve("in"));
-        Files.writeString(inputDir.resolve("doc.pdf"), "data");
+        Path file = inputDir.resolve("doc.pdf");
+        Files.writeString(file, "data");
 
-        List<ResolvedInput> work = source.resolve(InputSpec.folder(inputDir.toString()));
-        work.get(0).onComplete().accept(false);
+        List<ResolvedInput> work = source.resolve(InputSpec.folder(inputDir.toString()), ctx);
+        // The user saves a new version while the run is executing.
+        Files.writeString(file, "new data, different size");
+        work.get(0).onComplete().accept(true);
 
-        assertTrue(Files.exists(inputDir.resolve(".stirling").resolve("error").resolve("doc.pdf")));
+        // The delete is version-guarded: the replacement is not the file that ran, so it stays
+        // and is claimed as fresh work instead of being marked processed.
+        assertTrue(Files.exists(file));
+        assertEquals(1, source.resolve(InputSpec.folder(inputDir.toString()), ctx).size());
     }
 
     @Test
-    void snapshotReadsWithoutClaiming() throws IOException {
+    void aSharedFileIsRemovedOnlyOnceEveryPolicyHasProcessedIt() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Path file = inputDir.resolve("doc.pdf");
+        Files.writeString(file, "data");
+        InputSpec spec = InputSpec.folder(inputDir.toString());
+        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.
+        assertTrue(Files.exists(file));
+
+        theirs.get(0).onComplete().accept(true);
+        assertTrue(Files.notExists(file));
+    }
+
+    @Test
+    void aSharedFileStaysParkedWhenAnyPolicyFailsOnIt() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Path file = inputDir.resolve("doc.pdf");
+        Files.writeString(file, "data");
+        InputSpec spec = InputSpec.folder(inputDir.toString());
+        RecordingContext other = new RecordingContext("p2");
+
+        List<ResolvedInput> mine = source.resolve(spec, ctx);
+        List<ResolvedInput> theirs = source.resolve(spec, other);
+
+        theirs.get(0).onComplete().accept(false);
+        mine.get(0).onComplete().accept(true);
+
+        // The failure parks the file for everyone (retried when it changes), regardless of
+        // which policy settled last.
+        assertTrue(Files.exists(file));
+        assertTrue(source.resolve(spec, ctx).isEmpty());
+        assertTrue(source.resolve(spec, other).isEmpty());
+    }
+
+    @Test
+    void aReDroppedFileIsProcessedAgain() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Path file = inputDir.resolve("doc.pdf");
+        Files.writeString(file, "data");
+
+        source.resolve(InputSpec.folder(inputDir.toString()), ctx).get(0).onComplete().accept(true);
+        Files.writeString(file, "data again");
+
+        assertEquals(1, source.resolve(InputSpec.folder(inputDir.toString()), ctx).size());
+    }
+
+    @Test
+    void aFailedFileStaysInPlaceAndIsNotRetriedUntilItChanges() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Path file = inputDir.resolve("doc.pdf");
+        Files.writeString(file, "data");
+
+        source.resolve(InputSpec.folder(inputDir.toString()), ctx)
+                .get(0)
+                .onComplete()
+                .accept(false);
+
+        assertTrue(Files.exists(file));
+        assertTrue(source.resolve(InputSpec.folder(inputDir.toString()), ctx).isEmpty());
+
+        Files.setLastModifiedTime(file, FileTime.from(Instant.now().plusSeconds(60)));
+        assertEquals(1, source.resolve(InputSpec.folder(inputDir.toString()), ctx).size());
+    }
+
+    @Test
+    void statModeRetriesAFailureOnATouchButHashModeDoesNot() throws IOException {
+        Path statDir = Files.createDirectories(tempDir.resolve("stat"));
+        Path hashDir = Files.createDirectories(tempDir.resolve("hash"));
+        Path statFile = statDir.resolve("doc.pdf");
+        Path hashFile = hashDir.resolve("doc.pdf");
+        Files.writeString(statFile, "data");
+        Files.writeString(hashFile, "data");
+        InputSpec statSpec = InputSpec.folder(statDir.toString());
+        InputSpec hashSpec =
+                new InputSpec(
+                        "folder", Map.of("directory", hashDir.toString(), "identity", "hash"));
+
+        source.resolve(statSpec, ctx).get(0).onComplete().accept(false);
+        source.resolve(hashSpec, ctx).get(0).onComplete().accept(false);
+
+        FileTime touched = FileTime.from(Instant.now().plusSeconds(60));
+        Files.setLastModifiedTime(statFile, touched);
+        Files.setLastModifiedTime(hashFile, touched);
+
+        // Same content, new mtime: stat mode calls that a new version and retries; hash mode
+        // verifies the content is unchanged and keeps the failure parked.
+        assertEquals(1, source.resolve(statSpec, ctx).size());
+        assertTrue(source.resolve(hashSpec, ctx).isEmpty());
+    }
+
+    @Test
+    void hashModeRetriesAFailureOnARealContentChange() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Path file = inputDir.resolve("doc.pdf");
+        Files.writeString(file, "data");
+        InputSpec spec =
+                new InputSpec(
+                        "folder", Map.of("directory", inputDir.toString(), "identity", "hash"));
+
+        source.resolve(spec, ctx).get(0).onComplete().accept(false);
+        Files.writeString(file, "data v2 - longer");
+
+        assertEquals(1, source.resolve(spec, ctx).size());
+    }
+
+    @Test
+    void snapshotReadsStatelesslyEverySweep() throws IOException {
         Path inputDir = Files.createDirectories(tempDir.resolve("in"));
         Files.writeString(inputDir.resolve("doc.pdf"), "data");
+        InputSpec spec =
+                new InputSpec(
+                        "folder", Map.of("directory", inputDir.toString(), "mode", "snapshot"));
 
-        List<ResolvedInput> work =
-                source.resolve(
-                        new InputSpec(
-                                "folder",
-                                Map.of("directory", inputDir.toString(), "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()); // no ledger involvement: every run sees the full set
+        assertTrue(ctx.present.isEmpty());
+        assertTrue(Files.exists(inputDir.resolve("doc.pdf")));
+    }
+
+    @Test
+    void hiddenFilesAndTheLegacyWorkDirAreIgnored() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Files.writeString(inputDir.resolve("doc.pdf"), "data");
+        Files.writeString(inputDir.resolve(".hidden.pdf"), "secret");
+        Path legacy = Files.createDirectories(inputDir.resolve(".stirling").resolve("done"));
+        Files.writeString(legacy.resolve("old.pdf"), "processed long ago");
+
+        List<ResolvedInput> work = source.resolve(InputSpec.folder(inputDir.toString()), ctx);
 
         assertEquals(1, work.size());
-        // Not moved, and completing the run is a no-op.
-        assertTrue(Files.exists(inputDir.resolve("doc.pdf")));
-        work.get(0).onComplete().accept(true);
-        assertTrue(Files.exists(inputDir.resolve("doc.pdf")));
+        assertEquals(1, ctx.present.size());
+        assertTrue(ctx.present.get(0).endsWith("doc.pdf"));
+    }
+
+    @Test
+    void recursiveDiscoversSubdirectoriesButNotHiddenOnes() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Files.writeString(inputDir.resolve("top.pdf"), "a");
+        Path sub = Files.createDirectories(inputDir.resolve("sub"));
+        Files.writeString(sub.resolve("nested.pdf"), "b");
+        Path hiddenDir = Files.createDirectories(inputDir.resolve(".stirling"));
+        Files.writeString(hiddenDir.resolve("skipped.pdf"), "c");
+        // Sink staging inside a watched subdirectory is pruned at any depth.
+        Path nestedStaging = Files.createDirectories(sub.resolve(".stirling").resolve("tmp"));
+        Files.writeString(nestedStaging.resolve("half-delivered"), "d");
+
+        InputSpec flat = InputSpec.folder(inputDir.toString());
+        InputSpec recursive =
+                new InputSpec(
+                        "folder", Map.of("directory", inputDir.toString(), "recursive", "true"));
+
+        assertEquals(1, source.resolve(flat, ctx).size());
+        assertEquals(1, source.resolve(recursive, ctx).size()); // top.pdf already claimed above
+        assertTrue(ctx.present.stream().anyMatch(identity -> identity.endsWith("nested.pdf")));
+        assertTrue(ctx.present.stream().noneMatch(identity -> identity.endsWith("skipped.pdf")));
+        assertTrue(ctx.present.stream().noneMatch(identity -> identity.endsWith("half-delivered")));
+    }
+
+    @Test
+    void unreadyFilesAreReportedPresentButNotClaimed() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Path file = inputDir.resolve("mid-write.pdf");
+        Files.writeString(file, "partial");
+        when(readinessChecker.isReady(file)).thenReturn(false);
+
+        List<ResolvedInput> work = source.resolve(InputSpec.folder(inputDir.toString()), ctx);
+
+        assertTrue(work.isEmpty());
+        // Reported present so a full sweep does not prune its row while it settles on disk.
+        assertEquals(1, ctx.present.size());
+        assertTrue(ctx.present.get(0).endsWith("mid-write.pdf"));
+    }
+
+    @Test
+    void nestedSourcesShareThePolicysLedgerAndDoNotDoubleClaim() throws IOException {
+        Path parent = Files.createDirectories(tempDir.resolve("in"));
+        Path child = Files.createDirectories(parent.resolve("sub"));
+        Files.writeString(child.resolve("doc.pdf"), "data");
+        InputSpec parentRecursive =
+                new InputSpec(
+                        "folder", Map.of("directory", parent.toString(), "recursive", "true"));
+        InputSpec childFlat = InputSpec.folder(child.toString());
+
+        // Same sweep, same policy context: whichever source resolves first wins the file.
+        assertEquals(1, source.resolve(parentRecursive, ctx).size());
+        assertTrue(source.resolve(childFlat, ctx).isEmpty());
     }
 
     @Test
     void missingDirectoryOptionFails() {
         assertThrows(
                 IllegalArgumentException.class,
-                () -> source.resolve(new InputSpec("folder", Map.of())));
+                () -> source.resolve(new InputSpec("folder", Map.of()), ctx));
     }
 
     @Test
-    void nonexistentDirectoryYieldsNoWork() throws IOException {
-        List<ResolvedInput> work =
-                source.resolve(InputSpec.folder(tempDir.resolve("nope").toString()));
-        assertTrue(work.isEmpty());
+    void anUnknownIdentityModeIsRejected() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () ->
+                        source.validate(
+                                new InputSpec(
+                                        "folder",
+                                        Map.of(
+                                                "directory",
+                                                tempDir.toString(),
+                                                "identity",
+                                                "guesswork"))));
+    }
+
+    @Test
+    void nonexistentDirectoryFailsResolveSoTheSweepVetoesCleanup() {
+        // An unreachable directory (e.g. unmounted drive) must surface as a failed listing, not
+        // an empty one: the runner vetoes presence cleanup on failure, keeping the history that
+        // an empty listing would wipe.
+        assertThrows(
+                NoSuchFileException.class,
+                () -> source.resolve(InputSpec.folder(tempDir.resolve("nope").toString()), ctx));
     }
 
     @Test
@@ -126,7 +346,7 @@ class FolderInputSourceTest {
         Path outside = tempDir.resolveSibling("not-allowed");
         assertThrows(
                 IllegalArgumentException.class,
-                () -> source.resolve(InputSpec.folder(outside.toString())));
+                () -> source.resolve(InputSpec.folder(outside.toString()), ctx));
         assertThrows(
                 IllegalArgumentException.class,
                 () -> source.validate(InputSpec.folder(outside.toString())));
@@ -137,4 +357,40 @@ class FolderInputSourceTest {
         Path inputDir = tempDir.resolve("in");
         assertEquals(List.of(inputDir), source.watchTargets(InputSpec.folder(inputDir.toString())));
     }
+
+    /** Policy-scoped context backed by the in-process ledger, recording presence reports. */
+    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/ledger/FolderIdentitiesTest.java

@@ -0,0 +1,84 @@
+package stirling.software.proprietary.policy.ledger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
+import java.time.Instant;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Tests for {@link FolderIdentities}: identity derivation must agree for the same file, even
+ * through a symlinked alias of the directory.
+ */
+class FolderIdentitiesTest {
+
+    @TempDir Path tempDir;
+
+    @Test
+    void identityAgreesAcrossASymlinkedAliasOfTheDirectory() throws IOException {
+        Path real = Files.createDirectories(tempDir.resolve("real"));
+        Path alias = Files.createSymbolicLink(tempDir.resolve("alias"), real);
+        Files.writeString(real.resolve("doc.pdf"), "data");
+
+        String viaReal =
+                FolderIdentities.identity(
+                        FolderIdentities.canonicalDir(real), real, real.resolve("doc.pdf"));
+        String viaAlias =
+                FolderIdentities.identity(
+                        FolderIdentities.canonicalDir(alias), alias, alias.resolve("doc.pdf"));
+
+        assertEquals(viaReal, viaAlias);
+    }
+
+    @Test
+    void identityOfANestedFileKeepsItsRelativePath() throws IOException {
+        Path dir = Files.createDirectories(tempDir.resolve("in"));
+        Path nested = Files.createDirectories(dir.resolve("sub")).resolve("doc.pdf");
+        Files.writeString(nested, "data");
+
+        String identity =
+                FolderIdentities.identity(FolderIdentities.canonicalDir(dir), dir, nested);
+
+        assertTrue(identity.endsWith("sub" + java.io.File.separator + "doc.pdf"));
+    }
+
+    @Test
+    void theGateTracksSizeAndMtime() throws IOException {
+        Path file = tempDir.resolve("doc.pdf");
+        Files.writeString(file, "data");
+        String before = FolderIdentities.statGate(file);
+
+        Files.setLastModifiedTime(file, FileTime.from(Instant.now().plusSeconds(60)));
+
+        assertNotEquals(before, FolderIdentities.statGate(file));
+    }
+
+    @Test
+    void theContentHashIgnoresMtimeButTracksContent() throws IOException {
+        Path file = tempDir.resolve("doc.pdf");
+        Files.writeString(file, "data");
+        String before = FolderIdentities.contentHash(file);
+
+        Files.setLastModifiedTime(file, FileTime.from(Instant.now().plusSeconds(60)));
+        assertEquals(before, FolderIdentities.contentHash(file));
+
+        Files.writeString(file, "different");
+        assertNotEquals(before, FolderIdentities.contentHash(file));
+    }
+
+    @Test
+    void identityHashIsAStableFixedWidthKey() {
+        String hash = IdentityHasher.identityHash("/in/doc.pdf");
+
+        assertEquals(64, hash.length());
+        assertEquals(hash, IdentityHasher.identityHash("/in/doc.pdf"));
+        assertNotEquals(hash, IdentityHasher.identityHash("/in/other.pdf"));
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/InProcessProcessedLedgerTest.java

@@ -0,0 +1,12 @@
+package stirling.software.proprietary.policy.ledger;
+
+import java.util.function.Supplier;
+
+/** {@link InProcessProcessedLedger} against the shared {@link ProcessedLedger} contract. */
+class InProcessProcessedLedgerTest extends ProcessedLedgerContractTest {
+
+    @Override
+    ProcessedLedger newLedger(Supplier<Long> nowMillis) {
+        return new InProcessProcessedLedger(nowMillis);
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/JpaProcessedLedgerDbTest.java

@@ -0,0 +1,36 @@
+package stirling.software.proprietary.policy.ledger;
+
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.AfterEach;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.AutoConfigurationPackage;
+import org.springframework.boot.data.jpa.test.autoconfigure.DataJpaTest;
+
+/**
+ * {@link JpaProcessedLedger} against the shared contract on a real (H2) database. The inherited
+ * tests run outside {@code @DataJpaTest}'s per-test transaction (the transaction attribute resolves
+ * against the declaring class, the plain contract base), so every ledger call commits in its own
+ * transaction as at runtime; state is wiped explicitly instead of relying on rollback.
+ */
+@DataJpaTest
+class JpaProcessedLedgerDbTest extends ProcessedLedgerContractTest {
+
+    @Autowired private ProcessedFileRepository repository;
+
+    @AfterEach
+    void wipeLedger() {
+        // deleteAll() skips entities whose isNew() is hardcoded true, so use the bulk form.
+        repository.deleteAllInBatch();
+    }
+
+    @Override
+    ProcessedLedger newLedger(Supplier<Long> nowMillis) {
+        return new JpaProcessedLedger(repository, nowMillis);
+    }
+
+    @SpringBootConfiguration
+    @AutoConfigurationPackage
+    static class TestApp {}
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/ProcessedLedgerContractTest.java

@@ -0,0 +1,360 @@
+package stirling.software.proprietary.policy.ledger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** The {@link ProcessedLedger} contract, run against every implementation so they cannot drift. */
+abstract class ProcessedLedgerContractTest {
+
+    static final String POLICY = "p1";
+    static final String OTHER_POLICY = "p2";
+    static final String FILE = "/in/doc.pdf";
+    static final String GATE = "100:1111";
+    static final String NEW_GATE = "100:2222";
+    static final String HASH = "hash-aaa";
+    static final String NEW_HASH = "hash-bbb";
+
+    final AtomicLong clock = new AtomicLong(1_000_000L);
+
+    ProcessedLedger ledger;
+
+    abstract ProcessedLedger newLedger(Supplier<Long> nowMillis);
+
+    @BeforeEach
+    void createLedger() {
+        ledger = newLedger(clock::get);
+    }
+
+    @Test
+    void aFileIsClaimedOnceAndSkippedWhileInFlight() {
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null));
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null));
+        assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, null)); // new version waits for settle
+    }
+
+    @Test
+    void aSettledFileIsSkippedAtTheSameGate() {
+        ledger.claim(POLICY, FILE, GATE, null);
+        ledger.settle(POLICY, FILE, GATE, null, true);
+
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null));
+    }
+
+    @Test
+    void aMovedGateIsReclaimedInGateOnlyMode() {
+        ledger.claim(POLICY, FILE, GATE, null);
+        ledger.settle(POLICY, FILE, GATE, null, true);
+
+        assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null));
+    }
+
+    @Test
+    void settlingAtTheOutputsVersionStopsAnInPlaceOverwriteLooping() {
+        ledger.claim(POLICY, FILE, GATE, null);
+        // The run overwrote the input; settle re-reads and lands on the produced version.
+        ledger.settle(POLICY, FILE, NEW_GATE, null, true);
+
+        assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, null)); // own output: skip
+        assertTrue(ledger.claim(POLICY, FILE, "100:3333", null)); // later user edit: reprocess
+    }
+
+    @Test
+    void aFailedFileIsNotRetriedUntilItChanges() {
+        ledger.claim(POLICY, FILE, GATE, null);
+        ledger.settle(POLICY, FILE, GATE, null, false);
+
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null));
+        assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null));
+    }
+
+    @Test
+    void aTouchedButUnchangedFileRefreshesTheGateInsteadOfReprocessing() {
+        ledger.claim(POLICY, FILE, GATE, hash(HASH));
+        ledger.settle(POLICY, FILE, GATE, HASH, true);
+
+        // Same content under a new gate (touch / identical re-copy): verified, not reprocessed.
+        assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, hash(HASH)));
+
+        // The gate was refreshed, so the next sweep takes the cheap path: no content read at all.
+        CountingSupplier counting = new CountingSupplier(HASH);
+        assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, counting));
+        assertEquals(0, counting.invocations.get());
+    }
+
+    @Test
+    void theVerificationTierIsNotConsultedWhileTheGateMatches() {
+        ledger.claim(POLICY, FILE, GATE, hash(HASH));
+        ledger.settle(POLICY, FILE, GATE, HASH, true);
+
+        CountingSupplier counting = new CountingSupplier(HASH);
+        assertFalse(ledger.claim(POLICY, FILE, GATE, counting));
+        assertEquals(0, counting.invocations.get());
+    }
+
+    @Test
+    void aRealContentChangeUnderANewGateIsReprocessed() {
+        ledger.claim(POLICY, FILE, GATE, hash(HASH));
+        ledger.settle(POLICY, FILE, GATE, HASH, true);
+
+        assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, hash(NEW_HASH)));
+    }
+
+    @Test
+    void aGateOnlySettledRowCannotBeContentVerifiedSoItReprocesses() {
+        ledger.claim(POLICY, FILE, GATE, null);
+        ledger.settle(POLICY, FILE, GATE, null, true);
+
+        // The row stored no hash, so "same content" is unprovable: reprocess on gate change.
+        assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, hash(HASH)));
+    }
+
+    @Test
+    void aFailedFileStaysParkedThroughATouch() {
+        ledger.claim(POLICY, FILE, GATE, hash(HASH));
+        ledger.settle(POLICY, FILE, GATE, HASH, false);
+
+        // A touch must not resurrect an ERROR row; only a real content change does.
+        assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, hash(HASH)));
+        assertTrue(ledger.claim(POLICY, FILE, "100:3333", hash(NEW_HASH)));
+    }
+
+    @Test
+    void interruptedRunsAreRetriedABoundedNumberOfTimes() {
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null)); // attempt 1 dies with the JVM
+        ledger.recoverInterrupted();
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null)); // attempt 2
+        ledger.recoverInterrupted();
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null)); // attempt 3, the last
+        ledger.recoverInterrupted();
+
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null)); // parked: no crash-loop
+    }
+
+    @Test
+    void aNewGateResetsTheInterruptRetryBudgetInGateOnlyMode() {
+        for (int attempt = 0; attempt < ProcessedLedger.MAX_ATTEMPTS; attempt++) {
+            ledger.claim(POLICY, FILE, GATE, null);
+            ledger.recoverInterrupted();
+        }
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null));
+
+        assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null));
+        ledger.recoverInterrupted();
+        assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null)); // fresh budget at the new version
+    }
+
+    @Test
+    void aTouchDoesNotResetTheInterruptRetryBudgetWhenContentIsVerified() {
+        assertTrue(ledger.claim(POLICY, FILE, GATE, hash(HASH))); // attempt 1
+        ledger.recoverInterrupted();
+        // Same content, moved gate: still the interrupted work, still bounded.
+        assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, hash(HASH))); // attempt 2
+        ledger.recoverInterrupted();
+        assertTrue(ledger.claim(POLICY, FILE, "100:3333", hash(HASH))); // attempt 3
+        ledger.recoverInterrupted();
+
+        assertFalse(ledger.claim(POLICY, FILE, "100:4444", hash(HASH))); // parked
+        assertTrue(ledger.claim(POLICY, FILE, "100:5555", hash(NEW_HASH))); // real change: fresh
+    }
+
+    @Test
+    void recoveryOnlyTouchesInFlightRows() {
+        ledger.claim(POLICY, FILE, GATE, null);
+        ledger.settle(POLICY, FILE, GATE, null, true);
+        ledger.recoverInterrupted();
+
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null)); // still DONE, not retried
+    }
+
+    @Test
+    void anOutputIsSkippedByItsProducerButSeenByOtherPolicies() {
+        ledger.recordOutput(POLICY, FILE, GATE, HASH);
+
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null)); // producer skips its own output
+        assertTrue(ledger.claim(OTHER_POLICY, FILE, GATE, null)); // chaining still works
+    }
+
+    @Test
+    void anOutputIsSkippedByAHashVerifyingProducerEvenIfTheGateMoved() {
+        ledger.recordOutput(POLICY, FILE, GATE, HASH);
+
+        assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, hash(HASH)));
+    }
+
+    @Test
+    void policiesTrackTheSameFileIndependently() {
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null));
+        assertTrue(ledger.claim(OTHER_POLICY, FILE, GATE, null));
+    }
+
+    @Test
+    void statesForSnapshotsOnlyExistingRows() {
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null));
+
+        Map<String, ClaimState> states = ledger.statesFor(POLICY, List.of(FILE, "/in/other.pdf"));
+
+        assertEquals(1, states.size());
+        assertEquals(ProcessedFileStatus.PROCESSING, states.get(FILE).status());
+        assertEquals(GATE, states.get(FILE).gate());
+    }
+
+    @Test
+    void aStaleAbsentSnapshotLosesTheClaimRaceInsteadOfDoubleClaiming() {
+        ClaimState absent = ledger.statesFor(POLICY, List.of(FILE)).get(FILE); // no row yet
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null)); // another sweep wins meanwhile
+
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null, absent));
+    }
+
+    @Test
+    void aStaleSettledSnapshotCannotReclaimAnInFlightRow() {
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null));
+        ledger.settle(POLICY, FILE, GATE, null, true);
+        ClaimState settled = ledger.statesFor(POLICY, List.of(FILE)).get(FILE);
+        assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null)); // a fresh sweep reclaims first
+
+        assertFalse(ledger.claim(POLICY, FILE, "100:3333", null, settled));
+    }
+
+    @Test
+    void aForgottenOutputIsClaimableAtAnyVersion() {
+        ledger.recordOutput(POLICY, FILE, GATE, HASH);
+        ledger.forgetOutput(POLICY, FILE, GATE);
+
+        // Even a byte-identical file at that identity is fresh work: the record is gone.
+        assertTrue(ledger.claim(POLICY, FILE, GATE, hash(HASH)));
+    }
+
+    @Test
+    void forgetOutputLeavesARowReclaimedInTheMeantime() {
+        ledger.recordOutput(POLICY, FILE, GATE, HASH);
+        assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null)); // a real claim took the row over
+
+        ledger.forgetOutput(POLICY, FILE, GATE);
+
+        assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, null)); // still in flight, not deleted
+    }
+
+    @Test
+    void deletionConsensusNeedsEveryClaimantSettledDone() {
+        assertTrue(ledger.allSettledDone(FILE)); // vacuous: no rows yet
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null));
+        assertTrue(ledger.claim(OTHER_POLICY, FILE, GATE, null));
+        ledger.settle(POLICY, FILE, GATE, null, true);
+        assertFalse(ledger.allSettledDone(FILE)); // the other claim is still in flight
+        ledger.settle(OTHER_POLICY, FILE, GATE, null, true);
+        assertTrue(ledger.allSettledDone(FILE));
+    }
+
+    @Test
+    void aFailedClaimVetoesDeletionConsensus() {
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null));
+        assertTrue(ledger.claim(OTHER_POLICY, FILE, GATE, null));
+        ledger.settle(POLICY, FILE, GATE, null, true);
+        ledger.settle(OTHER_POLICY, FILE, GATE, null, false);
+        assertFalse(ledger.allSettledDone(FILE));
+    }
+
+    @Test
+    void anInterruptedClaimVetoesDeletionConsensus() {
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null));
+        ledger.recoverInterrupted();
+        assertFalse(ledger.allSettledDone(FILE));
+    }
+
+    @Test
+    void settleRecreatesARowRemovedMidRun() {
+        ledger.claim(POLICY, FILE, GATE, null);
+        ledger.clearPolicy(POLICY); // e.g. a clear-history while the run is in flight
+        ledger.settle(POLICY, FILE, GATE, null, true);
+
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null));
+    }
+
+    @Test
+    void presenceCleanupRemovesOnlyUnseenSettledRows() {
+        String inFlight = "/in/in-flight.pdf";
+        String stillPresent = "/in/still-present.pdf";
+        String deleted = "/in/deleted.pdf";
+        ledger.claim(POLICY, inFlight, GATE, null);
+        ledger.recordOutput(POLICY, stillPresent, GATE, HASH);
+        ledger.recordOutput(POLICY, deleted, GATE, HASH);
+
+        clock.addAndGet(10_000);
+        long sweepStart = clock.get();
+        ledger.markSeen(POLICY, List.of(inFlight, stillPresent)); // deleted.pdf is gone from disk
+        assertEquals(1, ledger.deleteUnseen(POLICY, sweepStart));
+
+        assertFalse(ledger.claim(POLICY, inFlight, GATE, null)); // in-flight row survived
+        assertFalse(ledger.claim(POLICY, stillPresent, GATE, null)); // stamped row survived
+        assertTrue(ledger.claim(POLICY, deleted, GATE, null)); // forgotten: a re-drop reprocesses
+    }
+
+    @Test
+    void presenceCleanupNeverRemovesInFlightRowsEvenUnstamped() {
+        ledger.claim(POLICY, FILE, GATE, null);
+        clock.addAndGet(10_000);
+
+        assertEquals(0, ledger.deleteUnseen(POLICY, clock.get()));
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null));
+    }
+
+    @Test
+    void rowsWrittenDuringTheSweepSurviveItsCleanup() {
+        long sweepStart = clock.get();
+        // Recorded after the sweep's cutoff: unseen by it, but newer, so it must survive.
+        clock.addAndGet(5);
+        ledger.recordOutput(POLICY, FILE, GATE, HASH);
+
+        assertEquals(0, ledger.deleteUnseen(POLICY, sweepStart));
+        assertFalse(ledger.claim(POLICY, FILE, GATE, null));
+    }
+
+    @Test
+    void markSeenOnUnknownIdentitiesIsANoOp() {
+        ledger.markSeen(POLICY, List.of("/never/claimed.pdf"));
+        assertEquals(0, ledger.deleteUnseen(POLICY, clock.get()));
+    }
+
+    @Test
+    void clearPolicyForgetsOnlyThatPolicy() {
+        ledger.claim(POLICY, FILE, GATE, null);
+        ledger.settle(POLICY, FILE, GATE, null, true);
+        ledger.claim(OTHER_POLICY, FILE, GATE, null);
+        ledger.settle(OTHER_POLICY, FILE, GATE, null, true);
+
+        ledger.clearPolicy(POLICY);
+
+        assertTrue(ledger.claim(POLICY, FILE, GATE, null));
+        assertFalse(ledger.claim(OTHER_POLICY, FILE, GATE, null));
+    }
+
+    static Supplier<String> hash(String value) {
+        return () -> value;
+    }
+
+    static final class CountingSupplier implements Supplier<String> {
+        final AtomicInteger invocations = new AtomicInteger();
+        private final String value;
+
+        CountingSupplier(String value) {
+            this.value = value;
+        }
+
+        @Override
+        public String get() {
+            invocations.incrementAndGet();
+            return value;
+        }
+    }
+}

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

@@ -10,6 +10,7 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Stream;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -21,24 +22,35 @@ import org.springframework.core.io.Resource;
 import stirling.software.common.model.ApplicationProperties;
 import stirling.software.common.model.job.ResultFile;
 import stirling.software.proprietary.policy.config.FolderAccessGuard;
+import stirling.software.proprietary.policy.ledger.FolderIdentities;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
 import stirling.software.proprietary.policy.model.OutputSpec;
 import stirling.software.proprietary.policy.source.InProcessSourceStore;
 
-/** Tests for {@link FolderOutputSink}: outputs are written to the configured directory on disk. */
+/**
+ * Tests for {@link FolderOutputSink}: outputs are staged hidden, recorded in the ledger, then
+ * atomically renamed into the configured directory.
+ */
 class FolderOutputSinkTest {
 
+    private static final OutputDelivery AD_HOC = new OutputDelivery("run-1", null);
+    private static final OutputDelivery POLICY_RUN = new OutputDelivery("run-1", "p1");
+
     @TempDir Path tempDir;
 
     private FolderOutputSink sink;
+    private InProcessProcessedLedger ledger;
 
     @BeforeEach
     void setUp() {
         ApplicationProperties properties = new ApplicationProperties();
         properties.getPolicies().setAllowedFolderRoots(List.of(tempDir.toString()));
+        ledger = new InProcessProcessedLedger();
         sink =
                 new FolderOutputSink(
                         new FolderAccessGuard(
-                                properties, new StandardEnvironment(), new InProcessSourceStore()));
+                                properties, new StandardEnvironment(), new InProcessSourceStore()),
+                        ledger);
     }
 
     @Test
@@ -46,13 +58,80 @@ class FolderOutputSinkTest {
         Path out = tempDir.resolve("out");
         List<Resource> outputs = List.of(named("a.pdf", "aaa"), named("b.pdf", "bb"));
 
-        List<ResultFile> results =
-                sink.deliver("run-1", outputs, OutputSpec.folder(out.toString()));
+        List<ResultFile> results = sink.deliver(AD_HOC, outputs, OutputSpec.folder(out.toString()));
 
         assertEquals(2, results.size());
         assertTrue(Files.exists(out.resolve("a.pdf")));
         assertEquals("aaa", Files.readString(out.resolve("a.pdf")));
         assertEquals("bb", Files.readString(out.resolve("b.pdf")));
+        // Nothing left behind in the staging dir.
+        try (Stream<Path> staged = Files.list(out.resolve(".stirling").resolve("tmp"))) {
+            assertEquals(0, staged.count());
+        }
+    }
+
+    @Test
+    void recordsThePolicysOutputsSoOnlyOtherPoliciesReprocessThem() throws IOException {
+        Path out = tempDir.resolve("out");
+
+        sink.deliver(POLICY_RUN, List.of(named("a.pdf", "aaa")), OutputSpec.folder(out.toString()));
+
+        Path delivered = FolderIdentities.canonicalDir(out).resolve("a.pdf");
+        String gate = FolderIdentities.statGate(delivered);
+        assertFalse(ledger.claim("p1", delivered.toString(), gate, null)); // producer skips it
+        assertTrue(ledger.claim("p2", delivered.toString(), gate, null)); // chaining still works
+    }
+
+    @Test
+    void aHashVerifyingProducerSkipsItsOwnOutputEvenIfTheGateMoved() throws IOException {
+        Path out = tempDir.resolve("out");
+
+        sink.deliver(POLICY_RUN, List.of(named("a.pdf", "aaa")), OutputSpec.folder(out.toString()));
+
+        // A hash-verifying reader matches on content even when the stat moved.
+        Path delivered = FolderIdentities.canonicalDir(out).resolve("a.pdf");
+        assertFalse(
+                ledger.claim(
+                        "p1",
+                        delivered.toString(),
+                        "999:12345",
+                        () -> {
+                            try {
+                                return FolderIdentities.contentHash(delivered);
+                            } catch (IOException e) {
+                                throw new java.io.UncheckedIOException(e);
+                            }
+                        }));
+    }
+
+    @Test
+    void recordsAnOutputBeforeItBecomesVisible() throws IOException {
+        Path out = tempDir.resolve("out");
+        VisibilityAssertingLedger orderedLedger = new VisibilityAssertingLedger();
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowedFolderRoots(List.of(tempDir.toString()));
+        FolderOutputSink orderedSink =
+                new FolderOutputSink(
+                        new FolderAccessGuard(
+                                properties, new StandardEnvironment(), new InProcessSourceStore()),
+                        orderedLedger);
+
+        orderedSink.deliver(
+                POLICY_RUN, List.of(named("a.pdf", "aaa")), OutputSpec.folder(out.toString()));
+
+        assertTrue(orderedLedger.recorded);
+        assertTrue(Files.exists(out.resolve("a.pdf")));
+    }
+
+    @Test
+    void adHocDeliveriesRecordNothing() throws IOException {
+        Path out = tempDir.resolve("out");
+
+        sink.deliver(AD_HOC, List.of(named("a.pdf", "aaa")), OutputSpec.folder(out.toString()));
+
+        Path delivered = FolderIdentities.canonicalDir(out).resolve("a.pdf");
+        // No row was recorded, so any policy (including a hypothetical producer) may claim it.
+        assertTrue(ledger.claim("p1", delivered.toString(), "any-gate", null));
     }
 
     @Test
@@ -60,7 +139,7 @@ class FolderOutputSinkTest {
         Path out = tempDir.resolve("out");
         List<Resource> outputs = List.of(named("a.pdf", "first"), named("a.pdf", "second"));
 
-        sink.deliver("run-1", outputs, OutputSpec.folder(out.toString()));
+        sink.deliver(AD_HOC, outputs, OutputSpec.folder(out.toString()));
 
         assertTrue(Files.exists(out.resolve("a.pdf")));
         assertTrue(Files.exists(out.resolve("a (1).pdf")));
@@ -72,7 +151,7 @@ class FolderOutputSinkTest {
         assertThrows(IllegalArgumentException.class, () -> sink.validate(noDir));
         assertThrows(
                 IllegalArgumentException.class,
-                () -> sink.deliver("run-1", List.of(named("a.pdf", "x")), noDir));
+                () -> sink.deliver(AD_HOC, List.of(named("a.pdf", "x")), noDir));
     }
 
     @Test
@@ -81,7 +160,7 @@ class FolderOutputSinkTest {
         assertThrows(IllegalArgumentException.class, () -> sink.validate(outside));
         assertThrows(
                 IllegalArgumentException.class,
-                () -> sink.deliver("run-1", List.of(named("a.pdf", "x")), outside));
+                () -> sink.deliver(AD_HOC, List.of(named("a.pdf", "x")), outside));
     }
 
     @Test
@@ -90,7 +169,7 @@ class FolderOutputSinkTest {
         List<Resource> outputs =
                 List.of(named("../escape.pdf", "x"), named("nested/deep.pdf", "y"));
 
-        sink.deliver("run-1", outputs, OutputSpec.folder(out.toString()));
+        sink.deliver(AD_HOC, outputs, OutputSpec.folder(out.toString()));
 
         // Each name is reduced to its bare form inside the target dir; nothing escapes.
         assertTrue(Files.exists(out.resolve("escape.pdf")));
@@ -106,4 +185,20 @@ class FolderOutputSinkTest {
             }
         };
     }
+
+    /** Fails the delivery if an output is visible at its final path before being recorded. */
+    private static class VisibilityAssertingLedger extends InProcessProcessedLedger {
+
+        private boolean recorded;
+
+        @Override
+        public synchronized void recordOutput(
+                String policyId, String identity, String gate, String contentHash) {
+            assertFalse(
+                    Files.exists(Path.of(identity)),
+                    "output must be recorded before it is visible at its final path");
+            recorded = true;
+            super.recordOutput(policyId, identity, gate, contentHash);
+        }
+    }
 }

app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/FolderWatchTriggerTest.java

@@ -3,6 +3,7 @@ package stirling.software.proprietary.policy.trigger;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
@@ -26,6 +27,7 @@ import org.mockito.junit.jupiter.MockitoExtension;
 
 import stirling.software.common.model.ApplicationProperties;
 import stirling.software.proprietary.policy.engine.PolicyRunner;
+import stirling.software.proprietary.policy.engine.SweepKind;
 import stirling.software.proprietary.policy.input.InputSource;
 import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.OutputSpec;
@@ -100,8 +102,8 @@ class FolderWatchTriggerTest {
 
         trigger.runForChangedDirs(Set.of(normalized("/in/a")));
 
-        verify(policyRunner).run(a);
-        verify(policyRunner, never()).run(b);
+        verify(policyRunner).run(a, SweepKind.LIGHT);
+        verify(policyRunner, never()).run(eq(b), any());
     }
 
     @Test
@@ -112,8 +114,8 @@ class FolderWatchTriggerTest {
 
         trigger.runForChangedDirs(Set.of(normalized("/in/a")));
 
-        verify(policyRunner).run(good);
-        verify(policyRunner, never()).run(bad);
+        verify(policyRunner).run(good, SweepKind.LIGHT);
+        verify(policyRunner, never()).run(eq(bad), any());
     }
 
     @Test

app/saas/src/main/resources/db/migration/saas/V32__policy_processed_files.sql

@@ -0,0 +1,30 @@
+-- Per-policy processed-file ledger:
+--
+--   policy_processed_files  one row per (policy, file identity) recording the version a policy
+--                           last settled that file at, so folder sources track files in place
+--                           instead of moving them into a work directory. signature is a cheap
+--                           version gate (folder: size:mtime); content_hash an optional strong
+--                           token consulted only when the gate moves. Rows are claimed into
+--                           PROCESSING, settled to DONE/ERROR, flipped to INTERRUPTED at boot if
+--                           a run died with the JVM, and pruned once the file is gone from all of
+--                           the policy's sources, so the table stays near the set of files
+--                           currently present.
+--
+-- Gated by policies.enabled like the rest of the subsystem; Hibernate ddl-auto would also create
+-- this, but the migration keeps the schema explicit for the Flyway-managed deployments.
+
+CREATE TABLE IF NOT EXISTS policy_processed_files (
+    policy_id     VARCHAR(255)  NOT NULL,
+    identity_hash VARCHAR(64)   NOT NULL,
+    identity      VARCHAR(4096),
+    signature     VARCHAR(255)  NOT NULL,
+    content_hash  VARCHAR(64),
+    status        VARCHAR(16)   NOT NULL,
+    attempts      SMALLINT      NOT NULL DEFAULT 1,
+    last_seen     BIGINT        NOT NULL DEFAULT 0,
+    updated_at    BIGINT        NOT NULL DEFAULT 0,
+    PRIMARY KEY (policy_id, identity_hash)
+);
+
+CREATE INDEX IF NOT EXISTS idx_processed_files_policy_seen
+    ON policy_processed_files (policy_id, last_seen);

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

@@ -7327,6 +7327,7 @@ showMore = "Show more"
 sources = "Sources"
 
 [portal.policies.detail.actions]
+clearHistory = "Clear processed history"
 delete = "Delete"
 editSettings = "Edit settings"
 pause = "Pause"
@@ -7896,6 +7897,14 @@ helperText = "Absolute path Stirling watches for files to process."
 label = "Directory path"
 placeholder = "/data/incoming"
 
+[portal.sources.types.folder.fields.identity]
+helperText = "Content check reads each changed file, so renames and touches that don't alter content are not reprocessed."
+label = "Change detection"
+
+[portal.sources.types.folder.fields.identity.options]
+hash = "Size, date and content check"
+stat = "Size and date modified"
+
 [portal.sources.types.folder.fields.mode]
 label = "Read mode"
 
@@ -7903,6 +7912,13 @@ label = "Read mode"
 consume = "Consume: process each file once"
 snapshot = "Snapshot: re-read the folder every run"
 
+[portal.sources.types.folder.fields.recursive]
+label = "Folder depth"
+
+[portal.sources.types.folder.fields.recursive.options]
+all = "Include subfolders"
+top = "Top level only"
+
 [portal.sources.types.unknown]
 label = "Source"
 

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

@@ -177,6 +177,19 @@ export async function deletePolicy(id: string): Promise<void> {
   );
 }
 
+/**
+ * DELETE /api/v1/policies/{id}/processed-history — forget which source files
+ * the policy has processed, so its next sweep reprocesses everything present.
+ */
+export async function clearProcessedHistory(id: string): Promise<void> {
+  await apiClient.local.json<void>(
+    `/api/v1/policies/${encodeURIComponent(id)}/processed-history`,
+    {
+      method: "DELETE",
+    },
+  );
+}
+
 // ── Wire-build helpers (so Policies.tsx doesn't need codec knowledge) ────────
 
 const DEFAULT_RETRIES = 3;

frontend/editor/src/portal/components/policies/PolicyDetailPanel.tsx

@@ -23,6 +23,7 @@ interface PolicyDetailPanelProps {
   onRun?: () => void;
   onTogglePause: () => void;
   onDelete: () => void;
+  onClearHistory?: () => void;
   onRetry?: (item: PolicyActivityItem) => void;
 }
 
@@ -106,6 +107,7 @@ export function PolicyDetailPanel({
   onRun,
   onTogglePause,
   onDelete,
+  onClearHistory,
   onRetry,
 }: PolicyDetailPanelProps) {
   const { t } = useTranslation();
@@ -113,6 +115,9 @@ export function PolicyDetailPanel({
   const { category, config, state, steps, stats, activity } = policy;
   const isPaused = state.status === "paused";
   const canDelete = state.isDefault !== true;
+  // Processed history only exists for watched sources; editor uploads are never ledgered.
+  const canClearHistory =
+    onClearHistory !== undefined && state.sources.some((s) => s !== "editor");
 
   const enforceItems = steps.length > 0 ? steps.map((s) => s.operation) : null;
   const hasEditorSource = state.sources.includes("editor");
@@ -161,6 +166,16 @@ export function PolicyDetailPanel({
               {t("portal.policies.detail.actions.runNow")}
             </Button>
           )}
+          {canClearHistory && (
+            <Button
+              variant="secondary"
+              size="sm"
+              onClick={onClearHistory}
+              disabled={busy}
+            >
+              {t("portal.policies.detail.actions.clearHistory")}
+            </Button>
+          )}
           <Button
             variant="secondary"
             size="sm"

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

@@ -61,7 +61,12 @@ describe("ConnectWizard", () => {
     expect(createSource).toHaveBeenCalledWith({
       name: "Claims intake",
       type: "folder",
-      options: { directory: "/data/incoming", mode: "consume" },
+      options: {
+        directory: "/data/incoming",
+        mode: "consume",
+        recursive: "false",
+        identity: "stat",
+      },
       enabled: true,
     });
     await waitFor(() => {
@@ -101,11 +106,17 @@ describe("ConnectWizard", () => {
     await waitFor(() => {
       expect(createSource).toHaveBeenCalledTimes(1);
     });
+    // Options absent from the stored source are submitted at their defaults.
     expect(createSource).toHaveBeenCalledWith({
       id: "s1",
       name: "James",
       type: "folder",
-      options: { directory: "/data/in", mode: "consume" },
+      options: {
+        directory: "/data/in",
+        mode: "consume",
+        recursive: "false",
+        identity: "stat",
+      },
       enabled: true,
     });
     await waitFor(() => {

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

@@ -91,6 +91,43 @@ export const CREATABLE_SOURCE_TYPES: CreatableSourceType[] = [
           },
         ],
       },
+      {
+        key: "recursive",
+        labelKey: "portal.sources.types.folder.fields.recursive.label",
+        control: "select",
+        defaultValue: "false",
+        options: [
+          {
+            value: "false",
+            labelKey:
+              "portal.sources.types.folder.fields.recursive.options.top",
+          },
+          {
+            value: "true",
+            labelKey:
+              "portal.sources.types.folder.fields.recursive.options.all",
+          },
+        ],
+      },
+      {
+        key: "identity",
+        labelKey: "portal.sources.types.folder.fields.identity.label",
+        control: "select",
+        defaultValue: "stat",
+        helperTextKey: "portal.sources.types.folder.fields.identity.helperText",
+        options: [
+          {
+            value: "stat",
+            labelKey:
+              "portal.sources.types.folder.fields.identity.options.stat",
+          },
+          {
+            value: "hash",
+            labelKey:
+              "portal.sources.types.folder.fields.identity.options.hash",
+          },
+        ],
+      },
     ],
   },
 ];

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

@@ -6,6 +6,7 @@ import { useAsync, useSectionFlags } from "@portal/hooks/useAsync";
 import {
   buildWireFromSetup,
   buildWireFromState,
+  clearProcessedHistory,
   deletePolicy,
   fetchPolicies,
   savePolicy,
@@ -101,6 +102,11 @@ export function Policies() {
     if (id) void runLifecycle(() => deletePolicy(id));
   }
 
+  function handleClearHistory() {
+    const id = detail?.policy?.state.backendId;
+    if (id) void runLifecycle(() => clearProcessedHistory(id));
+  }
+
   function handleEdit() {
     if (detail) {
       setWizard(detail);
@@ -159,6 +165,7 @@ export function Policies() {
         onEdit={handleEdit}
         onTogglePause={handleTogglePause}
         onDelete={handleDelete}
+        onClearHistory={handleClearHistory}
       />
 
       <PolicySetupWizard