Improve logic for tracking which files have already been processed in policies (Stirling-Tools#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.

Author: jbrunton96

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.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 ResponseEntity<Void> deletePolicy(@PathVariable String policyId) {
         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 ResponseEntity<Void> deletePolicy(@PathVariable String policyId) {
         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.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 @@ private void runToCompletion(
             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 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 @@
 /**
  * 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 List<String> run(Policy policy) {
                         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 PolicyRunHandle runAdHoc(
 
     /**
      * 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
+}