Fix pre-existing multi-node bugs: shared node keys, Valkey url handling, backplane health, policy shape

app/common/src/main/java/stirling/software/common/cluster/ClusterConfig.java

@@ -1,19 +1,22 @@
 package stirling.software.common.cluster;
 
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
 
 import jakarta.annotation.PostConstruct;
 
-import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.model.ApplicationProperties;
 import stirling.software.common.model.ApplicationProperties.Cluster;
+import stirling.software.common.util.GeneralUtils;
 
 /** Validates cluster config consistency. All guards are skipped when cluster.enabled=false. */
 @Slf4j
 @Configuration
-@RequiredArgsConstructor
 public class ClusterConfig {
 
     private static final String MISSING_URL_MESSAGE =
@@ -21,7 +24,64 @@ public class ClusterConfig {
                     + " cluster.valkey.url to be set (e.g."
                     + " redis://valkey:6379).";
 
+    private static final String SHARED_SECRET_MESSAGE_SUFFIX =
+            " must be set to the same UUID on every node when cluster.enabled=true (env"
+                    + " AUTOMATICALLYGENERATED_KEY and AUTOMATICALLYGENERATED_UUID). Otherwise every"
+                    + " node mints its own at first boot, so workflow metadata encrypted on one"
+                    + " node cannot be decrypted on another and licence seat signatures do not"
+                    + " verify across nodes. The value must be a UUID; anything else (the shipped"
+                    + " 'example' placeholder included) is replaced by a per-node random UUID.";
+
+    /** Default bean name of stirling.software.SPDF.config.InitialSetup, which lives in :core. */
+    private static final String INITIAL_SETUP_BEAN = "initialSetup";
+
     private final ApplicationProperties applicationProperties;
+    private final String automaticallyGeneratedKey;
+    private final String automaticallyGeneratedUuid;
+
+    // Read from config, not from ApplicationProperties: InitialSetup overwrites the bound values
+    // with per-node UUIDs in its own @PostConstruct, which would defeat the guard below.
+    public ClusterConfig(
+            ApplicationProperties applicationProperties,
+            @Value("${AutomaticallyGenerated.key:}") String automaticallyGeneratedKey,
+            @Value("${AutomaticallyGenerated.UUID:}") String automaticallyGeneratedUuid) {
+        this.applicationProperties = applicationProperties;
+        this.automaticallyGeneratedKey = automaticallyGeneratedKey;
+        this.automaticallyGeneratedUuid = automaticallyGeneratedUuid;
+    }
+
+    // InitialSetup's @PostConstruct usually wins the race against ours and would persist a
+    // per-node UUID before we can refuse; a BeanFactoryPostProcessor runs before either of them.
+    @Bean
+    static BeanFactoryPostProcessor clusterSharedSecretGuard(Environment environment) {
+        return beanFactory -> {
+            // InitialSetup is the only thing that mints per-node UUIDs; without it there is
+            // nothing to pre-empt, and slice tests that wire ClusterConfig alone stay usable.
+            if (!beanFactory.containsBeanDefinition(INITIAL_SETUP_BEAN)) {
+                return;
+            }
+            validateSharedCryptoMaterial(
+                    environment.getProperty("cluster.enabled", Boolean.class, false),
+                    environment.getProperty("AutomaticallyGenerated.key", ""),
+                    environment.getProperty("AutomaticallyGenerated.UUID", ""));
+        };
+    }
+
+    /** Cluster nodes derive metadata encryption and licence HMAC keys from these two values. */
+    static void validateSharedCryptoMaterial(boolean clusterEnabled, String key, String uuid) {
+        if (!clusterEnabled) {
+            return;
+        }
+        requireSharedUuid("AutomaticallyGenerated.key", key);
+        requireSharedUuid("AutomaticallyGenerated.UUID", uuid);
+    }
+
+    // InitialSetup replaces any non-UUID value, so only a valid UUID survives startup unchanged.
+    private static void requireSharedUuid(String property, String value) {
+        if (!GeneralUtils.isValidUUID(value)) {
+            throw new IllegalStateException(property + SHARED_SECRET_MESSAGE_SUFFIX);
+        }
+    }
 
     @PostConstruct
     void validate() {
@@ -29,6 +89,7 @@ public class ClusterConfig {
         if (!cluster.isEnabled()) {
             return;
         }
+        validateSharedCryptoMaterial(true, automaticallyGeneratedKey, automaticallyGeneratedUuid);
         String backplane = cluster.getBackplane();
         if ("valkey".equalsIgnoreCase(backplane)) {
             // getValkey() re-seeds a null block, so an absent 'valkey:' reads as a missing url.

app/common/src/test/java/stirling/software/common/cluster/ClusterConfigValidationTest.java

@@ -1,5 +1,6 @@
 package stirling.software.common.cluster;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -12,6 +13,11 @@ import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
 
 import stirling.software.common.model.ApplicationProperties;
 import stirling.software.common.model.ApplicationProperties.Cluster;
@@ -19,10 +25,19 @@ import stirling.software.common.model.ApplicationProperties.Cluster.Valkey;
 
 class ClusterConfigValidationTest {
 
+    /** Stand-ins for the shared AutomaticallyGenerated values every node must be given. */
+    private static final String SHARED_KEY = "11111111-1111-1111-1111-111111111111";
+
+    private static final String SHARED_UUID = "22222222-2222-2222-2222-222222222222";
+
+    private static ClusterConfig config(ApplicationProperties props) {
+        return new ClusterConfig(props, SHARED_KEY, SHARED_UUID);
+    }
+
     @Test
     void validationPassesWhenDisabled() {
         ApplicationProperties props = new ApplicationProperties();
-        ClusterConfig config = new ClusterConfig(props);
+        ClusterConfig config = config(props);
         assertDoesNotThrow(() -> invokeValidate(config));
     }
 
@@ -32,7 +47,7 @@ class ClusterConfigValidationTest {
         Cluster cluster = props.getCluster();
         cluster.setEnabled(true);
         cluster.setBackplane("valkey");
-        ClusterConfig config = new ClusterConfig(props);
+        ClusterConfig config = config(props);
         assertThrows(IllegalStateException.class, () -> invokeValidate(config));
     }
 
@@ -44,7 +59,7 @@ class ClusterConfigValidationTest {
         cluster.setEnabled(true);
         cluster.setBackplane("valkey");
         cluster.getValkey().setUrl("redis://localhost:6379");
-        ClusterConfig config = new ClusterConfig(props);
+        ClusterConfig config = config(props);
         assertDoesNotThrow(() -> invokeValidate(config));
     }
 
@@ -54,7 +69,7 @@ class ClusterConfigValidationTest {
         Cluster cluster = props.getCluster();
         cluster.setEnabled(true);
         cluster.setBackplane("inprocess");
-        ClusterConfig config = new ClusterConfig(props);
+        ClusterConfig config = config(props);
         assertDoesNotThrow(() -> invokeValidate(config));
     }
 
@@ -265,12 +280,12 @@ class ClusterConfigValidationTest {
         }
 
         private void assertPasses() {
-            ClusterConfig config = new ClusterConfig(props);
+            ClusterConfig config = config(props);
             assertDoesNotThrow(() -> invokeValidate(config));
         }
 
         private void assertMessage(String... expectedSubstrings) {
-            ClusterConfig config = new ClusterConfig(props);
+            ClusterConfig config = config(props);
             IllegalStateException ex =
                     assertThrows(IllegalStateException.class, () -> invokeValidate(config));
             for (String expected : expectedSubstrings) {
@@ -327,7 +342,7 @@ class ClusterConfigValidationTest {
             v.setUrl("redis://valkey:6379");
             v.setPool(null);
             v.setTls(null);
-            ClusterConfig config = new ClusterConfig(props);
+            ClusterConfig config = config(props);
             assertDoesNotThrow(() -> invokeValidate(config));
         }
 
@@ -339,7 +354,7 @@ class ClusterConfigValidationTest {
         }
 
         private void assertMessage(String expected) {
-            ClusterConfig config = new ClusterConfig(props);
+            ClusterConfig config = config(props);
             IllegalStateException ex =
                     assertThrows(IllegalStateException.class, () -> invokeValidate(config));
             assertTrue(
@@ -348,6 +363,156 @@ class ClusterConfigValidationTest {
         }
     }
 
+    /** AutomaticallyGenerated.key/.UUID feed metadata encryption and licence seat HMACs. */
+    @Nested
+    @DisplayName("shared AutomaticallyGenerated key/UUID guard")
+    class SharedCryptoMaterial {
+
+        private ApplicationProperties props;
+
+        @Test
+        @DisplayName("an unset key is rejected and names the env var to set")
+        void missingKeyRejected() {
+            enabled("valkey");
+            assertMessage(
+                    new ClusterConfig(props, "", SHARED_UUID),
+                    "AutomaticallyGenerated.key",
+                    "same UUID on every node",
+                    "AUTOMATICALLYGENERATED_KEY");
+        }
+
+        @Test
+        @DisplayName("an unset UUID is rejected even when the key is set")
+        void missingUuidRejected() {
+            enabled("valkey");
+            assertMessage(
+                    new ClusterConfig(props, SHARED_KEY, null),
+                    "AutomaticallyGenerated.UUID",
+                    "AUTOMATICALLYGENERATED_UUID");
+        }
+
+        @Test
+        @DisplayName("the settings.yml.template placeholder is rejected: InitialSetup replaces it")
+        void nonUuidPlaceholderRejected() {
+            enabled("valkey");
+            assertMessage(
+                    new ClusterConfig(props, "example", "example"),
+                    "AutomaticallyGenerated.key",
+                    "must be a UUID");
+        }
+
+        @Test
+        @DisplayName("the guard also applies to backplane=inprocess")
+        void inProcessBackplaneAlsoGuarded() {
+            enabled("inprocess");
+            assertMessage(new ClusterConfig(props, "", ""), "AutomaticallyGenerated.key");
+        }
+
+        @Test
+        @DisplayName("two explicit UUIDs pass")
+        void explicitSharedValuesPass() {
+            enabled("valkey");
+            props.getCluster().getValkey().setUrl("redis://valkey:6379");
+            assertDoesNotThrow(() -> invokeValidate(config(props)));
+        }
+
+        @Test
+        @DisplayName("cluster.enabled=false never requires the shared values")
+        void disabledClusterSkipsGuard() {
+            enabled("valkey");
+            props.getCluster().setEnabled(false);
+            assertDoesNotThrow(() -> invokeValidate(new ClusterConfig(props, "", "")));
+        }
+
+        @Test
+        @DisplayName("the pre-bean guard reads the same rule as the @PostConstruct one")
+        void staticGuardMatchesPostConstructGuard() {
+            assertDoesNotThrow(() -> ClusterConfig.validateSharedCryptoMaterial(false, "", ""));
+            assertDoesNotThrow(
+                    () ->
+                            ClusterConfig.validateSharedCryptoMaterial(
+                                    true, SHARED_KEY, SHARED_UUID));
+            assertThrows(
+                    IllegalStateException.class,
+                    () -> ClusterConfig.validateSharedCryptoMaterial(true, SHARED_KEY, ""));
+        }
+
+        private void enabled(String backplane) {
+            props = new ApplicationProperties();
+            props.getCluster().setEnabled(true);
+            props.getCluster().setBackplane(backplane);
+        }
+
+        private void assertMessage(ClusterConfig config, String... expectedSubstrings) {
+            IllegalStateException ex =
+                    assertThrows(IllegalStateException.class, () -> invokeValidate(config));
+            for (String expected : expectedSubstrings) {
+                assertTrue(
+                        ex.getMessage().contains(expected),
+                        "message must contain '" + expected + "'; got: " + ex.getMessage());
+            }
+        }
+    }
+
+    /** The guard must abort the context before InitialSetup's @PostConstruct can generate one. */
+    @Nested
+    @DisplayName("pre-bean shared key/UUID guard")
+    class SharedCryptoMaterialBootGuard {
+
+        private final ApplicationContextRunner runner =
+                new ApplicationContextRunner()
+                        .withConfiguration(
+                                AutoConfigurations.of(PropertyPlaceholderAutoConfiguration.class))
+                        .withUserConfiguration(TestAppPropertiesConfig.class, ClusterConfig.class)
+                        .withPropertyValues("cluster.enabled=true", "cluster.backplane=inprocess");
+
+        @Test
+        @DisplayName("boot fails before any bean is created when the shared values are missing")
+        void bootFailsWhenSharedValuesMissing() {
+            runner.withUserConfiguration(StubInitialSetupConfig.class)
+                    .run(
+                            context ->
+                                    assertThat(context)
+                                            .getFailure()
+                                            .hasMessageContaining("AutomaticallyGenerated.key")
+                                            .hasMessageContaining("AUTOMATICALLYGENERATED_KEY"));
+        }
+
+        @Test
+        @DisplayName("boot succeeds once both values are configured")
+        void bootSucceedsWhenSharedValuesConfigured() {
+            runner.withUserConfiguration(StubInitialSetupConfig.class)
+                    .withPropertyValues(
+                            "AutomaticallyGenerated.key=" + SHARED_KEY,
+                            "AutomaticallyGenerated.UUID=" + SHARED_UUID)
+                    .run(context -> assertThat(context).hasNotFailed());
+        }
+
+        @Test
+        @DisplayName("no InitialSetup in the context means nothing mints a per-node UUID")
+        void guardSkippedWithoutInitialSetup() {
+            runner.run(context -> assertThat(context).hasNotFailed());
+        }
+    }
+
+    /** Defaults-only bean: the production class loads YAML in {@code @PostConstruct}. */
+    @Configuration
+    static class TestAppPropertiesConfig {
+        @Bean
+        ApplicationProperties applicationProperties() {
+            return new ApplicationProperties();
+        }
+    }
+
+    /** Stands in for :core's InitialSetup, which the guard keys off by bean name. */
+    @Configuration
+    static class StubInitialSetupConfig {
+        @Bean(name = "initialSetup")
+        Object initialSetup() {
+            return new Object();
+        }
+    }
+
     private void invokeValidate(ClusterConfig config) throws Exception {
         Method m = ClusterConfig.class.getDeclaredMethod("validate");
         m.setAccessible(true);

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

@@ -329,9 +329,11 @@ metrics:
   enabled: true # 'true' to enable Info APIs (`/api/*`) endpoints, 'false' to disable
 
 # Automatically Generated Settings (Do Not Edit Directly)
+# Generated on first boot and written back here. In a cluster EVERY node must carry the SAME key and
+# UUID (env: AUTOMATICALLYGENERATED_KEY / AUTOMATICALLYGENERATED_UUID) or nodes cannot read each other's data.
 AutomaticallyGenerated:
-  key: example
-  UUID: example
+  key: example # AES key for stored metadata. Differs per node = one node cannot decrypt another's rows.
+  UUID: example # Part of the licence-seat HMAC. Differs per node = peers reject each other's seat count.
   appVersion: 0.35.0
 
 processExecutor:
@@ -463,7 +465,9 @@ cluster:
   # The Valkey/Redis server every Stirling PDF node shares.
   valkey:
     mode: "" # 'standalone' | 'sentinel' | 'cluster'. Blank = inferred from whichever block below is filled in.
-    url: "" # e.g. 'redis://valkey:6379'. STANDALONE ONLY - sentinel and cluster ignore it, so a 'rediss://' url there refuses to boot unless tls.enabled is true.
+    url: "" # e.g. 'redis://valkey:6379', optionally '/2' to pick a database. STANDALONE ONLY - sentinel and cluster ignore it, so a 'rediss://' url there refuses to boot unless tls.enabled is true.
+    # Two Stirling deployments sharing one Valkey MUST use different database indexes: there is no key
+    # prefix, so on the same index they merge each other's node registries and cross-route job traffic.
     username: "" # Valkey ACL user. Overrides any user in 'url'.
     password: "" # Valkey password. Overrides any password in 'url'. Never logged.
     commandTimeoutMs: 2000 # Per-command timeout, so a slow Valkey cannot stall request threads.

app/proprietary/src/main/java/stirling/software/proprietary/cluster/valkey/ValkeyBackplaneHealthIndicator.java

@@ -0,0 +1,33 @@
+package stirling.software.proprietary.cluster.valkey;
+
+import org.springframework.boot.health.contributor.Health;
+import org.springframework.boot.health.contributor.HealthIndicator;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Publishes backplane reachability as the {@code valkeyBackplane} contributor of {@code
+ * /actuator/health}. Without it a node whose Valkey is dead still reports UP, so a load balancer
+ * keeps sending it traffic it cannot serve.
+ */
+@Component
+@RequiredArgsConstructor
+@ConditionalOnValkeyBackplane
+public class ValkeyBackplaneHealthIndicator implements HealthIndicator {
+
+    private final ValkeyClusterBackplane backplane;
+
+    @Override
+    public Health health() {
+        try {
+            Health.Builder builder = backplane.isHealthy() ? Health.up() : Health.down();
+            return builder.withDetail("backplane", backplane.backplaneType())
+                    .withDetail("nodeId", backplane.localNodeId())
+                    .build();
+        } catch (RuntimeException ex) {
+            // isHealthy() swallows its own probe errors; this covers node-id resolution failing.
+            return Health.down(ex).withDetail("backplane", backplane.backplaneType()).build();
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/cluster/valkey/ValkeyConnectionConfiguration.java

@@ -145,12 +145,22 @@ public class ValkeyConnectionConfiguration {
                             + " cluster.valkey.username/password (and sentinel.password) instead",
                     mode);
         }
+        int database = databaseOrZero(uri.getPath());
+        if (database != 0) {
+            log.warn(
+                    "cluster.valkey.url selects database {} but mode={} ignores the url - this"
+                            + " deployment will use database 0",
+                    database,
+                    mode);
+        }
     }
 
     static RedisStandaloneConfiguration standaloneConfiguration(
             Endpoint endpoint, String username, String password) {
         RedisStandaloneConfiguration cfg =
                 new RedisStandaloneConfiguration(endpoint.host(), endpoint.port());
+        // The database is the only isolation between two deployments sharing one Valkey.
+        cfg.setDatabase(endpoint.database());
         applyAuth(cfg, username, password);
         return cfg;
     }
@@ -197,30 +207,38 @@ public class ValkeyConnectionConfiguration {
     }
 
     /** Parsed connection endpoint; username/password are null when absent. */
-    record Endpoint(String host, int port, boolean tls, String username, String password) {}
+    record Endpoint(
+            String host, int port, boolean tls, String username, String password, int database) {}
 
     /**
      * Reserved chars in the password ({@code @ : / # ?}) must be percent-encoded - {@link URI}
      * otherwise parses them structurally (e.g. {@code #} starts the fragment).
      */
-    static Endpoint parseUrl(String url) {
-        if (url == null || url.isBlank()) {
+    static Endpoint parseUrl(String rawUrl) {
+        if (rawUrl == null || rawUrl.isBlank()) {
             throw new IllegalStateException("cluster.valkey.url must be set when backplane=valkey");
         }
+        // A .env line or YAML block scalar leaves a trailing newline that URI rejects.
+        String url = rawUrl.trim();
         URI uri;
         try {
             uri = new URI(url);
         } catch (URISyntaxException ex) {
+            // Never attach ex: its own message echoes the url, credentials included.
             throw new IllegalStateException(
-                    "cluster.valkey.url is not a valid URI: " + url + " (" + ex.getMessage() + ")",
-                    ex);
+                    "cluster.valkey.url is not a valid URI: "
+                            + redactUserInfo(url)
+                            + " ("
+                            + ex.getReason()
+                            + (ex.getIndex() >= 0 ? " at index " + ex.getIndex() : "")
+                            + ")");
         }
         String host = uri.getHost();
         if (host == null || host.isBlank()) {
             throw new IllegalStateException(
                     "cluster.valkey.url has no host: "
-                            + url
-                            + " (expected redis://[user:password@]host[:port])");
+                            + redactUserInfo(url)
+                            + " (expected redis://[user:password@]host[:port][/database])");
         }
         boolean tls = "rediss".equalsIgnoreCase(uri.getScheme());
         int port = uri.getPort() <= 0 ? 6379 : uri.getPort();
@@ -236,7 +254,69 @@ public class ValkeyConnectionConfiguration {
                 password = parts[0];
             }
         }
-        return new Endpoint(host, port, tls, username, password);
+        int database;
+        try {
+            database = parseDatabaseSegment(uri.getPath());
+        } catch (NumberFormatException ex) {
+            throw new IllegalStateException(
+                    "cluster.valkey.url has an invalid database index: "
+                            + redactUserInfo(url)
+                            + " (the path must be a non-negative integer, e.g."
+                            + " redis://valkey:6379/2)");
+        }
+        return new Endpoint(host, port, tls, username, password, database);
+    }
+
+    /**
+     * Spring Boot's {@code spring.data.redis.url} reads the path as the database index, so an
+     * operator copying that syntax must not be silently downgraded to database 0.
+     */
+    private static int parseDatabaseSegment(String path) {
+        if (path == null || path.isBlank() || "/".equals(path)) {
+            return 0;
+        }
+        int database = Integer.parseInt(path.startsWith("/") ? path.substring(1) : path);
+        if (database < 0) {
+            throw new NumberFormatException("negative database index " + database);
+        }
+        return database;
+    }
+
+    /** Lenient variant: the ignored-url guard must never fail boot over an unused database. */
+    private static int databaseOrZero(String path) {
+        try {
+            return parseDatabaseSegment(path);
+        } catch (NumberFormatException ex) {
+            return 0;
+        }
+    }
+
+    /**
+     * Masks {@code user:password@} so a url can be named in an error. Bounded to the authority: an
+     * unbounded lastIndexOf('@') would mangle {@code redis://valkey:6379/0?tag=a@b}.
+     */
+    static String redactUserInfo(String url) {
+        if (url == null) {
+            return null;
+        }
+        int slashes = url.indexOf("//");
+        if (slashes < 0) {
+            return url;
+        }
+        int start = slashes + 2;
+        int end = url.length();
+        for (int i = start; i < url.length(); i++) {
+            char c = url.charAt(i);
+            if (c == '/' || c == '?' || c == '#') {
+                end = i;
+                break;
+            }
+        }
+        int at = end == start ? -1 : url.lastIndexOf('@', end - 1);
+        if (at < start) {
+            return url;
+        }
+        return url.substring(0, start) + "****" + url.substring(at);
     }
 
     /**
@@ -365,7 +445,12 @@ public class ValkeyConnectionConfiguration {
 
     static String describeTarget(Valkey.ValkeyMode mode, Endpoint endpoint, Valkey valkey) {
         return switch (mode) {
-            case STANDALONE -> endpoint.host() + ":" + endpoint.port();
+            // Database only when non-zero: the boot log stays host:port for the common case.
+            case STANDALONE ->
+                    endpoint.host()
+                            + ":"
+                            + endpoint.port()
+                            + (endpoint.database() == 0 ? "" : "/" + endpoint.database());
             case SENTINEL ->
                     "sentinels="
                             + String.join(",", valkey.getSentinel().getNodes())

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

@@ -84,6 +84,12 @@ import stirling.software.proprietary.policy.trigger.PolicyTriggerManager;
 import stirling.software.proprietary.policy.trigger.TriggerInfo;
 import stirling.software.proprietary.util.SecretMasker;
 
+import tools.jackson.core.JsonParser;
+import tools.jackson.databind.DeserializationContext;
+import tools.jackson.databind.JsonNode;
+import tools.jackson.databind.ValueDeserializer;
+import tools.jackson.databind.annotation.JsonDeserialize;
+
 /**
  * Policy CRUD plus pipeline runs (stored or ad-hoc). Runs are async: returns a run id, poll {@code
  * GET /run/{runId}} for status, download outputs via {@code GET /api/v1/general/files/{fileId}}.
@@ -271,8 +277,19 @@ public class PolicyController {
             summary = "Create or update a policy",
             description =
                     "Stores a policy (trigger config + steps + output + metadata). A blank id is"
-                            + " assigned; returns the stored policy with its id.")
-    public ResponseEntity<Policy> savePolicy(@RequestBody Policy policy) {
+                            + " assigned; returns the stored policy with its id. Sources and their"
+                            + " triggers live in 'inputs'; the pre-inputs body shape is rejected"
+                            + " rather than silently bound to nothing.")
+    public ResponseEntity<Policy> savePolicy(@RequestBody PolicySaveRequest request) {
+        requireInputsShape(request.legacyKeys());
+        return savePolicy(request.policy());
+    }
+
+    /**
+     * The save itself, once the body's shape is known good. Also the entry point for callers that
+     * already hold a typed {@link Policy}.
+     */
+    public ResponseEntity<Policy> savePolicy(Policy policy) {
         requirePolicyEditingAllowed();
         Policy owned = withStoredOutputSecrets(resolveOwnership(policy));
         requireAccessibleSources(owned);
@@ -310,6 +327,51 @@ public class PolicyController {
         return ResponseEntity.noContent().build();
     }
 
+    /**
+     * Reject the pre-inputs body shape rather than binding it to nothing: {@code sourceIds} and a
+     * populated {@code trigger} have no field on {@link Policy}, so Jackson drops them and the save
+     * would store a policy that references and watches nothing. Stored blobs in that shape are
+     * migrated on read by the policy store; an inbound request has no such path.
+     */
+    private static void requireInputsShape(List<String> legacyKeys) {
+        if (legacyKeys.isEmpty()) {
+            return;
+        }
+        throw new ResponseStatusException(
+                HttpStatus.BAD_REQUEST,
+                "Unsupported policy field(s): "
+                        + String.join(", ", legacyKeys)
+                        + ". Use 'inputs': sourceIds and trigger moved onto each input.");
+    }
+
+    /**
+     * The save body: the bound policy plus whichever dead pre-inputs keys it still carries. Binding
+     * straight to {@link Policy} would drop them silently, since unknown properties are ignored.
+     */
+    @JsonDeserialize(using = PolicySaveRequest.Deserializer.class)
+    public record PolicySaveRequest(Policy policy, List<String> legacyKeys) {
+
+        /** Reads the policy off the raw body, noting the dead top-level keys it carries. */
+        static final class Deserializer extends ValueDeserializer<PolicySaveRequest> {
+
+            @Override
+            public PolicySaveRequest deserialize(JsonParser parser, DeserializationContext ctxt) {
+                JsonNode root = ctxt.readTree(parser);
+                List<String> legacyKeys = new ArrayList<>();
+                // A null trigger drops nothing, and the editor and portal still send one on
+                // every save.
+                if (root.hasNonNull("trigger")) {
+                    legacyKeys.add("trigger");
+                }
+                if (root.has("sourceIds")) {
+                    legacyKeys.add("sourceIds");
+                }
+                return new PolicySaveRequest(
+                        ctxt.readTreeAsValue(root, Policy.class), List.copyOf(legacyKeys));
+            }
+        }
+    }
+
     /**
      * Every {@code sourceId} a policy references must resolve to a source in the caller's team, so
      * a client can neither reference a non-existent source nor reach across teams to use another

app/proprietary/src/main/java/stirling/software/proprietary/workflow/service/MetadataEncryptionService.java

@@ -109,7 +109,7 @@ public class MetadataEncryptionService {
 
     /**
      * Reverses {@link #encryptBytes}. Also accepts legacy values stored as plain Base64 before
-     * encryption was introduced — {@link #decrypt} returns those unchanged, so they still decode.
+     * encryption was introduced - {@link #decrypt} returns those unchanged, so they still decode.
      */
     public byte[] decryptBytes(String stored) {
         if (stored == null) {
@@ -124,7 +124,7 @@ public class MetadataEncryptionService {
         String rawKey = applicationProperties.getAutomaticallyGenerated().getKey();
         if (rawKey == null || rawKey.isBlank()) {
             throw new IllegalStateException(
-                    "AutomaticallyGenerated.key is not initialised — cannot derive encryption key");
+                    "AutomaticallyGenerated.key is not initialised - cannot derive encryption key");
         }
         // SHA-256 of the raw key gives a stable 32-byte AES-256 key
         byte[] hash =

app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/LiveValkeySentinelModeTest.java

@@ -0,0 +1,241 @@
+package stirling.software.proprietary.cluster.valkey;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIf;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import io.lettuce.core.RedisClient;
+
+import stirling.software.common.cluster.ClusterNode;
+import stirling.software.common.cluster.JobStoreEntry;
+import stirling.software.common.model.ApplicationProperties;
+
+// Lettuce takes the primary's address from the SENTINEL'S REPLY, not from the seed URI, so the
+// announced address must resolve identically inside the container and on the host.
+// Primary and sentinel therefore share one container (localhost = the primary) and both ports are
+// published host:container identically, making "127.0.0.1:<port>" true on both sides.
+@Testcontainers
+@EnabledIf("isDockerAvailable")
+class LiveValkeySentinelModeTest {
+
+    private static final String NODE_ID = "sentinel-node";
+    private static final String MASTER = "mymaster";
+    private static final String RUN = UUID.randomUUID().toString().substring(0, 8);
+
+    private static final int PRIMARY_PORT;
+    private static final int SENTINEL_PORT;
+
+    static {
+        int[] ports = reserveTwoPorts();
+        PRIMARY_PORT = ports[0];
+        SENTINEL_PORT = ports[1];
+    }
+
+    // Lifecycle is manual: SENTINEL MONITOR must name a host-reachable address, and the sentinel
+    // must have seen the primary, before any client connects.
+    static final GenericContainer<?> VALKEY =
+            new GenericContainer<>(DockerImageName.parse("valkey/valkey:8.0-alpine"))
+                    .withExposedPorts(PRIMARY_PORT, SENTINEL_PORT)
+                    .withCommand("sh", "-c", bootstrapScript());
+
+    static boolean isDockerAvailable() {
+        return DockerClientFactory.instance().isDockerAvailable();
+    }
+
+    private static LettuceConnectionFactory factory;
+    private static StringRedisTemplate template;
+
+    @BeforeAll
+    static void monitorPrimaryAndConnect() throws Exception {
+        VALKEY.setPortBindings(
+                List.of(PRIMARY_PORT + ":" + PRIMARY_PORT, SENTINEL_PORT + ":" + SENTINEL_PORT));
+        VALKEY.start();
+        String host = VALKEY.getHost();
+        String announceIp = "localhost".equalsIgnoreCase(host) ? "127.0.0.1" : host;
+
+        // resolve-hostnames stays off (set in the config): a hostname whose record disappears
+        // stalls sentinel's event loop, and Lettuce would then get an address it cannot use.
+        cli("sentinel", "monitor", MASTER, announceIp, String.valueOf(PRIMARY_PORT), "1");
+        awaitPrimaryMonitored();
+
+        ApplicationProperties p = new ApplicationProperties();
+        p.getCluster().setEnabled(true);
+        p.getCluster().setBackplane("valkey");
+        p.getCluster().getNode().setId(NODE_ID);
+        var valkey = p.getCluster().getValkey();
+        valkey.setMode("sentinel");
+        valkey.getSentinel().setMaster(MASTER);
+        valkey.getSentinel().setNodes(List.of(announceIp + ":" + SENTINEL_PORT));
+        factory = new ValkeyConnectionConfiguration(p).valkeyConnectionFactory();
+        template = new StringRedisTemplate(factory);
+    }
+
+    @AfterAll
+    static void disconnect() {
+        if (factory != null) {
+            factory.destroy();
+        }
+        VALKEY.stop();
+    }
+
+    /** Primary in the background, sentinel in the foreground: same netns, so localhost is both. */
+    private static String bootstrapScript() {
+        return "valkey-server --port "
+                + PRIMARY_PORT
+                + " --save '' --appendonly no --daemonize yes"
+                + " && printf 'port "
+                + SENTINEL_PORT
+                + "\\ndir /tmp\\nsentinel resolve-hostnames no\\n' > /tmp/sentinel.conf"
+                + " && exec valkey-sentinel /tmp/sentinel.conf";
+    }
+
+    /** Both sockets are held open at once so the two ports cannot collide. */
+    private static int[] reserveTwoPorts() {
+        try (ServerSocket first = new ServerSocket(0);
+                ServerSocket second = new ServerSocket(0)) {
+            return new int[] {first.getLocalPort(), second.getLocalPort()};
+        } catch (IOException ex) {
+            throw new IllegalStateException("could not reserve local ports for valkey", ex);
+        }
+    }
+
+    private static Container.ExecResult cli(String... args) throws Exception {
+        String[] cmd = new String[args.length + 3];
+        cmd[0] = "valkey-cli";
+        cmd[1] = "-p";
+        cmd[2] = String.valueOf(SENTINEL_PORT);
+        System.arraycopy(args, 0, cmd, 3, args.length);
+        Container.ExecResult res = VALKEY.execInContainer(cmd);
+        assertEquals(
+                0,
+                res.getExitCode(),
+                "valkey-cli " + String.join(" ", args) + " failed: " + res.getStderr());
+        return res;
+    }
+
+    /** SENTINEL MONITOR returns OK before sentinel has reached the primary; wait for both. */
+    private static void awaitPrimaryMonitored() throws Exception {
+        long deadline = System.currentTimeMillis() + 60_000;
+        String last = "";
+        while (System.currentTimeMillis() < deadline) {
+            last = cli("sentinel", "master", MASTER).getStdout();
+            boolean quorum = cli("sentinel", "ckquorum", MASTER).getStdout().startsWith("OK");
+            if (quorum && !last.contains("_down") && last.contains(String.valueOf(PRIMARY_PORT))) {
+                return;
+            }
+            Thread.sleep(250);
+        }
+        throw new IllegalStateException(
+                "sentinel never saw the primary; last SENTINEL MASTER:\n" + last);
+    }
+
+    @Test
+    @DisplayName("sentinel mode wires a standalone client against the sentinel-resolved primary")
+    void sentinelModeUsesSentinelAwareFactory() {
+        assertTrue(factory.isRedisSentinelAware(), "the factory must hold a sentinel config");
+        assertInstanceOf(RedisClient.class, factory.getNativeClient());
+    }
+
+    @Test
+    @DisplayName("backplane health probe succeeds through sentinel")
+    void backplaneHealthyThroughSentinel() {
+        ApplicationProperties p = new ApplicationProperties();
+        p.getCluster().getNode().setId(NODE_ID);
+        ValkeyClusterBackplane bp = new ValkeyClusterBackplane(p, template);
+        assertEquals("valkey", bp.backplaneType());
+        assertTrue(bp.isHealthy(), "sentinel must resolve a writable primary");
+    }
+
+    @Test
+    @DisplayName("JobStore round-trips through the sentinel-resolved primary")
+    void jobStoreRoundTripsThroughSentinel() throws Exception {
+        ValkeyJobStore store = new ValkeyJobStore(template);
+        String jobId = "sent-job-" + RUN;
+        String fileId = "sent-file-" + RUN;
+
+        store.put(
+                new JobStoreEntry(
+                        jobId,
+                        JobStoreEntry.JobState.RUNNING,
+                        NODE_ID,
+                        Instant.now(),
+                        null,
+                        null,
+                        List.of(fileId),
+                        Map.of("k", "v")),
+                Duration.ofSeconds(60));
+
+        Optional<JobStoreEntry> seen = store.get(jobId);
+        assertTrue(seen.isPresent(), "the write must land on the primary, not a read-only replica");
+        assertEquals(NODE_ID, seen.get().owningNodeId());
+        assertEquals(jobId, store.findJobIdByFileId(fileId).orElse(null));
+        assertTrue(
+                template.getExpire("stirling:job:" + jobId, java.util.concurrent.TimeUnit.SECONDS)
+                        > 0,
+                "the single-key script must arm the TTL through sentinel too");
+        // Proves the sentinel-announced address is the container's primary, not some other server.
+        assertEquals(
+                "1",
+                onPrimary("exists", "stirling:job:" + jobId).getStdout().trim(),
+                "the key must exist on the primary sentinel monitors");
+
+        store.delete(jobId);
+        assertFalse(store.exists(jobId));
+        assertFalse(store.findJobIdByFileId(fileId).isPresent());
+    }
+
+    @Test
+    @DisplayName("InstanceRegistry register/lookup/deregister works through sentinel")
+    void instanceRegistryThroughSentinel() {
+        ValkeyInstanceRegistry registry = new ValkeyInstanceRegistry(template);
+        String nodeId = "sent-node-" + RUN;
+        registry.register(
+                new ClusterNode(nodeId, "10.0.0.9:8080", Instant.now(), "BOTH"),
+                Duration.ofSeconds(60));
+
+        Optional<ClusterNode> looked = registry.lookup(nodeId);
+        assertTrue(looked.isPresent());
+        assertEquals("10.0.0.9:8080", looked.get().internalAddress());
+        assertTrue(registry.activeNodes().stream().anyMatch(n -> nodeId.equals(n.nodeId())));
+        assertTrue(
+                template.getExpire(
+                                "stirling:nodes:" + nodeId, java.util.concurrent.TimeUnit.SECONDS)
+                        > 0,
+                "a node hash without a TTL would mask a dead node as alive forever");
+
+        registry.deregister(nodeId);
+        assertFalse(registry.lookup(nodeId).isPresent());
+    }
+
+    private static Container.ExecResult onPrimary(String... args) throws Exception {
+        String[] cmd = new String[args.length + 3];
+        cmd[0] = "valkey-cli";
+        cmd[1] = "-p";
+        cmd[2] = String.valueOf(PRIMARY_PORT);
+        System.arraycopy(args, 0, cmd, 3, args.length);
+        return VALKEY.execInContainer(cmd);
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/ValkeyClusterBackplaneTest.java

@@ -1,5 +1,7 @@
 package stirling.software.proprietary.cluster.valkey;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
@@ -11,7 +13,12 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
+import org.springframework.boot.health.contributor.Health;
+import org.springframework.boot.health.contributor.Status;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.data.redis.RedisConnectionFailureException;
 import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.StringRedisTemplate;
 
@@ -42,11 +49,27 @@ class ValkeyClusterBackplaneTest {
         verify(template, never()).getConnectionFactory();
     }
 
+    // Replaces an assertion on a null hasKey() reply: Lettuce only returns null while queueing
+    // inside a pipeline or MULTI, and this template is neither, so that branch was unreachable.
     @Test
-    @DisplayName("unhealthy when the probe yields no reply at all")
-    void isHealthy_returnsFalseOnNullReply() {
+    @DisplayName("healthy whichever boolean the probe returns - only the round trip matters")
+    void isHealthy_ignoresProbeReplyValue() {
+        StringRedisTemplate exists = mock(StringRedisTemplate.class);
+        when(exists.hasKey(anyString())).thenReturn(Boolean.TRUE);
+        assertTrue(backplane(exists).isHealthy());
+
+        StringRedisTemplate missing = mock(StringRedisTemplate.class);
+        when(missing.hasKey(anyString())).thenReturn(Boolean.FALSE);
+        assertTrue(backplane(missing).isHealthy());
+    }
+
+    @Test
+    @DisplayName("unhealthy when the connection to Valkey is down")
+    void isHealthy_returnsFalseOnConnectionFailure() {
         StringRedisTemplate template = mock(StringRedisTemplate.class);
-        when(template.hasKey(anyString())).thenReturn(null);
+        // The exception a real outage surfaces, rather than a synthetic RuntimeException.
+        when(template.hasKey(anyString()))
+                .thenThrow(new RedisConnectionFailureException("connection refused"));
 
         assertFalse(backplane(template).isHealthy());
     }
@@ -64,4 +87,88 @@ class ValkeyClusterBackplaneTest {
     void shouldRunLocalCleanup_returnsFalse_valkeyOwnsTtlEviction() {
         assertFalse(backplane(mock(StringRedisTemplate.class)).shouldRunLocalCleanup());
     }
+
+    @Nested
+    @DisplayName("health indicator")
+    class BackplaneHealthIndicator {
+
+        private final ApplicationContextRunner runner =
+                new ApplicationContextRunner()
+                        .withUserConfiguration(ValkeyBackplaneHealthIndicator.class);
+
+        private static Health health(StringRedisTemplate template) {
+            return new ValkeyBackplaneHealthIndicator(backplane(template)).health();
+        }
+
+        @Test
+        @DisplayName("UP with backplane and node details when the probe answers")
+        void reportsUp() {
+            StringRedisTemplate template = mock(StringRedisTemplate.class);
+            when(template.hasKey(anyString())).thenReturn(Boolean.FALSE);
+
+            Health health = health(template);
+
+            assertEquals(Status.UP, health.getStatus());
+            assertEquals("valkey", health.getDetails().get("backplane"));
+            assertEquals("n-1", health.getDetails().get("nodeId"));
+        }
+
+        @Test
+        @DisplayName("DOWN when Valkey is unreachable, so the node drops out of rotation")
+        void reportsDownOnOutage() {
+            StringRedisTemplate template = mock(StringRedisTemplate.class);
+            when(template.hasKey(anyString()))
+                    .thenThrow(new RedisConnectionFailureException("connection refused"));
+
+            Health health = health(template);
+
+            assertEquals(Status.DOWN, health.getStatus());
+            assertEquals("valkey", health.getDetails().get("backplane"));
+        }
+
+        @Test
+        @DisplayName("DOWN rather than throwing when node-id resolution fails")
+        void reportsDownInsteadOfPropagating() {
+            ValkeyClusterBackplane broken = mock(ValkeyClusterBackplane.class);
+            when(broken.isHealthy()).thenReturn(true);
+            when(broken.backplaneType()).thenReturn("valkey");
+            when(broken.localNodeId()).thenThrow(new IllegalStateException("no node id"));
+
+            Health health = new ValkeyBackplaneHealthIndicator(broken).health();
+
+            assertEquals(Status.DOWN, health.getStatus());
+        }
+
+        @Test
+        @DisplayName("not registered on a single-node install")
+        void absentWhenClusterDisabled() {
+            runner.run(
+                    context ->
+                            assertThat(context)
+                                    .doesNotHaveBean(ValkeyBackplaneHealthIndicator.class));
+        }
+
+        @Test
+        @DisplayName("not registered when clustering uses the in-process backplane")
+        void absentWhenBackplaneIsInProcess() {
+            runner.withPropertyValues("cluster.enabled=true", "cluster.backplane=inprocess")
+                    .run(
+                            context ->
+                                    assertThat(context)
+                                            .doesNotHaveBean(ValkeyBackplaneHealthIndicator.class));
+        }
+
+        @Test
+        @DisplayName("registered when the Valkey backplane is active")
+        void presentWhenValkeyBackplaneActive() {
+            runner.withPropertyValues("cluster.enabled=true", "cluster.backplane=valkey")
+                    .withBean(
+                            ValkeyClusterBackplane.class,
+                            () -> backplane(mock(StringRedisTemplate.class)))
+                    .run(
+                            context ->
+                                    assertThat(context)
+                                            .hasSingleBean(ValkeyBackplaneHealthIndicator.class));
+        }
+    }
 }

app/proprietary/src/test/java/stirling/software/proprietary/cluster/valkey/ValkeyConnectionConfigurationTest.java

@@ -510,19 +510,31 @@ class ValkeyConnectionConfigurationTest {
         void hostPortAndCredentials() {
             RedisStandaloneConfiguration cfg =
                     ValkeyConnectionConfiguration.standaloneConfiguration(
-                            new Endpoint("valkey", 6380, false, null, null), "alice", "s3cret");
+                            new Endpoint("valkey", 6380, false, null, null, 0), "alice", "s3cret");
             assertEquals("valkey", cfg.getHostName());
             assertEquals(6380, cfg.getPort());
             assertEquals("alice", cfg.getUsername());
             assertEquals("s3cret", passwordOf(cfg.getPassword()));
         }
 
+        @Test
+        @DisplayName("the parsed database index reaches the connection configuration")
+        void databaseIsApplied() {
+            RedisStandaloneConfiguration cfg =
+                    ValkeyConnectionConfiguration.standaloneConfiguration(
+                            new Endpoint("valkey", 6379, false, null, null, 3), null, null);
+            assertEquals(
+                    3,
+                    cfg.getDatabase(),
+                    "database 0 for every deployment merges their node registries");
+        }
+
         @Test
         @DisplayName("null credentials leave the configuration unauthenticated")
         void nullCredentialsLeaveNoAuth() {
             RedisStandaloneConfiguration cfg =
                     ValkeyConnectionConfiguration.standaloneConfiguration(
-                            new Endpoint("valkey", 6379, false, "from-url", "from-url-pw"),
+                            new Endpoint("valkey", 6379, false, "from-url", "from-url-pw", 0),
                             null,
                             null);
             assertNull(cfg.getUsername());
@@ -801,6 +813,15 @@ class ValkeyConnectionConfigurationTest {
                     withUrl("redis://valkey:6379 with spaces"), ValkeyMode.CLUSTER, false);
         }
 
+        @Test
+        @DisplayName("a database index in an ignored url warns but still boots")
+        void databaseInIgnoredUrlOnlyWarns() {
+            ValkeyConnectionConfiguration.guardIgnoredUrl(
+                    withUrl("redis://valkey:6379/2"), ValkeyMode.CLUSTER, false);
+            ValkeyConnectionConfiguration.guardIgnoredUrl(
+                    withUrl("redis://valkey:6379/notadb"), ValkeyMode.SENTINEL, false);
+        }
+
         @Test
         @DisplayName("a blank url is nothing to guard")
         void blankUrlIsIgnored() {
@@ -819,7 +840,18 @@ class ValkeyConnectionConfigurationTest {
                     "valkey:6379",
                     ValkeyConnectionConfiguration.describeTarget(
                             ValkeyMode.STANDALONE,
-                            new Endpoint("valkey", 6379, false, null, null),
+                            new Endpoint("valkey", 6379, false, null, null, 0),
+                            new Valkey()));
+        }
+
+        @Test
+        @DisplayName("a non-zero database is surfaced so the boot log shows the isolation")
+        void standaloneWithDatabase() {
+            assertEquals(
+                    "valkey:6379/2",
+                    ValkeyConnectionConfiguration.describeTarget(
+                            ValkeyMode.STANDALONE,
+                            new Endpoint("valkey", 6379, false, null, null, 2),
                             new Valkey()));
         }
 
@@ -1013,5 +1045,100 @@ class ValkeyConnectionConfigurationTest {
             assertTrue(ex.getMessage().contains("not a valid URI"));
             assertTrue(ex.getMessage().contains("redis://ho st:6379"));
         }
+
+        @Test
+        @DisplayName("a trailing newline is trimmed, not rejected (.env / YAML block scalar)")
+        void trailingNewlineIsTrimmed() {
+            Endpoint e = ValkeyConnectionConfiguration.parseUrl("redis://valkey:6380\n");
+            assertEquals("valkey", e.host());
+            assertEquals(6380, e.port());
+        }
+
+        @Test
+        @DisplayName("surrounding whitespace is trimmed too")
+        void surroundingWhitespaceIsTrimmed() {
+            assertEquals(
+                    "valkey",
+                    ValkeyConnectionConfiguration.parseUrl("  redis://valkey:6379  ").host());
+        }
+
+        @Test
+        @DisplayName("credentials never reach the invalid-URI message")
+        void invalidUriRedactsCredentials() {
+            IllegalStateException ex =
+                    assertThrows(
+                            IllegalStateException.class,
+                            () ->
+                                    ValkeyConnectionConfiguration.parseUrl(
+                                            "redis://user:hunter2@ho st:6379"));
+            assertFalse(
+                    ex.getMessage().contains("hunter2"),
+                    "the password must never be logged; got: " + ex.getMessage());
+            assertNull(ex.getCause(), "the URISyntaxException message repeats the raw url");
+            assertTrue(ex.getMessage().contains("redis://****@ho st:6379"));
+        }
+
+        @Test
+        @DisplayName("credentials never reach the no-host message")
+        void noHostRedactsCredentials() {
+            IllegalStateException ex =
+                    assertThrows(
+                            IllegalStateException.class,
+                            () ->
+                                    ValkeyConnectionConfiguration.parseUrl(
+                                            "redis://user:hunter2@host:notaport"));
+            assertTrue(ex.getMessage().contains("has no host"));
+            assertFalse(
+                    ex.getMessage().contains("hunter2"),
+                    "the password must never be logged; got: " + ex.getMessage());
+        }
+
+        @Test
+        @DisplayName("an '@' in the query is not treated as userinfo when redacting")
+        void redactionIsBoundedToTheAuthority() {
+            assertEquals(
+                    "redis://valkey:6379/0?tag=a@b",
+                    ValkeyConnectionConfiguration.redactUserInfo("redis://valkey:6379/0?tag=a@b"));
+            assertEquals(
+                    "redis://****@valkey:6379/0?tag=a@b",
+                    ValkeyConnectionConfiguration.redactUserInfo(
+                            "redis://user:pw@valkey:6379/0?tag=a@b"));
+        }
+
+        @Test
+        @DisplayName("the path segment selects the database index")
+        void pathSelectsDatabase() {
+            assertEquals(
+                    2,
+                    ValkeyConnectionConfiguration.parseUrl("redis://valkey:6379/2").database(),
+                    "spring.data.redis.url honours the path, so operators expect this to");
+        }
+
+        @Test
+        @DisplayName("an absent path and a bare '/' both mean database 0")
+        void absentPathIsDatabaseZero() {
+            assertEquals(
+                    0, ValkeyConnectionConfiguration.parseUrl("redis://valkey:6379").database());
+            assertEquals(
+                    0, ValkeyConnectionConfiguration.parseUrl("redis://valkey:6379/").database());
+        }
+
+        @Test
+        @DisplayName("a non-numeric or negative database index is rejected, credentials redacted")
+        void invalidDatabaseIsRejected() {
+            IllegalStateException ex =
+                    assertThrows(
+                            IllegalStateException.class,
+                            () ->
+                                    ValkeyConnectionConfiguration.parseUrl(
+                                            "redis://user:hunter2@valkey:6379/notadb"));
+            assertTrue(
+                    ex.getMessage().contains("invalid database index"),
+                    "message must name the problem; got: " + ex.getMessage());
+            assertFalse(ex.getMessage().contains("hunter2"));
+            assertThrows(
+                    IllegalStateException.class,
+                    () -> ValkeyConnectionConfiguration.parseUrl("redis://valkey:6379/-1"));
+        }
     }
 }

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

@@ -0,0 +1,222 @@
+package stirling.software.proprietary.policy.controller;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.http.MediaType;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.MvcResult;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+import org.springframework.web.servlet.mvc.annotation.ResponseStatusExceptionResolver;
+import org.springframework.web.servlet.mvc.support.DefaultHandlerExceptionResolver;
+
+import stirling.software.common.cluster.JobStore;
+import stirling.software.common.cluster.inprocess.InProcessJobStore;
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.service.JobOwnershipService;
+import stirling.software.common.util.TempFileManager;
+import stirling.software.proprietary.policy.asset.PolicyAssetCleaner;
+import stirling.software.proprietary.policy.asset.PolicyAssetResolver;
+import stirling.software.proprietary.policy.config.PolicyAccessGuard;
+import stirling.software.proprietary.policy.config.PolicyManagementAuthority;
+import stirling.software.proprietary.policy.engine.PolicyRunRegistry;
+import stirling.software.proprietary.policy.engine.PolicyRunner;
+import stirling.software.proprietary.policy.engine.PolicyValidator;
+import stirling.software.proprietary.policy.ledger.ProcessedLedger;
+import stirling.software.proprietary.policy.model.Policy;
+import stirling.software.proprietary.policy.overview.PolicyOverviewService;
+import stirling.software.proprietary.policy.source.Source;
+import stirling.software.proprietary.policy.source.SourceAccessGuard;
+import stirling.software.proprietary.policy.source.SourceDocCounter;
+import stirling.software.proprietary.policy.source.SourceStore;
+import stirling.software.proprietary.policy.store.PolicyStore;
+import stirling.software.proprietary.policy.trigger.PolicyTriggerManager;
+
+/**
+ * The save endpoint's body shape. {@link Policy} has no {@code sourceIds} or {@code trigger} field,
+ * and unknown properties are ignored, so the pre-inputs body used to store a policy that referenced
+ * and watched nothing while still returning 200. It must be an error instead.
+ */
+@ExtendWith(MockitoExtension.class)
+@DisplayName("Policy save body shape")
+class PolicySaveShapeTest {
+
+    @Mock private PolicyRunner policyRunner;
+    @Mock private PolicyRunRegistry runRegistry;
+    @Mock private PolicyStore policyStore;
+    @Mock private SourceStore sourceStore;
+    @Mock private SourceAccessGuard sourceAccessGuard;
+    @Mock private SourceDocCounter docCounter;
+    @Mock private PolicyValidator policyValidator;
+    @Mock private PolicyAccessGuard policyAccessGuard;
+    @Mock private PolicyManagementAuthority policyManagementAuthority;
+    @Mock private PolicyTriggerManager policyTriggerManager;
+    @Mock private PolicyOverviewService policyOverviewService;
+    @Mock private PolicyAssetCleaner assetCleaner;
+    @Mock private PolicyAssetResolver assetResolver;
+    @Mock private ProcessedLedger processedLedger;
+    @Mock private TempFileManager tempFileManager;
+    @Mock private JobOwnershipService jobOwnershipService;
+
+    private final JobStore jobStore = new InProcessJobStore();
+    private MockMvc mockMvc;
+
+    @BeforeEach
+    void setUp() {
+        ApplicationProperties applicationProperties = new ApplicationProperties();
+        // Login off: policy editing is then open to the local operator, so the shape is what's
+        // under test rather than the role gate.
+        applicationProperties.getSecurity().setEnableLogin(false);
+        PolicyController controller =
+                new PolicyController(
+                        policyRunner,
+                        runRegistry,
+                        policyStore,
+                        sourceStore,
+                        sourceAccessGuard,
+                        docCounter,
+                        policyValidator,
+                        policyAccessGuard,
+                        policyManagementAuthority,
+                        policyTriggerManager,
+                        policyOverviewService,
+                        assetCleaner,
+                        assetResolver,
+                        processedLedger,
+                        List.of(),
+                        applicationProperties,
+                        tempFileManager,
+                        jobOwnershipService,
+                        jobStore);
+        mockMvc =
+                MockMvcBuilders.standaloneSetup(controller)
+                        // standaloneSetup's defaults don't handle ResponseStatusException.
+                        .setHandlerExceptionResolvers(
+                                new ResponseStatusExceptionResolver(),
+                                new DefaultHandlerExceptionResolver())
+                        .build();
+    }
+
+    private MvcResult save(String body) throws Exception {
+        return mockMvc.perform(
+                        post("/api/v1/policies")
+                                .contentType(MediaType.APPLICATION_JSON)
+                                .content(body))
+                .andReturn();
+    }
+
+    @Test
+    @DisplayName("the legacy sourceIds key is rejected instead of silently dropped")
+    void legacySourceIdsRejected() throws Exception {
+        MvcResult result =
+                save(
+                        """
+                        {"name":"legacy","enabled":true,"sourceIds":["src-1"],
+                         "steps":[{"operation":"/api/v1/misc/compress-pdf","parameters":{}}],
+                         "output":{"type":"inline","options":{}}}
+                        """);
+
+        assertThat(result.getResponse().getStatus()).isEqualTo(400);
+        assertThat(result.getResponse().getErrorMessage()).contains("sourceIds").contains("inputs");
+        verify(policyStore, never()).save(any());
+    }
+
+    @Test
+    @DisplayName("the legacy policy-level trigger is rejected instead of silently dropped")
+    void legacyTriggerRejected() throws Exception {
+        MvcResult result =
+                save(
+                        """
+                        {"name":"legacy","enabled":true,
+                         "trigger":{"type":"folder-watch","options":{}},
+                         "steps":[{"operation":"/api/v1/misc/compress-pdf","parameters":{}}],
+                         "output":{"type":"inline","options":{}}}
+                        """);
+
+        assertThat(result.getResponse().getStatus()).isEqualTo(400);
+        assertThat(result.getResponse().getErrorMessage()).contains("trigger").contains("inputs");
+        verify(policyStore, never()).save(any());
+    }
+
+    @Test
+    @DisplayName("the inputs shape is stored with its source reference intact")
+    void modernInputsAccepted() throws Exception {
+        Source source = new Source("src-1", "Incoming", "folder", Map.of(), true, "alice", 1L);
+        when(sourceStore.get("src-1")).thenReturn(Optional.of(source));
+        when(sourceAccessGuard.canAccess(source)).thenReturn(true);
+        when(policyStore.save(any())).thenAnswer(invocation -> invocation.getArgument(0));
+
+        mockMvc.perform(
+                        post("/api/v1/policies")
+                                .contentType(MediaType.APPLICATION_JSON)
+                                .content(
+                                        """
+                                        {"name":"modern","enabled":true,
+                                         "inputs":[{"sourceId":"src-1","trigger":null}],
+                                         "steps":[{"operation":"/api/v1/misc/compress-pdf",
+                                                   "parameters":{}}],
+                                         "output":{"type":"inline","options":{}}}
+                                        """))
+                .andExpect(status().isOk());
+
+        ArgumentCaptor<Policy> stored = ArgumentCaptor.forClass(Policy.class);
+        verify(policyStore).save(stored.capture());
+        assertThat(stored.getValue().sourceIds()).containsExactly("src-1");
+    }
+
+    @Test
+    @DisplayName("a body with neither key still saves")
+    void bodyWithNeitherKeyAccepted() throws Exception {
+        when(policyStore.save(any())).thenAnswer(invocation -> invocation.getArgument(0));
+
+        mockMvc.perform(
+                        post("/api/v1/policies")
+                                .contentType(MediaType.APPLICATION_JSON)
+                                .content(
+                                        """
+                                        {"name":"editor","enabled":true,
+                                         "steps":[{"operation":"/api/v1/misc/compress-pdf",
+                                                   "parameters":{}}],
+                                         "output":{"type":"inline","options":{}}}
+                                        """))
+                .andExpect(status().isOk());
+
+        verify(policyStore).save(any());
+    }
+
+    @Test
+    @DisplayName("a null trigger still saves - the editor and portal send one on every save")
+    void nullTriggerAccepted() throws Exception {
+        when(policyStore.save(any())).thenAnswer(invocation -> invocation.getArgument(0));
+
+        mockMvc.perform(
+                        post("/api/v1/policies")
+                                .contentType(MediaType.APPLICATION_JSON)
+                                .content(
+                                        """
+                                        {"name":"editor","enabled":true,"trigger":null,
+                                         "steps":[{"operation":"/api/v1/misc/compress-pdf",
+                                                   "parameters":{}}],
+                                         "output":{"type":"inline","options":{}}}
+                                        """))
+                .andExpect(status().isOk());
+
+        verify(policyStore).save(any());
+    }
+}

testing/compose/docker-compose-multinode.yml

@@ -81,6 +81,11 @@ x-stirling-node: &stirling-node
     # AES-256 key that encrypts stored integration/S3 secrets - must match on every node or secrets encrypted on one can't decrypt on another; boot fails if unset with cluster.enabled=true (test-only value; JWT keys persist separately in the shared DB).
     STIRLING_CREDENTIAL_ENCRYPTION_KEY: "dMobekyUEnEV7WHBah2FkbboP4Coqifd3JRXB00LiIY="
 
+    # --- Shared AutomaticallyGenerated key/UUID (REQUIRED in cluster mode) ---
+    # Left unset, each node mints its own UUID into its local settings.yml: workflow metadata encrypted on one node then fails to decrypt on another and licence-seat HMACs do not verify across nodes. Must be valid UUIDs (anything else is replaced at startup); boot fails if unset with cluster.enabled=true (test-only values).
+    AUTOMATICALLYGENERATED_KEY: "11111111-1111-1111-1111-111111111111"
+    AUTOMATICALLYGENERATED_UUID: "22222222-2222-2222-2222-222222222222"
+
     # --- Policy / processor subsystem (the thing under test) ---
     POLICIES_ENABLED: "true"
     # Let policy S3 sources/webhook-staging connections point at the in-cluster MinIO.