De-AI comments

# Conflicts:
#	app/core/src/main/resources/settings.yml.template

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

@@ -80,6 +80,7 @@ public class ApplicationProperties {
     private Mcp mcp = new Mcp();
     private InternalApi internalApi = new InternalApi();
     private Cluster cluster = new Cluster();
+    private Policies policies = new Policies();
 
     @Bean
     public PropertySource<?> dynamicYamlPropertySource(ConfigurableEnvironment environment)
@@ -203,6 +204,45 @@ public class ApplicationProperties {
         }
     }
 
+    @Data
+    public static class Policies {
+        /**
+         * Absolute directories that policy folder input sources and output sinks may read from or
+         * write to. Empty (the default) disables folder access entirely, so a policy can never be
+         * pointed at an arbitrary server path. Stirling's own config directory is always
+         * off-limits, and folder access is always disabled in SaaS mode regardless of this list.
+         */
+        private List<String> allowedFolderRoots = new java.util.ArrayList<>();
+
+        /** How often (seconds) the schedule trigger checks for policies whose schedule is due. */
+        private long scheduleSweepSeconds = 60;
+
+        /**
+         * How often (seconds) the folder-watch trigger reconciles its watch registrations and
+         * re-runs every folder-watch policy as a safety net for filesystem events that were missed
+         * (NFS, bind mounts, inotify-queue overflow).
+         */
+        private long watchReconcileSeconds = 300;
+
+        /**
+         * How long (milliseconds) the folder-watch trigger keeps draining filesystem events after
+         * the first, so a burst from a single file copy coalesces into one run instead of many.
+         */
+        private long watchQuietPeriodMs = 500;
+
+        /**
+         * SSE emitter timeout (milliseconds) for streamed runs; generous for long multi-step runs.
+         */
+        private long streamTimeoutMs = 1800000;
+
+        /**
+         * How long (minutes) a finished run's in-memory state is retained before eviction,
+         * mirroring the job-result expiry so rich run state does not outlive the process. Active
+         * and paused runs are kept regardless of age.
+         */
+        private int runExpiryMinutes = 30;
+    }
+
     @Data
     public static class PdfEditor {
         private Cache cache = new Cache();

app/core/src/main/resources/settings.yml.template

@@ -364,6 +364,18 @@ aiEngine:
   url: http://localhost:5001 # URL of the Python AI engine
   timeoutSeconds: 120 # Timeout in seconds for AI engine requests
 
+policies:
+  # Folder automations can read from and write to the directories you allow here, so treat this as a
+  # security boundary. Leave allowedFolderRoots empty (default) to disable folder sources/outputs
+  # entirely; list absolute directories to permit folder access only within them. Stirling's own
+  # config directory is always off-limits, and folder access is always disabled in SaaS mode.
+  allowedFolderRoots: [] # e.g. ["/data/inbox", "/data/outbox"]
+  scheduleSweepSeconds: 60 # How often (seconds) scheduled policies are checked for being due
+  watchReconcileSeconds: 300 # How often (seconds) folder-watch re-syncs watches and re-runs as a safety net for missed events
+  watchQuietPeriodMs: 500 # How long (ms) folder-watch coalesces a burst of file events into a single run
+  streamTimeoutMs: 1800000 # SSE timeout (ms) for live run-progress streams
+  runExpiryMinutes: 30 # How long (minutes) a finished run's in-memory state is kept before eviction
+
 # Model Context Protocol (MCP) server. Exposes Stirling's PDF tools (grouped by namespace)
 # plus the AI agents to MCP clients (Inspector, Claude Desktop, custom). OAuth-protected.
 # Disabled by default - enable explicitly per deployment after configuring mcp.auth.

app/proprietary/src/main/java/stirling/software/proprietary/policy/config/FolderAccessGuard.java

@@ -0,0 +1,92 @@
+package stirling.software.proprietary.policy.config;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.core.env.Environment;
+import org.springframework.stereotype.Component;
+
+import stirling.software.common.configuration.InstallationPathConfig;
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.policy.model.Policy;
+
+/**
+ * Authority on which filesystem locations a policy may read/write. Checked at save time and again
+ * at run time, fail-closed in order:
+ *
+ * <ol>
+ *   <li>denied entirely under the {@code saas} profile;
+ *   <li>Stirling's own config dir always rejected, even if an allowed root were misconfigured to
+ *       contain it;
+ *   <li>must resolve within {@code policies.allowedFolderRoots}; none configured means all denied.
+ * </ol>
+ *
+ * <p>Compared after normalisation so {@code ..} cannot escape a root. Symlink escape is not
+ * defended: an operator who roots an allowlist on a symlink to a sensitive location is trusted.
+ */
+@Component
+public class FolderAccessGuard {
+
+    public static final String FOLDER_TYPE = "folder";
+
+    private final boolean saasActive;
+    private final List<Path> allowedRoots;
+    private final List<Path> protectedRoots;
+
+    public FolderAccessGuard(ApplicationProperties applicationProperties, Environment environment) {
+        this.saasActive = Arrays.asList(environment.getActiveProfiles()).contains("saas");
+        this.allowedRoots =
+                normalizeAll(applicationProperties.getPolicies().getAllowedFolderRoots());
+        this.protectedRoots = List.of(normalize(Path.of(InstallationPathConfig.getConfigPath())));
+    }
+
+    /** Returns the normalised absolute path; throws if not permitted. */
+    public Path requirePermitted(Path dir) {
+        if (saasActive) {
+            throw new IllegalArgumentException(
+                    "folder sources and outputs are not available in SaaS mode");
+        }
+        Path normalized = normalize(dir);
+        for (Path protectedRoot : protectedRoots) {
+            if (normalized.startsWith(protectedRoot)) {
+                throw new IllegalArgumentException(
+                        "folder may not point inside a protected Stirling directory");
+            }
+        }
+        if (allowedRoots.isEmpty()) {
+            throw new IllegalArgumentException(
+                    "folder access is disabled; set policies.allowedFolderRoots to permit it");
+        }
+        boolean within = allowedRoots.stream().anyMatch(normalized::startsWith);
+        if (!within) {
+            throw new IllegalArgumentException(
+                    "folder '" + normalized + "' is outside the allowed folder roots");
+        }
+        return normalized;
+    }
+
+    /** Whether this policy touches a folder source/sink, and so is subject to these rules. */
+    public boolean usesFolderAccess(Policy policy) {
+        boolean readsFolder =
+                policy.sources().stream().anyMatch(spec -> FOLDER_TYPE.equals(spec.type()));
+        boolean writesFolder =
+                policy.output() != null && FOLDER_TYPE.equals(policy.output().type());
+        return readsFolder || writesFolder;
+    }
+
+    private static List<Path> normalizeAll(List<String> roots) {
+        List<Path> result = new ArrayList<>();
+        for (String root : roots) {
+            if (root != null && !root.isBlank()) {
+                result.add(normalize(Path.of(root)));
+            }
+        }
+        return result;
+    }
+
+    private static Path normalize(Path path) {
+        return path.toAbsolutePath().normalize();
+    }
+}

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

@@ -6,7 +6,6 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.core.io.FileSystemResource;
 import org.springframework.core.io.Resource;
 import org.springframework.http.HttpStatus;
@@ -34,11 +33,16 @@ import io.swagger.v3.oas.annotations.tags.Tag;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
+import stirling.software.common.model.ApplicationProperties;
 import stirling.software.common.model.job.JobResponse;
+import stirling.software.common.service.UserServiceInterface;
 import stirling.software.common.util.TempFile;
 import stirling.software.common.util.TempFileManager;
+import stirling.software.proprietary.policy.config.FolderAccessGuard;
 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.model.PipelineDefinition;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.model.PolicyInputs;
@@ -47,20 +51,14 @@ import stirling.software.proprietary.policy.model.PolicyRunStatus;
 import stirling.software.proprietary.policy.model.PolicyRunView;
 import stirling.software.proprietary.policy.progress.PolicyProgressListener;
 import stirling.software.proprietary.policy.store.PolicyStore;
-import stirling.software.proprietary.policy.trigger.ManualTrigger;
 import stirling.software.proprietary.security.config.PremiumEndpoint;
 
 import tools.jackson.core.JacksonException;
 import tools.jackson.databind.ObjectMapper;
 
 /**
- * Manages policies and runs pipelines. The premium backend entry point: CRUD for stored {@code
- * Policy} objects, running a stored policy by id, and running an ad-hoc pipeline (for AI/Automate
- * one-offs).
- *
- * <p>Runs execute asynchronously and return a run id immediately. Poll {@code GET /run/{runId}} for
- * status, and download outputs via the existing {@code GET /api/v1/general/files/{fileId}} using
- * the file ids in the run view.
+ * Policy CRUD plus pipeline runs (stored or ad-hoc). Runs are async: returns a run id, poll {@code
+ * GET /run/{runId}} for status, download outputs via {@code GET /api/v1/general/files/{fileId}}.
  */
 @Slf4j
 @RestController
@@ -71,16 +69,16 @@ import tools.jackson.databind.ObjectMapper;
 @Tag(name = "Policies", description = "Run tool pipelines on the backend")
 public class PolicyController {
 
-    private final ManualTrigger manualTrigger;
+    private final PolicyRunner policyRunner;
     private final PolicyRunRegistry runRegistry;
     private final PolicyStore policyStore;
+    private final PolicyValidator policyValidator;
+    private final FolderAccessGuard folderAccessGuard;
+    private final UserServiceInterface userService;
+    private final ApplicationProperties applicationProperties;
     private final ObjectMapper objectMapper;
     private final TempFileManager tempFileManager;
 
-    /** SSE emitter timeout, generous enough for long multi-step runs on large files. */
-    @Value("${stirling.policies.streamTimeoutMs:1800000}")
-    private long streamTimeoutMs;
-
     @PostMapping(value = "/run", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
     @Operation(
             summary = "Run a tool pipeline",
@@ -95,7 +93,8 @@ public class PolicyController {
             throws IOException {
         PipelineDefinition definition = parseDefinition(json);
         PolicyInputs inputs = collectInputs(request);
-        String runId = manualTrigger.fire(definition, inputs, PolicyProgressListener.NOOP).runId();
+        String runId =
+                policyRunner.runAdHoc(definition, inputs, PolicyProgressListener.NOOP).runId();
         return ResponseEntity.accepted().body(new JobResponse<>(true, runId, null));
     }
 
@@ -112,12 +111,13 @@ public class PolicyController {
         PipelineDefinition definition = parseDefinition(json);
         PolicyInputs inputs = collectInputs(request);
 
-        SseEmitter emitter = new SseEmitter(streamTimeoutMs);
+        SseEmitter emitter =
+                new SseEmitter(applicationProperties.getPolicies().getStreamTimeoutMs());
         emitter.onError(e -> log.warn("Policy run SSE emitter error", e));
 
-        PolicyRunHandle handle = manualTrigger.fire(definition, inputs, streamListener(emitter));
-        // Close the stream with a terminal event once the run finishes. whenComplete runs on the
-        // engine's worker thread after the run is done, so this never races the step events.
+        PolicyRunHandle handle = policyRunner.runAdHoc(definition, inputs, streamListener(emitter));
+        // whenComplete runs on the worker thread after the run finishes, so the terminal event
+        // never races the step events.
         handle.completion()
                 .whenComplete(
                         (run, throwable) -> {
@@ -155,7 +155,33 @@ public class PolicyController {
                     "Stores a policy (trigger config + steps + output + metadata). A blank id is"
                             + " assigned; returns the stored policy with its id.")
     public ResponseEntity<Policy> savePolicy(@RequestBody String json) {
-        return ResponseEntity.ok(policyStore.save(parsePolicy(json)));
+        Policy policy = parsePolicy(json);
+        requireAuthorizedForFolderAccess(policy);
+        try {
+            policyValidator.validate(policy);
+        } catch (IllegalArgumentException e) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, e.getMessage());
+        }
+        return ResponseEntity.ok(policyStore.save(policy));
+    }
+
+    /**
+     * Folder sources/outputs grant whoever saves the policy access to that path, so gate them to
+     * admins on multi-user deployments. Single-user (login disabled) trusts the local operator;
+     * {@link FolderAccessGuard} still enforces SaaS-off and the path allowlist at validation time.
+     */
+    private void requireAuthorizedForFolderAccess(Policy policy) {
+        if (!folderAccessGuard.usesFolderAccess(policy)) {
+            return;
+        }
+        if (!applicationProperties.getSecurity().isEnableLogin()) {
+            return;
+        }
+        if (!userService.isCurrentUserAdmin()) {
+            throw new ResponseStatusException(
+                    HttpStatus.FORBIDDEN,
+                    "Folder sources and outputs may only be configured by an administrator");
+        }
     }
 
     @GetMapping
@@ -199,7 +225,7 @@ public class PolicyController {
                                         new ResponseStatusException(
                                                 HttpStatus.NOT_FOUND, "No policy: " + policyId));
         PolicyInputs inputs = collectInputs(request);
-        String runId = manualTrigger.run(policy, inputs, PolicyProgressListener.NOOP).runId();
+        String runId = policyRunner.runWith(policy, inputs, PolicyProgressListener.NOOP).runId();
         return ResponseEntity.accepted().body(new JobResponse<>(true, runId, null));
     }
 
@@ -226,11 +252,8 @@ public class PolicyController {
         return definition;
     }
 
-    /**
-     * Split the multipart file parts into the primary document stream ("fileInput") and the named
-     * supporting-file store: every other file field becomes an asset keyed by its field name, which
-     * a step references from {@code fileParameters}.
-     */
+    // "fileInput" is the primary document stream; every other file field becomes a supporting
+    // asset keyed by field name, which a step references from fileParameters.
     private PolicyInputs collectInputs(MultipartHttpServletRequest request) throws IOException {
         MultiValueMap<String, MultipartFile> fileMap = request.getMultiFileMap();
         List<Resource> primary = toResources(fileMap.get("fileInput"));
@@ -247,9 +270,6 @@ public class PolicyController {
         return new PolicyInputs(primary, supportingFiles);
     }
 
-    /**
-     * A progress listener that forwards each step transition to the SSE stream as a "step" event.
-     */
     private PolicyProgressListener streamListener(SseEmitter emitter) {
         return new PolicyProgressListener() {
             @Override
@@ -288,8 +308,8 @@ public class PolicyController {
         try {
             emitter.send(SseEmitter.event().name(name).data(data, MediaType.APPLICATION_JSON));
         } catch (IOException | IllegalStateException e) {
-            // Client disconnected or the emitter already closed. The run continues and its results
-            // remain downloadable via the job endpoints; nothing useful left to stream.
+            // Client gone or emitter closed. The run continues and outputs stay downloadable via
+            // the job endpoints.
             log.debug("Dropping policy SSE event '{}': {}", name, e.getMessage());
         }
     }

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

@@ -33,31 +33,24 @@ import stirling.software.proprietary.policy.output.PolicyOutputSink;
 import stirling.software.proprietary.policy.progress.PolicyProgressListener;
 
 /**
- * Runs pipelines asynchronously as tracked jobs.
+ * Runs pipelines asynchronously as tracked jobs. {@link #submit} returns a run id immediately; the
+ * pipeline runs on a virtual thread (so a step blocked on a slow tool does not hold a platform
+ * thread). Drives {@link PolicyExecutor} for the step loop, projects status/outputs into {@link
+ * TaskManager} (existing job endpoints work unchanged), and keeps live state in {@link
+ * PolicyRunRegistry}.
  *
- * <p>Each run is the unit of async work: {@link #submit} returns a run id immediately and the
- * pipeline executes on a virtual thread, so a step blocking on a slow tool does not tie up a
- * platform thread. The run drives {@link PolicyExecutor} for the actual step loop, registers its
- * outputs and progress with {@link TaskManager} (so the existing job status/download endpoints work
- * unchanged), and keeps rich state in {@link PolicyRunRegistry}.
- *
- * <p>The engine deliberately manages its own virtual-thread execution rather than routing through
- * {@code JobExecutorService}: that path force-completes a job once its work returns, which is
- * incompatible with a run that suspends in {@code WAITING_FOR_INPUT}. It still applies the shared
- * {@link ResourceMonitor}/{@link JobQueue} admission control, so heavy runs queue under load
- * instead of oversubscribing.
+ * <p>Manages its own virtual-thread execution rather than {@code JobExecutorService}, which
+ * force-completes a job once its work returns: incompatible with a run that suspends in {@code
+ * WAITING_FOR_INPUT}. Still applies the shared {@link ResourceMonitor}/{@link JobQueue} admission
+ * control so heavy runs queue under load.
  */
 @Slf4j
 @Service
 @RequiredArgsConstructor
 public class PolicyEngine {
 
-    /**
-     * Resource weight of a pipeline run for admission control. A run chains many tools and holds
-     * intermediate files, so it is weighted as heavy work: the shared {@link ResourceMonitor}
-     * should let it start while the system is healthy but hold it back under memory/CPU pressure.
-     * See {@link ResourceMonitor#shouldQueueJob(int)} for how a weight maps to that decision.
-     */
+    // Admission weight for one run. Weighted heavy: a run chains many tools and holds intermediate
+    // files. See ResourceMonitor#shouldQueueJob(int).
     private static final int RUN_RESOURCE_WEIGHT = 50;
 
     private final PolicyExecutor stepExecutor;
@@ -72,16 +65,14 @@ public class PolicyEngine {
     private final ExecutorService asyncExecutor = ExecutorFactory.newVirtualThreadExecutor();
 
     /**
-     * Submit a pipeline to run asynchronously. The returned handle's run id scopes a job in {@link
-     * TaskManager}, so progress (notes), status, and result files are observable via the existing
-     * job endpoints as well as via {@link #getRun(String)}; its completion future resolves when the
-     * run reaches a terminal or paused state.
+     * Submit a pipeline to run asynchronously. The handle's run id scopes a {@link TaskManager} job
+     * (status/notes/results observable via the job endpoints); its future resolves when the run
+     * reaches a terminal or paused state.
      */
     public PolicyRunHandle submit(
             PipelineDefinition definition, PolicyInputs inputs, PolicyProgressListener listener) {
-        // Scope the run id to the current user (on this request thread) so the file-download
-        // ownership check passes; NoOpJobOwnershipService returns the id unchanged when security
-        // is off.
+        // Scope the run id to the current user (this request thread) so the file-download
+        // ownership check passes. No-op when security is off.
         String runId = jobOwnershipService.createScopedJobKey(UUID.randomUUID().toString());
         taskManager.createTask(runId);
         PolicyRun run = new PolicyRun(runId, definition);
@@ -90,9 +81,8 @@ public class PolicyEngine {
         PolicyProgressListener tracking = trackingListener(runId, run, listener);
         Runnable task = () -> runToCompletion(run, inputs, tracking, completion);
 
-        // Each run is one admission unit; steps run synchronously within it, so this gates heavy
-        // work under load without the pool-within-pool risk of queueing each tool call. Under
-        // resource pressure the run waits in the shared JobQueue; otherwise it starts immediately.
+        // One admission unit per run; steps run synchronously within it, so this gates heavy work
+        // without the pool-within-pool risk of queueing each tool call.
         if (resourceMonitor.shouldQueueJob(RUN_RESOURCE_WEIGHT)) {
             log.debug("Queueing policy run {} under resource pressure", runId);
             jobQueue.queueJob(
@@ -110,10 +100,7 @@ public class PolicyEngine {
         return new PolicyRunHandle(runId, completion);
     }
 
-    /**
-     * Run a stored policy on demand. Builds the policy's pipeline and submits it. {@code enabled}
-     * gates automatic triggering, not explicit runs, so this runs regardless of that flag.
-     */
+    /** Run a stored policy on demand. {@code enabled} gates triggers, not explicit runs. */
     public PolicyRunHandle runPolicy(
             Policy policy, PolicyInputs inputs, PolicyProgressListener listener) {
         return submit(policy.toDefinition(), inputs, listener);
@@ -124,8 +111,7 @@ public class PolicyEngine {
     }
 
     /**
-     * Request cancellation of a run. Stage 1 marks the run cancelled in the registry if it has not
-     * already finished; interrupting an in-flight tool call lands in a later stage.
+     * Mark a run cancelled if not already finished. Does not yet interrupt an in-flight tool call.
      */
     public boolean cancel(String runId) {
         PolicyRun run = registry.get(runId);
@@ -139,10 +125,7 @@ public class PolicyEngine {
         return cancelled;
     }
 
-    /**
-     * Resume a run paused in {@code WAITING_FOR_INPUT}. Not yet implemented; the run shape and
-     * {@link WaitState} snapshot are in place so this can be added without reworking the engine.
-     */
+    /** Resume a run paused in {@code WAITING_FOR_INPUT}. Not yet implemented. */
     public String resume(String runId, List<Resource> additionalInputs) {
         throw new UnsupportedOperationException("Pause/resume is not yet implemented");
     }
@@ -163,8 +146,8 @@ public class PolicyEngine {
             taskManager.setComplete(runId);
             run.complete(outputs);
         } catch (PolicyInputRequiredException e) {
-            // Designed-for path: suspend the run rather than fail it. Persist intermediates as
-            // fileIds so the run can resume after this worker thread is gone.
+            // Expected path: suspend rather than fail. Persist intermediates as fileIds so the run
+            // can resume after this worker thread is gone.
             WaitState wait = suspend(e);
             run.waitForInput(wait);
             taskManager.addNote(runId, "Waiting for input: " + e.getMessage());
@@ -183,15 +166,15 @@ public class PolicyEngine {
             run.fail(message);
             taskManager.setError(runId, message);
         } finally {
-            // Always resolve the handle with the run's final state so stream/await callers unblock.
+            // Always resolve so stream/await callers unblock.
             completion.complete(run);
         }
     }
 
     private ResponseEntity<?> failRejectedRun(
             PolicyRun run, CompletableFuture<PolicyRun> completion, Throwable ex) {
-        // Only reached if the run never started (e.g. the queue was full). A run that started
-        // always resolves its own completion in runToCompletion.
+        // Only reached if the run never started (e.g. queue full); a started run resolves its own
+        // completion in runToCompletion.
         if (!completion.isDone()) {
             String message = "Policy run could not be queued: " + ex.getMessage();
             log.error("Policy run {} was not admitted: {}", run.getRunId(), ex.getMessage());

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

@@ -7,11 +7,8 @@ import org.springframework.core.io.Resource;
 import tools.jackson.databind.JsonNode;
 
 /**
- * Result of running a pipeline through {@link PolicyExecutor}.
- *
- * <p>{@code files} are the final output resources (temp files, not yet stored to {@code
- * FileStorage}). {@code report} is the structured metadata payload captured from the last step that
- * produced one (a JSON body, or an {@code X-Stirling-Tool-Report} header), with {@code reportTool}
- * naming the step it came from; both are null when no step produced a report.
+ * Result of a {@link PolicyExecutor} run. {@code files} are final temp files (not yet stored).
+ * {@code report}/{@code reportTool} carry the last step's structured report and its operation, or
+ * null if no step produced one.
  */
 public record PolicyExecutionResult(List<Resource> files, JsonNode report, String reportTool) {}

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

@@ -35,15 +35,11 @@ import tools.jackson.databind.JsonNode;
 import tools.jackson.databind.ObjectMapper;
 
 /**
- * Runs an ordered chain of tool steps, chaining each step's output files into the next step's
- * input.
+ * Runs an ordered chain of tool steps, feeding each step's output files into the next.
  *
- * <p>This is the single execution loop for the proprietary surface (AI plans now;
- * manually-triggered runs and watched folders later). Each step is dispatched synchronously via
- * {@link InternalApiClient} loopback HTTP: the tool runs in its own handler and returns its file
- * inline. The caller decides how to run the executor itself (the AI turn loop calls it directly;
- * the engine runs it on a virtual thread for async runs). Files cross step boundaries as {@link
- * Resource} temp files; they are only persisted to durable storage at the run boundaries by the
+ * <p>Steps dispatch synchronously via {@link InternalApiClient} loopback HTTP (each tool runs in
+ * its own handler, returns its file inline). The caller controls threading. Files cross step
+ * boundaries as {@link Resource} temp files and are only persisted at the run boundaries by the
  * caller.
  */
 @Slf4j
@@ -58,25 +54,16 @@ public class PolicyExecutor {
     private final TempFileManager tempFileManager;
     private final ObjectMapper objectMapper;
 
-    /**
-     * Internal value-class for tool responses. {@code files} holds any result files (typically one;
-     * multiple for ZIP-response tools). {@code report} holds an optional structured metadata
-     * payload the tool chose to surface alongside (or instead of) a file.
-     */
+    // files: result files (one, or many for ZIP-response tools). report: optional structured
+    // payload the tool surfaced alongside or instead of a file.
     private record ToolResult(List<Resource> files, JsonNode report) {}
 
     /**
-     * Execute every step in {@code definition} in order, feeding each step's output into the next.
-     * Supporting files supplied in {@code inputs} are bound to steps' named file fields and never
-     * enter the document stream.
+     * Run every step in order, feeding each step's output into the next. Supporting files in {@code
+     * inputs} bind to named file fields and never enter the document stream.
      *
-     * @param definition the pipeline to run (must have at least one step)
-     * @param inputs the primary documents plus the named supporting-file store
-     * @param listener receives per-step progress
-     * @return the final output files plus the last structured report produced, if any
      * @throws InternalApiTimeoutException if a tool does not respond within its read timeout
-     * @throws IOException if a tool returns a non-OK response, references a missing supporting
-     *     file, or a file cannot be read
+     * @throws IOException on a non-OK tool response, a missing supporting file, or a read failure
      */
     public PolicyExecutionResult execute(
             PipelineDefinition definition, PolicyInputs inputs, PolicyProgressListener listener)
@@ -88,7 +75,7 @@ public class PolicyExecutor {
 
         List<Resource> currentFiles = inputs.primary();
         Map<String, List<Resource>> supportingFiles = inputs.supportingFiles();
-        // Propagate the *last* non-null report; the terminal step defines the output.
+        // Last non-null report wins: the terminal step defines the output.
         JsonNode lastReport = null;
         String lastReportTool = null;
 
@@ -113,13 +100,9 @@ public class PolicyExecutor {
     }
 
     /**
-     * Execute a single tool step. If the endpoint accepts multiple files, all files are sent in one
-     * call. Otherwise, the endpoint is called once per file. ZIP responses are unpacked so each
-     * inner file is treated as its own result (e.g. split outputs a ZIP of pages).
-     *
-     * <p>A structured {@code report} may be returned alongside (or instead of) files; see {@link
-     * ToolResult}. For per-file dispatch (single-input endpoints called once per input), the first
-     * non-null report wins.
+     * Multi-input endpoints get all files in one call; others are called once per file. ZIP
+     * responses are unpacked so each inner file is its own result (e.g. split). For per-file
+     * dispatch the first non-null report wins.
      */
     private ToolResult executeStep(
             PipelineStep step,
@@ -146,17 +129,10 @@ public class PolicyExecutor {
     }
 
     /**
-     * Call an endpoint and return its result files and optional report.
-     *
-     * <ul>
-     *   <li>JSON body (Content-Type: application/json): the entire body is the report, no files are
-     *       returned.
-     *   <li>File body (PDF etc.): the file is returned; if an {@link
-     *       AiToolResponseHeaders#TOOL_REPORT} header is present, its (minified JSON) value is
-     *       parsed as the report.
-     *   <li>ZIP responses declared by the tool metadata service are unpacked so callers always see
-     *       a flat list of result files.
-     * </ul>
+     * Call an endpoint, returning result files and optional report. Response handling: JSON body is
+     * the report with no file; a file body returns the file plus any {@link
+     * AiToolResponseHeaders#TOOL_REPORT} header report; ZIP responses (per tool metadata) are
+     * unpacked to a flat file list.
      */
     private ToolResult callEndpoint(
             PipelineStep step, List<Resource> files, Map<String, List<Resource>> supportingFiles)
@@ -166,8 +142,8 @@ public class PolicyExecutor {
         for (Resource file : files) {
             body.add("fileInput", file);
         }
-        // Bind supporting files to their named tool fields (e.g. stampImage, overlayFiles). These
-        // come from the run's named asset store, not the document stream.
+        // Bind supporting files to named tool fields (e.g. stampImage); from the asset store, not
+        // the document stream.
         for (Map.Entry<String, String> binding : step.fileParameters().entrySet()) {
             String fieldName = binding.getKey();
             String assetKey = binding.getValue();
@@ -189,9 +165,9 @@ public class PolicyExecutor {
         for (Map.Entry<String, Object> entry : step.parameters().entrySet()) {
             if (entry.getValue() instanceof List<?> list) {
                 if (containsStructuredElements(list)) {
-                    // Endpoints binding lists of structured objects (e.g. /security/redact's
-                    // redactions, /general/edit-text's edits) parse a single JSON string field via
-                    // a property editor. Pre-serialize the whole list so binding succeeds.
+                    // These endpoints (e.g. /security/redact redactions, /general/edit-text edits)
+                    // bind a list of structured objects from a single JSON string field via a
+                    // property editor, so pre-serialize the whole list.
                     body.add(entry.getKey(), objectMapper.writeValueAsString(list));
                 } else {
                     for (Object item : list) {
@@ -209,8 +185,8 @@ public class PolicyExecutor {
         }
         Resource resource = response.getBody();
 
-        // Filter operations return an empty body to signal the file was filtered out: drop it
-        // rather than forwarding a zero-byte document.
+        // Filter ops return an empty body to mean "filtered out": drop it rather than forward a
+        // zero-byte document.
         if (isFilterOperation(endpointPath) && isEmpty(resource)) {
             return new ToolResult(List.of(), null);
         }
@@ -218,7 +194,7 @@ public class PolicyExecutor {
         HttpHeaders headers = response.getHeaders();
         MediaType contentType = headers.getContentType();
 
-        // JSON-only response: the whole body is the structured report, no result file.
+        // JSON-only response: whole body is the report, no file.
         if (contentType != null && MediaType.APPLICATION_JSON.isCompatibleWith(contentType)) {
             try (InputStream is = resource.getInputStream()) {
                 JsonNode report = objectMapper.readTree(is);
@@ -233,10 +209,7 @@ public class PolicyExecutor {
         return new ToolResult(List.of(resource), report);
     }
 
-    /**
-     * Parse the optional {@link AiToolResponseHeaders#TOOL_REPORT} header into a {@link JsonNode},
-     * or return null.
-     */
+    /** Parse the optional {@link AiToolResponseHeaders#TOOL_REPORT} header, or null. */
     private JsonNode parseReportHeader(HttpHeaders headers, String endpointPath) {
         String raw = headers.getFirst(AiToolResponseHeaders.TOOL_REPORT);
         if (raw == null || raw.isBlank()) {
@@ -264,8 +237,7 @@ public class PolicyExecutor {
     }
 
     /**
-     * Fail the run if any document in the primary stream is not a file type the step accepts. An
-     * endpoint that declares no specific input type accepts anything.
+     * Fail if any primary-stream file is a type the step rejects. No declared type means anything.
      */
     private void requireAcceptedTypes(String operation, List<Resource> files) throws IOException {
         List<String> accepted = toolMetadataService.getExtensionTypes(false, operation);

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

@@ -7,15 +7,9 @@ import org.springframework.core.io.Resource;
 import lombok.Getter;
 
 /**
- * Thrown by a step to signal that the run cannot proceed without further user input, pausing the
- * run in {@code WAITING_FOR_INPUT} rather than failing it.
- *
- * <p>Carries everything needed to resume: a human-readable reason, the 0-based index of the step to
- * resume from, and the intermediate files produced so far. The engine persists those files and
- * suspends the run.
- *
- * <p>Defined now to fix the run shape; no step throws it yet, and the resume handshake is
- * implemented in a later stage.
+ * Thrown by a step that needs further user input, pausing the run in {@code WAITING_FOR_INPUT}
+ * instead of failing. Carries the resume reason, 0-based resume step index, and intermediate files;
+ * the engine persists those and suspends. Not yet thrown by any step.
  */
 @Getter
 public class PolicyInputRequiredException extends RuntimeException {

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

@@ -5,12 +5,9 @@ import java.util.concurrent.CompletableFuture;
 import stirling.software.proprietary.policy.model.PolicyRun;
 
 /**
- * Returned by {@link PolicyEngine#submit}: the run id (for status polling and result download) plus
- * a future that resolves when the run reaches a terminal or paused state.
- *
- * <p>The completion future lets callers react to the end of a run (e.g. an SSE endpoint sending a
- * final event and closing the stream) without polling. It carries the {@link PolicyRun} whose
- * status describes the outcome (completed, failed, cancelled, or waiting for input); it does not
- * complete exceptionally for ordinary run failures.
+ * Returned by {@link PolicyEngine#submit}: the run id (status polling, result download) plus a
+ * future that resolves when the run reaches a terminal or paused state. The future carries the
+ * {@link PolicyRun} whose status describes the outcome; it does not complete exceptionally for
+ * ordinary run failures.
  */
 public record PolicyRunHandle(String runId, CompletableFuture<PolicyRun> completion) {}

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

@@ -9,26 +9,22 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
 import jakarta.annotation.PreDestroy;
 
 import lombok.extern.slf4j.Slf4j;
 
+import stirling.software.common.model.ApplicationProperties;
 import stirling.software.proprietary.policy.model.PolicyRun;
 
 /**
- * In-memory store of live {@link PolicyRun} state, keyed by runId. Holds the authoritative run
- * state machine; durable status/files for download are projected separately into {@code
- * TaskManager}.
+ * In-memory store of live {@link PolicyRun} state, keyed by runId. Authoritative run state machine;
+ * durable status/files are projected separately into {@code TaskManager}.
  *
- * <p>Finished runs are evicted on a fixed interval once they age past {@code
- * stirling.policies.runExpiryMinutes}, mirroring the job-result expiry in {@code TaskManager} so a
- * run's rich in-memory state does not outlive the process. Only terminal runs are evicted; active
- * and paused ({@code WAITING_FOR_INPUT}) runs are retained regardless of age. Result files are not
- * touched here: a run shares its runId with a {@code TaskManager} job, which owns file-lifecycle
- * cleanup, so eviction only frees this map's entry.
+ * <p>A scheduled sweep evicts only terminal runs aged past {@code policies.runExpiryMinutes};
+ * active and paused runs are kept regardless of age. Eviction frees only this map's entry: the
+ * shared {@code TaskManager} job owns file-lifecycle cleanup.
  */
 @Slf4j
 @Service
@@ -41,8 +37,8 @@ public class PolicyRunRegistry {
             Executors.newSingleThreadScheduledExecutor(
                     Thread.ofVirtual().name("policy-run-cleanup-", 0).factory());
 
-    public PolicyRunRegistry(
-            @Value("${stirling.policies.runExpiryMinutes:30}") int runExpiryMinutes) {
+    public PolicyRunRegistry(ApplicationProperties applicationProperties) {
+        int runExpiryMinutes = applicationProperties.getPolicies().getRunExpiryMinutes();
         this.runExpiry = Duration.ofMinutes(runExpiryMinutes);
         cleanupExecutor.scheduleAtFixedRate(this::evictExpiredRuns, 10, 10, TimeUnit.MINUTES);
         log.debug(
@@ -61,7 +57,7 @@ public class PolicyRunRegistry {
         return runs.values();
     }
 
-    /** Scheduled hook: evict terminal runs that finished before the expiry window. */
+    /** Scheduled sweep entry point. */
     private void evictExpiredRuns() {
         try {
             evictExpired(Instant.now().minus(runExpiry));
@@ -71,9 +67,8 @@ public class PolicyRunRegistry {
     }
 
     /**
-     * Remove every terminal run last updated before {@code cutoff}; active and paused runs are kept
-     * regardless of age. Returns the number evicted. Package-visible so the scheduled sweep and
-     * tests exercise the same path with an explicit cutoff.
+     * Evict terminal runs last updated before {@code cutoff}, returning the count. Package-visible
+     * so the sweep and tests share one path with an explicit cutoff.
      */
     int evictExpired(Instant cutoff) {
         int removed = 0;

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

@@ -0,0 +1,107 @@
+package stirling.software.proprietary.policy.engine;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.function.Consumer;
+
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.proprietary.policy.input.InputSource;
+import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.PipelineDefinition;
+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.PolicyRunStatus;
+import stirling.software.proprietary.policy.progress.PolicyProgressListener;
+
+/**
+ * Turns a policy's configured {@link InputSpec sources} into runs. Triggers decide <em>when</em>
+ * and call {@link #run(Policy)}; they never touch sources. The controller uses the supplied-input
+ * and ad-hoc entry points for on-demand work. This is the seam keeping triggers and sources
+ * independent.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class PolicyRunner {
+
+    private final PolicyEngine policyEngine;
+    private final List<InputSource> inputSources;
+
+    /**
+     * Trigger entry point. Pulls every configured source; each yielded unit becomes its own run so
+     * one failure does not affect the others. No sources means one run with no input (generator
+     * pipeline).
+     */
+    public void run(Policy policy) {
+        List<InputSpec> sources = policy.sources();
+        if (sources.isEmpty()) {
+            startRun(policy, PolicyInputs.of(List.of()), unused -> {});
+            return;
+        }
+        for (InputSpec spec : sources) {
+            pullAndRun(policy, spec);
+        }
+    }
+
+    /** Run a stored policy on caller-supplied files (e.g. manual upload), bypassing its sources. */
+    public PolicyRunHandle runWith(
+            Policy policy, PolicyInputs inputs, PolicyProgressListener listener) {
+        return policyEngine.runPolicy(policy, inputs, listener);
+    }
+
+    /** Run an ad-hoc pipeline with no stored policy (AI/Automate one-offs). */
+    public PolicyRunHandle runAdHoc(
+            PipelineDefinition definition, PolicyInputs inputs, PolicyProgressListener listener) {
+        return policyEngine.submit(definition, inputs, listener);
+    }
+
+    private void pullAndRun(Policy policy, InputSpec spec) {
+        InputSource source = sourceFor(spec);
+        if (source == null) {
+            log.warn(
+                    "No input source for type '{}' (policy {}); skipping",
+                    spec.type(),
+                    policy.id());
+            return;
+        }
+        List<ResolvedInput> work;
+        try {
+            work = source.resolve(spec);
+        } catch (IOException | RuntimeException e) {
+            log.warn(
+                    "Failed to resolve source '{}' for policy {}: {}",
+                    spec.type(),
+                    policy.id(),
+                    e.getMessage());
+            return;
+        }
+        for (ResolvedInput unit : work) {
+            startRun(policy, unit.inputs(), unit.onComplete());
+        }
+    }
+
+    private void startRun(Policy policy, PolicyInputs inputs, Consumer<Boolean> onComplete) {
+        log.info("Running policy {} ({})", policy.id(), policy.name());
+        PolicyRunHandle handle =
+                policyEngine.runPolicy(policy, inputs, PolicyProgressListener.NOOP);
+        handle.completion()
+                .whenComplete((run, throwable) -> onComplete.accept(succeeded(run, throwable)));
+    }
+
+    private static boolean succeeded(PolicyRun run, Throwable throwable) {
+        return throwable == null && run != null && run.getStatus() == PolicyRunStatus.COMPLETED;
+    }
+
+    private InputSource sourceFor(InputSpec spec) {
+        return inputSources.stream()
+                .filter(source -> source.supports(spec))
+                .findFirst()
+                .orElse(null);
+    }
+}

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

@@ -0,0 +1,70 @@
+package stirling.software.proprietary.policy.engine;
+
+import java.util.List;
+
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+
+import stirling.software.proprietary.policy.input.InputSource;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.TriggerConfig;
+import stirling.software.proprietary.policy.output.PolicyOutputSink;
+import stirling.software.proprietary.policy.trigger.PolicyTrigger;
+
+/**
+ * Validates a policy at save time by delegating each facet (trigger, sources, output) to the bean
+ * that handles its type, so a misconfiguration fails fast rather than at run time. A null trigger
+ * is a manual-only policy and skips trigger validation.
+ */
+@Service
+@RequiredArgsConstructor
+public class PolicyValidator {
+
+    private final List<PolicyTrigger> triggers;
+    private final List<InputSource> inputSources;
+    private final List<PolicyOutputSink> outputSinks;
+
+    /**
+     * @throws IllegalArgumentException if any facet's type is unknown or its config is invalid
+     */
+    public void validate(Policy policy) {
+        if (policy.trigger() != null) {
+            triggerFor(policy.trigger()).validate(policy);
+        }
+        for (InputSpec source : policy.sources()) {
+            inputSourceFor(source).validate(source);
+        }
+        outputSinkFor(policy.output()).validate(policy.output());
+    }
+
+    private PolicyTrigger triggerFor(TriggerConfig config) {
+        return triggers.stream()
+                .filter(trigger -> trigger.type().equals(config.type()))
+                .findFirst()
+                .orElseThrow(
+                        () ->
+                                new IllegalArgumentException(
+                                        "unknown trigger type: " + config.type()));
+    }
+
+    private InputSource inputSourceFor(InputSpec spec) {
+        return inputSources.stream()
+                .filter(source -> source.supports(spec))
+                .findFirst()
+                .orElseThrow(
+                        () ->
+                                new IllegalArgumentException(
+                                        "unknown input source type: " + spec.type()));
+    }
+
+    private PolicyOutputSink outputSinkFor(OutputSpec spec) {
+        return outputSinks.stream()
+                .filter(sink -> sink.supports(spec))
+                .findFirst()
+                .orElseThrow(
+                        () -> new IllegalArgumentException("unknown output type: " + spec.type()));
+    }
+}

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

@@ -0,0 +1,177 @@
+package stirling.software.proprietary.policy.input;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import org.springframework.core.io.FileSystemResource;
+import org.springframework.core.io.Resource;
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.util.FileReadinessChecker;
+import stirling.software.proprietary.policy.config.FolderAccessGuard;
+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.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+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;
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    @Override
+    public boolean supports(InputSpec spec) {
+        return spec != null && TYPE.equals(spec.type());
+    }
+
+    @Override
+    public void validate(InputSpec spec) {
+        accessGuard.requirePermitted(FolderConfig.from(spec.options()).directory());
+    }
+
+    @Override
+    public List<Path> watchTargets(InputSpec spec) {
+        return List.of(FolderConfig.from(spec.options()).directory());
+    }
+
+    @Override
+    public List<ResolvedInput> resolve(InputSpec spec) 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();
+        }
+
+        List<Path> ready = new ArrayList<>();
+        try (Stream<Path> entries = Files.list(inputDir)) {
+            entries.filter(Files::isRegularFile)
+                    .filter(readinessChecker::isReady)
+                    .forEach(ready::add);
+        }
+
+        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)));
+            }
+        }
+        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) {
+        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;
+        } catch (IOException e) {
+            log.debug("Could not claim {}: {}", 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());
+        }
+    }
+
+    private static Path workDir(Path inputDir, String subdir) {
+        return inputDir.resolve(WORK_SUBDIR).resolve(subdir);
+    }
+
+    private static Resource fileResource(Path path) {
+        String name = path.getFileName().toString();
+        return new FileSystemResource(path.toFile()) {
+            @Override
+            public String getFilename() {
+                return name;
+            }
+        };
+    }
+
+    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) {
+
+        private static final String DIRECTORY_OPTION = "directory";
+        private static final String MODE_OPTION = "mode";
+        private static final String MODE_SNAPSHOT = "snapshot";
+
+        static FolderConfig from(Map<String, Object> options) {
+            Object directory = options.get(DIRECTORY_OPTION);
+            if (directory == null || directory.toString().isBlank()) {
+                throw new IllegalArgumentException("folder input requires a 'directory' option");
+            }
+            Object mode = options.get(MODE_OPTION);
+            boolean snapshot = mode != null && MODE_SNAPSHOT.equals(mode.toString());
+            return new FolderConfig(Path.of(directory.toString()), snapshot);
+        }
+    }
+}

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

@@ -0,0 +1,38 @@
+package stirling.software.proprietary.policy.input;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+
+import stirling.software.proprietary.policy.model.InputSpec;
+
+/**
+ * Resolves a policy {@link InputSpec} into the files to run on. Implementations are beans selected
+ * by {@link #supports(InputSpec)}, so a new source kind (folder, S3) is just a new bean. A manual
+ * run may supply files directly and bypass sources entirely.
+ */
+public interface InputSource {
+
+    /** Stable identifier for this source, matching {@code InputSpec.type()} (e.g. "folder"). */
+    String type();
+
+    /** Whether this source can handle the given spec. */
+    boolean supports(InputSpec spec);
+
+    /** Throws {@link IllegalArgumentException} on bad config. Called on save to fail fast. */
+    default void validate(InputSpec spec) {}
+
+    /**
+     * 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.
+     */
+    List<ResolvedInput> resolve(InputSpec spec) throws IOException;
+
+    /**
+     * Filesystem dirs this source draws from, for the folder-watch trigger. Advisory: resolving is
+     * still done by {@link #resolve}. Non-filesystem sources return empty and are not watchable.
+     */
+    default List<Path> watchTargets(InputSpec spec) {
+        return List.of();
+    }
+}

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

@@ -0,0 +1,22 @@
+package stirling.software.proprietary.policy.input;
+
+import java.util.function.Consumer;
+
+import stirling.software.proprietary.policy.model.PolicyInputs;
+
+/**
+ * One unit of work from an {@link InputSource}: the files to run plus a completion callback invoked
+ * with the run's success (e.g. a folder source routes the input to done/error). A source may return
+ * several of these, one per file.
+ */
+public record ResolvedInput(PolicyInputs inputs, Consumer<Boolean> onComplete) {
+
+    public ResolvedInput {
+        onComplete = onComplete == null ? success -> {} : onComplete;
+    }
+
+    /** No completion side effect. */
+    public static ResolvedInput of(PolicyInputs inputs) {
+        return new ResolvedInput(inputs, success -> {});
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/InputSpec.java

@@ -0,0 +1,19 @@
+package stirling.software.proprietary.policy.model;
+
+import java.util.Map;
+
+/**
+ * One input source for a policy. {@code type} keys an {@code InputSource} bean; a run pulls from
+ * every source.
+ */
+public record InputSpec(String type, Map<String, Object> options) {
+
+    public InputSpec {
+        options = options == null ? Map.of() : options;
+    }
+
+    /** Read input files from a directory on disk. */
+    public static InputSpec folder(String directory) {
+        return new InputSpec("folder", Map.of("directory", directory));
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/OutputSpec.java

@@ -2,20 +2,19 @@ package stirling.software.proprietary.policy.model;
 
 import java.util.Map;
 
-/**
- * Describes where a pipeline run's output files should be delivered. {@code type} selects a {@code
- * PolicyOutputSink} (e.g. "inline"); {@code options} carries sink-specific configuration.
- *
- * <p>New destinations (folder, S3) are added as new sink beans keyed on a new {@code type} without
- * changing this shape or the engine.
- */
+/** Where a run's outputs are delivered. {@code type} keys a {@code PolicyOutputSink} bean. */
 public record OutputSpec(String type, Map<String, Object> options) {
     public OutputSpec {
         options = options == null ? Map.of() : options;
     }
 
-    /** The default destination: store outputs and return them to the caller for download. */
+    /** Default sink: store outputs and return them to the caller for download. */
     public static OutputSpec inline() {
         return new OutputSpec("inline", Map.of());
     }
+
+    /** Write outputs to a directory on disk. */
+    public static OutputSpec folder(String directory) {
+        return new OutputSpec("folder", Map.of("directory", directory));
+    }
 }

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineDefinition.java

@@ -3,11 +3,10 @@ package stirling.software.proprietary.policy.model;
 import java.util.List;
 
 /**
- * An ordered chain of tool steps plus where the output should go.
+ * An ordered chain of tool steps plus an output destination; the unit the engine executes.
  *
- * <p>This is the single shape executed by the policy engine, shared by AI plans, manually-triggered
- * runs, and (later) watched folders. {@code output} may be null for callers that handle result
- * files themselves (e.g. the AI workflow, which builds its own response payload).
+ * <p>{@code output} may be null for callers that handle result files themselves (e.g. the AI
+ * workflow, which builds its own response payload).
  */
 public record PipelineDefinition(String name, List<PipelineStep> steps, OutputSpec output) {
     public PipelineDefinition {

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineStep.java

@@ -3,17 +3,13 @@ package stirling.software.proprietary.policy.model;
 import java.util.Map;
 
 /**
- * A single tool invocation in a pipeline: the API endpoint path to call and the inputs to pass.
+ * A single tool invocation. {@code operation} is a Stirling endpoint path (e.g. {@code
+ * /api/v1/misc/compress-pdf}) per the {@code InternalApiClient} convention; {@code parameters} are
+ * scalar form fields.
  *
- * <p>{@code operation} is a Stirling tool endpoint path (e.g. {@code /api/v1/misc/compress-pdf}),
- * matching the dispatch convention used by {@code InternalApiClient}. {@code parameters} are the
- * tool-specific scalar form fields.
- *
- * <p>{@code fileParameters} binds a tool's named file fields (beyond the primary {@code fileInput}
- * stream) to supporting files supplied with the run: it maps the form field name (e.g. {@code
- * stampImage}, {@code overlayFiles}) to an asset key in the run's supporting-file store. This keeps
- * supporting inputs (a stamp image, a certificate, an overlay) out of the document stream that
- * flows step to step.
+ * <p>{@code fileParameters} maps a tool's named file field (e.g. {@code stampImage}, beyond the
+ * primary {@code fileInput} stream) to an asset key in the run's supporting-file store, keeping
+ * supporting inputs out of the document stream that flows step to step.
  */
 public record PipelineStep(
         String operation, Map<String, Object> parameters, Map<String, String> fileParameters) {

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Policy.java

@@ -3,18 +3,11 @@ package stirling.software.proprietary.policy.model;
 import java.util.List;
 
 /**
- * A stored, owned automation: how it is triggered, the ordered tool steps to run, and where its
- * output goes, plus identity and metadata.
+ * A stored automation: ordered tool steps, input sources, and an output destination.
  *
- * <p>This is the central object of the feature. Everything that runs a chain of tools is a use of a
- * Policy: a watched folder is a Policy with a folder {@link TriggerConfig} and a folder {@link
- * OutputSpec}; a scheduled job is a Policy with a schedule trigger; manual/Automate/AI runs execute
- * a Policy (or an ad-hoc {@link PipelineDefinition}) on demand. The engine itself only ever
- * executes the {@link PipelineDefinition} this exposes via {@link #toDefinition()} - it is
- * trigger-agnostic.
- *
- * <p>{@code enabled} gates automatic triggering (a disabled policy is not picked up by its
- * trigger); it does not block an explicit manual run.
+ * <p>Always runnable on demand. An optional {@link TriggerConfig} fires it automatically; a {@code
+ * null} trigger means manual-only. Trigger decides when, {@link InputSpec sources} decide where
+ * files come from; a run pulls from every source.
  */
 public record Policy(
         String id,
@@ -22,16 +15,29 @@ public record Policy(
         String owner,
         boolean enabled,
         TriggerConfig trigger,
+        List<InputSpec> sources,
         List<PipelineStep> steps,
         OutputSpec output) {
 
     public Policy {
-        trigger = trigger == null ? TriggerConfig.manual() : trigger;
+        sources = sources == null ? List.of() : List.copyOf(sources);
         steps = steps == null ? List.of() : steps;
         output = output == null ? OutputSpec.inline() : output;
     }
 
-    /** The engine-level, trigger-agnostic view of this policy's pipeline. */
+    /** A policy with no configured sources (a generator, or files supplied directly to a run). */
+    public Policy(
+            String id,
+            String name,
+            String owner,
+            boolean enabled,
+            TriggerConfig trigger,
+            List<PipelineStep> steps,
+            OutputSpec output) {
+        this(id, name, owner, enabled, trigger, List.of(), steps, output);
+    }
+
+    /** This policy's pipeline as the engine sees it. */
     public PipelineDefinition toDefinition() {
         return new PipelineDefinition(name, steps, output);
     }

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PolicyInputs.java

@@ -6,17 +6,9 @@ import java.util.Map;
 import org.springframework.core.io.Resource;
 
 /**
- * The files a run operates on, split into two roles:
- *
- * <ul>
- *   <li>{@code primary} - the documents that flow through the pipeline, each step's output becoming
- *       the next step's input.
- *   <li>{@code supportingFiles} - a named store of auxiliary files (a stamp image, certificate,
- *       overlay, attachments) that steps bind to their named file fields via {@link
- *       PipelineStep#fileParameters()}. These never enter the document stream.
- * </ul>
- *
- * Asset values are lists so a single key can carry multi-file fields (e.g. attachments).
+ * A run's files. {@code primary} documents flow step to step; {@code supportingFiles} are auxiliary
+ * assets bound by key via {@link PipelineStep#fileParameters()} and never enter the document
+ * stream. Asset values are lists so one key can carry a multi-file field (e.g. attachments).
  */
 public record PolicyInputs(List<Resource> primary, Map<String, List<Resource>> supportingFiles) {
 

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PolicyRun.java

@@ -8,12 +8,10 @@ import lombok.Getter;
 import stirling.software.common.model.job.ResultFile;
 
 /**
- * Live, mutable state of a single pipeline run, held in memory by {@code PolicyRunRegistry}.
- *
- * <p>This carries the rich execution state (status, step cursor, wait state) that the job system's
- * {@code JobResult} does not model. The run is also projected into {@code TaskManager} for
- * cluster-visible status, progress notes, and file download; this object is the authoritative
- * source of the state machine.
+ * Live, mutable state of one pipeline run, held in memory by {@code PolicyRunRegistry} and the
+ * authoritative source of the state machine. Carries execution state ({@code JobResult} does not
+ * model status/step cursor/wait state); also projected into {@code TaskManager} for cluster-visible
+ * status and download.
  */
 @Getter
 public class PolicyRun {
@@ -69,9 +67,7 @@ public class PolicyRun {
         touch();
     }
 
-    /**
-     * Mark cancelled if the run has not already reached a terminal state. Returns whether it did.
-     */
+    /** Cancels unless already terminal; returns whether it transitioned. */
     public synchronized boolean cancel() {
         if (status.isTerminal()) {
             return false;

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PolicyRunStatus.java

@@ -1,11 +1,8 @@
 package stirling.software.proprietary.policy.model;
 
 /**
- * Lifecycle states of a {@link PolicyRun}.
- *
- * <p>{@code WAITING_FOR_INPUT} is modelled now so the engine and run shape support pausing a run
- * (e.g. a step that blocks for a human decision) without holding a thread; the resume handshake is
- * implemented in a later stage.
+ * Lifecycle states of a {@link PolicyRun}. {@code WAITING_FOR_INPUT} models a thread-free pause;
+ * the resume handshake lands in a later stage.
  */
 public enum PolicyRunStatus {
     PENDING,

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PolicyRunView.java

@@ -5,8 +5,8 @@ import java.util.List;
 import stirling.software.common.model.job.ResultFile;
 
 /**
- * Read-only view of a {@link PolicyRun} returned by the status endpoint. Output files are surfaced
- * as {@link ResultFile} so the caller can download each via {@code GET /api/v1/general/files/{id}}.
+ * Read-only view of a {@link PolicyRun} for the status endpoint. Outputs are {@link ResultFile}s,
+ * downloadable via {@code GET /api/v1/general/files/{id}}.
  */
 public record PolicyRunView(
         String runId,

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Schedule.java

@@ -0,0 +1,130 @@
+package stirling.software.proprietary.policy.model;
+
+import java.time.DayOfWeek;
+import java.time.LocalTime;
+import java.time.ZonedDateTime;
+import java.util.EnumSet;
+import java.util.Set;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+
+/**
+ * A scheduled policy's firing cadence; {@code type} is the JSON discriminator. Wall-clock kinds
+ * ({@link Daily}, {@link Weekly}, {@link Monthly}) evaluate in the {@code after} argument's zone;
+ * {@link Every} is a fixed offset and ignores wall-clock time.
+ */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
+@JsonSubTypes({
+    @JsonSubTypes.Type(value = Schedule.Every.class, name = "every"),
+    @JsonSubTypes.Type(value = Schedule.Daily.class, name = "daily"),
+    @JsonSubTypes.Type(value = Schedule.Weekly.class, name = "weekly"),
+    @JsonSubTypes.Type(value = Schedule.Monthly.class, name = "monthly"),
+})
+@JsonIgnoreProperties(ignoreUnknown = true)
+public sealed interface Schedule {
+
+    /** The next firing strictly after {@code after}, evaluated in {@code after}'s zone. */
+    ZonedDateTime nextAfter(ZonedDateTime after);
+
+    /** The granularities a fixed-interval schedule can repeat on. */
+    enum Unit {
+        MINUTES,
+        HOURS,
+        DAYS
+    }
+
+    /** A fixed offset from {@code after}: "every 15 minutes", "every 6 hours". No time of day. */
+    record Every(long count, Unit unit) implements Schedule {
+        public Every {
+            if (count <= 0) {
+                throw new IllegalArgumentException("'every' schedule needs a positive count");
+            }
+            if (unit == null) {
+                throw new IllegalArgumentException("'every' schedule needs a unit");
+            }
+        }
+
+        @Override
+        public ZonedDateTime nextAfter(ZonedDateTime after) {
+            return switch (unit) {
+                case MINUTES -> after.plusMinutes(count);
+                case HOURS -> after.plusHours(count);
+                case DAYS -> after.plusDays(count);
+            };
+        }
+    }
+
+    /** Once a day at a wall-clock time: "every day at 02:00". */
+    record Daily(LocalTime at) implements Schedule {
+        public Daily {
+            requireTime(at);
+        }
+
+        @Override
+        public ZonedDateTime nextAfter(ZonedDateTime after) {
+            ZonedDateTime today = after.with(at);
+            return today.isAfter(after) ? today : today.plusDays(1);
+        }
+    }
+
+    /** On chosen weekdays at a wall-clock time: "every Monday and Thursday at 09:00". */
+    record Weekly(Set<DayOfWeek> days, LocalTime at) implements Schedule {
+        public Weekly {
+            if (days == null || days.isEmpty()) {
+                throw new IllegalArgumentException("'weekly' schedule needs at least one day");
+            }
+            requireTime(at);
+            days = EnumSet.copyOf(days);
+        }
+
+        @Override
+        public ZonedDateTime nextAfter(ZonedDateTime after) {
+            // Soonest of the next 7 days landing on a chosen weekday, at the configured time.
+            for (int i = 0; i <= 7; i++) {
+                ZonedDateTime candidate = after.plusDays(i).with(at);
+                if (candidate.isAfter(after) && days.contains(candidate.getDayOfWeek())) {
+                    return candidate;
+                }
+            }
+            throw new IllegalStateException("unreachable: a chosen weekday recurs within 8 days");
+        }
+    }
+
+    /**
+     * On a day of the month at a wall-clock time: "the 1st at 00:00". Months too short for the
+     * chosen day (e.g. the 31st in February) are skipped, not clamped.
+     */
+    record Monthly(int dayOfMonth, LocalTime at) implements Schedule {
+        public Monthly {
+            if (dayOfMonth < 1 || dayOfMonth > 31) {
+                throw new IllegalArgumentException("'monthly' day-of-month must be 1-31");
+            }
+            requireTime(at);
+        }
+
+        @Override
+        public ZonedDateTime nextAfter(ZonedDateTime after) {
+            ZonedDateTime firstOfMonth = after.withDayOfMonth(1).with(at);
+            // Scan forward a few years' worth of months to skip ones without the chosen day.
+            for (int i = 0; i < 48; i++) {
+                ZonedDateTime month = firstOfMonth.plusMonths(i);
+                if (month.toLocalDate().lengthOfMonth() >= dayOfMonth) {
+                    ZonedDateTime fire = month.withDayOfMonth(dayOfMonth);
+                    if (fire.isAfter(after)) {
+                        return fire;
+                    }
+                }
+            }
+            throw new IllegalStateException(
+                    "unreachable: a month with the chosen day recurs yearly");
+        }
+    }
+
+    private static void requireTime(LocalTime at) {
+        if (at == null) {
+            throw new IllegalArgumentException("schedule needs a time of day ('at')");
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/TriggerConfig.java

@@ -3,23 +3,13 @@ package stirling.software.proprietary.policy.model;
 import java.util.Map;
 
 /**
- * How a {@link Policy} is automatically triggered. {@code type} selects a trigger kind ("manual",
- * "folder", "schedule", "s3"); {@code options} carries type-specific configuration (a folder path,
- * a cron expression, a bucket, ...).
- *
- * <p>Data-driven and parallel to {@link OutputSpec}: new trigger kinds are new {@code type} values
- * handled by a new trigger bean, with no change to the model. {@code "manual"} means there is no
- * automatic trigger - the policy is only ever run on demand.
+ * A {@link Policy}'s automatic trigger; {@code type} keys a trigger bean (e.g. "schedule"). Manual
+ * running is not a trigger kind: a manual-only policy carries a {@code null} {@code TriggerConfig}.
+ * Answers only "when"; file sources are the policy's {@link InputSpec}s.
  */
 public record TriggerConfig(String type, Map<String, Object> options) {
 
     public TriggerConfig {
-        type = type == null || type.isBlank() ? "manual" : type;
         options = options == null ? Map.of() : options;
     }
-
-    /** No automatic trigger; the policy is run on demand only. */
-    public static TriggerConfig manual() {
-        return new TriggerConfig("manual", Map.of());
-    }
 }

app/proprietary/src/main/java/stirling/software/proprietary/policy/model/WaitState.java

@@ -3,14 +3,10 @@ package stirling.software.proprietary.policy.model;
 import java.util.List;
 
 /**
- * Captured when a run pauses in {@link PolicyRunStatus#WAITING_FOR_INPUT}. Together with the run's
- * {@link PipelineDefinition} this is the resumable snapshot: {@code resumeStepIndex} is the 0-based
- * step to continue from, and {@code pendingFileIds} are the intermediate files (stored in {@code
- * FileStorage}, so they survive the worker thread ending or a node restart) that become the input
- * to the resumed run.
- *
- * <p>Stored as fileIds rather than in-memory resources by design: a paused run must be resumable
- * long after its worker thread has gone.
+ * Resumable snapshot captured when a run pauses ({@link PolicyRunStatus#WAITING_FOR_INPUT}). {@code
+ * resumeStepIndex} is the 0-based step to continue from; {@code pendingFileIds} are intermediate
+ * files held in {@code FileStorage} (not in-memory resources) so a pause survives the worker thread
+ * ending or a node restart.
  */
 public record WaitState(String reason, int resumeStepIndex, List<String> pendingFileIds) {
     public WaitState {

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

@@ -0,0 +1,123 @@
+package stirling.software.proprietary.policy.output;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.apache.commons.io.FilenameUtils;
+import org.springframework.core.io.Resource;
+import org.springframework.http.MediaType;
+import org.springframework.http.MediaTypeFactory;
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.model.job.ResultFile;
+import stirling.software.proprietary.policy.config.FolderAccessGuard;
+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}}.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class FolderOutputSink implements PolicyOutputSink {
+
+    static final String TYPE = FolderAccessGuard.FOLDER_TYPE;
+    static final String DIRECTORY_OPTION = "directory";
+
+    private final FolderAccessGuard accessGuard;
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    @Override
+    public boolean supports(OutputSpec spec) {
+        return spec != null && TYPE.equals(spec.type());
+    }
+
+    @Override
+    public void validate(OutputSpec spec) {
+        accessGuard.requirePermitted(directoryOf(spec));
+    }
+
+    @Override
+    public List<ResultFile> deliver(String runId, List<Resource> outputs, OutputSpec spec)
+            throws IOException {
+        Path targetDir = accessGuard.requirePermitted(directoryOf(spec));
+        Files.createDirectories(targetDir);
+
+        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);
+            String contentType =
+                    MediaTypeFactory.getMediaType(name)
+                            .orElse(MediaType.APPLICATION_OCTET_STREAM)
+                            .toString();
+            results.add(
+                    ResultFile.builder()
+                            .fileId(UUID.randomUUID().toString())
+                            .fileName(target.toString())
+                            .contentType(contentType)
+                            .fileSize(size)
+                            .build());
+            log.debug("Wrote policy run {} output to {}", runId, target);
+        }
+        return results;
+    }
+
+    private static Path directoryOf(OutputSpec spec) {
+        Object directory = spec.options().get(DIRECTORY_OPTION);
+        if (directory == null || directory.toString().isBlank()) {
+            throw new IllegalArgumentException(
+                    "folder output requires a '" + DIRECTORY_OPTION + "' option");
+        }
+        return Path.of(directory.toString());
+    }
+
+    // Strip any directory component / "../" so a crafted output name cannot escape targetDir.
+    private static String safeName(String filename, int index) {
+        if (filename == null || filename.isBlank()) {
+            return "output-" + index;
+        }
+        String name = FilenameUtils.getName(filename);
+        if (name.isBlank() || ".".equals(name) || "..".equals(name)) {
+            return "output-" + index;
+        }
+        return name;
+    }
+
+    // Non-colliding path, appending " (n)" before the extension.
+    private static Path uniqueTarget(Path dir, String filename) {
+        Path candidate = dir.resolve(filename);
+        if (!Files.exists(candidate)) {
+            return candidate;
+        }
+        String base = FilenameUtils.getBaseName(filename);
+        String ext = FilenameUtils.getExtension(filename);
+        String suffix = ext.isEmpty() ? "" : "." + ext;
+        for (int n = 1; ; n++) {
+            Path next = dir.resolve(base + " (" + n + ")" + suffix);
+            if (!Files.exists(next)) {
+                return next;
+            }
+        }
+    }
+}

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

@@ -17,9 +17,8 @@ import stirling.software.common.service.FileStorage;
 import stirling.software.proprietary.policy.model.OutputSpec;
 
 /**
- * Default output sink: stores each output file in {@code FileStorage} so it is downloadable via
- * {@code GET /api/v1/general/files/{fileId}}. This is the destination for manually-triggered runs
- * whose results are returned to the caller.
+ * Default sink: stores each output in {@code FileStorage} so it is downloadable via {@code GET
+ * /api/v1/general/files/{fileId}}. Used for manual runs whose results return to the caller.
  */
 @Service
 @RequiredArgsConstructor

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

@@ -9,11 +9,9 @@ import stirling.software.common.model.job.ResultFile;
 import stirling.software.proprietary.policy.model.OutputSpec;
 
 /**
- * Delivers a finished run's output files to a destination, returning durable {@link ResultFile}
- * descriptors (fileId + metadata) for the run record.
- *
- * <p>Implementations are Spring beans selected by {@link #supports(OutputSpec)}. New destinations
- * (folder, S3) are added as new beans without changing the engine.
+ * Delivers a finished run's outputs to a destination, returning {@link ResultFile} descriptors for
+ * the run record. Implementations are beans selected by {@link #supports(OutputSpec)}, so a new
+ * destination (folder, S3) is just a new bean.
  */
 public interface PolicyOutputSink {
 
@@ -23,13 +21,10 @@ public interface PolicyOutputSink {
     /** Whether this sink can handle the given output spec. */
     boolean supports(OutputSpec spec);
 
-    /**
-     * Persist/deliver the output files and return their descriptors.
-     *
-     * @param runId the run these outputs belong to
-     * @param outputs the final pipeline output resources
-     * @param spec the requested destination
-     */
+    /** Throws {@link IllegalArgumentException} on bad config. Called on save to fail fast. */
+    default void validate(OutputSpec spec) {}
+
+    /** Persist/deliver the output files and return their descriptors. */
     List<ResultFile> deliver(String runId, List<Resource> outputs, OutputSpec spec)
             throws IOException;
 }

app/proprietary/src/main/java/stirling/software/proprietary/policy/progress/PolicyProgressListener.java

@@ -1,22 +1,17 @@
 package stirling.software.proprietary.policy.progress;
 
 /**
- * Receives live progress as a pipeline run executes. Implementations forward to an SSE stream,
- * write job notes for polling, or both. Step indices are 1-based.
- *
- * <p>All methods default to no-ops so callers implement only what they surface.
+ * Receives live progress as a pipeline run executes (SSE stream, job notes, or both). Step indices
+ * are 1-based. All methods default to no-ops.
  */
 public interface PolicyProgressListener {
 
-    /** A listener that ignores all progress. */
     PolicyProgressListener NOOP = new PolicyProgressListener() {};
 
-    /** Called immediately before step {@code stepIndex} of {@code stepCount} begins. */
     default void onStepStart(int stepIndex, int stepCount, String operation) {}
 
-    /** Called immediately after step {@code stepIndex} of {@code stepCount} completes. */
     default void onStepComplete(int stepIndex, int stepCount, String operation) {}
 
-    /** Called on a keep-alive tick so downstream connections can detect disconnects promptly. */
+    /** Keep-alive tick so downstream connections can detect disconnects promptly. */
     default void onHeartbeat() {}
 }

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

@@ -9,9 +9,8 @@ import java.util.concurrent.ConcurrentHashMap;
 import stirling.software.proprietary.policy.model.Policy;
 
 /**
- * In-memory {@link PolicyStore}. Not the runtime bean - {@link JpaPolicyStore} is the durable
- * store. Kept as a lightweight, dependency-free implementation for tests and for any future no-
- * database mode.
+ * In-memory {@link PolicyStore} for tests and any future no-database mode. {@link JpaPolicyStore}
+ * is the runtime bean.
  */
 public class InProcessPolicyStore implements PolicyStore {
 
@@ -30,6 +29,7 @@ public class InProcessPolicyStore implements PolicyStore {
                         policy.owner(),
                         policy.enabled(),
                         policy.trigger(),
+                        policy.sources(),
                         policy.steps(),
                         policy.output());
         policies.put(id, stored);
@@ -50,6 +50,7 @@ public class InProcessPolicyStore implements PolicyStore {
     public List<Policy> findByTriggerType(String triggerType) {
         return policies.values().stream()
                 .filter(Policy::enabled)
+                .filter(policy -> policy.trigger() != null)
                 .filter(policy -> triggerType.equals(policy.trigger().type()))
                 .toList();
     }

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

@@ -13,9 +13,8 @@ import stirling.software.proprietary.policy.model.Policy;
 import tools.jackson.databind.ObjectMapper;
 
 /**
- * Durable {@link PolicyStore} backed by JPA. The runtime store whenever the proprietary module runs
- * (a datasource is always present). Policies are persisted as JSON via {@link PolicyEntity}; the
- * scalar columns are kept in sync for querying.
+ * Durable {@link PolicyStore} backed by JPA; the runtime store. Policies are persisted as JSON via
+ * {@link PolicyEntity}, with scalar columns kept in sync for querying.
  */
 @Service
 @RequiredArgsConstructor
@@ -37,6 +36,7 @@ public class JpaPolicyStore implements PolicyStore {
                         policy.owner(),
                         policy.enabled(),
                         policy.trigger(),
+                        policy.sources(),
                         policy.steps(),
                         policy.output());
 
@@ -45,7 +45,7 @@ public class JpaPolicyStore implements PolicyStore {
         entity.setName(stored.name());
         entity.setOwner(stored.owner());
         entity.setEnabled(stored.enabled());
-        entity.setTriggerType(stored.trigger().type());
+        entity.setTriggerType(stored.trigger() == null ? null : stored.trigger().type());
         entity.setPolicyJson(objectMapper.writeValueAsString(stored));
         repository.save(entity);
         return stored;

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

@@ -12,13 +12,11 @@ import lombok.NoArgsConstructor;
 import lombok.Setter;
 
 /**
- * JPA row for a {@link stirling.software.proprietary.policy.model.Policy}.
- *
- * <p>The whole policy is stored as JSON in {@code policyJson} (authoritative on read, and the same
- * serialization the API uses); the scalar columns are denormalized copies for querying - notably
- * {@code triggerType} + {@code enabled} so background triggers can fetch their policies. Ownership
- * is a plain {@code owner} string rather than a foreign key, to stay decoupled from the security
- * entities; richer team scoping can be layered on later.
+ * JPA row for a {@link stirling.software.proprietary.policy.model.Policy}. The whole policy lives
+ * as JSON in {@code policyJson} (authoritative on read); the scalar columns are denormalized copies
+ * for querying, notably {@code triggerType} + {@code enabled} so background triggers can fetch
+ * their policies. {@code owner} is a plain string, not a foreign key, to stay decoupled from the
+ * security entities.
  */
 @Entity
 @Table(name = "policies")

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

@@ -5,24 +5,19 @@ import java.util.Optional;
 
 import stirling.software.proprietary.policy.model.Policy;
 
-/**
- * Stores {@link Policy} definitions. The in-memory implementation backs simple deployments now; a
- * durable (JPA) implementation can replace it behind this interface without touching callers.
- */
+/** Stores {@link Policy} definitions. */
 public interface PolicyStore {
 
-    /** Create or update a policy. A blank/absent id is assigned; returns the stored policy. */
+    /** Create or update; a blank/absent id is assigned. Returns the stored policy. */
     Policy save(Policy policy);
 
     Optional<Policy> get(String id);
 
     List<Policy> all();
 
-    /**
-     * Enabled policies whose automatic trigger is of the given type (used by background triggers).
-     */
+    /** Enabled policies with the given trigger type, for background triggers. */
     List<Policy> findByTriggerType(String triggerType);
 
-    /** Remove a policy; returns whether it existed. */
+    /** Returns whether the policy existed. */
     boolean delete(String id);
 }

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

@@ -0,0 +1,288 @@
+package stirling.software.proprietary.policy.trigger;
+
+import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
+import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
+
+import java.io.IOException;
+import java.nio.file.ClosedWatchServiceException;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.WatchKey;
+import java.nio.file.WatchService;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.policy.engine.PolicyRunner;
+import stirling.software.proprietary.policy.input.InputSource;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.store.PolicyStore;
+
+/**
+ * Fires policies when a file lands in one of their folder sources, rather than polling on a timer.
+ *
+ * <p>The watch is a latency optimisation, not a source of truth: a periodic reconcile sweep ({@code
+ * watchReconcileSeconds}) re-syncs watched dirs and re-runs every policy, covering files that
+ * pre-dated the watch, dropped events, and filesystems that emit none (NFS, bind mounts). Redundant
+ * runs are harmless since {@link InputSource} does the claiming.
+ *
+ * <p>Watch state is in memory, so this assumes a single node and rebuilds registrations on restart.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class FolderWatchTrigger implements PolicyTrigger {
+
+    private static final String TYPE = "folder-watch";
+
+    private final PolicyStore policyStore;
+    private final PolicyRunner policyRunner;
+    private final List<InputSource> inputSources;
+    private final ApplicationProperties applicationProperties;
+
+    private final Map<Path, WatchKey> keysByDir = new ConcurrentHashMap<>();
+    private final Map<WatchKey, Path> dirByKey = new ConcurrentHashMap<>();
+
+    private volatile boolean running;
+
+    // Package-visible so tests can drive syncRegistrations() against a real service.
+    volatile WatchService watchService;
+
+    private volatile ScheduledExecutorService reconciler;
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    @Override
+    public void validate(Policy policy) {
+        if (watchDirsOf(policy).isEmpty()) {
+            throw new IllegalArgumentException(
+                    "folder-watch trigger requires at least one watchable (folder) input source");
+        }
+    }
+
+    @Override
+    public synchronized void start() {
+        if (watchService != null) {
+            return;
+        }
+        try {
+            watchService = FileSystems.getDefault().newWatchService();
+        } catch (IOException e) {
+            log.error("Could not start folder-watch trigger: {}", e.getMessage(), e);
+            return;
+        }
+        running = true;
+        Thread.ofVirtual().name("policy-folder-watch").start(this::watchLoop);
+        long reconcileSeconds = applicationProperties.getPolicies().getWatchReconcileSeconds();
+        reconciler =
+                Executors.newSingleThreadScheduledExecutor(
+                        Thread.ofVirtual().name("policy-folder-reconcile-", 0).factory());
+        // First reconcile runs immediately so pre-existing files are picked up at startup.
+        reconciler.scheduleAtFixedRate(this::safeReconcile, 0, reconcileSeconds, TimeUnit.SECONDS);
+        log.info("Folder-watch trigger started (reconcile every {}s)", reconcileSeconds);
+    }
+
+    @Override
+    public synchronized void stop() {
+        running = false;
+        if (reconciler != null) {
+            reconciler.shutdownNow();
+            reconciler = null;
+        }
+        if (watchService != null) {
+            try {
+                watchService.close(); // wakes the watch loop with ClosedWatchServiceException
+            } catch (IOException e) {
+                log.debug("Error closing folder watch service: {}", e.getMessage());
+            }
+            watchService = null;
+        }
+        keysByDir.clear();
+        dirByKey.clear();
+    }
+
+    private void watchLoop() {
+        // Capture once: stop() may null the field; close() still wakes take()/poll() on this local.
+        WatchService watcher = watchService;
+        if (watcher == null) {
+            return;
+        }
+        while (running) {
+            WatchKey first;
+            try {
+                first = watcher.take();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return;
+            } catch (ClosedWatchServiceException e) {
+                return;
+            }
+            runForChangedDirs(drainBurst(watcher, first));
+        }
+    }
+
+    /**
+     * Coalesce a burst of file-system events into one set of affected directories: drain everything
+     * arriving within the quiet period. Event kinds are irrelevant; any event means "go look".
+     */
+    private Set<Path> drainBurst(WatchService watcher, WatchKey first) {
+        long quietPeriodMs = applicationProperties.getPolicies().getWatchQuietPeriodMs();
+        Set<Path> changed = new HashSet<>();
+        WatchKey key = first;
+        while (key != null) {
+            key.pollEvents();
+            Path dir = dirByKey.get(key);
+            if (dir != null) {
+                changed.add(dir);
+            }
+            key.reset();
+            try {
+                key = watcher.poll(quietPeriodMs, TimeUnit.MILLISECONDS);
+            } catch (ClosedWatchServiceException | InterruptedException e) {
+                break;
+            }
+        }
+        return changed;
+    }
+
+    /** Run every folder-watch policy that draws from one of the changed directories. */
+    void runForChangedDirs(Set<Path> changedDirs) {
+        if (changedDirs.isEmpty()) {
+            return;
+        }
+        for (Policy policy : policyStore.findByTriggerType(TYPE)) {
+            List<Path> dirs;
+            try {
+                dirs = watchDirsOf(policy);
+            } catch (RuntimeException e) {
+                log.warn(
+                        "Folder-watch policy {} is misconfigured: {}", policy.id(), e.getMessage());
+                continue;
+            }
+            if (dirs.stream().anyMatch(changedDirs::contains)) {
+                log.debug("Folder-watch policy {} ({}) saw activity", policy.id(), policy.name());
+                policyRunner.run(policy);
+            }
+        }
+    }
+
+    private void safeReconcile() {
+        try {
+            syncRegistrations();
+            runAll();
+        } catch (RuntimeException e) {
+            log.error("Folder-watch reconcile failed: {}", e.getMessage(), e);
+        }
+    }
+
+    /** Reconcile safety net: run every folder-watch policy regardless of watch events. */
+    void runAll() {
+        for (Policy policy : policyStore.findByTriggerType(TYPE)) {
+            try {
+                policyRunner.run(policy);
+            } catch (RuntimeException e) {
+                log.warn(
+                        "Folder-watch reconcile run failed for policy {}: {}",
+                        policy.id(),
+                        e.getMessage());
+            }
+        }
+    }
+
+    /** Register newly-wanted dirs that exist on disk, cancel ones no longer wanted. */
+    synchronized void syncRegistrations() {
+        if (watchService == null) {
+            return;
+        }
+        Set<Path> desired = desiredDirs();
+
+        keysByDir
+                .entrySet()
+                .removeIf(
+                        entry -> {
+                            if (desired.contains(entry.getKey())) {
+                                return false;
+                            }
+                            entry.getValue().cancel();
+                            dirByKey.remove(entry.getValue());
+                            return true;
+                        });
+
+        for (Path dir : desired) {
+            if (keysByDir.containsKey(dir)) {
+                continue;
+            }
+            try {
+                WatchKey key = dir.register(watchService, ENTRY_CREATE, ENTRY_MODIFY);
+                keysByDir.put(dir, key);
+                dirByKey.put(key, dir);
+                log.info("Watching {} for folder-watch policies", dir);
+            } catch (IOException | RuntimeException e) {
+                log.warn("Could not watch {}: {}", dir, e.getMessage());
+            }
+        }
+    }
+
+    /** The directories currently registered with the watch service. Visible for tests. */
+    Set<Path> watchedDirs() {
+        return Set.copyOf(keysByDir.keySet());
+    }
+
+    /** Every existing directory any current folder-watch policy wants watched. */
+    private Set<Path> desiredDirs() {
+        Set<Path> dirs = new HashSet<>();
+        for (Policy policy : policyStore.findByTriggerType(TYPE)) {
+            try {
+                for (Path dir : watchDirsOf(policy)) {
+                    if (Files.isDirectory(dir)) {
+                        dirs.add(dir);
+                    }
+                }
+            } catch (RuntimeException e) {
+                log.warn(
+                        "Folder-watch policy {} is misconfigured: {}", policy.id(), e.getMessage());
+            }
+        }
+        return dirs;
+    }
+
+    // Absolute + normalised so registration keys and event-time matching compare regardless of how
+    // the path was configured.
+    private List<Path> watchDirsOf(Policy policy) {
+        List<Path> dirs = new ArrayList<>();
+        for (InputSpec spec : policy.sources()) {
+            InputSource source = sourceFor(spec);
+            if (source == null) {
+                continue;
+            }
+            for (Path dir : source.watchTargets(spec)) {
+                dirs.add(dir.toAbsolutePath().normalize());
+            }
+        }
+        return dirs;
+    }
+
+    private InputSource sourceFor(InputSpec spec) {
+        return inputSources.stream()
+                .filter(source -> source.supports(spec))
+                .findFirst()
+                .orElse(null);
+    }
+}

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

@@ -1,41 +0,0 @@
-package stirling.software.proprietary.policy.trigger;
-
-import org.springframework.stereotype.Service;
-
-import lombok.RequiredArgsConstructor;
-
-import stirling.software.proprietary.policy.engine.PolicyEngine;
-import stirling.software.proprietary.policy.engine.PolicyRunHandle;
-import stirling.software.proprietary.policy.model.PipelineDefinition;
-import stirling.software.proprietary.policy.model.Policy;
-import stirling.software.proprietary.policy.model.PolicyInputs;
-import stirling.software.proprietary.policy.progress.PolicyProgressListener;
-
-/**
- * Runs policies on demand, in response to a request (the {@code PolicyController} endpoints, an AI,
- * or another automation). It is the request-driven trigger: no background lifecycle, it just
- * forwards to the engine. Any policy can be run manually regardless of its configured trigger type.
- */
-@Service
-@RequiredArgsConstructor
-public class ManualTrigger implements PolicyTrigger {
-
-    private final PolicyEngine policyEngine;
-
-    @Override
-    public String type() {
-        return "manual";
-    }
-
-    /** Run a stored policy immediately and return its run handle. */
-    public PolicyRunHandle run(
-            Policy policy, PolicyInputs inputs, PolicyProgressListener listener) {
-        return policyEngine.runPolicy(policy, inputs, listener);
-    }
-
-    /** Run an ad-hoc pipeline (no stored policy), e.g. for AI or Automate one-offs. */
-    public PolicyRunHandle fire(
-            PipelineDefinition definition, PolicyInputs inputs, PolicyProgressListener listener) {
-        return policyEngine.submit(definition, inputs, listener);
-    }
-}

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

@@ -1,24 +1,23 @@
 package stirling.software.proprietary.policy.trigger;
 
+import stirling.software.proprietary.policy.model.Policy;
+
 /**
- * Activates policies of one trigger type. A trigger owns a {@link #type()} (matching {@code
- * TriggerConfig.type()}); when its condition fires it runs the relevant {@code Policy} through the
- * {@code PolicyEngine}.
- *
- * <p>Background triggers (folder watcher, schedule) are driven by configuration: on {@link
- * #start()} they begin watching/scheduling for the policies returned by {@code
- * PolicyStore.findByTriggerType(type())}, and stop on {@link #stop()}. Request-driven triggers
- * (manual) have no background lifecycle and run a policy directly in response to a call. New
- * trigger kinds are new beans of this type; the engine and the {@code Policy} model do not change.
+ * Decides <em>when</em> a policy runs. On firing it hands the policy to {@code PolicyRunner}; it
+ * never resolves sources itself. New trigger kinds are just new beans of this type.
  */
 public interface PolicyTrigger {
 
-    /** Stable identifier for this trigger kind, matching {@code TriggerConfig.type()}. */
+    /** Matches {@code TriggerConfig.type()}. */
     String type();
 
-    /** Begin activating policies of this type (e.g. start a folder watcher). No-op for manual. */
+    /**
+     * Validate at save time so misconfiguration fails fast, not at fire time. Receives the whole
+     * {@link Policy} so triggers that depend on the policy's sources (folder-watch) can check that.
+     */
+    default void validate(Policy policy) {}
+
     default void start() {}
 
-    /** Stop activating and release any resources. */
     default void stop() {}
 }

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

@@ -0,0 +1,49 @@
+package stirling.software.proprietary.policy.trigger;
+
+import java.util.List;
+
+import org.springframework.context.SmartLifecycle;
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+/** Starts and stops every {@link PolicyTrigger} with the application lifecycle. */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class PolicyTriggerManager implements SmartLifecycle {
+
+    private final List<PolicyTrigger> triggers;
+
+    private volatile boolean running;
+
+    @Override
+    public void start() {
+        for (PolicyTrigger trigger : triggers) {
+            try {
+                trigger.start();
+            } catch (RuntimeException e) {
+                log.error("Failed to start trigger '{}': {}", trigger.type(), e.getMessage(), e);
+            }
+        }
+        running = true;
+    }
+
+    @Override
+    public void stop() {
+        for (PolicyTrigger trigger : triggers) {
+            try {
+                trigger.stop();
+            } catch (RuntimeException e) {
+                log.error("Failed to stop trigger '{}': {}", trigger.type(), e.getMessage(), e);
+            }
+        }
+        running = false;
+    }
+
+    @Override
+    public boolean isRunning() {
+        return running;
+    }
+}

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

@@ -0,0 +1,149 @@
+package stirling.software.proprietary.policy.trigger;
+
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.policy.engine.PolicyRunner;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.Schedule;
+import stirling.software.proprietary.policy.store.PolicyStore;
+
+import tools.jackson.databind.ObjectMapper;
+
+/**
+ * Fires policies on a {@link Schedule}: a fixed-interval sweep runs each due "schedule" policy.
+ *
+ * <p>Last-fire times are in memory, so this assumes a single node and resets on restart.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class ScheduleTrigger implements PolicyTrigger {
+
+    private static final String TYPE = "schedule";
+
+    private final PolicyStore policyStore;
+    private final PolicyRunner policyRunner;
+    private final ObjectMapper objectMapper;
+    private final ApplicationProperties applicationProperties;
+
+    private final Map<String, Instant> lastFiredByPolicy = new ConcurrentHashMap<>();
+    private volatile ScheduledExecutorService scheduler;
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    @Override
+    public void validate(Policy policy) {
+        ScheduleConfig.from(objectMapper, policy.trigger().options());
+    }
+
+    @Override
+    public synchronized void start() {
+        if (scheduler != null) {
+            return;
+        }
+        long sweepSeconds = applicationProperties.getPolicies().getScheduleSweepSeconds();
+        scheduler =
+                Executors.newSingleThreadScheduledExecutor(
+                        Thread.ofVirtual().name("policy-schedule-", 0).factory());
+        scheduler.scheduleAtFixedRate(
+                this::safeSweep, sweepSeconds, sweepSeconds, TimeUnit.SECONDS);
+        log.info("Schedule trigger started (sweep every {}s)", sweepSeconds);
+    }
+
+    @Override
+    public synchronized void stop() {
+        if (scheduler != null) {
+            scheduler.shutdownNow();
+            scheduler = null;
+        }
+    }
+
+    private void safeSweep() {
+        try {
+            sweep(Instant.now());
+        } catch (RuntimeException e) {
+            log.error("Schedule sweep failed: {}", e.getMessage(), e);
+        }
+    }
+
+    /** Fire every scheduled policy that is due as of {@code now}. Package-visible for testing. */
+    void sweep(Instant now) {
+        for (Policy policy : policyStore.findByTriggerType(TYPE)) {
+            ScheduleConfig config;
+            try {
+                config = ScheduleConfig.from(objectMapper, policy.trigger().options());
+            } catch (IllegalArgumentException e) {
+                log.warn("Scheduled policy {} is misconfigured: {}", policy.id(), e.getMessage());
+                continue;
+            }
+
+            // Baseline a newly-seen policy to now so it does not fire immediately.
+            Instant last = lastFiredByPolicy.computeIfAbsent(policy.id(), id -> now);
+            ZonedDateTime next = config.schedule().nextAfter(last.atZone(config.zone()));
+            if (!next.toInstant().isAfter(now)) {
+                lastFiredByPolicy.put(policy.id(), now);
+                log.info("Scheduled policy {} ({}) is due", policy.id(), policy.name());
+                policyRunner.run(policy);
+            }
+        }
+    }
+
+    /**
+     * Validated schedule-trigger options: the {@link Schedule} and the zone it runs in (UTC by
+     * default).
+     */
+    record ScheduleConfig(Schedule schedule, ZoneId zone) {
+
+        private static final String SCHEDULE_OPTION = "schedule";
+        private static final String ZONE_OPTION = "zone";
+
+        static ScheduleConfig from(ObjectMapper mapper, Map<String, Object> options) {
+            Object scheduleNode = options.get(SCHEDULE_OPTION);
+            if (scheduleNode == null) {
+                throw new IllegalArgumentException("schedule trigger requires a 'schedule'");
+            }
+            Schedule schedule;
+            try {
+                schedule = mapper.convertValue(scheduleNode, Schedule.class);
+            } catch (RuntimeException e) {
+                throw new IllegalArgumentException("invalid schedule: " + rootMessage(e), e);
+            }
+
+            ZoneId zone = ZoneOffset.UTC;
+            Object zoneNode = options.get(ZONE_OPTION);
+            if (zoneNode != null && !zoneNode.toString().isBlank()) {
+                try {
+                    zone = ZoneId.of(zoneNode.toString());
+                } catch (RuntimeException e) {
+                    throw new IllegalArgumentException("invalid zone '" + zoneNode + "'");
+                }
+            }
+            return new ScheduleConfig(schedule, zone);
+        }
+
+        private static String rootMessage(Throwable t) {
+            Throwable cause = t;
+            while (cause.getCause() != null) {
+                cause = cause.getCause();
+            }
+            return cause.getMessage();
+        }
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/config/FolderAccessGuardTest.java

@@ -0,0 +1,99 @@
+package stirling.software.proprietary.policy.config;
+
+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 java.nio.file.Path;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.springframework.core.env.StandardEnvironment;
+
+import stirling.software.common.configuration.InstallationPathConfig;
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.Policy;
+
+/**
+ * Tests for {@link FolderAccessGuard}: folder access is fail-closed, confined to the configured
+ * allowed roots, never reaches Stirling's own config directory, and is off entirely under SaaS.
+ */
+class FolderAccessGuardTest {
+
+    @TempDir Path tempDir;
+
+    private FolderAccessGuard guard(List<String> allowedRoots, String... activeProfiles) {
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowedFolderRoots(allowedRoots);
+        StandardEnvironment environment = new StandardEnvironment();
+        environment.setActiveProfiles(activeProfiles);
+        return new FolderAccessGuard(properties, environment);
+    }
+
+    @Test
+    void permitsAndNormalisesADirectoryWithinAnAllowedRoot() {
+        FolderAccessGuard guard = guard(List.of(tempDir.toString()));
+        Path within = tempDir.resolve("inbox");
+
+        assertEquals(within.toAbsolutePath().normalize(), guard.requirePermitted(within));
+    }
+
+    @Test
+    void rejectsADirectoryOutsideEveryAllowedRoot() {
+        FolderAccessGuard guard = guard(List.of(tempDir.toString()));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> guard.requirePermitted(tempDir.resolveSibling("elsewhere")));
+    }
+
+    @Test
+    void rejectsTraversalThatWalksOutOfAnAllowedRoot() {
+        FolderAccessGuard guard = guard(List.of(tempDir.toString()));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> guard.requirePermitted(tempDir.resolve("..").resolve("escaped")));
+    }
+
+    @Test
+    void rejectsEverythingWhenNoRootsAreConfigured() {
+        FolderAccessGuard guard = guard(List.of());
+        assertThrows(IllegalArgumentException.class, () -> guard.requirePermitted(tempDir));
+    }
+
+    @Test
+    void rejectsTheStirlingConfigDirectoryEvenWhenItWouldBeInsideAnAllowedRoot() {
+        Path configDir =
+                Path.of(InstallationPathConfig.getConfigPath()).toAbsolutePath().normalize();
+        // Allow the config dir's parent, so only the protected-path rule can reject it.
+        FolderAccessGuard guard = guard(List.of(configDir.getParent().toString()));
+
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> guard.requirePermitted(configDir.resolve("settings.yml")));
+    }
+
+    @Test
+    void refusesAllFolderAccessUnderTheSaasProfile() {
+        FolderAccessGuard guard = guard(List.of(tempDir.toString()), "saas");
+        assertThrows(IllegalArgumentException.class, () -> guard.requirePermitted(tempDir));
+    }
+
+    @Test
+    void usesFolderAccessDetectsFolderSourcesAndOutputs() {
+        FolderAccessGuard guard = guard(List.of(tempDir.toString()));
+
+        assertTrue(
+                guard.usesFolderAccess(
+                        policy(List.of(InputSpec.folder("/in")), OutputSpec.inline())));
+        assertTrue(guard.usesFolderAccess(policy(List.of(), OutputSpec.folder("/out"))));
+        assertFalse(guard.usesFolderAccess(policy(List.of(), OutputSpec.inline())));
+    }
+
+    private static Policy policy(List<InputSpec> sources, OutputSpec output) {
+        return new Policy("p1", "p", "owner", true, null, sources, List.of(), output);
+    }
+}

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

@@ -50,7 +50,6 @@ 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.PolicyRunStatus;
-import stirling.software.proprietary.policy.model.TriggerConfig;
 import stirling.software.proprietary.policy.output.InlineOutputSink;
 import stirling.software.proprietary.policy.progress.PolicyProgressListener;
 
@@ -93,7 +92,7 @@ class PolicyEngineTest {
                         toolMetadataService,
                         tempFileManager,
                         JsonMapper.builder().build());
-        registry = new PolicyRunRegistry(30);
+        registry = new PolicyRunRegistry(new ApplicationProperties());
         InlineOutputSink sink = new InlineOutputSink(fileStorage);
         engine =
                 new PolicyEngine(
@@ -190,7 +189,7 @@ class PolicyEngineTest {
                         "rotate",
                         "owner",
                         true,
-                        TriggerConfig.manual(),
+                        null,
                         List.of(new PipelineStep(ROTATE, Map.of())),
                         OutputSpec.inline());
 

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

@@ -12,6 +12,7 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import stirling.software.common.model.ApplicationProperties;
 import stirling.software.proprietary.policy.model.PipelineDefinition;
 import stirling.software.proprietary.policy.model.PolicyRun;
 import stirling.software.proprietary.policy.model.WaitState;
@@ -25,7 +26,7 @@ class PolicyRunRegistryTest {
 
     @BeforeEach
     void setUp() {
-        registry = new PolicyRunRegistry(30);
+        registry = new PolicyRunRegistry(new ApplicationProperties());
     }
 
     @AfterEach

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

@@ -0,0 +1,159 @@
+package stirling.software.proprietary.policy.engine;
+
+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.eq;
+import static org.mockito.Mockito.mock;
+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.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import stirling.software.proprietary.policy.input.InputSource;
+import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineStep;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyInputs;
+import stirling.software.proprietary.policy.model.PolicyRun;
+import stirling.software.proprietary.policy.model.PolicyRunStatus;
+import stirling.software.proprietary.policy.progress.PolicyProgressListener;
+
+/**
+ * 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.
+ */
+@ExtendWith(MockitoExtension.class)
+class PolicyRunnerTest {
+
+    @Mock private PolicyEngine policyEngine;
+    @Mock private InputSource folderSource;
+
+    private PolicyRunner runner;
+
+    @BeforeEach
+    void setUp() {
+        runner = new PolicyRunner(policyEngine, List.of(folderSource));
+    }
+
+    @Test
+    void runsOnceWithNoFilesWhenThePolicyHasNoSources() {
+        Policy policy = policy(List.of());
+        when(policyEngine.runPolicy(eq(policy), any(), any()))
+                .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>()));
+
+        runner.run(policy);
+
+        ArgumentCaptor<PolicyInputs> inputs = ArgumentCaptor.forClass(PolicyInputs.class);
+        verify(policyEngine).runPolicy(eq(policy), inputs.capture(), any());
+        assertTrue(inputs.getValue().primary().isEmpty());
+    }
+
+    @Test
+    void pullsEverySourceAndRunsOnePerUnitOfWork() throws Exception {
+        InputSpec spec = InputSpec.folder("/in");
+        Policy policy = policy(List.of(spec));
+        when(folderSource.supports(spec)).thenReturn(true);
+        when(folderSource.resolve(spec))
+                .thenReturn(
+                        List.of(
+                                ResolvedInput.of(PolicyInputs.of(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, times(2)).runPolicy(eq(policy), any(), any());
+    }
+
+    @Test
+    void feedsEachUnitsCompletionHookTheRunOutcome() throws Exception {
+        InputSpec spec = InputSpec.folder("/in");
+        Policy policy = policy(List.of(spec));
+        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));
+        CompletableFuture<PolicyRun> completion = new CompletableFuture<>();
+        when(policyEngine.runPolicy(any(), any(), any()))
+                .thenReturn(new PolicyRunHandle("r", completion));
+
+        runner.run(policy);
+
+        PolicyRun run = mock(PolicyRun.class);
+        when(run.getStatus()).thenReturn(PolicyRunStatus.COMPLETED);
+        completion.complete(run);
+
+        assertTrue(outcome.get());
+    }
+
+    @Test
+    void reportsFailureToTheCompletionHookWhenTheRunDoesNotComplete() throws Exception {
+        InputSpec spec = InputSpec.folder("/in");
+        Policy policy = policy(List.of(spec));
+        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));
+        CompletableFuture<PolicyRun> completion = new CompletableFuture<>();
+        when(policyEngine.runPolicy(any(), any(), any()))
+                .thenReturn(new PolicyRunHandle("r", completion));
+
+        runner.run(policy);
+        completion.completeExceptionally(new RuntimeException("boom"));
+
+        assertFalse(outcome.get());
+    }
+
+    @Test
+    void skipsSourcesWithNoMatchingBean() {
+        InputSpec spec = new InputSpec("s3", Map.of());
+        Policy policy = policy(List.of(spec));
+        when(folderSource.supports(spec)).thenReturn(false);
+
+        runner.run(policy);
+
+        verifyNoInteractions(policyEngine);
+    }
+
+    @Test
+    void runWithSuppliedInputsBypassesSources() {
+        Policy policy = policy(List.of(InputSpec.folder("/in")));
+        PolicyInputs inputs = PolicyInputs.of(List.of());
+        PolicyRunHandle handle = new PolicyRunHandle("r", new CompletableFuture<>());
+        when(policyEngine.runPolicy(policy, inputs, PolicyProgressListener.NOOP))
+                .thenReturn(handle);
+
+        assertSame(handle, runner.runWith(policy, inputs, PolicyProgressListener.NOOP));
+        verifyNoInteractions(folderSource);
+    }
+
+    private static Policy policy(List<InputSpec> sources) {
+        return new Policy(
+                "p1",
+                "p",
+                "owner",
+                true,
+                null,
+                sources,
+                List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
+                OutputSpec.inline());
+    }
+}

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

@@ -0,0 +1,114 @@
+package stirling.software.proprietary.policy.engine;
+
+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.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import stirling.software.proprietary.policy.input.InputSource;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.TriggerConfig;
+import stirling.software.proprietary.policy.output.PolicyOutputSink;
+import stirling.software.proprietary.policy.trigger.PolicyTrigger;
+
+/** Tests for {@link PolicyValidator}: routes each facet to its handler and surfaces failures. */
+@ExtendWith(MockitoExtension.class)
+class PolicyValidatorTest {
+
+    @Mock private PolicyTrigger trigger;
+    @Mock private InputSource inputSource;
+    @Mock private PolicyOutputSink outputSink;
+
+    private PolicyValidator validator;
+
+    @BeforeEach
+    void setUp() {
+        validator =
+                new PolicyValidator(List.of(trigger), List.of(inputSource), List.of(outputSink));
+    }
+
+    @Test
+    void delegatesEachFacetToItsHandler() {
+        when(trigger.type()).thenReturn("schedule");
+        when(inputSource.supports(any())).thenReturn(true);
+        when(outputSink.supports(any())).thenReturn(true);
+        Policy policy = policy("schedule");
+
+        validator.validate(policy);
+
+        verify(trigger).validate(policy);
+        verify(inputSource).validate(policy.sources().get(0));
+        verify(outputSink).validate(policy.output());
+    }
+
+    @Test
+    void skipsTriggerValidationForAManualOnlyPolicy() {
+        when(inputSource.supports(any())).thenReturn(true);
+        when(outputSink.supports(any())).thenReturn(true);
+
+        validator.validate(manualOnly());
+
+        verify(trigger, never()).validate(any());
+    }
+
+    @Test
+    void surfacesAnInvalidConfigFromAHandler() {
+        when(trigger.type()).thenReturn("schedule");
+        doThrow(new IllegalArgumentException("invalid schedule")).when(trigger).validate(any());
+
+        IllegalArgumentException ex =
+                assertThrows(
+                        IllegalArgumentException.class,
+                        () -> validator.validate(policy("schedule")));
+        assertTrue(ex.getMessage().contains("schedule"));
+    }
+
+    @Test
+    void rejectsAnUnknownTriggerType() {
+        when(trigger.type()).thenReturn("schedule");
+
+        IllegalArgumentException ex =
+                assertThrows(
+                        IllegalArgumentException.class,
+                        () -> validator.validate(policy("mystery")));
+        assertTrue(ex.getMessage().contains("unknown trigger type"));
+    }
+
+    private static Policy policy(String triggerType) {
+        return new Policy(
+                "p1",
+                "p",
+                "owner",
+                true,
+                new TriggerConfig(triggerType, Map.of()),
+                List.of(InputSpec.folder("/in")),
+                List.of(),
+                OutputSpec.inline());
+    }
+
+    private static Policy manualOnly() {
+        return new Policy(
+                "p1",
+                "p",
+                "owner",
+                true,
+                null,
+                List.of(InputSpec.folder("/in")),
+                List.of(),
+                OutputSpec.inline());
+    }
+}

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

@@ -0,0 +1,137 @@
+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 java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+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.model.InputSpec;
+
+/** Tests for {@link FolderInputSource}: consume (claim + route) and snapshot (read-only) modes. */
+@ExtendWith(MockitoExtension.class)
+class FolderInputSourceTest {
+
+    @Mock private FileReadinessChecker readinessChecker;
+
+    @TempDir Path tempDir;
+
+    private FolderInputSource source;
+
+    @BeforeEach
+    void setUp() {
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowedFolderRoots(List.of(tempDir.toString()));
+        FolderAccessGuard guard = new FolderAccessGuard(properties, new StandardEnvironment());
+        source = new FolderInputSource(readinessChecker, guard);
+        // Lenient: the missing-dir / nonexistent-dir cases return before any readiness check.
+        lenient().when(readinessChecker.isReady(any())).thenReturn(true);
+    }
+
+    @Test
+    void consumeClaimsFilesAndRoutesToDoneOnSuccess() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Files.writeString(inputDir.resolve("doc.pdf"), "data");
+
+        List<ResolvedInput> work = source.resolve(InputSpec.folder(inputDir.toString()));
+
+        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")));
+
+        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")));
+    }
+
+    @Test
+    void consumeRoutesToErrorOnFailure() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Files.writeString(inputDir.resolve("doc.pdf"), "data");
+
+        List<ResolvedInput> work = source.resolve(InputSpec.folder(inputDir.toString()));
+        work.get(0).onComplete().accept(false);
+
+        assertTrue(Files.exists(inputDir.resolve(".stirling").resolve("error").resolve("doc.pdf")));
+    }
+
+    @Test
+    void snapshotReadsWithoutClaiming() throws IOException {
+        Path inputDir = Files.createDirectories(tempDir.resolve("in"));
+        Files.writeString(inputDir.resolve("doc.pdf"), "data");
+
+        List<ResolvedInput> work =
+                source.resolve(
+                        new InputSpec(
+                                "folder",
+                                Map.of("directory", inputDir.toString(), "mode", "snapshot")));
+
+        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")));
+    }
+
+    @Test
+    void missingDirectoryOptionFails() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> source.resolve(new InputSpec("folder", Map.of())));
+    }
+
+    @Test
+    void nonexistentDirectoryYieldsNoWork() throws IOException {
+        List<ResolvedInput> work =
+                source.resolve(InputSpec.folder(tempDir.resolve("nope").toString()));
+        assertTrue(work.isEmpty());
+    }
+
+    @Test
+    void validateRejectsMissingDirectory() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> source.validate(new InputSpec("folder", Map.of())));
+    }
+
+    @Test
+    void rejectsADirectoryOutsideTheAllowedRoots() {
+        Path outside = tempDir.resolveSibling("not-allowed");
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> source.resolve(InputSpec.folder(outside.toString())));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> source.validate(InputSpec.folder(outside.toString())));
+    }
+
+    @Test
+    void watchTargetsIsTheConfiguredDirectory() {
+        Path inputDir = tempDir.resolve("in");
+        assertEquals(List.of(inputDir), source.watchTargets(InputSpec.folder(inputDir.toString())));
+    }
+}

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

@@ -0,0 +1,105 @@
+package stirling.software.proprietary.policy.output;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.springframework.core.env.StandardEnvironment;
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.Resource;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.model.job.ResultFile;
+import stirling.software.proprietary.policy.config.FolderAccessGuard;
+import stirling.software.proprietary.policy.model.OutputSpec;
+
+/** Tests for {@link FolderOutputSink}: outputs are written to the configured directory on disk. */
+class FolderOutputSinkTest {
+
+    @TempDir Path tempDir;
+
+    private FolderOutputSink sink;
+
+    @BeforeEach
+    void setUp() {
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowedFolderRoots(List.of(tempDir.toString()));
+        sink = new FolderOutputSink(new FolderAccessGuard(properties, new StandardEnvironment()));
+    }
+
+    @Test
+    void writesEachOutputToTheDirectory() throws IOException {
+        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()));
+
+        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")));
+    }
+
+    @Test
+    void collidingNamesGetAUniqueSuffix() throws IOException {
+        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()));
+
+        assertTrue(Files.exists(out.resolve("a.pdf")));
+        assertTrue(Files.exists(out.resolve("a (1).pdf")));
+    }
+
+    @Test
+    void missingDirectoryOptionIsRejected() {
+        OutputSpec noDir = new OutputSpec("folder", Map.of());
+        assertThrows(IllegalArgumentException.class, () -> sink.validate(noDir));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> sink.deliver("run-1", List.of(named("a.pdf", "x")), noDir));
+    }
+
+    @Test
+    void aDirectoryOutsideTheAllowedRootsIsRejected() {
+        OutputSpec outside = OutputSpec.folder(tempDir.resolveSibling("not-allowed").toString());
+        assertThrows(IllegalArgumentException.class, () -> sink.validate(outside));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> sink.deliver("run-1", List.of(named("a.pdf", "x")), outside));
+    }
+
+    @Test
+    void filenamesWithPathTraversalAreConfinedToTheDirectory() throws IOException {
+        Path out = tempDir.resolve("out");
+        List<Resource> outputs =
+                List.of(named("../escape.pdf", "x"), named("nested/deep.pdf", "y"));
+
+        sink.deliver("run-1", 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")));
+        assertTrue(Files.exists(out.resolve("deep.pdf")));
+        assertFalse(Files.exists(tempDir.resolve("escape.pdf")));
+    }
+
+    private static ByteArrayResource named(String filename, String content) {
+        return new ByteArrayResource(content.getBytes()) {
+            @Override
+            public String getFilename() {
+                return filename;
+            }
+        };
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/store/InProcessPolicyStoreTest.java

@@ -28,7 +28,7 @@ class InProcessPolicyStoreTest {
 
     @Test
     void savedPolicyGetsAnIdAndIsRetrievable() {
-        Policy saved = store.save(policy(null, "compress", "manual", true));
+        Policy saved = store.save(policy(null, "compress", null, true));
 
         assertNotNull(saved.id());
         assertFalse(saved.id().isBlank());
@@ -37,7 +37,7 @@ class InProcessPolicyStoreTest {
 
     @Test
     void savingWithAnExistingIdUpdatesInPlace() {
-        Policy created = store.save(policy(null, "before", "manual", true));
+        Policy created = store.save(policy(null, "before", null, true));
 
         store.save(
                 new Policy(
@@ -45,7 +45,7 @@ class InProcessPolicyStoreTest {
                         "after",
                         "owner",
                         true,
-                        TriggerConfig.manual(),
+                        null,
                         List.of(),
                         OutputSpec.inline()));
 
@@ -55,19 +55,20 @@ class InProcessPolicyStoreTest {
 
     @Test
     void findByTriggerTypeReturnsOnlyEnabledMatches() {
-        store.save(policy(null, "watch", "folder", true));
-        store.save(policy(null, "watch-disabled", "folder", false));
         store.save(policy(null, "nightly", "schedule", true));
+        store.save(policy(null, "nightly-disabled", "schedule", false));
+        store.save(policy(null, "hooked", "webhook", true));
+        store.save(policy(null, "on-demand", null, true)); // manual-only: no trigger
 
-        List<Policy> folder = store.findByTriggerType("folder");
+        List<Policy> scheduled = store.findByTriggerType("schedule");
 
-        assertEquals(1, folder.size());
-        assertEquals("watch", folder.get(0).name());
+        assertEquals(1, scheduled.size());
+        assertEquals("nightly", scheduled.get(0).name());
     }
 
     @Test
     void deleteRemovesThePolicy() {
-        Policy saved = store.save(policy(null, "p", "manual", true));
+        Policy saved = store.save(policy(null, "p", null, true));
 
         assertTrue(store.delete(saved.id()));
         assertTrue(store.get(saved.id()).isEmpty());
@@ -75,12 +76,14 @@ class InProcessPolicyStoreTest {
     }
 
     private static Policy policy(String id, String name, String triggerType, boolean enabled) {
+        TriggerConfig trigger =
+                triggerType == null ? null : new TriggerConfig(triggerType, Map.of());
         return new Policy(
                 id,
                 name,
                 "owner",
                 enabled,
-                new TriggerConfig(triggerType, Map.of()),
+                trigger,
                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                 OutputSpec.inline());
     }

app/proprietary/src/test/java/stirling/software/proprietary/policy/store/JpaPolicyStoreTest.java

@@ -18,6 +18,7 @@ import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
+import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.OutputSpec;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
@@ -53,7 +54,8 @@ class JpaPolicyStoreTest {
                                 "compress incoming",
                                 "alice",
                                 true,
-                                new TriggerConfig("folder", Map.of("path", "/in")),
+                                new TriggerConfig("schedule", Map.of()),
+                                List.of(InputSpec.folder("/in")),
                                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                                 OutputSpec.inline()));
 
@@ -62,7 +64,7 @@ class JpaPolicyStoreTest {
         verify(repository).save(captor.capture());
         PolicyEntity entity = captor.getValue();
         assertEquals(saved.id(), entity.getId());
-        assertEquals("folder", entity.getTriggerType());
+        assertEquals("schedule", entity.getTriggerType());
         assertTrue(entity.isEnabled());
         // The stored JSON round-trips back to an equal policy.
         assertEquals(saved, objectMapper.readValue(entity.getPolicyJson(), Policy.class));
@@ -76,7 +78,7 @@ class JpaPolicyStoreTest {
                         "rotate",
                         "alice",
                         true,
-                        TriggerConfig.manual(),
+                        null, // manual-only: no automatic trigger
                         List.of(
                                 new PipelineStep(
                                         "/api/v1/general/rotate-pdf", Map.of("angle", 90))),
@@ -94,16 +96,16 @@ class JpaPolicyStoreTest {
                         "watch",
                         "alice",
                         true,
-                        new TriggerConfig("folder", Map.of()),
+                        new TriggerConfig("schedule", Map.of()),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline());
-        when(repository.findByTriggerTypeAndEnabledTrue("folder"))
+        when(repository.findByTriggerTypeAndEnabledTrue("schedule"))
                 .thenReturn(List.of(entityFor(policy)));
 
-        List<Policy> folder = store.findByTriggerType("folder");
+        List<Policy> scheduled = store.findByTriggerType("schedule");
 
-        assertEquals(1, folder.size());
-        assertEquals("p1", folder.get(0).id());
+        assertEquals(1, scheduled.size());
+        assertEquals("p1", scheduled.get(0).id());
     }
 
     @Test
@@ -122,7 +124,7 @@ class JpaPolicyStoreTest {
         entity.setName(policy.name());
         entity.setOwner(policy.owner());
         entity.setEnabled(policy.enabled());
-        entity.setTriggerType(policy.trigger().type());
+        entity.setTriggerType(policy.trigger() == null ? null : policy.trigger().type());
         entity.setPolicyJson(objectMapper.writeValueAsString(policy));
         return entity;
     }

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

@@ -0,0 +1,178 @@
+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.Mockito.lenient;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.WatchService;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.policy.engine.PolicyRunner;
+import stirling.software.proprietary.policy.input.InputSource;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineStep;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.TriggerConfig;
+import stirling.software.proprietary.policy.store.PolicyStore;
+
+/**
+ * Tests for {@link FolderWatchTrigger}'s dispatch logic via the package-visible {@code
+ * runForChangedDirs}/{@code runAll}, plus its cross-facet validation. The OS watch loop and
+ * scheduled reconcile are thin glue around these and are not exercised here (a real {@code
+ * WatchService} is timing-dependent), mirroring how {@link ScheduleTriggerTest} drives {@code
+ * sweep} directly. The folder source is stubbed to mirror {@code FolderInputSource.watchTargets}.
+ */
+@ExtendWith(MockitoExtension.class)
+class FolderWatchTriggerTest {
+
+    @Mock private PolicyStore policyStore;
+    @Mock private PolicyRunner policyRunner;
+    @Mock private InputSource folderSource;
+
+    @TempDir Path tempDir;
+
+    private FolderWatchTrigger trigger;
+
+    @BeforeEach
+    void setUp() {
+        trigger =
+                new FolderWatchTrigger(
+                        policyStore,
+                        policyRunner,
+                        List.of(folderSource),
+                        new ApplicationProperties());
+        lenient().when(folderSource.supports(any())).thenReturn(true);
+        lenient()
+                .when(folderSource.watchTargets(any()))
+                .thenAnswer(
+                        invocation -> {
+                            InputSpec spec = invocation.getArgument(0);
+                            Object dir = spec.options().get("directory");
+                            if (dir == null) {
+                                throw new IllegalArgumentException(
+                                        "folder input requires a 'directory' option");
+                            }
+                            return List.of(Path.of(dir.toString()));
+                        });
+    }
+
+    @Test
+    void validateRejectsPolicyWithNoWatchableSource() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> trigger.validate(folderWatch("p1", List.of())));
+    }
+
+    @Test
+    void validateAcceptsPolicyWithAFolderSource() {
+        trigger.validate(folderWatch("p1", List.of(InputSpec.folder("/in"))));
+    }
+
+    @Test
+    void runsOnlyPoliciesDrawingFromTheChangedDirectory() {
+        Policy a = folderWatch("a", List.of(InputSpec.folder("/in/a")));
+        Policy b = folderWatch("b", List.of(InputSpec.folder("/in/b")));
+        when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(a, b));
+
+        trigger.runForChangedDirs(Set.of(normalized("/in/a")));
+
+        verify(policyRunner).run(a);
+        verify(policyRunner, never()).run(b);
+    }
+
+    @Test
+    void skipsAMisconfiguredPolicyButStillRunsTheOthers() {
+        Policy bad = folderWatch("bad", List.of(new InputSpec("folder", Map.of())));
+        Policy good = folderWatch("good", List.of(InputSpec.folder("/in/a")));
+        when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(bad, good));
+
+        trigger.runForChangedDirs(Set.of(normalized("/in/a")));
+
+        verify(policyRunner).run(good);
+        verify(policyRunner, never()).run(bad);
+    }
+
+    @Test
+    void anEmptyChangeSetDoesNothing() {
+        trigger.runForChangedDirs(Set.of());
+
+        verifyNoInteractions(policyStore, policyRunner);
+    }
+
+    @Test
+    void reconcileRunsEveryFolderWatchPolicyAsASafetyNet() {
+        Policy a = folderWatch("a", List.of(InputSpec.folder("/in/a")));
+        Policy b = folderWatch("b", List.of(InputSpec.folder("/in/b")));
+        when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(a, b));
+
+        trigger.runAll();
+
+        verify(policyRunner).run(a);
+        verify(policyRunner).run(b);
+    }
+
+    @Test
+    void syncRegistrationsWatchesExistingDirsAndCancelsRemovedOnes() throws Exception {
+        Path dirA = Files.createDirectories(tempDir.resolve("a"));
+        Path dirB = Files.createDirectories(tempDir.resolve("b"));
+        Path missing = tempDir.resolve("missing"); // never created on disk
+
+        Policy a = folderWatch("a", List.of(InputSpec.folder(dirA.toString())));
+        Policy b = folderWatch("b", List.of(InputSpec.folder(dirB.toString())));
+        Policy m = folderWatch("m", List.of(InputSpec.folder(missing.toString())));
+
+        WatchService service = FileSystems.getDefault().newWatchService();
+        try {
+            trigger.watchService = service;
+
+            when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(a, b, m));
+            trigger.syncRegistrations();
+            // Existing dirs are watched; the non-existent one is skipped.
+            assertEquals(
+                    Set.of(normalized(dirA.toString()), normalized(dirB.toString())),
+                    trigger.watchedDirs());
+
+            // b's policy is removed: its registration is cancelled, a remains.
+            when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(a));
+            trigger.syncRegistrations();
+            assertEquals(Set.of(normalized(dirA.toString())), trigger.watchedDirs());
+        } finally {
+            service.close();
+        }
+    }
+
+    private static Path normalized(String dir) {
+        return Path.of(dir).toAbsolutePath().normalize();
+    }
+
+    private static Policy folderWatch(String id, List<InputSpec> sources) {
+        return new Policy(
+                id,
+                "watcher",
+                "owner",
+                true,
+                new TriggerConfig("folder-watch", Map.of()),
+                sources,
+                List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
+                OutputSpec.inline());
+    }
+}

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

@@ -0,0 +1,51 @@
+package stirling.software.proprietary.policy.trigger;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+/**
+ * Tests for {@link PolicyTriggerManager}: starts/stops every trigger, tolerating individual
+ * failures.
+ */
+@ExtendWith(MockitoExtension.class)
+class PolicyTriggerManagerTest {
+
+    @Mock private PolicyTrigger triggerA;
+    @Mock private PolicyTrigger triggerB;
+
+    @Test
+    void startsAndStopsAllTriggers() {
+        PolicyTriggerManager manager = new PolicyTriggerManager(List.of(triggerA, triggerB));
+        assertFalse(manager.isRunning());
+
+        manager.start();
+        verify(triggerA).start();
+        verify(triggerB).start();
+        assertTrue(manager.isRunning());
+
+        manager.stop();
+        verify(triggerA).stop();
+        verify(triggerB).stop();
+        assertFalse(manager.isRunning());
+    }
+
+    @Test
+    void oneTriggerFailingToStartDoesNotBlockTheOthers() {
+        doThrow(new RuntimeException("boom")).when(triggerA).start();
+        PolicyTriggerManager manager = new PolicyTriggerManager(List.of(triggerA, triggerB));
+
+        manager.start();
+
+        verify(triggerB).start();
+        assertTrue(manager.isRunning());
+    }
+}

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

@@ -0,0 +1,147 @@
+package stirling.software.proprietary.policy.trigger;
+
+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.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.time.DayOfWeek;
+import java.time.Instant;
+import java.time.LocalTime;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.policy.engine.PolicyRunner;
+import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineStep;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.Schedule;
+import stirling.software.proprietary.policy.model.TriggerConfig;
+import stirling.software.proprietary.policy.store.PolicyStore;
+
+import tools.jackson.databind.json.JsonMapper;
+
+/**
+ * Tests for {@link ScheduleTrigger}'s due-firing logic via the package-visible {@code
+ * sweep(Instant)}. The trigger only decides when a policy is due; pulling sources and starting runs
+ * is the {@link PolicyRunner}'s job, so these assert it delegates to the runner. Schedules default
+ * to UTC, so explicit UTC instants make these deterministic.
+ */
+@ExtendWith(MockitoExtension.class)
+class ScheduleTriggerTest {
+
+    @Mock private PolicyStore policyStore;
+    @Mock private PolicyRunner policyRunner;
+
+    private ScheduleTrigger trigger;
+
+    @BeforeEach
+    void setUp() {
+        trigger =
+                new ScheduleTrigger(
+                        policyStore,
+                        policyRunner,
+                        JsonMapper.builder().build(),
+                        new ApplicationProperties());
+    }
+
+    @Test
+    void firesOncePerScheduleWhenItComesDue() {
+        Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES));
+        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+
+        Instant t0 = Instant.parse("2026-06-05T10:00:30Z");
+        trigger.sweep(t0); // first sight: baseline, must not fire immediately
+        verify(policyRunner, never()).run(any());
+
+        trigger.sweep(t0.plusSeconds(120)); // the one-minute mark has passed
+        verify(policyRunner, times(1)).run(eq(policy));
+    }
+
+    @Test
+    void doesNotFireBeforeTheNextScheduledTime() {
+        Policy policy = scheduled("p1", new Schedule.Daily(LocalTime.of(3, 0))); // 03:00 UTC daily
+        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+
+        Instant t0 = Instant.parse("2026-06-05T10:00:00Z");
+        trigger.sweep(t0);
+        trigger.sweep(t0.plusSeconds(60)); // next 03:00 is far away
+
+        verify(policyRunner, never()).run(any());
+    }
+
+    @Test
+    void firesWeeklyOnAChosenDay() {
+        // 2026-06-05 is a Friday; the next Monday 09:00 is the soonest firing.
+        Policy policy =
+                scheduled("p1", new Schedule.Weekly(Set.of(DayOfWeek.MONDAY), LocalTime.of(9, 0)));
+        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+
+        Instant friday = Instant.parse("2026-06-05T10:00:00Z");
+        trigger.sweep(friday); // baseline
+        trigger.sweep(Instant.parse("2026-06-08T09:00:00Z")); // Monday 09:00
+
+        verify(policyRunner, times(1)).run(eq(policy));
+    }
+
+    @Test
+    void skipsPoliciesWithAnInvalidSchedule() {
+        Policy policy = scheduledWithRawOptions("p1", Map.of()); // no schedule
+        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+
+        trigger.sweep(Instant.parse("2026-06-05T10:00:00Z"));
+
+        verify(policyRunner, never()).run(any());
+    }
+
+    @Test
+    void validateRejectsMissingSchedule() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> trigger.validate(scheduledWithRawOptions("p1", Map.of())));
+    }
+
+    @Test
+    void validateRejectsAnInvalidSchedule() {
+        Map<String, Object> options =
+                Map.of("schedule", Map.of("type", "every", "count", -5, "unit", "MINUTES"));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> trigger.validate(scheduledWithRawOptions("p1", options)));
+    }
+
+    @Test
+    void validateAcceptsAValidScheduleAndZone() {
+        Map<String, Object> options = new LinkedHashMap<>();
+        options.put("schedule", new Schedule.Daily(LocalTime.of(2, 0)));
+        options.put("zone", "Europe/London");
+        trigger.validate(scheduledWithRawOptions("p1", options));
+    }
+
+    private static Policy scheduled(String id, Schedule schedule) {
+        return scheduledWithRawOptions(id, Map.of("schedule", schedule));
+    }
+
+    private static Policy scheduledWithRawOptions(String id, Map<String, Object> options) {
+        return new Policy(
+                id,
+                "nightly",
+                "owner",
+                true,
+                new TriggerConfig("schedule", options),
+                List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
+                OutputSpec.inline());
+    }
+}