Change pipelines to have 1 input and 1 output (#7121)

# Description of Changes

Change pipelines so that sources and triggers are grouped into a list of
inputs, so you can have a different trigger for each source in the list.
This is necessary because triggers are not universally supported by all
source types. If you wanted to have a pipeline pull from both a folder
and an S3 bucket, the current system allows you to choose "Folder Watch"
as the trigger, which will either do nothing or crash when it's paired
with the S3 bucket.

I've got reservations about actually allowing different triggers for
every source because it allows for user workflows that I don't believe
exist, like "I want this folder to be polled every minute and this other
one to be polled every hour, but they should run the same tools and
should output to the same place". Because of this (with agreement from
Connor, Anthony and Matt) I've changed this PR to artificially limit
pipelines to having 1 input & output at this stage. The backend is still
shaped to support multiple inputs & outputs so it should be trivial to
re-add support for them in the future if we decide we want to, but the
UI can be much simpler and easier to understand with just 1 input and
output.

<img width="1262" height="521" alt="image"
src="https://github.com/user-attachments/assets/809e6803-9f99-436d-9aeb-52dddf0906ff"
/>

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

@@ -355,8 +355,7 @@ public class PolicyController {
                 policy.name(),
                 owner,
                 policy.enabled(),
-                policy.trigger(),
-                policy.sourceIds(),
+                policy.inputs(),
                 policy.steps(),
                 policy.output(),
                 policy.outputIds(),

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

@@ -15,6 +15,7 @@ import stirling.software.proprietary.policy.input.ResolvedInput;
 import stirling.software.proprietary.policy.ledger.ProcessedLedger;
 import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.PipelineDefinition;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.model.PolicyInputs;
 import stirling.software.proprietary.policy.model.PolicyRun;
@@ -42,30 +43,46 @@ public class PolicyRunner {
     private final SourceDocCounter docCounter;
     private final ProcessedLedger processedLedger;
 
-    /** Full-listing sweep: resolve every source, then reconcile the ledger. */
+    /** Full-listing sweep over every input: resolve each source, then reconcile the ledger. */
     public SweepOutcome run(Policy policy) {
         return run(policy, SweepKind.FULL);
     }
 
-    /**
-     * Trigger entry point. Pulls every referenced source; each yielded unit becomes its own run so
-     * one failure does not affect the others. No sources means one run with no input (generator
-     * pipeline). Missing or disabled sources are skipped so one broken reference does not stop the
-     * rest. Returns the ids of the runs it started plus what the sweep skipped, so a manual trigger
-     * can report which runs to follow or why nothing ran.
-     */
+    /** Sweep every input of the policy at the given listing depth. */
     public SweepOutcome run(Policy policy, SweepKind sweep) {
+        return run(policy, policy.inputs(), sweep);
+    }
+
+    /**
+     * Fire one input binding: a background trigger pulling its own source without touching the
+     * policy's other inputs. Never reconciles the ledger (it sees a single source, so pruning would
+     * wrongly forget the rest); a full-policy sweep handles that.
+     */
+    public SweepOutcome runInput(Policy policy, PipelineInput input, SweepKind sweep) {
+        return run(policy, List.of(input), sweep);
+    }
+
+    /**
+     * Core sweep: pulls each of the given inputs' sources; each yielded unit becomes its own run so
+     * one failure does not affect the others. No inputs means one run with no input (generator
+     * pipeline). Missing or disabled sources are skipped so one broken reference does not stop the
+     * rest. Presence cleanup only runs when the sweep covered every input of the policy - a
+     * single-binding fire cannot reconcile the whole policy's ledger. Returns the ids of the runs
+     * it started plus what the sweep skipped, so a manual trigger can report which runs to follow
+     * or why nothing ran.
+     */
+    public SweepOutcome run(Policy policy, List<PipelineInput> inputs, SweepKind sweep) {
         long sweepStart = System.currentTimeMillis();
         PolicySweep context = new PolicySweep(policy.id(), sweep, processedLedger);
         List<String> runIds = new ArrayList<>();
-        List<String> sourceIds = policy.sourceIds();
-        if (sourceIds.isEmpty()) {
+        if (inputs.isEmpty()) {
             // Generator pipeline: one run with no input. Still fall through to the cleanup
             // below so rows recorded for its folder outputs are pruned like anything else,
             // instead of accumulating until the policy is deleted.
             runIds.add(startRun(policy, PolicyInputs.of(List.of()), unused -> {}));
         }
-        for (String sourceId : sourceIds) {
+        for (PipelineInput input : inputs) {
+            String sourceId = input.sourceId();
             Source source = sourceStore.get(sourceId).orElse(null);
             if (source == null) {
                 // No veto: a deleted source's rows should age out via the cleanup below.
@@ -84,7 +101,8 @@ public class PolicyRunner {
             }
             runIds.addAll(pullAndRun(policy, sourceId, source.toInputSpec(), context));
         }
-        if (context.cleanupAllowed()) {
+        boolean fullPolicy = inputs.size() == policy.inputs().size();
+        if (fullPolicy && context.cleanupAllowed()) {
             processedLedger.markSeen(policy.id(), context.presentIdentities());
             int removed = processedLedger.deleteUnseen(policy.id(), sweepStart);
             if (removed > 0) {

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

@@ -9,6 +9,7 @@ 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.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.model.TriggerConfig;
@@ -18,10 +19,11 @@ import stirling.software.proprietary.policy.source.SourceStore;
 import stirling.software.proprietary.policy.trigger.PolicyTrigger;
 
 /**
- * Validates a policy at save time by delegating each facet (trigger, sources, steps, 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. Each referenced {@code sourceId}
- * must resolve to a persisted {@link Source} whose config its {@link InputSource} bean accepts.
+ * Validates a policy at save time by delegating each facet (inputs, their triggers, output) to the
+ * bean that handles its type, so a misconfiguration fails fast rather than at run time. Each
+ * input's {@code sourceId} must resolve to a persisted {@link Source} whose config its {@link
+ * InputSource} bean accepts; its optional trigger must be a known type compatible with that source.
+ * A null trigger is a manual-only input and skips trigger validation.
  */
 @Service
 @RequiredArgsConstructor
@@ -34,21 +36,31 @@ public class PolicyValidator {
     private final SourceStore sourceStore;
 
     /**
-     * @throws IllegalArgumentException if any facet's type is unknown, a referenced source does not
-     *     exist, or any config is invalid
+     * @throws IllegalArgumentException if the policy has more than one input or output, any facet's
+     *     type is unknown, a referenced source does not exist, a trigger is incompatible with its
+     *     input's source, or any config is invalid
      */
     public void validate(Policy policy) {
-        if (policy.trigger() != null) {
-            triggerFor(policy.trigger()).validate(policy);
+        // Deliberate product cap, not a model limit: the lists stay lists so multiple
+        // inputs/outputs can be supported later, but today a policy carries at most one of
+        // each (zero of either remains fine - run on demand / inline output).
+        if (policy.inputs().size() > 1) {
+            throw new IllegalArgumentException("a policy supports at most one input");
         }
-        for (String sourceId : policy.sourceIds()) {
+        if (policy.outputIds().size() > 1) {
+            throw new IllegalArgumentException("a policy supports at most one output");
+        }
+        for (PipelineInput input : policy.inputs()) {
             Source source =
                     sourceStore
-                            .get(sourceId)
+                            .get(input.sourceId())
                             .orElseThrow(
                                     () ->
                                             new IllegalArgumentException(
-                                                    "unknown source: " + sourceId));
+                                                    "unknown source: " + input.sourceId()));
+            if (input.trigger() != null) {
+                validateTrigger(policy, input, source);
+            }
             InputSpec spec = source.toInputSpec();
             inputSourceFor(spec).validate(spec);
         }
@@ -73,6 +85,24 @@ public class PolicyValidator {
         }
     }
 
+    /**
+     * Check an input's trigger is a known type whose source constraints its source satisfies (e.g.
+     * folder-watch only on a folder source), then let the trigger validate its own options.
+     */
+    private void validateTrigger(Policy policy, PipelineInput input, Source source) {
+        PolicyTrigger trigger = triggerFor(input.trigger());
+        if (!trigger.supportedSourceTypes().isEmpty()
+                && !trigger.supportedSourceTypes().contains(source.type())) {
+            throw new IllegalArgumentException(
+                    "trigger '"
+                            + trigger.type()
+                            + "' is not compatible with source type '"
+                            + source.type()
+                            + "'");
+        }
+        trigger.validate(policy, input);
+    }
+
     /**
      * Validate an output spec against its sink. Must be called on a request thread (caller's
      * principal present) so an S3 output's connection is authorization-checked against the caller -

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

@@ -0,0 +1,16 @@
+package stirling.software.proprietary.policy.model;
+
+/**
+ * One input of a policy: a reference to a persisted {@code Source} paired with the {@link
+ * TriggerConfig} that decides when <em>this</em> source is pulled. The trigger lives on the
+ * binding, not on the source (so one connection can feed many policies on different schedules) and
+ * not on the policy (so a folder input can be watched while an S3 input on the same policy polls).
+ * A {@code null} trigger means this input is pulled only when the policy is run on demand.
+ */
+public record PipelineInput(String sourceId, TriggerConfig trigger) {
+
+    /** An input with no automatic trigger: pulled only on a manual run. */
+    public static PipelineInput manual(String sourceId) {
+        return new PipelineInput(sourceId, null);
+    }
+}

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

@@ -3,29 +3,31 @@ package stirling.software.proprietary.policy.model;
 import java.util.List;
 
 /**
- * A stored automation: ordered tool steps, input sources, and output destinations.
+ * A stored automation: ordered tool steps, input bindings, and output destinations.
  *
- * <p>Always runnable on demand. An optional {@link TriggerConfig} fires it automatically; a {@code
- * null} trigger means manual-only. Trigger decides when; {@code sourceIds} reference the persisted
- * {@code Source} locations (resolved live at run time) files come from; a run pulls from every
- * referenced source. {@code outputIds} reference the {@code Source} locations (resolved live) a
- * run's files are delivered to - a run is delivered to every one; when empty the inline {@link
- * #output} is used (results returned to the caller), the case for editor and one-off policies.
+ * <p>Always runnable on demand. Each {@link PipelineInput} references a persisted {@code Source}
+ * connection (resolved live at run time) and carries its own optional {@link TriggerConfig}: the
+ * trigger decides when that source is pulled, so one input can be watched while another polls, and
+ * a {@code null} trigger makes that input manual-only. An input with no trigger, or a policy with
+ * no triggered inputs, still runs when the policy is run on demand; a manual run pulls every input.
+ *
+ * <p>{@code outputIds} reference the {@code Source} locations (resolved live) a run's files are
+ * delivered to - a run is delivered to every one; when empty the inline {@link #output} is used
+ * (results returned to the caller), the case for editor and one-off policies.
  */
 public record Policy(
         String id,
         String name,
         String owner,
         boolean enabled,
-        TriggerConfig trigger,
-        List<String> sourceIds,
+        List<PipelineInput> inputs,
         List<PipelineStep> steps,
         OutputSpec output,
         List<String> outputIds,
         Long teamId) {
 
     public Policy {
-        sourceIds = sourceIds == null ? List.of() : List.copyOf(sourceIds);
+        inputs = inputs == null ? List.of() : List.copyOf(inputs);
         steps = steps == null ? List.of() : steps;
         output = output == null ? OutputSpec.inline() : output;
         outputIds = outputIds == null ? List.of() : List.copyOf(outputIds);
@@ -41,12 +43,11 @@ public record Policy(
             String name,
             String owner,
             boolean enabled,
-            TriggerConfig trigger,
-            List<String> sourceIds,
+            List<PipelineInput> inputs,
             List<PipelineStep> steps,
             OutputSpec output,
             Long teamId) {
-        this(id, name, owner, enabled, trigger, sourceIds, steps, output, List.of(), teamId);
+        this(id, name, owner, enabled, inputs, steps, output, List.of(), teamId);
     }
 
     /**
@@ -58,35 +59,35 @@ public record Policy(
             String name,
             String owner,
             boolean enabled,
-            TriggerConfig trigger,
-            List<String> sourceIds,
+            List<PipelineInput> inputs,
             List<PipelineStep> steps,
             OutputSpec output) {
-        this(id, name, owner, enabled, trigger, sourceIds, steps, output, List.of(), null);
+        this(id, name, owner, enabled, inputs, steps, output, List.of(), null);
     }
 
-    /** 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, List.of(), null);
+    /** The source ids this policy pulls from, in input order; a derived view for reads. */
+    public List<String> sourceIds() {
+        return inputs.stream().map(PipelineInput::sourceId).toList();
+    }
+
+    /** The distinct trigger types configured across this policy's inputs (manual inputs aside). */
+    public List<String> triggerTypes() {
+        return inputs.stream()
+                .map(PipelineInput::trigger)
+                .filter(trigger -> trigger != null)
+                .map(TriggerConfig::type)
+                .distinct()
+                .toList();
     }
 
     /** A copy with the inline output replaced (e.g. resolved for the engine, or migrated). */
     public Policy withOutput(OutputSpec resolved) {
-        return new Policy(
-                id, name, owner, enabled, trigger, sourceIds, steps, resolved, outputIds, teamId);
+        return new Policy(id, name, owner, enabled, inputs, steps, resolved, outputIds, teamId);
     }
 
     /** A copy referencing the given saved output destinations. */
     public Policy withOutputIds(List<String> newOutputIds) {
-        return new Policy(
-                id, name, owner, enabled, trigger, sourceIds, steps, output, newOutputIds, teamId);
+        return new Policy(id, name, owner, enabled, inputs, steps, output, newOutputIds, teamId);
     }
 
     /**

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

@@ -0,0 +1,31 @@
+package stirling.software.proprietary.policy.model;
+
+import java.util.List;
+
+/**
+ * A policy paired with one of its {@link PipelineInput}s: the unit a background trigger fires. A
+ * policy with two triggered inputs yields two bindings, so each fires independently on its own
+ * trigger and pulls only its own source.
+ */
+public record PolicyBinding(Policy policy, PipelineInput input) {
+
+    /**
+     * The bindings across these policies whose input carries a trigger of the given type. Shared by
+     * the {@code PolicyStore} implementations so every backend derives a trigger's bindings the
+     * same way. Callers pass the policies a background trigger should consider (i.e. the enabled
+     * ones).
+     */
+    public static List<PolicyBinding> matching(List<Policy> policies, String triggerType) {
+        return policies.stream()
+                .flatMap(
+                        policy ->
+                                policy.inputs().stream()
+                                        .filter(
+                                                input ->
+                                                        input.trigger() != null
+                                                                && triggerType.equals(
+                                                                        input.trigger().type()))
+                                        .map(input -> new PolicyBinding(policy, input)))
+                .toList();
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/overview/PolicyOverviewService.java

@@ -14,7 +14,6 @@ import stirling.software.proprietary.policy.config.PolicyAccessGuard;
 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.source.Source;
 import stirling.software.proprietary.policy.source.SourceAccessGuard;
 import stirling.software.proprietary.policy.source.SourceStore;
@@ -73,7 +72,7 @@ public class PolicyOverviewService {
                 policy.name(),
                 policy.enabled(),
                 policy.enabled() ? "active" : "paused",
-                triggerSummary(policy.trigger()),
+                triggerSummary(policy),
                 sources,
                 steps,
                 outputSummary(policy, sourceNames),
@@ -95,9 +94,13 @@ public class PolicyOverviewService {
         return outputSummary(policy.output());
     }
 
-    /** A null trigger is a manual-only policy; otherwise the trigger's type keys the summary. */
-    private static String triggerSummary(TriggerConfig trigger) {
-        return trigger == null ? "manual" : trigger.type();
+    /**
+     * Summarise a policy's triggers for the overview row: "manual" when no input is triggered,
+     * otherwise the distinct trigger types across its inputs (e.g. "folder-watch, schedule").
+     */
+    private static String triggerSummary(Policy policy) {
+        List<String> types = policy.triggerTypes();
+        return types.isEmpty() ? "manual" : String.join(", ", types);
     }
 
     private static String outputSummary(OutputSpec output) {

app/proprietary/src/main/java/stirling/software/proprietary/policy/seed/DefaultClassificationPolicySeeder.java

@@ -87,7 +87,6 @@ public class DefaultClassificationPolicySeeder {
                 POLICY_NAME,
                 "system",
                 true,
-                null,
                 List.of(),
                 List.of(new PipelineStep(CLASSIFY_ENDPOINT, Map.of())),
                 new OutputSpec("inline", options),

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

@@ -9,6 +9,7 @@ import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 
 /**
  * In-memory {@link PolicyStore} for tests and any future no-database mode. {@link JpaPolicyStore}
@@ -32,8 +33,7 @@ public class InProcessPolicyStore implements PolicyStore {
                         policy.name(),
                         policy.owner(),
                         policy.enabled(),
-                        policy.trigger(),
-                        policy.sourceIds(),
+                        policy.inputs(),
                         policy.steps(),
                         policy.output(),
                         policy.outputIds(),
@@ -77,12 +77,9 @@ public class InProcessPolicyStore implements PolicyStore {
     }
 
     @Override
-    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();
+    public List<PolicyBinding> findBindingsByTriggerType(String triggerType) {
+        List<Policy> enabled = policies.values().stream().filter(Policy::enabled).toList();
+        return PolicyBinding.matching(enabled, triggerType);
     }
 
     @Override

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

@@ -12,8 +12,12 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 
+import tools.jackson.databind.JsonNode;
 import tools.jackson.databind.ObjectMapper;
+import tools.jackson.databind.node.ArrayNode;
+import tools.jackson.databind.node.ObjectNode;
 
 /**
  * Durable {@link PolicyStore} backed by JPA; the runtime store. Policies are persisted as JSON via
@@ -40,8 +44,7 @@ public class JpaPolicyStore implements PolicyStore {
                         policy.name(),
                         policy.owner(),
                         policy.enabled(),
-                        policy.trigger(),
-                        policy.sourceIds(),
+                        policy.inputs(),
                         policy.steps(),
                         policy.output(),
                         policy.outputIds(),
@@ -52,7 +55,6 @@ public class JpaPolicyStore implements PolicyStore {
         entity.setName(stored.name());
         entity.setOwner(stored.owner());
         entity.setEnabled(stored.enabled());
-        entity.setTriggerType(stored.trigger() == null ? null : stored.trigger().type());
         entity.setTeamId(stored.teamId());
         // Preserve an existing policy's run-order position; append a new one to the end of its
         // team's queue (max + 1), so setting up a policy adds it last by default.
@@ -119,11 +121,13 @@ public class JpaPolicyStore implements PolicyStore {
     }
 
     @Override
-    public List<Policy> findByTriggerType(String triggerType) {
-        return repository.findByTriggerTypeAndEnabledTrue(triggerType).stream()
+    public List<PolicyBinding> findBindingsByTriggerType(String triggerType) {
+        List<Policy> enabled =
+                repository.findByEnabledTrue().stream()
                         .map(this::toPolicy)
                         .flatMap(Optional::stream)
                         .toList();
+        return PolicyBinding.matching(enabled, triggerType);
     }
 
     @Override
@@ -139,7 +143,8 @@ public class JpaPolicyStore implements PolicyStore {
     // One unreadable row must never abort a bulk read or crash startup.
     private Optional<Policy> toPolicy(PolicyEntity entity) {
         try {
-            return Optional.of(objectMapper.readValue(entity.getPolicyJson(), Policy.class));
+            JsonNode node = upgradeLegacyShape(objectMapper.readTree(entity.getPolicyJson()));
+            return Optional.of(objectMapper.treeToValue(node, Policy.class));
         } catch (Exception e) {
             log.error(
                     "Skipping unreadable policy id={} name={}: stored JSON could not be parsed"
@@ -150,4 +155,35 @@ public class JpaPolicyStore implements PolicyStore {
             return Optional.empty();
         }
     }
+
+    /**
+     * Migrate a policy JSON blob written before triggers moved onto inputs. The old shape carried a
+     * single policy-level {@code trigger} and a {@code sourceIds} list; pair each source with that
+     * trigger so an upgraded policy keeps firing. A trigger incompatible with a source
+     * (folder-watch on an S3 source) is simply inert at run time, matching the old behaviour where
+     * such a source was never watched. New-shape blobs (already carrying {@code inputs}) are
+     * returned untouched.
+     */
+    private JsonNode upgradeLegacyShape(JsonNode root) {
+        if (!(root instanceof ObjectNode obj) || obj.has("inputs")) {
+            return root;
+        }
+        JsonNode trigger = obj.get("trigger");
+        JsonNode sourceIds = obj.get("sourceIds");
+        ArrayNode inputs = objectMapper.createArrayNode();
+        if (sourceIds != null && sourceIds.isArray()) {
+            for (JsonNode sourceId : sourceIds) {
+                ObjectNode input = objectMapper.createObjectNode();
+                input.set("sourceId", sourceId);
+                if (trigger != null && !trigger.isNull()) {
+                    input.set("trigger", trigger);
+                }
+                inputs.add(input);
+            }
+        }
+        obj.set("inputs", inputs);
+        obj.remove("trigger");
+        obj.remove("sourceIds");
+        return obj;
+    }
 }

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

@@ -17,10 +17,10 @@ import stirling.software.proprietary.integration.crypto.LegacyDecryptStringConve
 /**
  * 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, and {@code teamId} so the caller's team can be loaded without scanning every
- * team's rows. {@code owner} and {@code teamId} are plain values, not foreign keys, to stay
- * decoupled from the security entities.
+ * for querying, notably {@code enabled} so background triggers can scan the active policies, and
+ * {@code teamId} so the caller's team can be loaded without scanning every team's rows. {@code
+ * owner} and {@code teamId} are plain values, not foreign keys, to stay decoupled from the security
+ * entities.
  */
 @Entity
 @Table(name = "policies")
@@ -44,9 +44,6 @@ public class PolicyEntity implements Serializable {
     @Column(name = "enabled")
     private boolean enabled;
 
-    @Column(name = "trigger_type")
-    private String triggerType;
-
     @Column(name = "team_id")
     private Long teamId;
 

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

@@ -13,8 +13,12 @@ import jakarta.persistence.LockModeType;
 @Repository
 public interface PolicyRepository extends JpaRepository<PolicyEntity, String> {
 
-    /** Enabled policies of a given trigger type, for background triggers to activate. */
-    List<PolicyEntity> findByTriggerTypeAndEnabledTrue(String triggerType);
+    /**
+     * Enabled policies, for background triggers to scan for inputs of their trigger type. Which
+     * inputs (and their trigger types) a policy carries lives in the JSON blob, so the type filter
+     * is applied after parsing rather than in SQL.
+     */
+    List<PolicyEntity> findByEnabledTrue();
 
     /**
      * Policies belonging to a team, in run order (ascending {@code sortOrder}; a null order sorts

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

@@ -4,6 +4,7 @@ import java.util.List;
 import java.util.Optional;
 
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 
 /** Stores {@link Policy} definitions. */
 public interface PolicyStore {
@@ -18,8 +19,11 @@ public interface PolicyStore {
     /** Policies owned by the given team, loaded scoped rather than fetched globally. */
     List<Policy> findByTeam(Long teamId);
 
-    /** Enabled policies with the given trigger type, for background triggers. */
-    List<Policy> findByTriggerType(String triggerType);
+    /**
+     * Enabled inputs with the given trigger type, as {@code (policy, input)} bindings, so a
+     * background trigger fires each input independently and pulls only its own source.
+     */
+    List<PolicyBinding> findBindingsByTriggerType(String triggerType);
 
     /**
      * Set the team's run order from {@code orderedIds} (position → sortOrder). Only policies that

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

@@ -31,7 +31,9 @@ import stirling.software.proprietary.policy.engine.PolicyRunner;
 import stirling.software.proprietary.policy.engine.SweepKind;
 import stirling.software.proprietary.policy.input.InputSource;
 import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 import stirling.software.proprietary.policy.source.Source;
 import stirling.software.proprietary.policy.source.SourceStore;
 import stirling.software.proprietary.policy.store.PolicyStore;
@@ -85,10 +87,10 @@ public class FolderWatchTrigger implements PolicyTrigger {
     }
 
     @Override
-    public void validate(Policy policy) {
-        if (watchDirsOf(policy).isEmpty()) {
+    public void validate(Policy policy, PipelineInput input) {
+        if (watchDirsOf(input).isEmpty()) {
             throw new IllegalArgumentException(
-                    "folder-watch trigger requires at least one watchable (folder) input source");
+                    "folder-watch trigger requires a watchable (folder) input source");
         }
     }
 
@@ -185,24 +187,30 @@ public class FolderWatchTrigger implements PolicyTrigger {
         return changed;
     }
 
-    /** Run every folder-watch policy that draws from one of the changed directories. */
+    /** Fire every folder-watch input that draws from one of the changed directories. */
     void runForChangedDirs(Set<Path> changedDirs) {
         if (changedDirs.isEmpty()) {
             return;
         }
-        for (Policy policy : policyStore.findByTriggerType(TYPE)) {
+        for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) {
             List<Path> dirs;
             try {
-                dirs = watchDirsOf(policy);
+                dirs = watchDirsOf(binding.input());
             } catch (RuntimeException e) {
                 log.warn(
-                        "Folder-watch policy {} is misconfigured: {}", policy.id(), e.getMessage());
+                        "Folder-watch input {}/{} is misconfigured: {}",
+                        binding.policy().id(),
+                        binding.input().sourceId(),
+                        e.getMessage());
                 continue;
             }
             if (dirs.stream().anyMatch(changedDirs::contains)) {
-                log.debug("Folder-watch policy {} ({}) saw activity", policy.id(), policy.name());
+                log.debug(
+                        "Folder-watch input {}/{} saw activity",
+                        binding.policy().id(),
+                        binding.input().sourceId());
                 // Light: the periodic reconcile does the full sweep.
-                policyRunner.run(policy, SweepKind.LIGHT);
+                policyRunner.runInput(binding.policy(), binding.input(), SweepKind.LIGHT);
             }
         }
     }
@@ -216,15 +224,16 @@ public class FolderWatchTrigger implements PolicyTrigger {
         }
     }
 
-    /** Reconcile safety net: run every folder-watch policy regardless of watch events. */
+    /** Reconcile safety net: run every folder-watch input regardless of watch events. */
     void runAll() {
-        for (Policy policy : policyStore.findByTriggerType(TYPE)) {
+        for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) {
             try {
-                policyRunner.run(policy);
+                policyRunner.runInput(binding.policy(), binding.input(), SweepKind.FULL);
             } catch (RuntimeException e) {
                 log.warn(
-                        "Folder-watch reconcile run failed for policy {}: {}",
-                        policy.id(),
+                        "Folder-watch reconcile run failed for input {}/{}: {}",
+                        binding.policy().id(),
+                        binding.input().sourceId(),
                         e.getMessage());
             }
         }
@@ -269,42 +278,44 @@ public class FolderWatchTrigger implements PolicyTrigger {
         return Set.copyOf(keysByDir.keySet());
     }
 
-    /** Every existing directory any current folder-watch policy wants watched. */
+    /** Every existing directory any current folder-watch input wants watched. */
     private Set<Path> desiredDirs() {
         Set<Path> dirs = new HashSet<>();
-        for (Policy policy : policyStore.findByTriggerType(TYPE)) {
+        for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) {
             try {
-                for (Path dir : watchDirsOf(policy)) {
+                for (Path dir : watchDirsOf(binding.input())) {
                     if (Files.isDirectory(dir)) {
                         dirs.add(dir);
                     }
                 }
             } catch (RuntimeException e) {
                 log.warn(
-                        "Folder-watch policy {} is misconfigured: {}", policy.id(), e.getMessage());
+                        "Folder-watch input {}/{} is misconfigured: {}",
+                        binding.policy().id(),
+                        binding.input().sourceId(),
+                        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) {
+    // the path was configured. Empty for a non-folder or missing source (that input is never
+    // watched), so a folder-watch trigger paired with an S3 input is simply inert.
+    private List<Path> watchDirsOf(PipelineInput input) {
         List<Path> dirs = new ArrayList<>();
-        for (String sourceId : policy.sourceIds()) {
-            Source source = sourceStore.get(sourceId).orElse(null);
+        Source source = sourceStore.get(input.sourceId()).orElse(null);
         if (source == null) {
-                continue;
+            return dirs;
         }
         InputSpec spec = source.toInputSpec();
         InputSource inputSource = sourceFor(spec);
         if (inputSource == null) {
-                continue;
+            return dirs;
         }
         for (Path dir : inputSource.watchTargets(spec)) {
             dirs.add(dir.toAbsolutePath().normalize());
         }
-        }
         return dirs;
     }
 

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

@@ -2,11 +2,13 @@ package stirling.software.proprietary.policy.trigger;
 
 import java.util.Set;
 
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.Policy;
 
 /**
- * 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.
+ * Decides <em>when</em> a policy input runs. On firing it hands the binding to {@code
+ * PolicyRunner}, which pulls only that input's source; it never resolves sources itself. New
+ * trigger kinds are just new beans of this type.
  */
 public interface PolicyTrigger {
 
@@ -32,10 +34,11 @@ public interface PolicyTrigger {
     }
 
     /**
-     * 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.
+     * Validate one input's use of this trigger at save time so misconfiguration fails fast, not at
+     * fire time. Receives the owning {@link Policy} and the specific {@link PipelineInput} so a
+     * trigger that depends on the input's source (folder-watch) can check it.
      */
-    default void validate(Policy policy) {}
+    default void validate(Policy policy, PipelineInput input) {}
 
     default void start() {}
 

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

@@ -17,14 +17,18 @@ import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.model.ApplicationProperties;
 import stirling.software.proprietary.policy.engine.PolicyRunner;
+import stirling.software.proprietary.policy.engine.SweepKind;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 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.
+ * Fires policy inputs on a {@link Schedule}: a fixed-interval sweep pulls each due "schedule"
+ * input, independently of the policy's other inputs.
  *
  * <p>Last-fire times are in memory, so this assumes a single node and resets on restart.
  */
@@ -40,17 +44,20 @@ public class ScheduleTrigger implements PolicyTrigger {
     private final ObjectMapper objectMapper;
     private final ApplicationProperties applicationProperties;
 
-    private final Map<String, Instant> lastFiredByPolicy = new ConcurrentHashMap<>();
+    private final Map<BindingKey, Instant> lastFiredByBinding = new ConcurrentHashMap<>();
     private volatile ScheduledExecutorService scheduler;
 
+    /** Identifies a schedule binding: one input (by source) of one policy. */
+    private record BindingKey(String policyId, String sourceId) {}
+
     @Override
     public String type() {
         return TYPE;
     }
 
     @Override
-    public void validate(Policy policy) {
-        ScheduleConfig.from(objectMapper, policy.trigger().options());
+    public void validate(Policy policy, PipelineInput input) {
+        ScheduleConfig.from(objectMapper, input.trigger().options());
     }
 
     @Override
@@ -83,19 +90,26 @@ public class ScheduleTrigger implements PolicyTrigger {
         }
     }
 
-    /** Fire every scheduled policy that is due as of {@code now}. Package-visible for testing. */
+    /** Fire every scheduled input that is due as of {@code now}. Package-visible for testing. */
     void sweep(Instant now) {
-        for (Policy policy : policyStore.findByTriggerType(TYPE)) {
+        for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) {
+            Policy policy = binding.policy();
+            PipelineInput input = binding.input();
             ScheduleConfig config;
             try {
-                config = ScheduleConfig.from(objectMapper, policy.trigger().options());
+                config = ScheduleConfig.from(objectMapper, input.trigger().options());
             } catch (IllegalArgumentException e) {
-                log.warn("Scheduled policy {} is misconfigured: {}", policy.id(), e.getMessage());
+                log.warn(
+                        "Scheduled input {}/{} is misconfigured: {}",
+                        policy.id(),
+                        input.sourceId(),
+                        e.getMessage());
                 continue;
             }
 
-            // Baseline a newly-seen policy to now so it does not fire immediately.
-            Instant last = lastFiredByPolicy.computeIfAbsent(policy.id(), id -> now);
+            // Baseline a newly-seen binding to now so it does not fire immediately.
+            BindingKey key = new BindingKey(policy.id(), input.sourceId());
+            Instant last = lastFiredByBinding.computeIfAbsent(key, id -> now);
             ZonedDateTime next = config.schedule().nextAfter(last.atZone(config.zone()));
             if (next.toInstant().isAfter(now)) {
                 continue;
@@ -105,9 +119,13 @@ public class ScheduleTrigger implements PolicyTrigger {
                 next = later;
                 later = config.schedule().nextAfter(later);
             }
-            lastFiredByPolicy.put(policy.id(), next.toInstant());
-            log.info("Scheduled policy {} ({}) is due", policy.id(), policy.name());
-            policyRunner.run(policy);
+            lastFiredByBinding.put(key, next.toInstant());
+            log.info(
+                    "Scheduled input {}/{} ({}) is due",
+                    policy.id(),
+                    input.sourceId(),
+                    policy.name());
+            policyRunner.runInput(policy, input, SweepKind.FULL);
         }
     }
 

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

@@ -13,7 +13,9 @@ import lombok.extern.slf4j.Slf4j;
 import stirling.software.common.model.ApplicationProperties;
 import stirling.software.proprietary.policy.engine.PolicyRunner;
 import stirling.software.proprietary.policy.engine.SweepKind;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 import stirling.software.proprietary.policy.source.Source;
 import stirling.software.proprietary.policy.source.SourceStore;
 import stirling.software.proprietary.policy.store.PolicyStore;
@@ -50,15 +52,14 @@ public class WebhookTrigger implements PolicyTrigger {
     }
 
     @Override
-    public void validate(Policy policy) {
-        boolean hasWebhookSource =
-                policy.sourceIds().stream()
-                        .map(sourceStore::get)
-                        .flatMap(java.util.Optional::stream)
-                        .anyMatch(source -> WEBHOOK_SOURCE_TYPE.equals(source.type()));
-        if (!hasWebhookSource) {
-            throw new IllegalArgumentException(
-                    "webhook trigger requires at least one webhook input source");
+    public void validate(Policy policy, PipelineInput input) {
+        boolean isWebhookSource =
+                sourceStore
+                        .get(input.sourceId())
+                        .filter(source -> WEBHOOK_SOURCE_TYPE.equals(source.type()))
+                        .isPresent();
+        if (!isWebhookSource) {
+            throw new IllegalArgumentException("webhook trigger requires a webhook input source");
         }
     }
 
@@ -83,29 +84,38 @@ public class WebhookTrigger implements PolicyTrigger {
         }
     }
 
+    /** Fire every webhook input fed by this webhook, pulling only that input's source. */
     public void fireForWebhook(String webhookId) {
-        for (Policy policy : policyStore.findByTriggerType(TYPE)) {
-            if (!referencesWebhook(policy, webhookId)) {
+        for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) {
+            if (!referencesWebhook(binding.input(), webhookId)) {
                 continue;
             }
             try {
-                log.debug("Webhook policy {} ({}) saw a delivery", policy.id(), policy.name());
-                policyRunner.run(policy, SweepKind.LIGHT);
+                log.debug(
+                        "Webhook input {}/{} saw a delivery",
+                        binding.policy().id(),
+                        binding.input().sourceId());
+                policyRunner.runInput(binding.policy(), binding.input(), SweepKind.LIGHT);
             } catch (RuntimeException e) {
-                log.warn("Webhook run failed for policy {}: {}", policy.id(), e.getMessage());
+                log.warn(
+                        "Webhook run failed for input {}/{}: {}",
+                        binding.policy().id(),
+                        binding.input().sourceId(),
+                        e.getMessage());
             }
         }
     }
 
     private void safeReconcile() {
         try {
-            for (Policy policy : policyStore.findByTriggerType(TYPE)) {
+            for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) {
                 try {
-                    policyRunner.run(policy);
+                    policyRunner.runInput(binding.policy(), binding.input(), SweepKind.FULL);
                 } catch (RuntimeException e) {
                     log.warn(
-                            "Webhook reconcile run failed for policy {}: {}",
-                            policy.id(),
+                            "Webhook reconcile run failed for input {}/{}: {}",
+                            binding.policy().id(),
+                            binding.input().sourceId(),
                             e.getMessage());
                 }
             }
@@ -114,17 +124,13 @@ public class WebhookTrigger implements PolicyTrigger {
         }
     }
 
-    private boolean referencesWebhook(Policy policy, String webhookId) {
-        for (String sourceId : policy.sourceIds()) {
-            Source source = sourceStore.get(sourceId).orElse(null);
+    /** Whether this input draws from the webhook source the delivery arrived on. */
+    private boolean referencesWebhook(PipelineInput input, String webhookId) {
+        Source source = sourceStore.get(input.sourceId()).orElse(null);
         if (source == null || !WEBHOOK_SOURCE_TYPE.equals(source.type())) {
-                continue;
-            }
-            Object configured = source.options().get(WebhookConfig.WEBHOOK_ID_OPTION);
-            if (configured != null && configured.toString().equals(webhookId)) {
-                return true;
-            }
-        }
             return false;
         }
+        Object configured = source.options().get(WebhookConfig.WEBHOOK_ID_OPTION);
+        return configured != null && configured.toString().equals(webhookId);
+    }
 }

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

@@ -17,6 +17,7 @@ import stirling.software.common.configuration.RuntimePathConfig;
 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.PipelineInput;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.source.InProcessSourceStore;
 import stirling.software.proprietary.policy.source.Source;
@@ -189,6 +190,13 @@ class FolderAccessGuardTest {
                                                                 null))
                                                 .id())
                         .toList();
-        return new Policy("p1", "p", "owner", true, null, sourceIds, List.of(), output);
+        return new Policy(
+                "p1",
+                "p",
+                "owner",
+                true,
+                sourceIds.stream().map(PipelineInput::manual).toList(),
+                List.of(),
+                output);
     }
 }

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

@@ -90,6 +90,6 @@ class PolicyAccessGuardTest {
 
     private static Policy inTeam(Long teamId) {
         return new Policy(
-                null, "p", "owner", true, null, List.of(), List.of(), OutputSpec.inline(), teamId);
+                null, "p", "owner", true, List.of(), List.of(), OutputSpec.inline(), teamId);
     }
 }

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

@@ -139,7 +139,7 @@ class PolicyControllerTest {
     }
 
     private static Policy policy(String id, Long teamId) {
-        return new Policy(id, "name", "owner", true, null, List.of(), List.of(), null, teamId);
+        return new Policy(id, "name", "owner", true, List.of(), List.of(), null, teamId);
     }
 
     private static Policy s3OutputPolicy(String id, String secret) {
@@ -150,7 +150,7 @@ class PolicyControllerTest {
                                 "bucket", "outbox",
                                 "accessKeyId", "AKIAEXAMPLE",
                                 "secretAccessKey", secret));
-        return new Policy(id, "name", "owner", true, null, List.of(), List.of(), output, 1L);
+        return new Policy(id, "name", "owner", true, List.of(), List.of(), output, 1L);
     }
 
     private static PolicyRunHandle handle(String runId) {
@@ -401,8 +401,7 @@ class PolicyControllerTest {
         void updatePreservesOwnership() {
             applicationProperties.getSecurity().setEnableLogin(false);
             Policy existing =
-                    new Policy(
-                            "p2", "name", "origOwner", true, null, List.of(), List.of(), null, 3L);
+                    new Policy("p2", "name", "origOwner", true, List.of(), List.of(), null, 3L);
             when(policyStore.get("p2")).thenReturn(Optional.of(existing));
             when(policyAccessGuard.canAccess(existing)).thenReturn(true);
             when(policyStore.save(any())).thenAnswer(i -> i.getArgument(0));
@@ -410,8 +409,7 @@ class PolicyControllerTest {
             ResponseEntity<Policy> response =
                     controller.savePolicy(
                             new Policy(
-                                    "p2", "name", "forged", true, null, List.of(), List.of(), null,
-                                    77L));
+                                    "p2", "name", "forged", true, List.of(), List.of(), null, 77L));
 
             assertThat(response.getBody().owner()).isEqualTo("origOwner");
             assertThat(response.getBody().teamId()).isEqualTo(3L);

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

@@ -262,7 +262,7 @@ class PolicyEngineTest {
                         "rotate",
                         "owner",
                         true,
-                        null,
+                        List.of(),
                         List.of(new PipelineStep(ROTATE, Map.of())),
                         OutputSpec.inline());
 
@@ -306,7 +306,7 @@ class PolicyEngineTest {
                         "rotate",
                         "alice",
                         true,
-                        null,
+                        List.of(),
                         List.of(new PipelineStep(ROTATE, Map.of())),
                         OutputSpec.inline());
 

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

@@ -36,6 +36,7 @@ import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
 import stirling.software.proprietary.policy.ledger.ProcessedLedger;
 import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.model.PolicyInputs;
@@ -306,7 +307,6 @@ class PolicyRunnerTest {
                         "p",
                         "owner",
                         true,
-                        null,
                         List.of(),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline(),
@@ -338,8 +338,7 @@ class PolicyRunnerTest {
                 "p",
                 "owner",
                 true,
-                null,
-                sourceIds,
+                sourceIds.stream().map(PipelineInput::manual).toList(),
                 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

@@ -20,6 +20,7 @@ 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.PipelineInput;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.model.TriggerConfig;
 import stirling.software.proprietary.policy.output.PolicyOutputSink;
@@ -60,25 +61,27 @@ class PolicyValidatorTest {
 
         validator.validate(policy);
 
-        verify(trigger).validate(policy);
+        verify(trigger).validate(policy, policy.inputs().get(0));
         verify(inputSource).validate(InputSpec.folder("/in"));
         verify(outputSink).validate(policy.output());
     }
 
     @Test
-    void skipsTriggerValidationForAManualOnlyPolicy() {
+    void skipsTriggerValidationForAManualOnlyInput() {
         when(inputSource.supports(any())).thenReturn(true);
         when(outputSink.supports(any())).thenReturn(true);
 
         validator.validate(manualOnly());
 
-        verify(trigger, never()).validate(any());
+        verify(trigger, never()).validate(any(), any());
     }
 
     @Test
     void surfacesAnInvalidConfigFromAHandler() {
         when(trigger.type()).thenReturn("schedule");
-        doThrow(new IllegalArgumentException("invalid schedule")).when(trigger).validate(any());
+        doThrow(new IllegalArgumentException("invalid schedule"))
+                .when(trigger)
+                .validate(any(), any());
 
         IllegalArgumentException ex =
                 assertThrows(
@@ -120,14 +123,55 @@ class PolicyValidatorTest {
         assertTrue(ex.getMessage().contains("unknown trigger type"));
     }
 
+    // The one-input/one-output caps are a product decision, not a model limit: the lists stay so
+    // multiple can be supported later, but saving more than one of either is rejected today.
+
+    @Test
+    void rejectsMoreThanOneInput() {
+        Policy twoInputs =
+                new Policy(
+                        "p1",
+                        "p",
+                        "owner",
+                        true,
+                        List.of(
+                                PipelineInput.manual(folderSourceId()),
+                                PipelineInput.manual(folderSourceId())),
+                        List.of(),
+                        OutputSpec.inline());
+
+        IllegalArgumentException ex =
+                assertThrows(IllegalArgumentException.class, () -> validator.validate(twoInputs));
+        assertTrue(ex.getMessage().contains("at most one input"));
+    }
+
+    @Test
+    void rejectsMoreThanOneOutput() {
+        Policy twoOutputs = manualOnly().withOutputIds(List.of("out-a", "out-b"));
+
+        IllegalArgumentException ex =
+                assertThrows(IllegalArgumentException.class, () -> validator.validate(twoOutputs));
+        assertTrue(ex.getMessage().contains("at most one output"));
+    }
+
+    @Test
+    void allowsZeroInputsAndZeroOutputs() {
+        when(outputSink.supports(any())).thenReturn(true);
+        Policy bare =
+                new Policy("p1", "p", "owner", true, List.of(), List.of(), OutputSpec.inline());
+
+        validator.validate(bare);
+    }
+
     private Policy policy(String triggerType) {
         return new Policy(
                 "p1",
                 "p",
                 "owner",
                 true,
-                new TriggerConfig(triggerType, Map.of()),
-                List.of(folderSourceId()),
+                List.of(
+                        new PipelineInput(
+                                folderSourceId(), new TriggerConfig(triggerType, Map.of()))),
                 List.of(),
                 OutputSpec.inline());
     }
@@ -138,8 +182,7 @@ class PolicyValidatorTest {
                 "p",
                 "owner",
                 true,
-                null,
-                List.of(folderSourceId()),
+                List.of(PipelineInput.manual(folderSourceId())),
                 List.of(),
                 OutputSpec.inline());
     }

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

@@ -54,7 +54,6 @@ class PolicyInlineOutputMigrationTest {
                                 "Editor run",
                                 "owner",
                                 true,
-                                null,
                                 List.of(),
                                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                                 OutputSpec.inline()));
@@ -128,7 +127,6 @@ class PolicyInlineOutputMigrationTest {
                 name,
                 "owner",
                 true,
-                null,
                 List.of(),
                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                 OutputSpec.folder(directory));
@@ -140,7 +138,6 @@ class PolicyInlineOutputMigrationTest {
                 name,
                 "owner",
                 true,
-                null,
                 List.of(),
                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                 OutputSpec.folder(directory),

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

@@ -65,7 +65,6 @@ class PolicyOutputResolverTest {
                 "Pipeline",
                 "owner",
                 true,
-                null,
                 List.of(),
                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                 OutputSpec.inline());

app/proprietary/src/test/java/stirling/software/proprietary/policy/overview/PolicyOverviewServiceTest.java

@@ -16,6 +16,7 @@ import stirling.software.common.service.UserServiceInterface;
 import stirling.software.proprietary.policy.config.PolicyAccessGuard;
 import stirling.software.proprietary.policy.config.PolicyManagementAuthority;
 import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.model.TriggerConfig;
@@ -57,8 +58,9 @@ class PolicyOverviewServiceTest {
                         "Redaction",
                         "owner",
                         true,
-                        new TriggerConfig("schedule", Map.of()),
-                        List.of(claims.id()),
+                        List.of(
+                                new PipelineInput(
+                                        claims.id(), new TriggerConfig("schedule", Map.of()))),
                         List.of(new PipelineStep("/api/v1/security/auto-redact", Map.of())),
                         OutputSpec.inline()));
         policyStore.save(
@@ -67,7 +69,6 @@ class PolicyOverviewServiceTest {
                         "Archive (paused)",
                         "owner",
                         false,
-                        null,
                         List.of(),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline()));
@@ -102,8 +103,7 @@ class PolicyOverviewServiceTest {
                         "Orphan",
                         "owner",
                         true,
-                        null,
-                        List.of("src-missing"),
+                        List.of(PipelineInput.manual("src-missing")),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline()));
 
@@ -170,8 +170,7 @@ class PolicyOverviewServiceTest {
                         name,
                         "owner",
                         true,
-                        null,
-                        List.of(sourceIds),
+                        List.of(sourceIds).stream().map(PipelineInput::manual).toList(),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline(),
                         teamId));

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

@@ -25,6 +25,7 @@ import stirling.software.proprietary.integration.repository.IntegrationConfigRep
 import stirling.software.proprietary.model.Team;
 import stirling.software.proprietary.policy.migration.InProcessCompletedMigrations;
 import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.source.InProcessSourceStore;
@@ -97,8 +98,7 @@ class EmbeddedS3CredentialMigrationTest {
                                 "Rotate",
                                 "alice",
                                 true,
-                                null,
-                                List.of(source.id()),
+                                List.of(PipelineInput.manual(source.id())),
                                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                                 new OutputSpec(
                                         "s3",

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

@@ -39,7 +39,6 @@ class PolicyS3ConnectionUsageCheckTest {
                         "Rotate",
                         "alice",
                         true,
-                        null,
                         List.of(),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         new OutputSpec("s3", Map.of("connectionId", "5")),

app/proprietary/src/test/java/stirling/software/proprietary/policy/seed/DefaultClassificationPolicySeederTest.java

@@ -41,7 +41,6 @@ class DefaultClassificationPolicySeederTest {
                 "Classification Policy",
                 "system",
                 true,
-                null,
                 List.of(),
                 List.of(),
                 new OutputSpec("inline", Map.of("categoryId", "classification")),

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

@@ -30,6 +30,7 @@ import stirling.software.proprietary.policy.config.PolicyManagementAuthority;
 import stirling.software.proprietary.policy.input.InputSource;
 import stirling.software.proprietary.policy.input.WebhookInputSource;
 import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.store.InProcessPolicyStore;
@@ -274,8 +275,7 @@ class SourceControllerTest {
                 name,
                 "owner",
                 true,
-                null,
-                List.of(sourceId),
+                List.of(PipelineInput.manual(sourceId)),
                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                 OutputSpec.inline());
     }

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

@@ -16,6 +16,7 @@ import stirling.software.common.service.UserServiceInterface;
 import stirling.software.proprietary.policy.config.PolicyAccessGuard;
 import stirling.software.proprietary.policy.config.PolicyManagementAuthority;
 import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
 import stirling.software.proprietary.policy.store.InProcessPolicyStore;
@@ -216,8 +217,7 @@ class SourceOverviewServiceTest {
                         name,
                         "owner",
                         true,
-                        null,
-                        List.of(sourceIds),
+                        List.of(sourceIds).stream().map(PipelineInput::manual).toList(),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline()));
     }
@@ -232,7 +232,6 @@ class SourceOverviewServiceTest {
                         name,
                         "owner",
                         true,
-                        null,
                         List.of(),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         new OutputSpec("inline", Map.of("sources", List.of("editor")))));
@@ -245,8 +244,7 @@ class SourceOverviewServiceTest {
                         name,
                         "owner",
                         true,
-                        null,
-                        List.of(sourceIds),
+                        List.of(sourceIds).stream().map(PipelineInput::manual).toList(),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline(),
                         teamId));

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

@@ -12,11 +12,13 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 import stirling.software.proprietary.policy.model.TriggerConfig;
 
-/** Tests for {@link InProcessPolicyStore}: id assignment, upsert, trigger-type lookup, delete. */
+/** Tests for {@link InProcessPolicyStore}: id assignment, upsert, binding lookup, delete. */
 class InProcessPolicyStoreTest {
 
     private PolicyStore store;
@@ -45,7 +47,7 @@ class InProcessPolicyStoreTest {
                         "after",
                         "owner",
                         true,
-                        null,
+                        List.of(),
                         List.of(),
                         OutputSpec.inline()));
 
@@ -54,16 +56,16 @@ class InProcessPolicyStoreTest {
     }
 
     @Test
-    void findByTriggerTypeReturnsOnlyEnabledMatches() {
+    void findBindingsByTriggerTypeReturnsOnlyEnabledMatches() {
         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> scheduled = store.findByTriggerType("schedule");
+        List<PolicyBinding> scheduled = store.findBindingsByTriggerType("schedule");
 
         assertEquals(1, scheduled.size());
-        assertEquals("nightly", scheduled.get(0).name());
+        assertEquals("nightly", scheduled.get(0).policy().name());
     }
 
     @Test
@@ -76,14 +78,16 @@ class InProcessPolicyStoreTest {
     }
 
     private static Policy policy(String id, String name, String triggerType, boolean enabled) {
-        TriggerConfig trigger =
-                triggerType == null ? null : new TriggerConfig(triggerType, Map.of());
+        PipelineInput input =
+                triggerType == null
+                        ? PipelineInput.manual("src")
+                        : new PipelineInput("src", new TriggerConfig(triggerType, Map.of()));
         return new Policy(
                 id,
                 name,
                 "owner",
                 enabled,
-                trigger,
+                List.of(input),
                 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

@@ -19,8 +19,10 @@ import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
 import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 import stirling.software.proprietary.policy.model.TriggerConfig;
 
 import tools.jackson.databind.ObjectMapper;
@@ -53,8 +55,9 @@ class JpaPolicyStoreTest {
                                 "compress incoming",
                                 "alice",
                                 true,
-                                new TriggerConfig("schedule", Map.of()),
-                                List.of("src-in"),
+                                List.of(
+                                        new PipelineInput(
+                                                "src-in", new TriggerConfig("schedule", Map.of()))),
                                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                                 OutputSpec.inline()));
 
@@ -63,7 +66,6 @@ class JpaPolicyStoreTest {
         verify(repository).save(captor.capture());
         PolicyEntity entity = captor.getValue();
         assertEquals(saved.id(), entity.getId());
-        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));
@@ -77,7 +79,7 @@ class JpaPolicyStoreTest {
                         "rotate",
                         "alice",
                         true,
-                        null, // manual-only: no automatic trigger
+                        List.of(), // no inputs: run on demand only
                         List.of(
                                 new PipelineStep(
                                         "/api/v1/general/rotate-pdf", Map.of("angle", 90))),
@@ -87,6 +89,30 @@ class JpaPolicyStoreTest {
         assertEquals(policy, store.get("p1").orElseThrow());
     }
 
+    @Test
+    void getUpgradesLegacyTriggerAndSourceIdsToPerInputTriggers() {
+        // A blob written before triggers moved onto inputs: one policy-level trigger + sourceIds.
+        String legacyJson =
+                "{\"id\":\"p1\",\"name\":\"legacy\",\"owner\":\"alice\",\"enabled\":true,"
+                        + "\"trigger\":{\"type\":\"schedule\",\"options\":{}},"
+                        + "\"sourceIds\":[\"s1\",\"s2\"],\"steps\":[],"
+                        + "\"output\":{\"type\":\"inline\",\"options\":{}}}";
+        PolicyEntity entity = new PolicyEntity();
+        entity.setId("p1");
+        entity.setName("legacy");
+        entity.setEnabled(true);
+        entity.setPolicyJson(legacyJson);
+        when(repository.findById("p1")).thenReturn(Optional.of(entity));
+
+        Policy upgraded = store.get("p1").orElseThrow();
+
+        assertEquals(
+                List.of(
+                        new PipelineInput("s1", new TriggerConfig("schedule", Map.of())),
+                        new PipelineInput("s2", new TriggerConfig("schedule", Map.of()))),
+                upgraded.inputs());
+    }
+
     @Test
     void saveDenormalizesTeamIdForScopedQueries() {
         store.save(
@@ -95,7 +121,6 @@ class JpaPolicyStoreTest {
                         "scoped",
                         "alice",
                         true,
-                        null,
                         List.of(),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline(),
@@ -114,7 +139,6 @@ class JpaPolicyStoreTest {
                         "ours",
                         "alice",
                         true,
-                        null,
                         List.of(),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline(),
@@ -128,23 +152,28 @@ class JpaPolicyStoreTest {
     }
 
     @Test
-    void findByTriggerTypeUsesTheEnabledQuery() {
+    void findBindingsByTriggerTypeScansEnabledPoliciesForMatchingInputs() {
         Policy policy =
                 new Policy(
                         "p1",
                         "watch",
                         "alice",
                         true,
-                        new TriggerConfig("schedule", Map.of()),
+                        List.of(
+                                new PipelineInput(
+                                        "src-in", new TriggerConfig("schedule", Map.of())),
+                                PipelineInput.manual("src-manual")),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline());
-        when(repository.findByTriggerTypeAndEnabledTrue("schedule"))
-                .thenReturn(List.of(entityFor(policy)));
+        when(repository.findByEnabledTrue()).thenReturn(List.of(entityFor(policy)));
 
-        List<Policy> scheduled = store.findByTriggerType("schedule");
+        List<PolicyBinding> scheduled = store.findBindingsByTriggerType("schedule");
 
+        // Only the scheduled input yields a binding; the manual input on the same policy does not.
         assertEquals(1, scheduled.size());
-        assertEquals("p1", scheduled.get(0).id());
+        assertEquals("p1", scheduled.get(0).policy().id());
+        assertEquals("src-in", scheduled.get(0).input().sourceId());
+        assertEquals("schedule", scheduled.get(0).input().trigger().type());
     }
 
     @Test
@@ -163,7 +192,6 @@ class JpaPolicyStoreTest {
         entity.setName(policy.name());
         entity.setOwner(policy.owner());
         entity.setEnabled(policy.enabled());
-        entity.setTriggerType(policy.trigger() == null ? null : policy.trigger().type());
         entity.setTeamId(policy.teamId());
         entity.setPolicyJson(objectMapper.writeValueAsString(policy));
         return entity;

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

@@ -14,6 +14,7 @@ import java.nio.file.FileSystems;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.WatchService;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -31,8 +32,10 @@ import stirling.software.proprietary.policy.engine.SweepKind;
 import stirling.software.proprietary.policy.input.InputSource;
 import stirling.software.proprietary.policy.model.InputSpec;
 import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 import stirling.software.proprietary.policy.model.TriggerConfig;
 import stirling.software.proprietary.policy.source.InProcessSourceStore;
 import stirling.software.proprietary.policy.source.Source;
@@ -41,10 +44,10 @@ 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}.
+ * runForChangedDirs}/{@code runAll}, plus its per-input 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 {
@@ -83,39 +86,43 @@ class FolderWatchTriggerTest {
     }
 
     @Test
-    void validateRejectsPolicyWithNoWatchableSource() {
+    void validateRejectsInputWithNoWatchableSource() {
+        PolicyBinding binding =
+                bindings(folderWatch("p1", List.of(new InputSpec("folder", Map.of())))).get(0);
         assertThrows(
                 IllegalArgumentException.class,
-                () -> trigger.validate(folderWatch("p1", List.of())));
+                () -> trigger.validate(binding.policy(), binding.input()));
     }
 
     @Test
-    void validateAcceptsPolicyWithAFolderSource() {
-        trigger.validate(folderWatch("p1", List.of(InputSpec.folder("/in"))));
+    void validateAcceptsAFolderSource() {
+        PolicyBinding binding =
+                bindings(folderWatch("p1", List.of(InputSpec.folder("/in")))).get(0);
+        trigger.validate(binding.policy(), binding.input());
     }
 
     @Test
-    void runsOnlyPoliciesDrawingFromTheChangedDirectory() {
+    void runsOnlyInputsDrawingFromTheChangedDirectory() {
         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));
+        when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(bindings(a, b));
 
         trigger.runForChangedDirs(Set.of(normalized("/in/a")));
 
-        verify(policyRunner).run(a, SweepKind.LIGHT);
-        verify(policyRunner, never()).run(eq(b), any());
+        verify(policyRunner).runInput(a, a.inputs().get(0), SweepKind.LIGHT);
+        verify(policyRunner, never()).runInput(eq(b), any(), any());
     }
 
     @Test
-    void skipsAMisconfiguredPolicyButStillRunsTheOthers() {
+    void skipsAMisconfiguredInputButStillRunsTheOthers() {
         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));
+        when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(bindings(bad, good));
 
         trigger.runForChangedDirs(Set.of(normalized("/in/a")));
 
-        verify(policyRunner).run(good, SweepKind.LIGHT);
-        verify(policyRunner, never()).run(eq(bad), any());
+        verify(policyRunner).runInput(good, good.inputs().get(0), SweepKind.LIGHT);
+        verify(policyRunner, never()).runInput(eq(bad), any(), any());
     }
 
     @Test
@@ -126,15 +133,15 @@ class FolderWatchTriggerTest {
     }
 
     @Test
-    void reconcileRunsEveryFolderWatchPolicyAsASafetyNet() {
+    void reconcileRunsEveryFolderWatchInputAsASafetyNet() {
         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));
+        when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(bindings(a, b));
 
         trigger.runAll();
 
-        verify(policyRunner).run(a);
-        verify(policyRunner).run(b);
+        verify(policyRunner).runInput(a, a.inputs().get(0), SweepKind.FULL);
+        verify(policyRunner).runInput(b, b.inputs().get(0), SweepKind.FULL);
     }
 
     @Test
@@ -151,15 +158,16 @@ class FolderWatchTriggerTest {
         try {
             trigger.watchService = service;
 
-            when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(a, b, m));
+            when(policyStore.findBindingsByTriggerType("folder-watch"))
+                    .thenReturn(bindings(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));
+            // b's input is removed: its registration is cancelled, a remains.
+            when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(bindings(a));
             trigger.syncRegistrations();
             assertEquals(Set.of(normalized(dirA.toString())), trigger.watchedDirs());
         } finally {
@@ -175,15 +183,15 @@ class FolderWatchTriggerTest {
         WatchService service = FileSystems.getDefault().newWatchService();
         try {
             trigger.watchService = service;
-            when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(p));
+            when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(bindings(p));
 
-            // The mutation hook registers the new policy's directory without waiting for a
+            // The mutation hook registers the new input's directory without waiting for a
             // reconcile.
             trigger.onPoliciesChanged();
             assertEquals(Set.of(normalized(dir.toString())), trigger.watchedDirs());
 
-            // Once the policy is gone, the same hook cancels its registration.
-            when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of());
+            // Once the input is gone, the same hook cancels its registration.
+            when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(List.of());
             trigger.onPoliciesChanged();
             assertEquals(Set.of(), trigger.watchedDirs());
         } finally {
@@ -195,12 +203,21 @@ class FolderWatchTriggerTest {
         return Path.of(dir).toAbsolutePath().normalize();
     }
 
+    /** Every (policy, input) binding across the given policies, as the store would return them. */
+    private static List<PolicyBinding> bindings(Policy... policies) {
+        return Arrays.stream(policies)
+                .flatMap(
+                        policy -> policy.inputs().stream().map(in -> new PolicyBinding(policy, in)))
+                .toList();
+    }
+
     /** Persists each spec as a source and returns a folder-watch policy referencing them by id. */
     private Policy folderWatch(String id, List<InputSpec> sources) {
-        List<String> sourceIds =
+        List<PipelineInput> inputs =
                 sources.stream()
                         .map(
                                 spec ->
+                                        new PipelineInput(
                                                 sourceStore
                                                         .save(
                                                                 new Source(
@@ -211,15 +228,15 @@ class FolderWatchTriggerTest {
                                                                         true,
                                                                         "owner",
                                                                         null))
-                                                .id())
+                                                        .id(),
+                                                new TriggerConfig("folder-watch", Map.of())))
                         .toList();
         return new Policy(
                 id,
                 "watcher",
                 "owner",
                 true,
-                new TriggerConfig("folder-watch", Map.of()),
-                sourceIds,
+                inputs,
                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                 OutputSpec.inline());
     }

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

@@ -25,8 +25,10 @@ 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.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 import stirling.software.proprietary.policy.model.Schedule;
 import stirling.software.proprietary.policy.model.TriggerConfig;
 import stirling.software.proprietary.policy.store.PolicyStore;
@@ -35,9 +37,9 @@ 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.
+ * sweep(Instant)}. The trigger only decides when an input is due; pulling the source and starting
+ * runs is the {@link PolicyRunner}'s job, so these assert it delegates to the runner per binding.
+ * Schedules default to UTC, so explicit UTC instants make these deterministic.
  */
 @ExtendWith(MockitoExtension.class)
 class ScheduleTriggerTest {
@@ -59,102 +61,105 @@ class ScheduleTriggerTest {
 
     @Test
     void firesOncePerScheduleWhenItComesDue() {
-        Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES));
-        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+        PolicyBinding binding = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES));
+        when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding));
 
         Instant t0 = Instant.parse("2026-06-05T10:00:30Z");
         trigger.sweep(t0); // first sight: baseline, must not fire immediately
-        verify(policyRunner, never()).run(any());
+        verify(policyRunner, never()).runInput(any(), any(), any());
 
         trigger.sweep(t0.plusSeconds(120)); // the one-minute mark has passed
-        verify(policyRunner, times(1)).run(eq(policy));
+        verify(policyRunner, times(1)).runInput(eq(binding.policy()), eq(binding.input()), any());
     }
 
     @Test
     void anIntervalMatchingTheSweepPeriodFiresEverySweepDespiteJitter() {
-        Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES));
-        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+        PolicyBinding binding = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES));
+        when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding));
 
         Instant t0 = Instant.parse("2026-06-05T10:00:00Z");
         trigger.sweep(t0); // baseline
         // The sweep that fires runs a few ms late (scheduler jitter)...
         trigger.sweep(t0.plusSeconds(60).plusMillis(5));
-        verify(policyRunner, times(1)).run(eq(policy));
+        verify(policyRunner, times(1)).runInput(eq(binding.policy()), eq(binding.input()), any());
 
         // ...and the next sweep lands exactly on the 60s grid. Anchoring lastFired to the due
         // time (not the jittered observation) means this must still fire, not alias to skip.
         trigger.sweep(t0.plusSeconds(120));
-        verify(policyRunner, times(2)).run(eq(policy));
+        verify(policyRunner, times(2)).runInput(eq(binding.policy()), eq(binding.input()), any());
     }
 
     @Test
     void aGapFiresOnceNotOncePerMissedInterval() {
-        Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES));
-        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+        PolicyBinding binding = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES));
+        when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding));
 
         Instant t0 = Instant.parse("2026-06-05T10:00:00Z");
         trigger.sweep(t0); // baseline
         // Ten minutes of downtime: nine missed due points collapse into one firing.
         trigger.sweep(t0.plusSeconds(600));
-        verify(policyRunner, times(1)).run(eq(policy));
+        verify(policyRunner, times(1)).runInput(eq(binding.policy()), eq(binding.input()), any());
 
         // Not due again until a full interval after the latest due point.
         trigger.sweep(t0.plusSeconds(630));
-        verify(policyRunner, times(1)).run(eq(policy));
+        verify(policyRunner, times(1)).runInput(eq(binding.policy()), eq(binding.input()), any());
         trigger.sweep(t0.plusSeconds(660));
-        verify(policyRunner, times(2)).run(eq(policy));
+        verify(policyRunner, times(2)).runInput(eq(binding.policy()), eq(binding.input()), any());
     }
 
     @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));
+        PolicyBinding binding =
+                scheduled("p1", new Schedule.Daily(LocalTime.of(3, 0))); // 03:00 UTC daily
+        when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding));
 
         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());
+        verify(policyRunner, never()).runInput(any(), any(), any());
     }
 
     @Test
     void firesWeeklyOnAChosenDay() {
         // 2026-06-05 is a Friday; the next Monday 09:00 is the soonest firing.
-        Policy policy =
+        PolicyBinding binding =
                 scheduled("p1", new Schedule.Weekly(Set.of(DayOfWeek.MONDAY), LocalTime.of(9, 0)));
-        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+        when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding));
 
         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));
+        verify(policyRunner, times(1)).runInput(eq(binding.policy()), eq(binding.input()), any());
     }
 
     @Test
-    void skipsPoliciesWithAnInvalidSchedule() {
-        Policy policy = scheduledWithRawOptions("p1", Map.of()); // no schedule
-        when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy));
+    void skipsInputsWithAnInvalidSchedule() {
+        PolicyBinding binding = scheduledWithRawOptions("p1", Map.of()); // no schedule
+        when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding));
 
         trigger.sweep(Instant.parse("2026-06-05T10:00:00Z"));
 
-        verify(policyRunner, never()).run(any());
+        verify(policyRunner, never()).runInput(any(), any(), any());
     }
 
     @Test
     void validateRejectsMissingSchedule() {
+        PolicyBinding binding = scheduledWithRawOptions("p1", Map.of());
         assertThrows(
                 IllegalArgumentException.class,
-                () -> trigger.validate(scheduledWithRawOptions("p1", Map.of())));
+                () -> trigger.validate(binding.policy(), binding.input()));
     }
 
     @Test
     void validateRejectsAnInvalidSchedule() {
         Map<String, Object> options =
                 Map.of("schedule", Map.of("type", "every", "count", -5, "unit", "MINUTES"));
+        PolicyBinding binding = scheduledWithRawOptions("p1", options);
         assertThrows(
                 IllegalArgumentException.class,
-                () -> trigger.validate(scheduledWithRawOptions("p1", options)));
+                () -> trigger.validate(binding.policy(), binding.input()));
     }
 
     @Test
@@ -162,21 +167,25 @@ class ScheduleTriggerTest {
         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));
+        PolicyBinding binding = scheduledWithRawOptions("p1", options);
+        trigger.validate(binding.policy(), binding.input());
     }
 
-    private static Policy scheduled(String id, Schedule schedule) {
+    private static PolicyBinding 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(
+    private static PolicyBinding scheduledWithRawOptions(String id, Map<String, Object> options) {
+        PipelineInput input = new PipelineInput("s1", new TriggerConfig("schedule", options));
+        Policy policy =
+                new Policy(
                         id,
                         "nightly",
                         "owner",
                         true,
-                new TriggerConfig("schedule", options),
+                        List.of(input),
                         List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                         OutputSpec.inline());
+        return new PolicyBinding(policy, input);
     }
 }

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

@@ -2,10 +2,12 @@ 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.verify;
 import static org.mockito.Mockito.when;
 
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
@@ -19,8 +21,10 @@ import stirling.software.common.model.ApplicationProperties;
 import stirling.software.proprietary.policy.engine.PolicyRunner;
 import stirling.software.proprietary.policy.engine.SweepKind;
 import stirling.software.proprietary.policy.model.OutputSpec;
+import stirling.software.proprietary.policy.model.PipelineInput;
 import stirling.software.proprietary.policy.model.PipelineStep;
 import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.model.PolicyBinding;
 import stirling.software.proprietary.policy.model.TriggerConfig;
 import stirling.software.proprietary.policy.source.InProcessSourceStore;
 import stirling.software.proprietary.policy.source.Source;
@@ -46,37 +50,45 @@ class WebhookTriggerTest {
     }
 
     @Test
-    void firesOnlyPoliciesReferencingTheDeliveredWebhook() {
+    void firesOnlyInputsReferencingTheDeliveredWebhook() {
         Policy matching = webhookPolicy("a", "whkA");
         Policy other = webhookPolicy("b", "whkB");
-        when(policyStore.findByTriggerType(TYPE)).thenReturn(List.of(matching, other));
+        when(policyStore.findBindingsByTriggerType(TYPE)).thenReturn(bindings(matching, other));
 
         trigger.fireForWebhook("whkA");
 
-        verify(policyRunner).run(matching, SweepKind.LIGHT);
-        verify(policyRunner, never()).run(other, SweepKind.LIGHT);
+        verify(policyRunner).runInput(matching, matching.inputs().get(0), SweepKind.LIGHT);
+        verify(policyRunner, never()).runInput(eq(other), any(), any());
     }
 
     @Test
     void ignoresADeliveryForAnUnknownWebhookId() {
         Policy policy = webhookPolicy("a", "whkA");
-        when(policyStore.findByTriggerType(TYPE)).thenReturn(List.of(policy));
+        when(policyStore.findBindingsByTriggerType(TYPE)).thenReturn(bindings(policy));
 
         trigger.fireForWebhook("whkZ");
 
-        verify(policyRunner, never()).run(any(), any(SweepKind.class));
+        verify(policyRunner, never()).runInput(any(), any(), any());
     }
 
     @Test
     void validateRequiresAWebhookSource() {
+        // A non-webhook source (here: an id that resolves to nothing) is rejected.
+        Policy notWebhook = policy("p", PipelineInput.manual("missing-source"));
         assertThrows(
                 IllegalArgumentException.class,
-                () -> trigger.validate(policy("p", webhookTriggerConfig(), List.of())));
-        trigger.validate(webhookPolicy("p", "whkA"));
+                () -> trigger.validate(notWebhook, notWebhook.inputs().get(0)));
+
+        Policy hooked = webhookPolicy("p", "whkA");
+        trigger.validate(hooked, hooked.inputs().get(0));
     }
 
-    private static TriggerConfig webhookTriggerConfig() {
-        return new TriggerConfig(TYPE, Map.of());
+    /** Every (policy, input) binding across the given policies, as the store would return them. */
+    private static List<PolicyBinding> bindings(Policy... policies) {
+        return Arrays.stream(policies)
+                .flatMap(
+                        policy -> policy.inputs().stream().map(in -> new PolicyBinding(policy, in)))
+                .toList();
     }
 
     private Policy webhookPolicy(String id, String webhookId) {
@@ -98,17 +110,16 @@ class WebhookTriggerTest {
                                         "owner",
                                         null))
                         .id();
-        return policy(id, webhookTriggerConfig(), List.of(sourceId));
+        return policy(id, new PipelineInput(sourceId, new TriggerConfig(TYPE, Map.of())));
     }
 
-    private static Policy policy(String id, TriggerConfig trigger, List<String> sourceIds) {
+    private static Policy policy(String id, PipelineInput input) {
         return new Policy(
                 id,
                 "hook",
                 "owner",
                 true,
-                trigger,
-                sourceIds,
+                List.of(input),
                 List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())),
                 OutputSpec.inline());
     }

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

@@ -7745,11 +7745,17 @@ newPipeline = "New pipeline"
 addStep = "Add tool"
 back = "Back to pipelines"
 chooseAccount = "Choose an account"
+chooseDestination = "Choose a destination"
 chooseOperation = "Choose what this step does"
+chooseSource = "Choose a source"
 discard = "Discard changes"
 enabled = "Enabled"
+inputs = "Input"
+inputSource = "Input source"
+inputTrigger = "Trigger"
 keepEditing = "Keep editing"
 needsUpload = "Needs an uploaded file"
+noSources = "No sources connected yet. Create one to use it as an input."
 noToolMatches = "No tools match your search."
 pipelineSettings = "Pipeline settings"
 searchTools = "Search tools"
@@ -7777,7 +7783,7 @@ namePlaceholder = "e.g. Redaction sweep"
 noToolSettings = "This tool has no configurable settings."
 operations_one = "Operation ({{count}})"
 operations_other = "Operations ({{count}})"
-output = "Destinations"
+output = "Destination"
 removeStep = "Remove operation"
 save = "Save changes"
 scheduleEvery = "Run every"

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

@@ -17,12 +17,22 @@ export interface PipelineStep {
   fileParameters?: Record<string, string>;
 }
 
-/** When a policy fires automatically. `type` keys a trigger bean (e.g. "schedule"). */
+/** When a policy input fires automatically. `type` keys a trigger bean (e.g. "schedule"). */
 export interface TriggerConfig {
   type: string;
   options: Record<string, unknown>;
 }
 
+/**
+ * One input of a pipeline: a persisted source paired with the trigger that decides when that
+ * source is pulled. A `null` trigger means the input is pulled only on a manual run. Mirrors the
+ * backend `PipelineInput`.
+ */
+export interface PipelineInput {
+  sourceId: string;
+  trigger: TriggerConfig | null;
+}
+
 /** Where a run's outputs are delivered. `type` keys an output sink (e.g. "inline"). */
 export interface OutputSpec {
   type: string;
@@ -35,15 +45,15 @@ export type PipelineOutputMode = "folder" | "s3";
 /**
  * The stored policy record: the create/update body (`id` blank on create) and what
  * the backend returns from GET/POST. Mirrors Policy.java exactly; `owner`/`teamId`
- * are stamped server-side. A `null` trigger means manual-only.
+ * are stamped server-side. Each input pairs a source with its own trigger; an input
+ * with a `null` trigger (or a policy with no triggered inputs) runs only on demand.
  */
 export interface Policy {
   id?: string;
   name: string;
   owner?: string | null;
   enabled: boolean;
-  trigger: TriggerConfig | null;
-  sourceIds: string[];
+  inputs: PipelineInput[];
   steps: PipelineStep[];
   /**
    * Inline output, used only when no destinations are referenced (editor/one-off runs that return

frontend/editor/src/portal/components/pipelines/DestinationPicker.tsx

@@ -1,14 +1,15 @@
 import { useTranslation } from "react-i18next";
 import AddRoundedIcon from "@mui/icons-material/AddRounded";
-import { Button, Checkbox } from "@app/ui";
+import { Button, Select } from "@app/ui";
 
 /**
- * Picks the saved sources a pipeline delivers its output to. A destination is just
- * a source used as a write target, and a pipeline may write to several, so this is
- * a checklist over the same locations the builder loaded (filtered to writable
- * types by the caller) - mirroring the input-sources checklist. Creating a new one
- * is delegated to {@code onCreateNew} (the builder navigates to the source builder,
- * prompting about unsaved edits first).
+ * Picks the saved source a pipeline delivers its output to. A destination is just a
+ * source used as a write target. The value stays a list ({@code outputIds}) because
+ * the model supports several, but the product caps a pipeline at one destination
+ * today, so this renders a single dropdown over the same locations the builder
+ * loaded (filtered to writable types by the caller). Creating a new one is delegated
+ * to {@code onCreateNew} (the builder navigates to the source builder, prompting
+ * about unsaved edits first).
  */
 interface DestinationOption {
   id: string;
@@ -31,23 +32,21 @@ export function DestinationPicker({
 }: DestinationPickerProps) {
   const { t } = useTranslation();
 
-  function toggle(id: string, checked: boolean) {
-    onChange(
-      checked ? [...value, id] : value.filter((existing) => existing !== id),
-    );
-  }
-
   return (
-    <>
-      <div className="portal-pipelines__source-list">
-        {sources.map((source) => (
-          <Checkbox
-            key={source.id}
-            checked={value.includes(source.id)}
-            onChange={(e) => toggle(source.id, e.target.checked)}
-            label={source.name}
+    <div className="portal-builder__input-row">
+      <div className="portal-builder__input-field">
+        <Select
+          inputSize="sm"
+          aria-label={t("portal.pipelines.composer.output")}
+          placeholder={t("portal.pipelines.builder.chooseDestination")}
+          value={value[0] ?? null}
+          invalid={value.length !== 1}
+          onChange={(id) => onChange(id ? [id] : [])}
+          options={sources.map((source) => ({
+            value: source.id,
+            label: source.name,
+          }))}
         />
-        ))}
       </div>
       <Button
         variant="tertiary"
@@ -57,6 +56,6 @@ export function DestinationPicker({
       >
         {t("portal.sources.actions.connectSource")}
       </Button>
-    </>
+    </div>
   );
 }

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

@@ -40,11 +40,15 @@ function seedPipelines(): StoredPolicy[] {
       name: "Redaction sweep",
       owner: "security@acme.com",
       enabled: true,
+      inputs: [
+        {
+          sourceId: "src-claims",
           trigger: {
             type: "schedule",
             options: { schedule: { type: "every", count: 6, unit: "HOURS" } },
           },
-      sourceIds: ["src-claims"],
+        },
+      ],
       steps: [
         {
           operation: "/api/v1/security/auto-redact",
@@ -53,15 +57,14 @@ function seedPipelines(): StoredPolicy[] {
         { operation: "/api/v1/security/sanitize-pdf", parameters: {} },
       ],
       output: { type: "inline", options: {} },
-      outputIds: ["src-archive", "src-contracts"],
+      outputIds: ["src-archive"],
     },
     {
       id: "plc-archive",
       name: "Archive compressor",
       owner: "data-eng@acme.com",
       enabled: true,
-      trigger: null,
-      sourceIds: ["src-contracts", "src-archive"],
+      inputs: [{ sourceId: "src-contracts", trigger: null }],
       steps: [{ operation: "/api/v1/misc/compress-pdf", parameters: {} }],
       output: { type: "inline", options: {} },
       outputIds: ["src-contracts"],
@@ -71,8 +74,7 @@ function seedPipelines(): StoredPolicy[] {
       name: "Onboarding OCR (paused)",
       owner: "ops@acme.com",
       enabled: false,
-      trigger: null,
-      sourceIds: [],
+      inputs: [],
       steps: [
         { operation: "/api/v1/misc/ocr-pdf", parameters: {} },
         { operation: "/api/v1/misc/flatten", parameters: {} },
@@ -95,16 +97,28 @@ function deriveStatus(policy: StoredPolicy): PipelineStatus {
   return policy.enabled ? "active" : "paused";
 }
 
+// Distinct trigger types across a policy's inputs, or "manual" when none is triggered.
+function triggerSummary(policy: StoredPolicy): string {
+  const types = [
+    ...new Set(
+      policy.inputs
+        .map((input) => input.trigger?.type)
+        .filter((type): type is string => type != null),
+    ),
+  ];
+  return types.length === 0 ? "manual" : types.join(", ");
+}
+
 function toView(policy: StoredPolicy): PipelineView {
   return {
     id: policy.id,
     name: policy.name,
     enabled: policy.enabled,
     status: deriveStatus(policy),
-    trigger: policy.trigger?.type ?? "manual",
-    sources: policy.sourceIds.map((id) => ({
-      id,
-      name: SOURCE_NAMES[id] ?? id,
+    trigger: triggerSummary(policy),
+    sources: policy.inputs.map((input) => ({
+      id: input.sourceId,
+      name: SOURCE_NAMES[input.sourceId] ?? input.sourceId,
     })),
     steps: policy.steps.map((s) => s.operation),
     output:

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

@@ -242,6 +242,29 @@
   gap: 0.5rem;
 }
 
+/* Input and destination each span the full settings width so their row has room. */
+.portal-builder__inputs-col {
+  grid-column: 1 / -1;
+}
+
+/* The input row (source + trigger + optional schedule) and the destination row. */
+.portal-builder__input-row {
+  display: flex;
+  flex-wrap: wrap;
+  align-items: center;
+  gap: 0.5rem;
+}
+
+.portal-builder__input-field {
+  flex: 1 1 11rem;
+  min-width: 10rem;
+}
+
+/* The connect-source button trails to the end of the row. */
+.portal-builder__input-row > button:last-child {
+  margin-left: auto;
+}
+
 /* Inspector: heading sits outside the card so it aligns with the operations heading. */
 .portal-builder__inspector-col {
   position: sticky;

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

@@ -126,8 +126,7 @@ const POLICY: Policy = {
   id: "plc-1",
   name: "Existing pipeline",
   enabled: true,
-  trigger: null,
-  sourceIds: [],
+  inputs: [],
   steps: [],
   output: { type: "inline", options: {} },
   outputIds: [],
@@ -195,21 +194,53 @@ describe("PipelineBuilder", () => {
     createIntegration.mockReset();
   });
 
-  it("builds a new pipeline: name it, add a tool, and save", async () => {
+  // Choose the given source in the (pre-seeded) input row's dropdown.
+  async function pickInputSource(sourceName: string) {
+    fireEvent.click(
+      await screen.findByRole("textbox", {
+        name: "portal.pipelines.builder.inputSource",
+      }),
+    );
+    fireEvent.click(await screen.findByText(sourceName));
+  }
+
+  it("always shows exactly one input row, with no add or remove controls", async () => {
     renderBuilder("/processor/pipelines/new");
 
-    // The name field is the only textbox before the picker opens.
-    fireEvent.change(await screen.findByRole("textbox"), {
-      target: { value: "Nightly compress" },
+    // The input row is a fixed part of the form: its source dropdown is present from the
+    // start, and there is nothing to add or remove.
+    expect(
+      await screen.findAllByRole("textbox", {
+        name: "portal.pipelines.builder.inputSource",
+      }),
+    ).toHaveLength(1);
+    expect(
+      screen.queryByText("portal.pipelines.builder.addInput"),
+    ).not.toBeInTheDocument();
+    expect(
+      screen.queryByRole("button", {
+        name: "portal.pipelines.builder.removeInput",
+      }),
+    ).not.toBeInTheDocument();
   });
 
+  it("builds a new pipeline: name it, add a tool, an input, a destination, and save", async () => {
+    renderBuilder("/processor/pipelines/new");
+
+    fireEvent.change(
+      await screen.findByRole("textbox", {
+        name: "portal.pipelines.composer.name",
+      }),
+      {
+        target: { value: "Nightly compress" },
+      },
+    );
+
     fireEvent.click(screen.getByRole("button", { name: /addTool/ }));
     fireEvent.click(await screen.findByText("Compress"));
 
     // A pipeline must have at least one input source and one output destination.
-    fireEvent.click(
-      await screen.findByRole("checkbox", { name: "Claims intake" }),
-    );
+    await pickInputSource("Claims intake");
     fireEvent.click(screen.getByText("pick output"));
 
     fireEvent.click(screen.getByText("portal.pipelines.composer.create"));
@@ -218,8 +249,8 @@ describe("PipelineBuilder", () => {
     expect(savePipeline).toHaveBeenCalledWith(
       expect.objectContaining({
         name: "Nightly compress",
-        trigger: null,
-        sourceIds: ["src-in"],
+        // The input pairs the chosen source with its trigger (manual by default).
+        inputs: [{ sourceId: "src-in", trigger: null }],
         outputIds: ["src-1"],
         steps: [
           expect.objectContaining({ operation: "/api/v1/misc/compress-pdf" }),
@@ -232,19 +263,22 @@ describe("PipelineBuilder", () => {
   it("requires at least one source and one destination before saving", async () => {
     renderBuilder("/processor/pipelines/new");
 
-    fireEvent.change(await screen.findByRole("textbox"), {
+    fireEvent.change(
+      await screen.findByRole("textbox", {
+        name: "portal.pipelines.composer.name",
+      }),
+      {
         target: { value: "Needs both" },
-    });
+      },
+    );
     const saveButton = () =>
       screen.getByText("portal.pipelines.composer.create").closest("button");
 
     // Name only: blocked (no source, no destination).
     expect(saveButton()).toBeDisabled();
 
-    // A source but still no destination: blocked.
-    fireEvent.click(
-      await screen.findByRole("checkbox", { name: "Claims intake" }),
-    );
+    // An input with a source but still no destination: blocked.
+    await pickInputSource("Claims intake");
     expect(saveButton()).toBeDisabled();
 
     // Both chosen: allowed, and both are sent.
@@ -252,7 +286,10 @@ describe("PipelineBuilder", () => {
     fireEvent.click(screen.getByText("portal.pipelines.composer.create"));
     await waitFor(() => expect(savePipeline).toHaveBeenCalledTimes(1));
     expect(savePipeline).toHaveBeenCalledWith(
-      expect.objectContaining({ sourceIds: ["src-in"], outputIds: ["src-1"] }),
+      expect.objectContaining({
+        inputs: [{ sourceId: "src-in", trigger: null }],
+        outputIds: ["src-1"],
+      }),
     );
   });
 
@@ -309,9 +346,14 @@ describe("PipelineBuilder", () => {
   it("blocks saving a step that needs an uploaded file", async () => {
     renderBuilder("/processor/pipelines/new");
 
-    fireEvent.change(await screen.findByRole("textbox"), {
+    fireEvent.change(
+      await screen.findByRole("textbox", {
+        name: "portal.pipelines.composer.name",
+      }),
+      {
         target: { value: "Watermarked" },
-    });
+      },
+    );
     fireEvent.click(screen.getByRole("button", { name: /addTool/ }));
     fireEvent.click(await screen.findByText("Compress"));
     // The tool's settings upload a file, which a stored pipeline can't persist yet.
@@ -330,9 +372,14 @@ describe("PipelineBuilder", () => {
     // rejection; the builder must refuse to save it and say why, where the fix is one click away.
     renderBuilder("/processor/pipelines/new");
 
-    fireEvent.change(await screen.findByRole("textbox"), {
+    fireEvent.change(
+      await screen.findByRole("textbox", {
+        name: "portal.pipelines.composer.name",
+      }),
+      {
         target: { value: "Notify only" },
-    });
+      },
+    );
     fireEvent.click(screen.getByRole("button", { name: /addTool/ }));
     fireEvent.click(
       await screen.findByText("portal.policies.operations.discordNotify.label"),
@@ -360,9 +407,14 @@ describe("PipelineBuilder", () => {
   it("prompts to save or discard when leaving with unsaved edits", async () => {
     renderBuilder("/processor/pipelines/new");
 
-    fireEvent.change(await screen.findByRole("textbox"), {
+    fireEvent.change(
+      await screen.findByRole("textbox", {
+        name: "portal.pipelines.composer.name",
+      }),
+      {
         target: { value: "Draft" },
-    });
+      },
+    );
     fireEvent.click(screen.getByText("portal.pipelines.composer.cancel"));
 
     expect(
@@ -385,9 +437,14 @@ describe("PipelineBuilder", () => {
     ]);
     renderBuilder("/processor/pipelines/new");
 
-    fireEvent.change(await screen.findByRole("textbox"), {
+    fireEvent.change(
+      await screen.findByRole("textbox", {
+        name: "portal.pipelines.composer.name",
+      }),
+      {
         target: { value: "Notify on processed" },
-    });
+      },
+    );
 
     fireEvent.click(screen.getByRole("button", { name: /addTool/ }));
     fireEvent.click(
@@ -401,10 +458,8 @@ describe("PipelineBuilder", () => {
     );
     fireEvent.click(await screen.findByText("Ops alerts"));
 
-    // Saving needs at least one input source and one destination.
-    fireEvent.click(
-      await screen.findByRole("checkbox", { name: "Claims intake" }),
-    );
+    // Saving needs the input's source and a destination.
+    await pickInputSource("Claims intake");
     fireEvent.click(screen.getByText("pick output"));
 
     fireEvent.click(screen.getByText("portal.pipelines.composer.create"));
@@ -424,7 +479,9 @@ describe("PipelineBuilder", () => {
   it("leaves immediately when there are no unsaved edits", async () => {
     renderBuilder("/processor/pipelines/new");
 
-    await screen.findByRole("textbox");
+    await screen.findByRole("textbox", {
+      name: "portal.pipelines.composer.name",
+    });
     fireEvent.click(screen.getByText("portal.pipelines.composer.cancel"));
 
     expect(await screen.findByText("pipelines list")).toBeInTheDocument();

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

@@ -16,7 +16,6 @@ import {
   EmptyState,
   Input,
   Modal,
-  RadioGroup,
   Select,
   Spinner,
 } from "@app/ui";
@@ -72,6 +71,12 @@ type ScheduleUnit = "MINUTES" | "HOURS" | "DAYS";
 const SCHEDULE_UNITS: ScheduleUnit[] = ["MINUTES", "HOURS", "DAYS"];
 /** Empty trigger type = manual-only (no automatic trigger). */
 const MANUAL = "";
+/**
+ * Sentinel value for the manual choice in the trigger dropdown. Mantine's Select treats an empty
+ * string as "no selection" (it shows the placeholder, not the option), so the manual option needs a
+ * real value; it maps to/from the empty {@link MANUAL} trigger type at the edges.
+ */
+const MANUAL_OPTION = "manual";
 
 const TERMINAL_STATUSES = new Set(["COMPLETED", "FAILED", "CANCELLED"]);
 const POLL_INTERVAL_MS = 1500;
@@ -105,6 +110,42 @@ function parseTrigger(trigger: TriggerConfig | null): {
   return { triggerType: trigger.type, count: "1", unit: "HOURS" };
 }
 
+/** One input row in the builder: a source paired with its own trigger config. */
+interface WorkingInput {
+  sourceId: string;
+  triggerType: string;
+  scheduleCount: string;
+  scheduleUnit: ScheduleUnit;
+}
+
+/** The input row with nothing chosen yet: no source, manual trigger. */
+function blankInput(): WorkingInput {
+  return {
+    sourceId: "",
+    triggerType: MANUAL,
+    scheduleCount: "1",
+    scheduleUnit: "HOURS",
+  };
+}
+
+/** The trigger config for the input row, or null for a manual (on-demand) input. */
+function buildTriggerFor(input: WorkingInput): TriggerConfig | null {
+  if (input.triggerType === MANUAL) return null;
+  if (input.triggerType === "schedule") {
+    return {
+      type: "schedule",
+      options: {
+        schedule: {
+          type: "every",
+          count: Number(input.scheduleCount),
+          unit: input.scheduleUnit,
+        },
+      },
+    };
+  }
+  return { type: input.triggerType, options: {} };
+}
+
 /**
  * Full-page pipeline builder (route: /pipelines/new and /pipelines/:id). Pipeline-level settings
  * (sources, trigger, output) sit above the operation list; the operation list and the selected
@@ -166,13 +207,12 @@ export function PipelineBuilder() {
 
   const [name, setName] = useState("");
   const [enabled, setEnabled] = useState(true);
-  const [sourceIds, setSourceIds] = useState<string[]>([]);
+  // Exactly one input: the row is always present, so the working state is a single object; the
+  // wire shape stays a list (see save()).
+  const [input, setInput] = useState<WorkingInput>(blankInput);
   const [steps, setSteps] = useState<WorkingToolStep[]>([]);
   const [selectedIndex, setSelectedIndex] = useState<number | null>(null);
   const [pickerOpen, setPickerOpen] = useState(false);
-  const [triggerType, setTriggerType] = useState<string>(MANUAL);
-  const [scheduleCount, setScheduleCount] = useState("1");
-  const [scheduleUnit, setScheduleUnit] = useState<ScheduleUnit>("HOURS");
   const [outputIds, setOutputIds] = useState<string[]>([]);
   const [submitting, setSubmitting] = useState(false);
   const [error, setError] = useState<string | null>(null);
@@ -197,16 +237,26 @@ export function PipelineBuilder() {
     if (seeded) return;
     if (isEdit && !policyState.data) return;
     const policy = policyState.data ?? undefined;
-    const trigger = parseTrigger(policy?.trigger ?? null);
     setName(policy?.name ?? "");
     setEnabled(policy?.enabled ?? true);
-    setSourceIds(policy?.sourceIds ?? []);
+    // The one input row is always present: blank for a new pipeline (or a legacy policy saved
+    // without inputs), the stored input for an edit. A legacy multi-input policy shows only its
+    // first input; saving persists just that one (the backend rejects more anyway).
+    const stored = policy?.inputs[0];
+    if (stored) {
+      const trigger = parseTrigger(stored.trigger);
+      setInput({
+        sourceId: stored.sourceId,
+        triggerType: trigger.triggerType,
+        scheduleCount: trigger.count,
+        scheduleUnit: trigger.unit,
+      });
+    } else {
+      setInput(blankInput());
+    }
     setSteps(
       (policy?.steps ?? []).map((step) => deserializeToolStep(step, allTools)),
     );
-    setTriggerType(trigger.triggerType);
-    setScheduleCount(trigger.count);
-    setScheduleUnit(trigger.unit);
     setOutputIds(policy?.outputIds ?? []);
     setSeeded(true);
   }, [isEdit, policyState.data, allTools, seeded]);
@@ -219,37 +269,62 @@ export function PipelineBuilder() {
     }
   }, [seeded, selectedIndex, steps.length]);
 
-  const selectedSourceTypes = useMemo(
-    () =>
-      new Set(
-        availableSources
-          .filter((s) => sourceIds.includes(s.id))
-          .map((s) => s.type),
-      ),
-    [availableSources, sourceIds],
-  );
+  const sourceType = (sourceId: string) =>
+    availableSources.find((s) => s.id === sourceId)?.type;
 
-  const triggerAvailable = useMemo(
-    () => (trigger: TriggerInfo) =>
-      !trigger.requiresSource ||
-      trigger.supportedSourceTypes.some((type) =>
-        selectedSourceTypes.has(type),
-      ),
-    [selectedSourceTypes],
-  );
+  // A trigger fits a source when it needs no source, or the source's type is one it supports
+  // (folder-watch → folder; schedule → any). Drives the per-input trigger dropdown.
+  const triggerFitsType = (trigger: TriggerInfo, type: string) =>
+    !trigger.requiresSource || trigger.supportedSourceTypes.includes(type);
 
-  useEffect(() => {
-    if (triggerType === MANUAL) return;
-    const selected = triggers.find((trigger) => trigger.type === triggerType);
-    if (selected && !triggerAvailable(selected)) setTriggerType(MANUAL);
-  }, [triggerType, triggers, triggerAvailable]);
+  const sourceOptions = availableSources.map((source) => ({
+    value: source.id,
+    label: source.name,
+  }));
 
-  function toggleSource(sourceId: string, checked: boolean) {
-    setSourceIds((ids) =>
-      checked
-        ? [...ids, sourceId]
-        : ids.filter((existing) => existing !== sourceId),
-    );
+  // Manual plus every trigger compatible with this row's source. Manual only until a source is set.
+  function triggerOptionsFor(sourceId: string) {
+    const options = [
+      {
+        value: MANUAL_OPTION,
+        label: t("portal.pipelines.composer.triggerManual"),
+      },
+    ];
+    const type = sourceType(sourceId);
+    if (type) {
+      for (const trigger of triggers) {
+        if (triggerFitsType(trigger, type)) {
+          options.push({
+            value: trigger.type,
+            label: t(`portal.pipelines.trigger.${trigger.type}`, {
+              defaultValue: trigger.type,
+            }),
+          });
+        }
+      }
+    }
+    return options;
+  }
+
+  function updateInput(patch: Partial<WorkingInput>) {
+    setInput((current) => ({ ...current, ...patch }));
+  }
+
+  // Changing the source may make the current trigger incompatible (folder-watch on a non-folder);
+  // drop it back to manual when that happens so the row can't hold an invalid pairing.
+  function changeInputSource(sourceId: string) {
+    setInput((current) => {
+      const type = sourceType(sourceId);
+      const trigger = triggers.find((tr) => tr.type === current.triggerType);
+      const keepTrigger =
+        current.triggerType === MANUAL ||
+        (type != null && trigger != null && triggerFitsType(trigger, type));
+      return {
+        ...current,
+        sourceId,
+        triggerType: keepTrigger ? current.triggerType : MANUAL,
+      };
+    });
   }
 
   function addOperationStep(op: (typeof STEP_OPERATIONS)[number]) {
@@ -326,12 +401,9 @@ export function PipelineBuilder() {
   const snapshot = JSON.stringify({
     name: name.trim(),
     enabled,
-    sourceIds: [...sourceIds].sort(),
+    input,
     steps: steps.map((step) => serializeToolStep(step, allTools)),
     uploads: steps.map(stepRequiresUpload),
-    triggerType,
-    scheduleCount,
-    scheduleUnit,
     outputIds: [...outputIds].sort(),
   });
   const baseline = useRef<string | null>(null);
@@ -340,48 +412,20 @@ export function PipelineBuilder() {
   }, [seeded, snapshot]);
   const dirty = baseline.current !== null && baseline.current !== snapshot;
 
-  const scheduleCountValid =
-    triggerType !== "schedule" || Number(scheduleCount) > 0;
-  // A pipeline must have at least one input source and at least one output destination.
-  const sourceValid = sourceIds.length > 0;
-  const outputValid = outputIds.length > 0;
+  // The input needs a source, and a scheduled input needs a positive interval; the pipeline
+  // needs exactly one output destination.
+  const inputValid =
+    input.sourceId !== "" &&
+    (input.triggerType !== "schedule" || Number(input.scheduleCount) > 0);
+  const outputValid = outputIds.length === 1;
   const canSave =
     name.trim() !== "" &&
-    sourceValid &&
-    scheduleCountValid &&
+    inputValid &&
     outputValid &&
     !hasUploadSteps &&
     !hasUnconfiguredSteps &&
     !submitting;
 
-  const triggerOptions = [
-    { value: MANUAL, label: t("portal.pipelines.composer.triggerManual") },
-    ...triggers.map((trigger) => ({
-      value: trigger.type,
-      label: t(`portal.pipelines.trigger.${trigger.type}`, {
-        defaultValue: trigger.type,
-      }),
-      disabled: !triggerAvailable(trigger),
-    })),
-  ];
-
-  function buildTrigger(): TriggerConfig | null {
-    if (triggerType === MANUAL) return null;
-    if (triggerType === "schedule") {
-      return {
-        type: "schedule",
-        options: {
-          schedule: {
-            type: "every",
-            count: Number(scheduleCount),
-            unit: scheduleUnit,
-          },
-        },
-      };
-    }
-    return { type: triggerType, options: {} };
-  }
-
   const listPath = toPortalPath(VIEW_PATHS.pipelines);
   const sourcesPath = `${toPortalPath(VIEW_PATHS.sources)}/new`;
 
@@ -409,8 +453,8 @@ export function PipelineBuilder() {
       id: policyState.data?.id ?? undefined,
       name: name.trim(),
       enabled,
-      trigger: buildTrigger(),
-      sourceIds,
+      // The wire shape stays a list; canSave guarantees the one input has a source.
+      inputs: [{ sourceId: input.sourceId, trigger: buildTriggerFor(input) }],
       steps: steps.map((step) => serializeToolStep(step, allTools)),
       // Destinations are the referenced saved sources; the inline output field is
       // preserved as-is (e.g. an editor policy's membership metadata) or defaults to inline.
@@ -664,47 +708,48 @@ export function PipelineBuilder() {
           {t("portal.pipelines.builder.pipelineSettings")}
         </div>
         <div className="portal-builder__settings-grid">
-          <div className="portal-builder__settings-col">
+          <div className="portal-builder__settings-col portal-builder__inputs-col">
             <span className="portal-pipelines__detail-heading">
-              {t("portal.pipelines.composer.sources")}
+              {t("portal.pipelines.builder.inputs")}
             </span>
             {sourcesState.loading ? (
               <p className="portal-pipelines__muted">
                 {t("portal.pipelines.composer.sourcesLoading")}
               </p>
             ) : (
-              <div className="portal-pipelines__source-list">
-                {availableSources.map((source) => (
-                  <Checkbox
-                    key={source.id}
-                    checked={sourceIds.includes(source.id)}
-                    onChange={(e) => toggleSource(source.id, e.target.checked)}
-                    label={source.name}
+              <>
+                <div className="portal-builder__input-row">
+                  <div className="portal-builder__input-field">
+                    <Select
+                      inputSize="sm"
+                      aria-label={t("portal.pipelines.builder.inputSource")}
+                      placeholder={t("portal.pipelines.builder.chooseSource")}
+                      value={input.sourceId || null}
+                      invalid={input.sourceId === ""}
+                      onChange={(value) => changeInputSource(value ?? "")}
+                      options={sourceOptions}
                     />
-                ))}
                   </div>
-            )}
-            <Button
-              variant="tertiary"
-              size="sm"
-              onClick={goToSources}
-              leftSection={<AddRoundedIcon style={{ fontSize: "1.125rem" }} />}
-            >
-              {t("portal.sources.actions.connectSource")}
-            </Button>
-          </div>
-
-          <div className="portal-builder__settings-col">
-            <span className="portal-pipelines__detail-heading">
-              {t("portal.pipelines.composer.trigger")}
-            </span>
-            <RadioGroup<string>
-              name="pipeline-trigger"
-              value={triggerType}
-              onChange={setTriggerType}
-              options={triggerOptions}
+                  <div className="portal-builder__input-field">
+                    <Select
+                      inputSize="sm"
+                      aria-label={t("portal.pipelines.builder.inputTrigger")}
+                      value={
+                        input.triggerType === MANUAL
+                          ? MANUAL_OPTION
+                          : input.triggerType
+                      }
+                      disabled={input.sourceId === ""}
+                      onChange={(value) =>
+                        updateInput({
+                          triggerType:
+                            value && value !== MANUAL_OPTION ? value : MANUAL,
+                        })
+                      }
+                      options={triggerOptionsFor(input.sourceId)}
                     />
-            {triggerType === "schedule" && (
+                  </div>
+                  {input.triggerType === "schedule" && (
                     <div className="portal-pipelines__schedule">
                       <span className="portal-pipelines__muted">
                         {t("portal.pipelines.composer.scheduleEvery")}
@@ -713,16 +758,21 @@ export function PipelineBuilder() {
                         inputSize="sm"
                         type="number"
                         min={1}
-                  value={scheduleCount}
-                  invalid={!scheduleCountValid}
-                  onChange={(e) => setScheduleCount(e.target.value)}
+                        value={input.scheduleCount}
+                        invalid={Number(input.scheduleCount) <= 0}
+                        onChange={(e) =>
+                          updateInput({ scheduleCount: e.target.value })
+                        }
                         className="portal-pipelines__schedule-count"
                       />
                       <Select
                         inputSize="sm"
-                  value={scheduleUnit}
+                        value={input.scheduleUnit}
                         onChange={(value) =>
-                    value && setScheduleUnit(value as ScheduleUnit)
+                          value &&
+                          updateInput({
+                            scheduleUnit: value as ScheduleUnit,
+                          })
                         }
                         options={SCHEDULE_UNITS.map((unit) => ({
                           value: unit,
@@ -733,9 +783,27 @@ export function PipelineBuilder() {
                       />
                     </div>
                   )}
+                  <Button
+                    variant="tertiary"
+                    size="sm"
+                    onClick={goToSources}
+                    leftSection={
+                      <AddRoundedIcon style={{ fontSize: "1.125rem" }} />
+                    }
+                  >
+                    {t("portal.sources.actions.connectSource")}
+                  </Button>
+                </div>
+                {availableSources.length === 0 && (
+                  <p className="portal-pipelines__muted">
+                    {t("portal.pipelines.builder.noSources")}
+                  </p>
+                )}
+              </>
+            )}
           </div>
 
-          <div className="portal-builder__settings-col">
+          <div className="portal-builder__settings-col portal-builder__inputs-col">
             <span className="portal-pipelines__detail-heading">
               {t("portal.pipelines.composer.output")}
             </span>