Add AI engine load balancer authentication

app/common/build.gradle

@@ -78,6 +78,9 @@ dependencies {
         runtimeOnly "com.stirling:jpdfium-natives-${platform}:1.0.1"
     }
 
+    // Bucket4j (local in-process token bucket for RateLimitStore default impl)
+    implementation 'com.bucket4j:bucket4j_jdk17-core:8.19.0'
+
     // ArchUnit: enforces module dependency direction (see ArchitectureTest)
     testImplementation 'com.tngtech.archunit:archunit-junit5:1.4.2'
 }

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

@@ -0,0 +1,24 @@
+package stirling.software.common.cluster;
+
+/** Health and identity facade for the active cluster backplane. */
+public interface ClusterBackplane {
+
+    /** Returns {@code true} when the backplane is reachable; used for health endpoints. */
+    boolean isHealthy();
+
+    /** Returns {@code "inprocess"} or {@code "valkey"}. */
+    String backplaneType();
+
+    /** Returns this JVM's stable node id (matches {@code Cluster.resolvedNodeId()}). */
+    String localNodeId();
+
+    /**
+     * Whether this JVM should run the local {@link
+     * stirling.software.common.service.TaskManager#cleanupOldJobs()} loop. Distributed backplanes
+     * own job expiry via their own TTL, so they should override this to return {@code false}.
+     * Defaults to {@code true} so in-process behavior is preserved without an explicit override.
+     */
+    default boolean shouldRunLocalCleanup() {
+        return true;
+    }
+}

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

@@ -0,0 +1,63 @@
+package stirling.software.common.cluster;
+
+import org.springframework.context.annotation.Configuration;
+
+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;
+
+/**
+ * Validates that cluster mode is internally consistent.
+ *
+ * <p>Cluster settings are bound on the central {@link ApplicationProperties} under {@code
+ * cluster.*}; this class reads {@link ApplicationProperties#getCluster()} and runs guards in {@link
+ * PostConstruct}. When {@code cluster.enabled=false} (the default) all checks are skipped so a
+ * single-instance install needs no new config.
+ */
+@Slf4j
+@Configuration
+@RequiredArgsConstructor
+public class ClusterConfig {
+
+    private final ApplicationProperties applicationProperties;
+
+    @PostConstruct
+    void validate() {
+        Cluster cluster = applicationProperties.getCluster();
+        if (!cluster.isEnabled()) {
+            return;
+        }
+        String backplane = cluster.getBackplane();
+        if ("valkey".equalsIgnoreCase(backplane)) {
+            String url = cluster.getValkey() == null ? null : cluster.getValkey().getUrl();
+            if (url == null || url.isBlank()) {
+                throw new IllegalStateException(
+                        "cluster.enabled=true with backplane=valkey requires"
+                                + " cluster.valkey.url to be set (e.g."
+                                + " redis://valkey:6379).");
+            }
+        } else if ("inprocess".equalsIgnoreCase(backplane)) {
+            // enabled+inprocess only coordinates the local JVM; cross-node lookups will 410.
+            log.warn(
+                    "cluster.enabled=true with backplane=inprocess - only the local"
+                            + " JVM is coordinated. Cross-node lookups and the file proxy will fail."
+                            + " Use backplane=valkey for real multi-node deployments.");
+        } else {
+            // Fail fast on typos like "valky" so Spring doesn't later report a cryptic
+            // "no ClusterBackplane bean" - the operator-facing error names the bad value.
+            throw new IllegalStateException(
+                    "cluster.enabled=true with unknown backplane '"
+                            + backplane
+                            + "'. Valid values: inprocess | valkey.");
+        }
+        log.info(
+                "Cluster mode enabled (backplane={}, nodeRole={}, nodeId={}).",
+                backplane,
+                cluster.resolvedRole(),
+                cluster.resolvedNodeId());
+    }
+}

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

@@ -0,0 +1,12 @@
+package stirling.software.common.cluster;
+
+import java.time.Instant;
+
+/**
+ * Snapshot of a peer node as recorded in the {@link InstanceRegistry}.
+ *
+ * @param internalAddress {@code host:port} the node listens on for {@code /internal/cluster/**}
+ * @param role one of {@code WEB}, {@code WORKER}, {@code BOTH}
+ */
+public record ClusterNode(
+        String nodeId, String internalAddress, Instant lastHeartbeat, String role) {}

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

@@ -0,0 +1,21 @@
+package stirling.software.common.cluster;
+
+import java.time.Duration;
+import java.util.Optional;
+
+/** Cluster-wide mutual exclusion primitive; non-reentrant by contract. */
+public interface DistributedLock {
+
+    Optional<LockHandle> tryAcquire(String lockKey, Duration leaseTime);
+
+    interface LockHandle extends AutoCloseable {
+        void release();
+
+        boolean renew(Duration leaseTime);
+
+        @Override
+        default void close() {
+            release();
+        }
+    }
+}

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

@@ -0,0 +1,45 @@
+package stirling.software.common.cluster;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+/** Low-level storage seam for result/job files. */
+public interface FileStore {
+
+    /** Stored file record. */
+    record Stored(String fileId, long size) {}
+
+    /** Store the given stream and return a generated file id and total bytes written. */
+    Stored store(InputStream in, String originalName) throws IOException;
+
+    /**
+     * Store the file at {@code source} and return a generated file id and total bytes written.
+     *
+     * <p>Default implementation opens {@code source} as a stream and delegates to {@link
+     * #store(InputStream, String)}. Local-disk implementations should override to use a direct
+     * file-to-file copy ({@code Files.copy(source, dest)} can use {@code sendfile(2)} on Linux),
+     * which avoids the two-memory-copy hit of streaming a disk-backed upload through the JVM heap.
+     */
+    default Stored store(Path source, String originalName) throws IOException {
+        try (InputStream in = Files.newInputStream(source)) {
+            return store(in, originalName);
+        }
+    }
+
+    /** Open the stored file for streaming reads. Caller closes. */
+    InputStream retrieve(String fileId) throws IOException;
+
+    /** Load the stored file into a byte array. */
+    byte[] retrieveBytes(String fileId) throws IOException;
+
+    /** Size of the stored file in bytes. */
+    long size(String fileId) throws IOException;
+
+    /** Delete the stored file. Returns true if a file was removed. */
+    boolean delete(String fileId);
+
+    /** Whether the file id exists in the store. */
+    boolean exists(String fileId);
+}

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

@@ -0,0 +1,18 @@
+package stirling.software.common.cluster;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Optional;
+
+/** Maps {@code nodeId} to its internal cluster address, with TTL'd heartbeats. */
+public interface InstanceRegistry {
+
+    /** Register or refresh this node. Idempotent so a wiped backplane self-heals on next tick. */
+    void register(ClusterNode node, Duration heartbeatTtl);
+
+    Optional<ClusterNode> lookup(String nodeId);
+
+    Collection<ClusterNode> activeNodes();
+
+    void deregister(String nodeId);
+}

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

@@ -0,0 +1,24 @@
+package stirling.software.common.cluster;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Optional;
+
+/** Cluster-visible storage for job status and result metadata, with TTL'd entries. */
+public interface JobStore {
+
+    /** Persist or overwrite a job entry. {@code ttl} sets the lifetime of the entry. */
+    void put(JobStoreEntry entry, Duration ttl);
+
+    Optional<JobStoreEntry> get(String jobId);
+
+    void delete(String jobId);
+
+    boolean exists(String jobId);
+
+    /** Reverse lookup: which job owns this result file id? */
+    Optional<String> findJobIdByFileId(String fileId);
+
+    /** Snapshot of every active entry. Used by admin/stats endpoints; may be O(n). */
+    Collection<JobStoreEntry> all();
+}

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

@@ -0,0 +1,30 @@
+package stirling.software.common.cluster;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Cluster-visible projection of a job's status and result metadata, as persisted in {@link
+ * JobStore}.
+ *
+ * @param owningNodeId the node id that originally executed the job
+ */
+public record JobStoreEntry(
+        String jobId,
+        JobState state,
+        String owningNodeId,
+        Instant createdAt,
+        Instant completedAt,
+        String error,
+        List<String> fileIds,
+        Map<String, String> resultMeta) {
+
+    /** Lifecycle states for a job as observed by the cluster. */
+    public enum JobState {
+        PENDING,
+        RUNNING,
+        COMPLETE,
+        FAILED
+    }
+}

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

@@ -0,0 +1,16 @@
+package stirling.software.common.cluster;
+
+import java.time.Duration;
+import java.util.Optional;
+
+/** Short-TTL namespaced key/value cache backed by the cluster backplane. */
+public interface KeyValueCache {
+
+    void put(String namespace, String key, String value, Duration ttl);
+
+    Optional<String> get(String namespace, String key);
+
+    void evict(String namespace, String key);
+
+    void evictNamespace(String namespace);
+}

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

@@ -0,0 +1,18 @@
+package stirling.software.common.cluster;
+
+import java.time.Duration;
+
+/** Token-bucket rate limiting backed by the cluster backplane. */
+public interface RateLimitStore {
+
+    /**
+     * Attempt to consume one token from the bucket identified by {@code bucketKey}.
+     *
+     * @param bucketKey opaque key identifying the bucket (e.g. {@code api:user:123})
+     * @param capacity bucket capacity
+     * @param refillPeriod time window over which {@code capacity} tokens refill
+     */
+    RateLimitDecision tryConsume(String bucketKey, long capacity, Duration refillPeriod);
+
+    record RateLimitDecision(boolean allowed, long remainingTokens, long nanosToWaitForRefill) {}
+}

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

@@ -0,0 +1,7 @@
+package stirling.software.common.cluster;
+
+/** Records one increment per sticky-session miss (a 410 Gone for a job owned by another node). */
+@FunctionalInterface
+public interface StickyMissRecorder {
+    void recordStickyMiss();
+}

app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessClusterBackplane.java

@@ -0,0 +1,31 @@
+package stirling.software.common.cluster.inprocess;
+
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.cluster.ClusterBackplane;
+import stirling.software.common.model.ApplicationProperties;
+
+@Slf4j
+public class InProcessClusterBackplane implements ClusterBackplane {
+
+    private final ApplicationProperties applicationProperties;
+
+    public InProcessClusterBackplane(ApplicationProperties applicationProperties) {
+        this.applicationProperties = applicationProperties;
+    }
+
+    @Override
+    public boolean isHealthy() {
+        return true;
+    }
+
+    @Override
+    public String backplaneType() {
+        return "inprocess";
+    }
+
+    @Override
+    public String localNodeId() {
+        return applicationProperties.getCluster().resolvedNodeId();
+    }
+}

app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessClusterConfiguration.java

@@ -0,0 +1,65 @@
+package stirling.software.common.cluster.inprocess;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.cluster.ClusterBackplane;
+import stirling.software.common.cluster.DistributedLock;
+import stirling.software.common.cluster.InstanceRegistry;
+import stirling.software.common.cluster.JobStore;
+import stirling.software.common.cluster.KeyValueCache;
+import stirling.software.common.cluster.RateLimitStore;
+import stirling.software.common.model.ApplicationProperties;
+
+/**
+ * Default cluster backplane wiring: every interface gets an {@code InProcess*} bean. Active when
+ * cluster mode is off or {@code cluster.backplane=inprocess}.
+ */
+@Slf4j
+@Configuration
+@ConditionalOnExpression(
+        "!${cluster.enabled:false} ||"
+                + " '${cluster.backplane:inprocess}'.equalsIgnoreCase('inprocess')")
+public class InProcessClusterConfiguration {
+
+    @Bean
+    @ConditionalOnMissingBean
+    public ClusterBackplane clusterBackplane(ApplicationProperties applicationProperties) {
+        log.info("Cluster backplane: in-process (single node)");
+        return new InProcessClusterBackplane(applicationProperties);
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public JobStore jobStore() {
+        return new InProcessJobStore();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public RateLimitStore rateLimitStore() {
+        return new InProcessRateLimitStore();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public DistributedLock distributedLock() {
+        return new InProcessDistributedLock();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public KeyValueCache keyValueCache() {
+        return new InProcessKeyValueCache();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public InstanceRegistry instanceRegistry() {
+        return new InProcessInstanceRegistry();
+    }
+}

app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessDistributedLock.java

@@ -0,0 +1,128 @@
+package stirling.software.common.cluster.inprocess;
+
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import stirling.software.common.cluster.DistributedLock;
+
+/**
+ * In-process {@link DistributedLock}, non-reentrant per the interface contract, with lease-expiry
+ * semantics that mirror a SET-NX-EX style distributed backend.
+ *
+ * <p>Each lock state carries a per-acquire {@code ownerToken} and an {@code expiryNanos}; another
+ * caller can take over once the lease has elapsed even if the original holder never called {@link
+ * LockHandle#release()}. This matters mostly for parity with the Valkey-backed implementation
+ * (Redis {@code SETEX} auto-expires the key); within a single JVM a crashed holder takes its lock
+ * state with it, but tests and code that rely on the {@code leaseTime} parameter still need it to
+ * be honored.
+ *
+ * <p>Expiry is lazy: an expired lock state lingers in the map until the next acquire attempt for
+ * the same key replaces it. Per-key cleanup also happens on explicit {@link LockHandle#release()},
+ * so a balanced acquire/release workload keeps the map size bounded.
+ */
+public class InProcessDistributedLock implements DistributedLock {
+
+    private final ConcurrentHashMap<String, LockState> locks = new ConcurrentHashMap<>();
+    private final AtomicLong tokenSeq = new AtomicLong();
+
+    /**
+     * Lease state for a single acquired lock. {@code ownerToken} prevents a former holder from
+     * releasing or renewing a lock now owned by someone else after lease expiry; {@code
+     * expiryNanos} is read/written only inside {@link ConcurrentHashMap#compute} so the bin lock
+     * provides the necessary happens-before guarantee.
+     */
+    private static final class LockState {
+        final long ownerToken;
+        long expiryNanos;
+
+        LockState(long ownerToken, long expiryNanos) {
+            this.ownerToken = ownerToken;
+            this.expiryNanos = expiryNanos;
+        }
+    }
+
+    @Override
+    public Optional<LockHandle> tryAcquire(String lockKey, Duration leaseTime) {
+        long token = tokenSeq.incrementAndGet();
+        long nowNanos = System.nanoTime();
+        long expiryNanos = nowNanos + leaseTime.toNanos();
+        boolean[] acquired = {false};
+        locks.compute(
+                lockKey,
+                (k, existing) -> {
+                    if (existing == null || existing.expiryNanos - nowNanos <= 0L) {
+                        // No lock, or the previous lease has expired - we take it. Subtraction
+                        // form avoids the long-overflow trap that would bite a naive
+                        // expiryNanos <= nowNanos comparison around System.nanoTime() rollover.
+                        acquired[0] = true;
+                        return new LockState(token, expiryNanos);
+                    }
+                    return existing;
+                });
+        if (!acquired[0]) {
+            return Optional.empty();
+        }
+        return Optional.of(new InProcessHandle(lockKey, token));
+    }
+
+    private void releaseInternal(String lockKey, long token) {
+        locks.compute(
+                lockKey,
+                (k, existing) -> {
+                    if (existing == null || existing.ownerToken != token) {
+                        // Already removed, expired-and-replaced, or never ours.
+                        return existing;
+                    }
+                    return null;
+                });
+    }
+
+    private boolean renewInternal(String lockKey, long token, Duration leaseTime) {
+        long nowNanos = System.nanoTime();
+        boolean[] renewed = {false};
+        locks.compute(
+                lockKey,
+                (k, existing) -> {
+                    if (existing == null
+                            || existing.ownerToken != token
+                            || existing.expiryNanos - nowNanos <= 0L) {
+                        // Lock is gone or expired; renewal is a no-op so the caller can detect it.
+                        return existing;
+                    }
+                    existing.expiryNanos = nowNanos + leaseTime.toNanos();
+                    renewed[0] = true;
+                    return existing;
+                });
+        return renewed[0];
+    }
+
+    private final class InProcessHandle implements LockHandle {
+        private final String lockKey;
+        private final long token;
+        private boolean released;
+
+        InProcessHandle(String lockKey, long token) {
+            this.lockKey = lockKey;
+            this.token = token;
+        }
+
+        @Override
+        public synchronized void release() {
+            if (released) {
+                return;
+            }
+            released = true;
+            releaseInternal(lockKey, token);
+        }
+
+        @Override
+        public synchronized boolean renew(Duration leaseTime) {
+            if (released) {
+                return false;
+            }
+            return renewInternal(lockKey, token, leaseTime);
+        }
+    }
+}

app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessInstanceRegistry.java

@@ -0,0 +1,42 @@
+package stirling.software.common.cluster.inprocess;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+import stirling.software.common.cluster.ClusterNode;
+import stirling.software.common.cluster.InstanceRegistry;
+
+public class InProcessInstanceRegistry implements InstanceRegistry {
+
+    private final AtomicReference<ClusterNode> self = new AtomicReference<>();
+
+    @Override
+    public void register(ClusterNode node, Duration heartbeatTtl) {
+        self.set(node);
+    }
+
+    @Override
+    public Optional<ClusterNode> lookup(String nodeId) {
+        ClusterNode current = self.get();
+        return current != null && current.nodeId().equals(nodeId)
+                ? Optional.of(current)
+                : Optional.empty();
+    }
+
+    @Override
+    public Collection<ClusterNode> activeNodes() {
+        ClusterNode current = self.get();
+        return current == null ? Collections.emptyList() : Collections.singletonList(current);
+    }
+
+    @Override
+    public void deregister(String nodeId) {
+        ClusterNode current = self.get();
+        if (current != null && current.nodeId().equals(nodeId)) {
+            self.set(null);
+        }
+    }
+}

app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessJobStore.java

@@ -0,0 +1,96 @@
+package stirling.software.common.cluster.inprocess;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.cluster.JobStore;
+import stirling.software.common.cluster.JobStoreEntry;
+
+@Slf4j
+public class InProcessJobStore implements JobStore {
+
+    private final ConcurrentHashMap<String, Holder> entries = new ConcurrentHashMap<>();
+
+    @Override
+    public void put(JobStoreEntry entry, Duration ttl) {
+        Instant expiry = ttl == null ? Instant.MAX : Instant.now().plus(ttl);
+        entries.put(entry.jobId(), new Holder(entry, expiry));
+    }
+
+    @Override
+    public Optional<JobStoreEntry> get(String jobId) {
+        Holder h = entries.get(jobId);
+        if (h == null) {
+            return Optional.empty();
+        }
+        if (h.isExpired()) {
+            entries.remove(jobId, h);
+            return Optional.empty();
+        }
+        return Optional.of(h.entry);
+    }
+
+    @Override
+    public void delete(String jobId) {
+        entries.remove(jobId);
+    }
+
+    @Override
+    public boolean exists(String jobId) {
+        return get(jobId).isPresent();
+    }
+
+    @Override
+    public Optional<String> findJobIdByFileId(String fileId) {
+        for (Map.Entry<String, Holder> e : entries.entrySet()) {
+            Holder h = e.getValue();
+            if (h.isExpired()) {
+                continue;
+            }
+            List<String> fileIds = h.entry.fileIds();
+            if (fileIds != null && fileIds.contains(fileId)) {
+                return Optional.of(e.getKey());
+            }
+        }
+        return Optional.empty();
+    }
+
+    @Override
+    public Collection<JobStoreEntry> all() {
+        List<JobStoreEntry> result = new ArrayList<>(entries.size());
+        for (Holder h : entries.values()) {
+            if (!h.isExpired()) {
+                result.add(h.entry);
+            }
+        }
+        return result;
+    }
+
+    /** Drop entries whose TTL has elapsed. Called by the {@code TaskManager} cleanup scheduler. */
+    public int purgeExpired() {
+        int removed = 0;
+        Instant now = Instant.now();
+        for (Map.Entry<String, Holder> e : entries.entrySet()) {
+            if (!e.getValue().expiry.equals(Instant.MAX) && e.getValue().expiry.isBefore(now)) {
+                if (entries.remove(e.getKey(), e.getValue())) {
+                    removed++;
+                }
+            }
+        }
+        return removed;
+    }
+
+    private record Holder(JobStoreEntry entry, Instant expiry) {
+        boolean isExpired() {
+            return !expiry.equals(Instant.MAX) && expiry.isBefore(Instant.now());
+        }
+    }
+}

app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessKeyValueCache.java

@@ -0,0 +1,55 @@
+package stirling.software.common.cluster.inprocess;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+
+import stirling.software.common.cluster.KeyValueCache;
+
+public class InProcessKeyValueCache implements KeyValueCache {
+
+    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Expiring>> namespaces =
+            new ConcurrentHashMap<>();
+
+    @Override
+    public void put(String namespace, String key, String value, Duration ttl) {
+        Instant expiry = ttl == null ? Instant.MAX : Instant.now().plus(ttl);
+        namespaces
+                .computeIfAbsent(namespace, n -> new ConcurrentHashMap<>())
+                .put(key, new Expiring(value, expiry));
+    }
+
+    @Override
+    public Optional<String> get(String namespace, String key) {
+        Map<String, Expiring> ns = namespaces.get(namespace);
+        if (ns == null) {
+            return Optional.empty();
+        }
+        Expiring e = ns.get(key);
+        if (e == null) {
+            return Optional.empty();
+        }
+        if (e.expiry.isBefore(Instant.now())) {
+            ns.remove(key, e);
+            return Optional.empty();
+        }
+        return Optional.of(e.value);
+    }
+
+    @Override
+    public void evict(String namespace, String key) {
+        Map<String, Expiring> ns = namespaces.get(namespace);
+        if (ns != null) {
+            ns.remove(key);
+        }
+    }
+
+    @Override
+    public void evictNamespace(String namespace) {
+        namespaces.remove(namespace);
+    }
+
+    private record Expiring(String value, Instant expiry) {}
+}

app/common/src/main/java/stirling/software/common/cluster/inprocess/InProcessRateLimitStore.java

@@ -0,0 +1,49 @@
+package stirling.software.common.cluster.inprocess;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import io.github.bucket4j.Bandwidth;
+import io.github.bucket4j.Bucket;
+import io.github.bucket4j.ConsumptionProbe;
+import io.github.bucket4j.local.LocalBucketBuilder;
+
+import stirling.software.common.cluster.RateLimitStore;
+
+/** Bucket4j-backed token bucket implementation of {@link RateLimitStore}. */
+public class InProcessRateLimitStore implements RateLimitStore {
+
+    /** Cap to bound memory; oldest accessed buckets are evicted. */
+    private static final int MAX_BUCKETS = 10_000;
+
+    private final Map<String, Bucket> buckets =
+            Collections.synchronizedMap(
+                    new LinkedHashMap<String, Bucket>(256, 0.75f, true) {
+                        @Override
+                        protected boolean removeEldestEntry(Map.Entry<String, Bucket> eldest) {
+                            return size() > MAX_BUCKETS;
+                        }
+                    });
+
+    @Override
+    public RateLimitDecision tryConsume(String bucketKey, long capacity, Duration refillPeriod) {
+        String compositeKey = bucketKey + "|" + capacity + "|" + refillPeriod.toNanos();
+        Bucket bucket =
+                buckets.computeIfAbsent(compositeKey, k -> buildBucket(capacity, refillPeriod));
+        ConsumptionProbe probe = bucket.tryConsumeAndReturnRemaining(1);
+        return new RateLimitDecision(
+                probe.isConsumed(),
+                probe.getRemainingTokens(),
+                probe.isConsumed() ? 0L : probe.getNanosToWaitForRefill());
+    }
+
+    private static Bucket buildBucket(long capacity, Duration refillPeriod) {
+        Bandwidth limit =
+                Bandwidth.builder().capacity(capacity).refillGreedy(capacity, refillPeriod).build();
+        LocalBucketBuilder builder = Bucket.builder();
+        builder.addLimit(limit);
+        return builder.build();
+    }
+}

app/common/src/main/java/stirling/software/common/cluster/inprocess/LocalDiskFileStore.java

@@ -0,0 +1,128 @@
+package stirling.software.common.cluster.inprocess;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.UUID;
+
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.common.cluster.FileStore;
+
+/** Local-disk {@link FileStore} storing files under a base directory keyed by a UUID file id. */
+@Slf4j
+public class LocalDiskFileStore implements FileStore {
+
+    private final String baseDirPath;
+
+    public LocalDiskFileStore(String baseDirPath) {
+        this.baseDirPath = baseDirPath;
+    }
+
+    @Override
+    public Stored store(InputStream in, String originalName) throws IOException {
+        String fileId = UUID.randomUUID().toString();
+        Path filePath = resolve(fileId);
+        Files.createDirectories(filePath.getParent());
+        boolean success = false;
+        try {
+            long size = Files.copy(in, filePath);
+            success = true;
+            return new Stored(fileId, size);
+        } finally {
+            if (!success) {
+                try {
+                    Files.deleteIfExists(filePath);
+                } catch (IOException cleanupEx) {
+                    log.warn(
+                            "Failed to clean up partial file {} after store failure",
+                            filePath,
+                            cleanupEx);
+                }
+            }
+        }
+    }
+
+    /**
+     * File-to-file copy. {@link Files#copy(Path, Path, java.nio.file.CopyOption...)} can use {@code
+     * sendfile(2)} on Linux for a zero-copy kernel transfer when source and destination share a
+     * filesystem, avoiding the streaming overhead of pulling the bytes through the JVM heap. Reads
+     * the source size before copying so the post-copy stat is unnecessary.
+     */
+    @Override
+    public Stored store(Path source, String originalName) throws IOException {
+        String fileId = UUID.randomUUID().toString();
+        Path filePath = resolve(fileId);
+        Files.createDirectories(filePath.getParent());
+        long size = Files.size(source);
+        boolean success = false;
+        try {
+            Files.copy(source, filePath);
+            success = true;
+            return new Stored(fileId, size);
+        } finally {
+            if (!success) {
+                try {
+                    Files.deleteIfExists(filePath);
+                } catch (IOException cleanupEx) {
+                    log.warn(
+                            "Failed to clean up partial file {} after store failure",
+                            filePath,
+                            cleanupEx);
+                }
+            }
+        }
+    }
+
+    @Override
+    public InputStream retrieve(String fileId) throws IOException {
+        return new BufferedInputStream(Files.newInputStream(resolve(fileId)));
+    }
+
+    @Override
+    public byte[] retrieveBytes(String fileId) throws IOException {
+        Path filePath = resolve(fileId);
+        if (!Files.exists(filePath)) {
+            throw new IOException("File not found with ID: " + fileId);
+        }
+        return Files.readAllBytes(filePath);
+    }
+
+    @Override
+    public long size(String fileId) throws IOException {
+        Path filePath = resolve(fileId);
+        if (!Files.exists(filePath)) {
+            throw new IOException("File not found with ID: " + fileId);
+        }
+        return Files.size(filePath);
+    }
+
+    @Override
+    public boolean delete(String fileId) {
+        try {
+            return Files.deleteIfExists(resolve(fileId));
+        } catch (IOException e) {
+            log.error("Error deleting file with ID: {}", fileId, e);
+            return false;
+        }
+    }
+
+    @Override
+    public boolean exists(String fileId) {
+        return Files.exists(resolve(fileId));
+    }
+
+    public Path resolve(String fileId) {
+        if (fileId.contains("..") || fileId.contains("/") || fileId.contains("\\")) {
+            throw new IllegalArgumentException("Invalid file ID");
+        }
+        Path basePath = Path.of(baseDirPath).normalize().toAbsolutePath();
+        Path resolvedPath = basePath.resolve(fileId).normalize();
+        if (!resolvedPath.startsWith(basePath)) {
+            throw new IllegalArgumentException("File ID resolves to an invalid path");
+        }
+        return resolvedPath;
+    }
+}

app/common/src/main/java/stirling/software/common/cluster/inprocess/LocalDiskFileStoreConfiguration.java

@@ -0,0 +1,29 @@
+package stirling.software.common.cluster.inprocess;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import stirling.software.common.cluster.FileStore;
+
+/**
+ * Always-on wiring for the per-node local-disk {@link FileStore}. Active when {@code
+ * cluster.artifactStore=local} (the default; {@code matchIfMissing=true}). The S3 artifact-store
+ * supplies its own bean when {@code cluster.artifactStore=s3}.
+ */
+@Configuration
+@ConditionalOnProperty(
+        prefix = "cluster",
+        name = "artifactStore",
+        havingValue = "local",
+        matchIfMissing = true)
+public class LocalDiskFileStoreConfiguration {
+
+    @Bean
+    @ConditionalOnMissingBean
+    public FileStore fileStore(@Value("${stirling.tempDir:/tmp/stirling-files}") String tempDir) {
+        return new LocalDiskFileStore(tempDir);
+    }
+}

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

@@ -13,6 +13,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Locale;
+import java.util.UUID;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Bean;
@@ -77,6 +78,7 @@ public class ApplicationProperties {
     private PdfEditor pdfEditor = new PdfEditor();
     private AiEngine aiEngine = new AiEngine();
     private InternalApi internalApi = new InternalApi();
+    private Cluster cluster = new Cluster();
 
     @Bean
     public PropertySource<?> dynamicYamlPropertySource(ConfigurableEnvironment environment)
@@ -254,6 +256,117 @@ public class ApplicationProperties {
         private int longRunningTimeoutSeconds = 600;
     }
 
+    /**
+     * Cluster backplane configuration. All keys live under the top-level {@code cluster.*} prefix
+     * (e.g. env var {@code CLUSTER_ENABLED}). The master switch is {@link #enabled} and defaults to
+     * off; when off the in-process backplane is wired and no other cluster keys are required.
+     */
+    @Data
+    public static class Cluster {
+
+        /** Master switch. When {@code false} (default) the in-process backplane is wired. */
+        private boolean enabled = false;
+
+        /** Backplane implementation selector. Valid values: {@code inprocess} | {@code valkey}. */
+        private String backplane = "inprocess";
+
+        /**
+         * Transient cluster job-artifact store selector. Valid values: {@code local} | {@code s3}.
+         *
+         * <p>This is distinct from {@code storage.provider}, which selects the backend for
+         * persistent user-uploaded files. The two switches exist because the user-facing storage
+         * feature is optional ({@code storage.enabled=false} is common) but every multi-node
+         * cluster still needs a shared artifact store to serve cross-node downloads. Both
+         * implementations share credentials from {@code storage.s3.*} when set to {@code s3}.
+         */
+        private String artifactStore = "local";
+
+        private Valkey valkey = new Valkey();
+        private Node node = new Node();
+        private Engine engine = new Engine();
+
+        private transient String cachedNodeId;
+
+        public NodeRole resolvedRole() {
+            if (node == null || node.getRole() == null) {
+                return NodeRole.BOTH;
+            }
+            String value = node.getRole().trim().toUpperCase(Locale.ROOT);
+            try {
+                return NodeRole.valueOf(value);
+            } catch (IllegalArgumentException ex) {
+                return NodeRole.BOTH;
+            }
+        }
+
+        public synchronized String resolvedNodeId() {
+            if (node != null && node.getId() != null && !node.getId().isBlank()) {
+                return node.getId();
+            }
+            if (cachedNodeId == null) {
+                cachedNodeId = UUID.randomUUID().toString();
+            }
+            return cachedNodeId;
+        }
+
+        public enum NodeRole {
+            WEB,
+            WORKER,
+            BOTH
+        }
+
+        @Data
+        public static class Valkey {
+            /**
+             * {@code redis://host:6379} or {@code rediss://...} for TLS. Required when cluster mode
+             * is on and backplane is valkey.
+             */
+            private String url = "";
+
+            private Tls tls = new Tls();
+
+            @Data
+            public static class Tls {
+                /**
+                 * When {@code true}, skip Valkey/Redis TLS certificate verification (dev/test
+                 * only). Leave {@code false} in production.
+                 */
+                private boolean skipCertVerification = false;
+            }
+        }
+
+        @Data
+        public static class Node {
+            /** Optional explicit node id. Blank = auto-generated UUID at startup. */
+            private String id = "";
+
+            /** {@code web} | {@code worker} | {@code both}. */
+            private String role = "both";
+
+            /**
+             * Internal cluster address advertised in the instance registry (host:port). Blank =
+             * derived at startup.
+             */
+            private String internalAddress = "";
+
+            /** {@code http} | {@code https} - scheme used when peers call this node. */
+            private String scheme = "http";
+
+            /** Heartbeat publish interval for the instance registry, in milliseconds. */
+            private long heartbeatIntervalMs = 5000;
+        }
+
+        @Data
+        public static class Engine {
+            /**
+             * Shared secret for the {@code X-Engine-Auth} header on Java -&gt; Python engine calls.
+             * Must match {@code STIRLING_ENGINE_SHARED_SECRET} on the engine pod. Blank = dev mode
+             * (no auth) - never leave blank in production.
+             */
+            private String sharedSecret = "";
+        }
+    }
+
     /**
      * HTTP timeouts for loopback calls to internal Stirling API endpoints, used by the AI workflow
      * executor and the pipeline processor. A bounded read timeout prevents a hung tool (e.g. an

app/common/src/main/java/stirling/software/common/service/FileStorage.java

@@ -1,15 +1,13 @@
 package stirling.software.common.service;
 
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.OutputStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.UUID;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
 
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.core.io.Resource;
 import org.springframework.stereotype.Service;
 import org.springframework.web.multipart.MultipartFile;
@@ -18,9 +16,11 @@ import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBo
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
+import stirling.software.common.cluster.FileStore;
+
 /**
  * Service for storing and retrieving files with unique file IDs. Used by the AutoJobPostMapping
- * system to handle file references.
+ * system to handle file references. Disk I/O is delegated to the injected {@link FileStore} bean.
  */
 @Service
 @RequiredArgsConstructor
@@ -30,251 +30,150 @@ public class FileStorage {
     /** Holds the result of a stream-to-disk store operation: the file ID and the bytes written. */
     public record StoredFile(String fileId, long size) {}
 
-    @Value("${stirling.tempDir:/tmp/stirling-files}")
-    private String tempDirPath;
-
     private final FileOrUploadService fileOrUploadService;
+    private final FileStore fileStore;
 
-    /**
-     * Store a file and return its unique ID
-     *
-     * @param file The file to store
-     * @return The unique ID assigned to the file
-     * @throws IOException If there is an error storing the file
-     */
     public String storeFile(MultipartFile file) throws IOException {
-        String fileId = generateFileId();
-        Path filePath = getFilePath(fileId);
-
-        // Ensure the directory exists
-        Files.createDirectories(filePath.getParent());
-
-        // Transfer the file to the storage location
-        file.transferTo(filePath.toFile());
-
-        log.debug("Stored file with ID: {}", fileId);
-        return fileId;
-    }
-
-    /**
-     * Store a byte array as a file and return its unique ID
-     *
-     * @param bytes The byte array to store
-     * @param originalName The original name of the file (for extension)
-     * @return The unique ID assigned to the file
-     * @throws IOException If there is an error storing the file
-     */
-    public String storeBytes(byte[] bytes, String originalName) throws IOException {
-        String fileId = generateFileId();
-        Path filePath = getFilePath(fileId);
-
-        // Ensure the directory exists
-        Files.createDirectories(filePath.getParent());
-
-        // Write the bytes to the file
-        Files.write(filePath, bytes);
-
-        log.debug("Stored byte array with ID: {}", fileId);
-        return fileId;
-    }
-
-    /**
-     * Retrieve a file by its ID as a MultipartFile
-     *
-     * @param fileId The ID of the file to retrieve
-     * @return The file as a MultipartFile
-     * @throws IOException If the file doesn't exist or can't be read
-     */
-    public MultipartFile retrieveFile(String fileId) throws IOException {
-        Path filePath = getFilePath(fileId);
-
-        if (!Files.exists(filePath)) {
-            throw new IOException("File not found with ID: " + fileId);
+        // Fast path: when Spring buffered the multipart to disk (typical for large uploads), the
+        // backing Resource exposes a real File. Hand the Path to the FileStore so it can do a
+        // file-to-file copy (Linux sendfile, no copy through Java heap) rather than streaming
+        // the bytes through an 8K buffer. Falls back to the InputStream path for in-memory
+        // multiparts, exotic Resource impls, and anything that does not back onto a File.
+        Resource res;
+        try {
+            res = file.getResource();
+        } catch (RuntimeException ignored) {
+            res = null;
         }
+        if (res != null && res.isFile()) {
+            try {
+                FileStore.Stored stored =
+                        fileStore.store(res.getFile().toPath(), file.getOriginalFilename());
+                log.debug("Stored file with ID: {} (fast path)", stored.fileId());
+                return stored.fileId();
+            } catch (IOException ex) {
+                // Some Resource impls advertise isFile()=true but throw on getFile(); fall through.
+                log.debug("Resource fast path failed, falling back to stream copy", ex);
+            }
+        }
+        try (InputStream in = file.getInputStream()) {
+            FileStore.Stored stored = fileStore.store(in, file.getOriginalFilename());
+            log.debug("Stored file with ID: {}", stored.fileId());
+            return stored.fileId();
+        }
+    }
 
-        byte[] fileData = Files.readAllBytes(filePath);
+    public String storeBytes(byte[] bytes, String originalName) throws IOException {
+        FileStore.Stored stored = fileStore.store(new ByteArrayInputStream(bytes), originalName);
+        log.debug("Stored byte array with ID: {}", stored.fileId());
+        return stored.fileId();
+    }
+
+    public MultipartFile retrieveFile(String fileId) throws IOException {
+        byte[] fileData = fileStore.retrieveBytes(fileId);
         return fileOrUploadService.toMockMultipartFile(fileId, fileData);
     }
 
-    /**
-     * Retrieve a file by its ID as a byte array
-     *
-     * @param fileId The ID of the file to retrieve
-     * @return The file as a byte array
-     * @throws IOException If the file doesn't exist or can't be read
-     */
     public byte[] retrieveBytes(String fileId) throws IOException {
-        Path filePath = getFilePath(fileId);
-
-        if (!Files.exists(filePath)) {
-            throw new IOException("File not found with ID: " + fileId);
-        }
-
-        return Files.readAllBytes(filePath);
+        return fileStore.retrieveBytes(fileId);
     }
 
-    /**
-     * Retrieve a file by its ID as a streaming InputStream. The caller is responsible for closing
-     * the returned stream.
-     *
-     * @param fileId The ID of the file to retrieve
-     * @return A buffered InputStream for the file
-     * @throws IOException If the file doesn't exist or can't be read
-     */
     public InputStream retrieveInputStream(String fileId) throws IOException {
-        Path filePath = getFilePath(fileId);
-        // Let Files.newInputStream throw NoSuchFileException naturally — avoids TOCTOU race
-        // between exists-check and open when another thread may delete concurrently.
-        return new BufferedInputStream(Files.newInputStream(filePath));
+        return fileStore.retrieve(fileId);
     }
 
-    /**
-     * Store data from an InputStream as a file and return its unique ID and byte count. Streams
-     * directly to disk without buffering the entire content in heap.
-     *
-     * @param inputStream The input stream to read from
-     * @param originalName The original name of the file (unused, kept for API symmetry)
-     * @return A {@link StoredFile} containing the file ID and the number of bytes written
-     * @throws IOException If there is an error storing the file
-     */
     public StoredFile storeInputStream(InputStream inputStream, String originalName)
             throws IOException {
-        String fileId = generateFileId();
-        Path filePath = getFilePath(fileId);
-        Files.createDirectories(filePath.getParent());
-        long size = Files.copy(inputStream, filePath);
-        log.debug("Stored input stream with ID: {}", fileId);
-        return new StoredFile(fileId, size);
+        FileStore.Stored stored = fileStore.store(inputStream, originalName);
+        log.debug("Stored input stream with ID: {}", stored.fileId());
+        return new StoredFile(stored.fileId(), stored.size());
     }
 
     public String storeFromStreamingBody(StreamingResponseBody body, String originalName)
             throws IOException {
-        String fileId = generateFileId();
-        Path filePath = getFilePath(fileId);
-        Files.createDirectories(filePath.getParent());
-        boolean success = false;
-        try (OutputStream os = new BufferedOutputStream(Files.newOutputStream(filePath))) {
-            body.writeTo(os);
-            success = true;
-        } finally {
-            if (!success) {
+        // Hold Throwable not IOException: an unchecked failure (NPE, IllegalState, OOM, etc.)
+        // from the body writer would otherwise close the pipe with EOF and the consumer would
+        // return a truncated file with no error surfaced to the caller.
+        AtomicReference<Throwable> bodyError = new AtomicReference<>();
+        try (PipedOutputStream out = new PipedOutputStream();
+                PipedInputStream in = new PipedInputStream(out, 8192)) {
+            var executor = Executors.newSingleThreadExecutor(Thread.ofVirtual().factory());
+            java.util.concurrent.Future<?> task = null;
+            try {
+                task =
+                        executor.submit(
+                                () -> {
+                                    try {
+                                        body.writeTo(out);
+                                    } catch (Throwable ex) {
+                                        bodyError.set(ex);
+                                    } finally {
+                                        try {
+                                            out.close();
+                                        } catch (IOException ignored) {
+                                            // closed on the consumer side too
+                                        }
+                                    }
+                                });
+                FileStore.Stored stored = fileStore.store(in, originalName);
+                Throwable writerErr = bodyError.get();
+                if (writerErr != null) {
+                    // Body failed mid-write: the FileStore persisted a truncated entry.
+                    // Best-effort delete so we don't leak partial files; never let cleanup
+                    // mask the original writer error.
+                    try {
+                        fileStore.delete(stored.fileId());
+                    } catch (RuntimeException cleanupEx) {
+                        log.warn(
+                                "Failed to delete partial file {} after writer error: {}",
+                                stored.fileId(),
+                                cleanupEx.getMessage());
+                    }
+                    if (writerErr instanceof IOException ioe) {
+                        throw ioe;
+                    }
+                    throw new IOException(
+                            "StreamingResponseBody writer failed: " + writerErr.getMessage(),
+                            writerErr);
+                }
+                log.debug("Stored StreamingResponseBody with ID: {}", stored.fileId());
+                return stored.fileId();
+            } finally {
+                // Interrupt and join the writer task: shutdown() alone returns immediately and a
+                // failed store leaves the writer running, leaking a thread per failed upload.
+                if (task != null && !task.isDone()) {
+                    task.cancel(true);
+                }
+                executor.shutdown();
                 try {
-                    Files.deleteIfExists(filePath);
-                } catch (IOException cleanupEx) {
-                    log.warn(
-                            "Failed to clean up partial file {} after store failure",
-                            filePath,
-                            cleanupEx);
+                    if (!executor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
+                        executor.shutdownNow();
+                    }
+                } catch (InterruptedException ie) {
+                    executor.shutdownNow();
+                    Thread.currentThread().interrupt();
                 }
             }
         }
-        log.debug("Stored StreamingResponseBody with ID: {}", fileId);
-        return fileId;
     }
 
-    /**
-     * Persist a {@link Resource} body to disk, returning the generated file ID. Used by the async
-     * job pipeline to capture {@code ResponseEntity<Resource>} results produced by controllers.
-     */
     public String storeFromResource(Resource resource, String originalName) throws IOException {
-        String fileId = generateFileId();
-        Path filePath = getFilePath(fileId);
-        Files.createDirectories(filePath.getParent());
-        boolean success = false;
         try (InputStream in = resource.getInputStream()) {
-            Files.copy(in, filePath);
-            success = true;
-        } finally {
-            if (!success) {
-                try {
-                    Files.deleteIfExists(filePath);
-                } catch (IOException cleanupEx) {
-                    log.warn(
-                            "Failed to clean up partial file {} after store failure",
-                            filePath,
-                            cleanupEx);
-                }
-            }
+            FileStore.Stored stored = fileStore.store(in, originalName);
+            log.debug("Stored Resource with ID: {}", stored.fileId());
+            return stored.fileId();
         }
-        log.debug("Stored Resource with ID: {}", fileId);
-        return fileId;
     }
 
-    /**
-     * Delete a file by its ID
-     *
-     * @param fileId The ID of the file to delete
-     * @return true if the file was deleted, false otherwise
-     */
     public boolean deleteFile(String fileId) {
-        try {
-            Path filePath = getFilePath(fileId);
-            return Files.deleteIfExists(filePath);
-        } catch (IOException e) {
-            log.error("Error deleting file with ID: {}", fileId, e);
-            return false;
-        }
+        return fileStore.delete(fileId);
     }
 
-    /**
-     * Check if a file exists by its ID
-     *
-     * @param fileId The ID of the file to check
-     * @return true if the file exists, false otherwise
-     */
     public boolean fileExists(String fileId) {
-        Path filePath = getFilePath(fileId);
-        return Files.exists(filePath);
+        return fileStore.exists(fileId);
     }
 
-    /**
-     * Get the size of a file by its ID without loading the content into memory
-     *
-     * @param fileId The ID of the file
-     * @return The size of the file in bytes
-     * @throws IOException If the file doesn't exist or can't be read
-     */
     public long getFileSize(String fileId) throws IOException {
-        Path filePath = getFilePath(fileId);
-
-        if (!Files.exists(filePath)) {
-            throw new IOException("File not found with ID: " + fileId);
-        }
-
-        return Files.size(filePath);
-    }
-
-    /**
-     * Get the path for a file ID
-     *
-     * @param fileId The ID of the file
-     * @return The path to the file
-     * @throws IllegalArgumentException if fileId contains path traversal characters or resolves
-     *     outside base directory
-     */
-    private Path getFilePath(String fileId) {
-        // Validate fileId to prevent path traversal
-        if (fileId.contains("..") || fileId.contains("/") || fileId.contains("\\")) {
-            throw new IllegalArgumentException("Invalid file ID");
-        }
-
-        Path basePath = Path.of(tempDirPath).normalize().toAbsolutePath();
-        Path resolvedPath = basePath.resolve(fileId).normalize();
-
-        // Ensure resolved path is within the base directory
-        if (!resolvedPath.startsWith(basePath)) {
-            throw new IllegalArgumentException("File ID resolves to an invalid path");
-        }
-
-        return resolvedPath;
-    }
-
-    /**
-     * Generate a unique file ID
-     *
-     * @return A unique file ID
-     */
-    private String generateFileId() {
-        return UUID.randomUUID().toString();
+        return fileStore.size(fileId);
     }
 }

app/common/src/main/java/stirling/software/common/service/TaskManager.java

@@ -3,9 +3,13 @@ package stirling.software.common.service;
 import java.io.BufferedInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.time.Duration;
+import java.time.Instant;
 import java.time.LocalDateTime;
+import java.time.ZoneId;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -16,6 +20,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipInputStream;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.MediaType;
 import org.springframework.stereotype.Service;
@@ -26,6 +31,10 @@ import jakarta.annotation.PreDestroy;
 
 import lombok.extern.slf4j.Slf4j;
 
+import stirling.software.common.cluster.ClusterBackplane;
+import stirling.software.common.cluster.JobStore;
+import stirling.software.common.cluster.JobStoreEntry;
+import stirling.software.common.cluster.JobStoreEntry.JobState;
 import stirling.software.common.model.job.JobResult;
 import stirling.software.common.model.job.JobStats;
 import stirling.software.common.model.job.ResultFile;
@@ -40,20 +49,20 @@ public class TaskManager {
     private int jobResultExpiryMinutes = 30;
 
     private final FileStorage fileStorage;
+    private final JobStore jobStore;
+    private final ClusterBackplane clusterBackplane;
     private final ScheduledExecutorService cleanupExecutor =
             Executors.newSingleThreadScheduledExecutor(
                     Thread.ofVirtual().name("task-cleanup-", 0).factory());
 
-    /** Initialize the task manager and start the cleanup scheduler */
-    public TaskManager(FileStorage fileStorage) {
+    @Autowired
+    public TaskManager(
+            FileStorage fileStorage, JobStore jobStore, ClusterBackplane clusterBackplane) {
         this.fileStorage = fileStorage;
+        this.jobStore = jobStore;
+        this.clusterBackplane = clusterBackplane;
 
-        // Schedule periodic cleanup of old job results
-        cleanupExecutor.scheduleAtFixedRate(
-                this::cleanupOldJobs,
-                10, // Initial delay
-                10, // Interval
-                TimeUnit.MINUTES);
+        cleanupExecutor.scheduleAtFixedRate(this::cleanupOldJobs, 10, 10, TimeUnit.MINUTES);
 
         log.debug(
                 "Task manager initialized with job result expiry of {} minutes",
@@ -66,7 +75,9 @@ public class TaskManager {
      * @param jobId The job ID
      */
     public void createTask(String jobId) {
-        jobResults.put(jobId, JobResult.createNew(jobId));
+        JobResult result = JobResult.createNew(jobId);
+        jobResults.put(jobId, result);
+        writeThrough(jobId, result);
         log.debug("Created task with job ID: {}", jobId);
     }
 
@@ -79,6 +90,7 @@ public class TaskManager {
     public void setResult(String jobId, Object result) {
         JobResult jobResult = getOrCreateJobResult(jobId);
         jobResult.completeWithResult(result);
+        writeThrough(jobId, jobResult);
         log.debug("Set result for job ID: {}", jobId);
     }
 
@@ -101,6 +113,7 @@ public class TaskManager {
                         extractZipToIndividualFiles(fileId, originalFileName);
                 if (!extractedFiles.isEmpty()) {
                     jobResult.completeWithFiles(extractedFiles);
+                    writeThrough(jobId, jobResult);
                     log.debug(
                             "Set multiple file results for job ID: {} with {} files extracted from"
                                     + " ZIP",
@@ -127,6 +140,7 @@ public class TaskManager {
                     "Failed to get file size for job {}: {}. Using size 0.", jobId, e.getMessage());
             jobResult.completeWithSingleFile(fileId, originalFileName, contentType, 0);
         }
+        writeThrough(jobId, jobResult);
     }
 
     /**
@@ -138,6 +152,7 @@ public class TaskManager {
     public void setMultipleFileResults(String jobId, List<ResultFile> resultFiles) {
         JobResult jobResult = getOrCreateJobResult(jobId);
         jobResult.completeWithFiles(resultFiles);
+        writeThrough(jobId, jobResult);
         log.debug(
                 "Set multiple file results for job ID: {} with {} files",
                 jobId,
@@ -153,6 +168,7 @@ public class TaskManager {
     public void setError(String jobId, String error) {
         JobResult jobResult = getOrCreateJobResult(jobId);
         jobResult.failWithError(error);
+        writeThrough(jobId, jobResult);
         log.debug("Set error for job ID: {}: {}", jobId, error);
     }
 
@@ -169,6 +185,7 @@ public class TaskManager {
             // If no result or error has been set, mark it as complete with an empty result
             jobResult.completeWithResult("Task completed successfully");
         }
+        writeThrough(jobId, jobResult);
         log.debug("Marked job ID: {} as complete", jobId);
     }
 
@@ -205,6 +222,7 @@ public class TaskManager {
         JobResult jobResult = jobResults.get(jobId);
         if (jobResult != null) {
             jobResult.addNote(note);
+            writeThrough(jobId, jobResult);
             log.debug("Added note to job ID: {}: {}", jobId, note);
             return true;
         }
@@ -295,8 +313,11 @@ public class TaskManager {
         return jobResults.computeIfAbsent(jobId, JobResult::createNew);
     }
 
-    /** Clean up old completed job results */
+    /** Clean up old completed job results. No-op in cluster mode; the backplane TTL owns expiry. */
     public void cleanupOldJobs() {
+        if (clusterBackplane != null && !clusterBackplane.shouldRunLocalCleanup()) {
+            return;
+        }
         LocalDateTime expiryThreshold =
                 LocalDateTime.now().minus(jobResultExpiryMinutes, ChronoUnit.MINUTES);
         int removedCount = 0;
@@ -315,6 +336,9 @@ public class TaskManager {
 
                     // Remove the job result
                     jobResults.remove(entry.getKey());
+                    if (jobStore != null) {
+                        jobStore.delete(entry.getKey());
+                    }
                     removedCount++;
                 }
             }
@@ -327,6 +351,53 @@ public class TaskManager {
         }
     }
 
+    /** Mirror the in-memory {@code JobResult} into the cluster-visible {@link JobStore}. */
+    private void writeThrough(String jobId, JobResult result) {
+        if (jobStore == null) {
+            return;
+        }
+        try {
+            jobStore.put(toEntry(jobId, result), Duration.ofMinutes(jobResultExpiryMinutes));
+        } catch (RuntimeException ex) {
+            log.warn("JobStore write-through failed for job {}: {}", jobId, ex.getMessage());
+        }
+    }
+
+    private JobStoreEntry toEntry(String jobId, JobResult result) {
+        JobState state;
+        if (result.isComplete()) {
+            state = result.getError() != null ? JobState.FAILED : JobState.COMPLETE;
+        } else {
+            state = JobState.PENDING;
+        }
+        Instant createdAt = toInstant(result.getCreatedAt());
+        Instant completedAt = toInstant(result.getCompletedAt());
+        List<String> fileIds = new ArrayList<>();
+        if (result.hasFiles()) {
+            for (ResultFile rf : result.getAllResultFiles()) {
+                fileIds.add(rf.getFileId());
+            }
+        }
+        Map<String, String> meta = new HashMap<>();
+        if (result.getNotes() != null && !result.getNotes().isEmpty()) {
+            meta.put("notesCount", Integer.toString(result.getNotes().size()));
+        }
+        String owningNodeId = clusterBackplane == null ? "local" : clusterBackplane.localNodeId();
+        return new JobStoreEntry(
+                jobId,
+                state,
+                owningNodeId,
+                createdAt,
+                completedAt,
+                result.getError(),
+                fileIds,
+                meta);
+    }
+
+    private Instant toInstant(LocalDateTime ldt) {
+        return ldt == null ? null : ldt.atZone(ZoneId.systemDefault()).toInstant();
+    }
+
     /** Shutdown the cleanup executor */
     @PreDestroy
     public void shutdown() {
@@ -370,7 +441,7 @@ public class TaskManager {
             while ((entry = zipIn.getNextEntry()) != null) {
                 if (!entry.isDirectory()) {
                     String contentType = determineContentType(entry.getName());
-                    // storeInputStream returns the fileId and byte count — no extra stat needed
+                    // storeInputStream returns the fileId and byte count - no extra stat needed
                     FileStorage.StoredFile stored =
                             fileStorage.storeInputStream(zipIn, entry.getName());
 
@@ -458,7 +529,8 @@ public class TaskManager {
     }
 
     /**
-     * Find the job key that owns a given file ID.
+     * Find the job key that owns a given file ID. Checks the local in-memory map first, then falls
+     * back to the cluster-visible {@link JobStore}.
      *
      * @param fileId file identifier to look up
      * @return scoped job key if found, otherwise null
@@ -474,6 +546,18 @@ public class TaskManager {
                 }
             }
         }
+        if (jobStore != null) {
+            // Propagate JobStore failures: returning null on a backplane outage would conflate
+            // "no such file" with "lookup unavailable" and the caller would respond 404 to a
+            // transient blip that should be retried. Let Spring's exception handler surface a
+            // 5xx so clients know to retry.
+            try {
+                return jobStore.findJobIdByFileId(fileId).orElse(null);
+            } catch (RuntimeException e) {
+                log.warn("JobStore findJobIdByFileId failed for {}: {}", fileId, e.getMessage());
+                throw e;
+            }
+        }
         return null;
     }
 }

app/common/src/test/java/stirling/software/common/architecture/ArchitectureTest.java

@@ -56,4 +56,17 @@ class ArchitectureTest {
                         .resideInAPackage("stirling.software.saas..");
         rule.check(commonClasses);
     }
+
+    @Test
+    void clusterInterfacesHaveNoImplementationDependencies() {
+        ArchRule rule =
+                noClasses()
+                        .that()
+                        .resideInAPackage("stirling.software.common.cluster..")
+                        .should()
+                        .dependOnClassesThat()
+                        .resideInAnyPackage(
+                                "stirling.software.proprietary..", "stirling.software.saas..");
+        rule.check(commonClasses);
+    }
 }

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

@@ -0,0 +1,51 @@
+package stirling.software.common.cluster;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+class BackplaneContractCompilationTest {
+
+    @Test
+    void jobStoreEntryRecordRoundTrips() {
+        Instant now = Instant.now();
+        JobStoreEntry entry =
+                new JobStoreEntry(
+                        "job-1",
+                        JobStoreEntry.JobState.PENDING,
+                        "node-a",
+                        now,
+                        null,
+                        null,
+                        List.of("file-1"),
+                        Map.of("k", "v"));
+        assertEquals("job-1", entry.jobId());
+        assertEquals(JobStoreEntry.JobState.PENDING, entry.state());
+        assertEquals("node-a", entry.owningNodeId());
+        assertEquals(now, entry.createdAt());
+        assertEquals(List.of("file-1"), entry.fileIds());
+        assertEquals("v", entry.resultMeta().get("k"));
+    }
+
+    @Test
+    void clusterNodeRecordRoundTrips() {
+        Instant heartbeat = Instant.now();
+        ClusterNode node = new ClusterNode("node-a", "10.0.0.1:8080", heartbeat, "BOTH");
+        assertEquals("node-a", node.nodeId());
+        assertEquals("10.0.0.1:8080", node.internalAddress());
+        assertEquals(heartbeat, node.lastHeartbeat());
+        assertEquals("BOTH", node.role());
+    }
+
+    @Test
+    void rateLimitDecisionRecordRoundTrips() {
+        RateLimitStore.RateLimitDecision d = new RateLimitStore.RateLimitDecision(true, 7, 0L);
+        assertEquals(true, d.allowed());
+        assertEquals(7, d.remainingTokens());
+        assertEquals(0L, d.nanosToWaitForRefill());
+    }
+}

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

@@ -0,0 +1,65 @@
+package stirling.software.common.cluster;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.lang.reflect.Method;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.model.ApplicationProperties.Cluster;
+
+class ClusterConfigValidationTest {
+
+    @Test
+    void validationPassesWhenDisabled() {
+        ApplicationProperties props = new ApplicationProperties();
+        ClusterConfig config = new ClusterConfig(props);
+        assertDoesNotThrow(() -> invokeValidate(config));
+    }
+
+    @Test
+    void validationFailsWhenValkeyEnabledWithoutUrl() {
+        ApplicationProperties props = new ApplicationProperties();
+        Cluster cluster = props.getCluster();
+        cluster.setEnabled(true);
+        cluster.setBackplane("valkey");
+        ClusterConfig config = new ClusterConfig(props);
+        assertThrows(IllegalStateException.class, () -> invokeValidate(config));
+    }
+
+    @Test
+    void validationPassesWhenValkeyEnabledWithUrl() {
+        ApplicationProperties props = new ApplicationProperties();
+        Cluster cluster = props.getCluster();
+        cluster.setEnabled(true);
+        cluster.setBackplane("valkey");
+        cluster.getValkey().setUrl("redis://localhost:6379");
+        ClusterConfig config = new ClusterConfig(props);
+        assertDoesNotThrow(() -> invokeValidate(config));
+    }
+
+    @Test
+    void validationPassesWhenInProcessEnabled() {
+        ApplicationProperties props = new ApplicationProperties();
+        Cluster cluster = props.getCluster();
+        cluster.setEnabled(true);
+        cluster.setBackplane("inprocess");
+        ClusterConfig config = new ClusterConfig(props);
+        assertDoesNotThrow(() -> invokeValidate(config));
+    }
+
+    private void invokeValidate(ClusterConfig config) throws Exception {
+        Method m = ClusterConfig.class.getDeclaredMethod("validate");
+        m.setAccessible(true);
+        try {
+            m.invoke(config);
+        } catch (java.lang.reflect.InvocationTargetException ex) {
+            if (ex.getCause() instanceof RuntimeException re) {
+                throw re;
+            }
+            throw ex;
+        }
+    }
+}

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

@@ -0,0 +1,62 @@
+package stirling.software.common.cluster;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.model.ApplicationProperties.Cluster;
+
+class ClusterPropertiesTest {
+
+    @Test
+    void defaultsAreDisabledAndInprocess() {
+        Cluster props = new ApplicationProperties().getCluster();
+        assertFalse(props.isEnabled());
+        assertEquals("inprocess", props.getBackplane());
+        assertEquals("local", props.getArtifactStore());
+        assertEquals(Cluster.NodeRole.BOTH, props.resolvedRole());
+        assertEquals("", props.getValkey().getUrl());
+        assertFalse(props.getValkey().getTls().isSkipCertVerification());
+        assertEquals("both", props.getNode().getRole());
+        assertEquals("http", props.getNode().getScheme());
+        assertEquals(5000L, props.getNode().getHeartbeatIntervalMs());
+    }
+
+    @Test
+    void resolvedRoleParsesCaseInsensitively() {
+        Cluster props = new ApplicationProperties().getCluster();
+        props.getNode().setRole("WEB");
+        assertEquals(Cluster.NodeRole.WEB, props.resolvedRole());
+
+        props.getNode().setRole("web");
+        assertEquals(Cluster.NodeRole.WEB, props.resolvedRole());
+
+        props.getNode().setRole("Worker");
+        assertEquals(Cluster.NodeRole.WORKER, props.resolvedRole());
+
+        props.getNode().setRole("garbage");
+        assertEquals(Cluster.NodeRole.BOTH, props.resolvedRole());
+
+        props.getNode().setRole(null);
+        assertEquals(Cluster.NodeRole.BOTH, props.resolvedRole());
+    }
+
+    @Test
+    void resolvedNodeIdIsStableAcrossCalls() {
+        Cluster props = new ApplicationProperties().getCluster();
+        String first = props.resolvedNodeId();
+        String second = props.resolvedNodeId();
+        assertNotNull(first);
+        assertEquals(first, second);
+    }
+
+    @Test
+    void resolvedNodeIdHonoursExplicitId() {
+        Cluster props = new ApplicationProperties().getCluster();
+        props.getNode().setId("abc");
+        assertEquals("abc", props.resolvedNodeId());
+    }
+}

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

@@ -0,0 +1,90 @@
+package stirling.software.common.cluster;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+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.cluster.inprocess.InProcessClusterConfiguration;
+import stirling.software.common.model.ApplicationProperties;
+
+/**
+ * Verifies the {@link InProcessClusterConfiguration} conditional wiring: in-process beans wire when
+ * cluster mode is off or {@code backplane=inprocess}, and are skipped when {@code
+ * backplane=valkey}.
+ */
+class InProcessConfigurationConditionalTest {
+
+    private final ApplicationContextRunner runner =
+            new ApplicationContextRunner()
+                    .withConfiguration(
+                            org.springframework.boot.autoconfigure.AutoConfigurations.of(
+                                    PropertyPlaceholderAutoConfiguration.class))
+                    .withUserConfiguration(
+                            TestAppPropertiesConfig.class,
+                            ClusterConfig.class,
+                            InProcessClusterConfiguration.class);
+
+    @Test
+    void inProcessBeansWireWhenClusterDisabled() {
+        runner.run(
+                context ->
+                        assertThat(context)
+                                .hasNotFailed()
+                                .hasSingleBean(ClusterBackplane.class)
+                                .hasSingleBean(JobStore.class)
+                                .hasSingleBean(RateLimitStore.class)
+                                .hasSingleBean(DistributedLock.class)
+                                .hasSingleBean(KeyValueCache.class)
+                                .hasSingleBean(InstanceRegistry.class));
+    }
+
+    @Test
+    void inProcessBeansWireWhenEnabledWithInProcessBackplane() {
+        runner.withPropertyValues("cluster.enabled=true", "cluster.backplane=inprocess")
+                .run(
+                        context ->
+                                assertThat(context)
+                                        .hasNotFailed()
+                                        .hasSingleBean(ClusterBackplane.class)
+                                        .hasSingleBean(JobStore.class)
+                                        .hasSingleBean(RateLimitStore.class)
+                                        .hasSingleBean(DistributedLock.class)
+                                        .hasSingleBean(KeyValueCache.class)
+                                        .hasSingleBean(InstanceRegistry.class));
+    }
+
+    @Test
+    void inProcessBeansSkippedWhenEnabledWithDistributedBackplane() {
+        runner.withPropertyValues(
+                        "cluster.enabled=true",
+                        "cluster.backplane=valkey",
+                        "cluster.valkey.url=redis://localhost:6379")
+                .run(
+                        context ->
+                                assertThat(context)
+                                        .hasNotFailed()
+                                        .doesNotHaveBean(ClusterBackplane.class)
+                                        .doesNotHaveBean(JobStore.class)
+                                        .doesNotHaveBean(RateLimitStore.class)
+                                        .doesNotHaveBean(DistributedLock.class)
+                                        .doesNotHaveBean(KeyValueCache.class)
+                                        .doesNotHaveBean(InstanceRegistry.class));
+    }
+
+    /**
+     * Hand-rolled {@link ApplicationProperties} bean: the production class loads YAML at startup
+     * via a {@code @PostConstruct} hook that isn't appropriate for the slice runner, so we wire a
+     * defaults-only instance here.
+     */
+    @Configuration
+    static class TestAppPropertiesConfig {
+        @Bean
+        ApplicationProperties applicationProperties() {
+            return new ApplicationProperties();
+        }
+    }
+}

app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessDistributedLockTest.java

@@ -0,0 +1,167 @@
+package stirling.software.common.cluster.inprocess;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.cluster.DistributedLock;
+
+class InProcessDistributedLockTest {
+
+    @Test
+    void acquireReleaseAcquire() {
+        DistributedLock lock = new InProcessDistributedLock();
+        DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofSeconds(30)).orElseThrow();
+        h1.release();
+        assertTrue(lock.tryAcquire("k", Duration.ofSeconds(30)).isPresent());
+    }
+
+    @Test
+    void reentryFromSameThreadFails() {
+        DistributedLock lock = new InProcessDistributedLock();
+        DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofSeconds(30)).orElseThrow();
+        Optional<DistributedLock.LockHandle> reentry = lock.tryAcquire("k", Duration.ofSeconds(30));
+        assertFalse(reentry.isPresent(), "in-process lock must be non-reentrant");
+        h1.release();
+        // After release, anyone can acquire again.
+        assertTrue(lock.tryAcquire("k", Duration.ofSeconds(30)).isPresent());
+    }
+
+    @Test
+    void secondAcquireFromAnotherThreadFails() throws InterruptedException {
+        DistributedLock lock = new InProcessDistributedLock();
+        DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofSeconds(30)).orElseThrow();
+
+        CountDownLatch done = new CountDownLatch(1);
+        AtomicBoolean acquired = new AtomicBoolean(true);
+        Thread t =
+                new Thread(
+                        () -> {
+                            Optional<DistributedLock.LockHandle> attempt =
+                                    lock.tryAcquire("k", Duration.ofSeconds(30));
+                            acquired.set(attempt.isPresent());
+                            attempt.ifPresent(DistributedLock.LockHandle::release);
+                            done.countDown();
+                        });
+        t.start();
+        assertTrue(done.await(2, TimeUnit.SECONDS));
+        assertFalse(acquired.get());
+        h1.release();
+    }
+
+    @Test
+    void leaseExpiryAllowsTakeoverEvenWithoutRelease() throws InterruptedException {
+        // Acquire with a short lease, never call release, then try to acquire again after the
+        // lease has elapsed. Matches Redis SET-NX-EX semantics - the second caller gets the lock
+        // because the first lease auto-expired. 250ms lease + 350ms wait gives CI generous slack.
+        DistributedLock lock = new InProcessDistributedLock();
+        DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofMillis(250)).orElseThrow();
+        Thread.sleep(350);
+        Optional<DistributedLock.LockHandle> takeover =
+                lock.tryAcquire("k", Duration.ofSeconds(30));
+        assertTrue(
+                takeover.isPresent(),
+                "expired lease must release the lock so a new caller can take over");
+        // Calling release() on the original handle after takeover must be a no-op (token check).
+        h1.release();
+        // The takeover holder is still the legitimate owner.
+        assertFalse(lock.tryAcquire("k", Duration.ofSeconds(30)).isPresent());
+        takeover.get().release();
+    }
+
+    @Test
+    void renewExtendsLease() throws InterruptedException {
+        // Acquire with a short lease, renew it before it expires, then verify the lock is still
+        // held past the original expiry point. 200ms initial + renew to 2s + wait 350ms.
+        DistributedLock lock = new InProcessDistributedLock();
+        DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofMillis(200)).orElseThrow();
+        assertTrue(h1.renew(Duration.ofSeconds(2)), "renew on a held lease must succeed");
+        Thread.sleep(350);
+        assertFalse(
+                lock.tryAcquire("k", Duration.ofSeconds(30)).isPresent(),
+                "renew should have pushed expiry well past the original 200ms");
+        h1.release();
+    }
+
+    @Test
+    void renewAfterReleaseFails() {
+        DistributedLock lock = new InProcessDistributedLock();
+        DistributedLock.LockHandle h1 = lock.tryAcquire("k", Duration.ofSeconds(30)).orElseThrow();
+        h1.release();
+        assertFalse(h1.renew(Duration.ofSeconds(30)), "renew on a released handle must fail");
+    }
+
+    /**
+     * Concurrency stress: many threads contending on the same key with each holder respecting the
+     * lease (hold &lt;&lt; lease). The lock behaves as a strict mutex in this regime so asserting
+     * mutual exclusion is meaningful. A separate test ({@link
+     * #leaseExpiryAllowsTakeoverEvenWithoutRelease}) covers the takeover-across-expiry branch,
+     * which legitimately allows two holders momentarily and is split-brain behaviour inherent to
+     * any lease-based lock.
+     */
+    @Test
+    void concurrentContentionPreservesMutualExclusion() throws InterruptedException {
+        DistributedLock lock = new InProcessDistributedLock();
+        int threads = 16;
+        int attemptsPerThread = 200;
+        // Lease far exceeds any plausible hold time, so the takeover branch never triggers in
+        // this test and the lock acts as a strict mutex.
+        Duration lease = Duration.ofSeconds(5);
+        java.util.concurrent.atomic.AtomicInteger concurrentHolders =
+                new java.util.concurrent.atomic.AtomicInteger();
+        java.util.concurrent.atomic.AtomicInteger maxConcurrent =
+                new java.util.concurrent.atomic.AtomicInteger();
+        java.util.concurrent.atomic.AtomicInteger acquires =
+                new java.util.concurrent.atomic.AtomicInteger();
+        java.util.concurrent.atomic.AtomicReference<Throwable> firstFailure =
+                new java.util.concurrent.atomic.AtomicReference<>();
+        CountDownLatch start = new CountDownLatch(1);
+        CountDownLatch done = new CountDownLatch(threads);
+
+        for (int i = 0; i < threads; i++) {
+            new Thread(
+                            () -> {
+                                try {
+                                    start.await();
+                                    for (int j = 0; j < attemptsPerThread; j++) {
+                                        Optional<DistributedLock.LockHandle> h =
+                                                lock.tryAcquire("hot", lease);
+                                        if (h.isPresent()) {
+                                            int now = concurrentHolders.incrementAndGet();
+                                            maxConcurrent.accumulateAndGet(now, Math::max);
+                                            acquires.incrementAndGet();
+                                            // Trivial critical section; well within lease.
+                                            concurrentHolders.decrementAndGet();
+                                            h.get().release();
+                                        }
+                                    }
+                                } catch (Throwable t) {
+                                    firstFailure.compareAndSet(null, t);
+                                } finally {
+                                    done.countDown();
+                                }
+                            },
+                            "lock-stress-" + i)
+                    .start();
+        }
+        start.countDown();
+        assertTrue(done.await(30, TimeUnit.SECONDS), "stress workers must finish in time");
+        org.junit.jupiter.api.Assertions.assertNull(firstFailure.get(), "no worker may throw");
+        org.junit.jupiter.api.Assertions.assertEquals(
+                1,
+                maxConcurrent.get(),
+                "mutual exclusion violated: more than one holder observed simultaneously");
+        assertTrue(
+                acquires.get() > 0,
+                "at least some acquires must succeed under contention (saw "
+                        + acquires.get()
+                        + ")");
+    }
+}

app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessInstanceRegistryTest.java

@@ -0,0 +1,28 @@
+package stirling.software.common.cluster.inprocess;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import java.time.Instant;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.cluster.ClusterNode;
+
+class InProcessInstanceRegistryTest {
+
+    @Test
+    void registerThenLookupAndActiveNodes() {
+        InProcessInstanceRegistry registry = new InProcessInstanceRegistry();
+        ClusterNode node = new ClusterNode("node-1", "127.0.0.1:8080", Instant.now(), "BOTH");
+        registry.register(node, Duration.ofSeconds(30));
+
+        assertTrue(registry.lookup("node-1").isPresent());
+        assertEquals("node-1", registry.lookup("node-1").get().nodeId());
+        assertEquals(1, registry.activeNodes().size());
+
+        registry.deregister("node-1");
+        assertTrue(registry.lookup("node-1").isEmpty());
+    }
+}

app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessJobStoreTest.java

@@ -0,0 +1,91 @@
+package stirling.software.common.cluster.inprocess;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.cluster.JobStoreEntry;
+
+class InProcessJobStoreTest {
+
+    private final InProcessJobStore store = new InProcessJobStore();
+
+    @Test
+    void putGetDeleteExistsRoundTrip() {
+        JobStoreEntry entry = entry("job-1");
+        store.put(entry, Duration.ofMinutes(30));
+
+        assertTrue(store.exists("job-1"));
+        assertEquals(entry, store.get("job-1").orElseThrow());
+
+        store.delete("job-1");
+        assertFalse(store.exists("job-1"));
+    }
+
+    @Test
+    void ttlExpiry() throws InterruptedException {
+        store.put(entry("job-2"), Duration.ofMillis(50));
+        Thread.sleep(100);
+        assertFalse(store.get("job-2").isPresent());
+    }
+
+    @Test
+    void purgeExpiredRemovesOnlyStaleEntries() throws InterruptedException {
+        store.put(entry("job-fresh"), Duration.ofMinutes(30));
+        store.put(entry("job-stale"), Duration.ofMillis(20));
+        Thread.sleep(80);
+
+        int removed = store.purgeExpired();
+        assertEquals(1, removed);
+        assertTrue(store.exists("job-fresh"));
+    }
+
+    @Test
+    void findJobIdByFileIdReturnsTheRightJob() {
+        store.put(
+                new JobStoreEntry(
+                        "job-a",
+                        JobStoreEntry.JobState.COMPLETE,
+                        "node-1",
+                        Instant.now(),
+                        Instant.now(),
+                        null,
+                        List.of("file-1", "file-2"),
+                        Map.of()),
+                Duration.ofMinutes(30));
+        store.put(
+                new JobStoreEntry(
+                        "job-b",
+                        JobStoreEntry.JobState.COMPLETE,
+                        "node-1",
+                        Instant.now(),
+                        Instant.now(),
+                        null,
+                        List.of("file-3"),
+                        Map.of()),
+                Duration.ofMinutes(30));
+
+        assertEquals("job-a", store.findJobIdByFileId("file-1").orElseThrow());
+        assertEquals("job-b", store.findJobIdByFileId("file-3").orElseThrow());
+        assertFalse(store.findJobIdByFileId("missing").isPresent());
+    }
+
+    private JobStoreEntry entry(String id) {
+        return new JobStoreEntry(
+                id,
+                JobStoreEntry.JobState.PENDING,
+                "node-1",
+                Instant.now(),
+                null,
+                null,
+                List.of(),
+                Map.of());
+    }
+}

app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessKeyValueCacheTest.java

@@ -0,0 +1,41 @@
+package stirling.software.common.cluster.inprocess;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.cluster.KeyValueCache;
+
+class InProcessKeyValueCacheTest {
+
+    @Test
+    void putGetEvict() {
+        KeyValueCache cache = new InProcessKeyValueCache();
+        cache.put("apikey", "a", "userA", Duration.ofMinutes(1));
+        assertEquals("userA", cache.get("apikey", "a").orElseThrow());
+
+        cache.evict("apikey", "a");
+        assertFalse(cache.get("apikey", "a").isPresent());
+    }
+
+    @Test
+    void ttlExpiry() throws InterruptedException {
+        KeyValueCache cache = new InProcessKeyValueCache();
+        cache.put("ns", "k", "v", Duration.ofMillis(40));
+        Thread.sleep(80);
+        assertFalse(cache.get("ns", "k").isPresent());
+    }
+
+    @Test
+    void evictNamespace() {
+        KeyValueCache cache = new InProcessKeyValueCache();
+        cache.put("ns", "a", "1", Duration.ofMinutes(1));
+        cache.put("ns", "b", "2", Duration.ofMinutes(1));
+        cache.evictNamespace("ns");
+        assertFalse(cache.get("ns", "a").isPresent());
+        assertFalse(cache.get("ns", "b").isPresent());
+    }
+}

app/common/src/test/java/stirling/software/common/cluster/inprocess/InProcessRateLimitStoreTest.java

@@ -0,0 +1,56 @@
+package stirling.software.common.cluster.inprocess;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.cluster.RateLimitStore;
+import stirling.software.common.cluster.RateLimitStore.RateLimitDecision;
+
+class InProcessRateLimitStoreTest {
+
+    @Test
+    void firstNConsumesAllowed() {
+        RateLimitStore store = new InProcessRateLimitStore();
+        for (int i = 0; i < 5; i++) {
+            assertTrue(store.tryConsume("k", 5, Duration.ofSeconds(60)).allowed(), "i=" + i);
+        }
+        assertFalse(store.tryConsume("k", 5, Duration.ofSeconds(60)).allowed());
+    }
+
+    @Test
+    void remainingTokensDecrements() {
+        RateLimitStore store = new InProcessRateLimitStore();
+        RateLimitDecision d1 = store.tryConsume("k", 5, Duration.ofSeconds(60));
+        RateLimitDecision d2 = store.tryConsume("k", 5, Duration.ofSeconds(60));
+        assertTrue(d1.allowed());
+        assertTrue(d2.allowed());
+        assertEquals(4, d1.remainingTokens());
+        assertEquals(3, d2.remainingTokens());
+    }
+
+    @Test
+    void refillRestoresTokens() throws InterruptedException {
+        RateLimitStore store = new InProcessRateLimitStore();
+        // Capacity 2 with smooth refill over 100 ms -> ~1 token per 50 ms.
+        for (int i = 0; i < 2; i++) {
+            assertTrue(store.tryConsume("k", 2, Duration.ofMillis(100)).allowed());
+        }
+        assertFalse(store.tryConsume("k", 2, Duration.ofMillis(100)).allowed());
+        Thread.sleep(150);
+        assertTrue(store.tryConsume("k", 2, Duration.ofMillis(100)).allowed());
+    }
+
+    @Test
+    void deniedConsumeReportsWaitNanos() {
+        RateLimitStore store = new InProcessRateLimitStore();
+        assertTrue(store.tryConsume("wait", 1, Duration.ofSeconds(10)).allowed());
+        RateLimitDecision denied = store.tryConsume("wait", 1, Duration.ofSeconds(10));
+        assertFalse(denied.allowed());
+        assertTrue(denied.nanosToWaitForRefill() > 0L);
+    }
+}

app/common/src/test/java/stirling/software/common/cluster/inprocess/LocalDiskFileStoreTest.java

@@ -0,0 +1,42 @@
+package stirling.software.common.cluster.inprocess;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.file.Path;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import stirling.software.common.cluster.FileStore;
+
+class LocalDiskFileStoreTest {
+
+    @Test
+    void storeRetrieveSizeDeleteExistsRoundTrip(@TempDir Path dir) throws IOException {
+        LocalDiskFileStore store = new LocalDiskFileStore(dir.toString());
+        byte[] payload = "hello-bytes".getBytes();
+
+        FileStore.Stored stored = store.store(new ByteArrayInputStream(payload), "x.txt");
+        assertEquals(payload.length, stored.size());
+        assertTrue(store.exists(stored.fileId()));
+        assertEquals(payload.length, store.size(stored.fileId()));
+        assertArrayEquals(payload, store.retrieveBytes(stored.fileId()));
+
+        assertTrue(store.delete(stored.fileId()));
+        assertFalse(store.exists(stored.fileId()));
+    }
+
+    @Test
+    void traversalIdsAreRejected(@TempDir Path dir) {
+        LocalDiskFileStore store = new LocalDiskFileStore(dir.toString());
+        assertThrows(IllegalArgumentException.class, () -> store.resolve("../foo"));
+        assertThrows(IllegalArgumentException.class, () -> store.resolve("a/b"));
+        assertThrows(IllegalArgumentException.class, () -> store.resolve("a\\b"));
+    }
+}

app/common/src/test/java/stirling/software/common/service/FileStorageDelegationTest.java

@@ -0,0 +1,27 @@
+package stirling.software.common.service;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.mockito.Mockito.mock;
+
+import java.io.IOException;
+import java.nio.file.Path;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import stirling.software.common.cluster.inprocess.LocalDiskFileStore;
+
+class FileStorageDelegationTest {
+
+    @Test
+    void storeBytesThenRetrieveBytesRoundTripsThroughFileStore(@TempDir Path tempDir)
+            throws IOException {
+        FileStorage fs =
+                new FileStorage(
+                        mock(FileOrUploadService.class),
+                        new LocalDiskFileStore(tempDir.toString()));
+        byte[] payload = "round-trip".getBytes();
+        String id = fs.storeBytes(payload, "x.bin");
+        assertArrayEquals(payload, fs.retrieveBytes(id));
+    }
+}

app/common/src/test/java/stirling/software/common/service/FileStorageTest.java

@@ -3,6 +3,7 @@ package stirling.software.common.service;
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
@@ -13,29 +14,30 @@ import java.util.stream.Stream;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
-import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 import org.springframework.core.io.ByteArrayResource;
 import org.springframework.core.io.Resource;
 import org.springframework.http.MediaType;
-import org.springframework.test.util.ReflectionTestUtils;
 import org.springframework.web.multipart.MultipartFile;
 
+import stirling.software.common.cluster.inprocess.LocalDiskFileStore;
+
 class FileStorageTest {
 
     @TempDir Path tempDir;
 
     @Mock private FileOrUploadService fileOrUploadService;
 
-    @InjectMocks private FileStorage fileStorage;
+    private FileStorage fileStorage;
 
     private MultipartFile mockFile;
 
     @BeforeEach
-    void setUp() {
+    void setUp() throws IOException {
         MockitoAnnotations.openMocks(this);
-        ReflectionTestUtils.setField(fileStorage, "tempDirPath", tempDir.toString());
+        fileStorage =
+                new FileStorage(fileOrUploadService, new LocalDiskFileStore(tempDir.toString()));
 
         // Create a mock MultipartFile
         mockFile = mock(MultipartFile.class);
@@ -47,17 +49,7 @@ class FileStorageTest {
     void testStoreFile() throws IOException {
         // Arrange
         byte[] fileContent = "Test PDF content".getBytes();
-        when(mockFile.getBytes()).thenReturn(fileContent);
-
-        // Set up mock to handle transferTo by writing the file
-        doAnswer(
-                        invocation -> {
-                            java.io.File file = invocation.getArgument(0);
-                            Files.write(file.toPath(), fileContent);
-                            return null;
-                        })
-                .when(mockFile)
-                .transferTo(any(java.io.File.class));
+        when(mockFile.getInputStream()).thenReturn(new ByteArrayInputStream(fileContent));
 
         // Act
         String fileId = fileStorage.storeFile(mockFile);
@@ -65,7 +57,7 @@ class FileStorageTest {
         // Assert
         assertNotNull(fileId);
         assertTrue(Files.exists(tempDir.resolve(fileId)));
-        verify(mockFile).transferTo(any(java.io.File.class));
+        assertArrayEquals(fileContent, Files.readAllBytes(tempDir.resolve(fileId)));
     }
 
     @Test
@@ -247,11 +239,11 @@ class FileStorageTest {
             filesBefore = s.count();
         }
 
-        // Act + Assert: IOException must propagate out — not be swallowed.
+        // Act + Assert: IOException must propagate out - not be swallowed.
         assertThrows(
                 IOException.class, () -> fileStorage.storeFromResource(flakyResource, "n.pdf"));
 
-        // Assert: no partial file lingers under the storage directory — the finally
+        // Assert: no partial file lingers under the storage directory - the finally
         // branch's deleteIfExists must have cleaned it up.
         long filesAfter;
         try (Stream<Path> s = Files.list(tempDir)) {

app/common/src/test/java/stirling/software/common/service/TaskManagerJobStoreDelegationTest.java

@@ -0,0 +1,120 @@
+package stirling.software.common.service;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+import java.time.LocalDateTime;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import stirling.software.common.cluster.ClusterBackplane;
+import stirling.software.common.cluster.JobStoreEntry;
+import stirling.software.common.cluster.JobStoreEntry.JobState;
+import stirling.software.common.cluster.inprocess.InProcessClusterBackplane;
+import stirling.software.common.cluster.inprocess.InProcessJobStore;
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.model.job.JobResult;
+
+class TaskManagerJobStoreDelegationTest {
+
+    @Mock private FileStorage fileStorage;
+
+    private InProcessJobStore jobStore;
+    private ClusterBackplane backplane;
+    private TaskManager taskManager;
+
+    @BeforeEach
+    void setUp() {
+        MockitoAnnotations.openMocks(this);
+        jobStore = spy(new InProcessJobStore());
+        backplane = new InProcessClusterBackplane(new ApplicationProperties());
+        taskManager = new TaskManager(fileStorage, jobStore, backplane);
+        ReflectionTestUtils.setField(taskManager, "jobResultExpiryMinutes", 30);
+    }
+
+    @Test
+    void createTaskWritesPendingEntry() {
+        taskManager.createTask("job-1");
+        JobStoreEntry entry = jobStore.get("job-1").orElseThrow();
+        assertEquals(JobState.PENDING, entry.state());
+        assertEquals(backplane.localNodeId(), entry.owningNodeId());
+    }
+
+    @Test
+    void setCompleteFlipsToComplete() {
+        taskManager.createTask("job-2");
+        taskManager.setResult("job-2", "ok");
+        taskManager.setComplete("job-2");
+        JobStoreEntry entry = jobStore.get("job-2").orElseThrow();
+        assertEquals(JobState.COMPLETE, entry.state());
+    }
+
+    @Test
+    void setErrorFlipsToFailed() {
+        taskManager.createTask("job-3");
+        taskManager.setError("job-3", "boom");
+        JobStoreEntry entry = jobStore.get("job-3").orElseThrow();
+        assertEquals(JobState.FAILED, entry.state());
+        assertEquals("boom", entry.error());
+    }
+
+    @Test
+    void cleanupOldJobsIsNoopWhenBackplaneIsNotInProcess() {
+        ClusterBackplane mockedValkeyBackplane =
+                new ClusterBackplane() {
+                    @Override
+                    public boolean isHealthy() {
+                        return true;
+                    }
+
+                    @Override
+                    public String backplaneType() {
+                        return "valkey";
+                    }
+
+                    @Override
+                    public String localNodeId() {
+                        return "node-1";
+                    }
+
+                    @Override
+                    public boolean shouldRunLocalCleanup() {
+                        return false;
+                    }
+                };
+        TaskManager tm = new TaskManager(fileStorage, jobStore, mockedValkeyBackplane);
+        ReflectionTestUtils.setField(tm, "jobResultExpiryMinutes", 30);
+        tm.createTask("job-4");
+        tm.setComplete("job-4");
+        ageJobPastExpiry(tm, "job-4");
+        tm.cleanupOldJobs();
+        // cleanup must short-circuit before touching jobStore in cluster mode; the backplane
+        // TTL owns expiry there. If the gate fired correctly, delete is never called.
+        verify(jobStore, never()).delete(any());
+    }
+
+    @Test
+    void cleanupOldJobsDeletesFromJobStoreWhenBackplaneIsInProcess() {
+        taskManager.createTask("job-5");
+        taskManager.setComplete("job-5");
+        ageJobPastExpiry(taskManager, "job-5");
+        taskManager.cleanupOldJobs();
+        verify(jobStore).delete("job-5");
+    }
+
+    @SuppressWarnings("unchecked")
+    private static void ageJobPastExpiry(TaskManager tm, String jobId) {
+        var jobResults =
+                (java.util.Map<String, JobResult>) ReflectionTestUtils.getField(tm, "jobResults");
+        JobResult result = jobResults.get(jobId);
+        ReflectionTestUtils.setField(result, "completedAt", LocalDateTime.now().minusHours(2));
+        ReflectionTestUtils.setField(result, "complete", true);
+    }
+}

app/common/src/test/java/stirling/software/common/service/TaskManagerTest.java

@@ -1,20 +1,28 @@
 package stirling.software.common.service;
 
 import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.*;
 
 import java.time.LocalDateTime;
 import java.util.Map;
+import java.util.Optional;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 import org.springframework.http.MediaType;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import stirling.software.common.cluster.ClusterBackplane;
+import stirling.software.common.cluster.JobStore;
+import stirling.software.common.cluster.JobStoreEntry;
+import stirling.software.common.cluster.JobStoreEntry.JobState;
 import stirling.software.common.model.job.JobResult;
 import stirling.software.common.model.job.JobStats;
 import stirling.software.common.model.job.ResultFile;
@@ -22,6 +30,8 @@ import stirling.software.common.model.job.ResultFile;
 class TaskManagerTest {
 
     @Mock private FileStorage fileStorage;
+    @Mock private JobStore jobStore;
+    @Mock private ClusterBackplane clusterBackplane;
 
     @InjectMocks private TaskManager taskManager;
 
@@ -30,6 +40,10 @@ class TaskManagerTest {
     @BeforeEach
     void setUp() {
         closeable = MockitoAnnotations.openMocks(this);
+        // Treat the backplane as in-process so cleanupOldJobs is not short-circuited.
+        lenient().when(clusterBackplane.backplaneType()).thenReturn("inprocess");
+        lenient().when(clusterBackplane.localNodeId()).thenReturn("test-node");
+        lenient().when(clusterBackplane.shouldRunLocalCleanup()).thenReturn(true);
         ReflectionTestUtils.setField(taskManager, "jobResultExpiryMinutes", 30);
     }
 
@@ -270,6 +284,33 @@ class TaskManagerTest {
         verify(fileStorage).deleteFile("file-id");
     }
 
+    @Test
+    void testCleanupOldJobs_NoOpWhenBackplaneOwnsExpiry() {
+        // When the backplane reports it should NOT run local cleanup (e.g. a distributed
+        // backplane with its own TTL), the cleanup loop must leave local state untouched.
+        when(clusterBackplane.shouldRunLocalCleanup()).thenReturn(false);
+
+        // Seed an old completed job that would normally be removed.
+        String oldJobId = "old-job-distributed";
+        taskManager.createTask(oldJobId);
+        JobResult oldJob = taskManager.getJobResult(oldJobId);
+        ReflectionTestUtils.setField(oldJob, "completedAt", LocalDateTime.now().minusHours(1));
+        ReflectionTestUtils.setField(oldJob, "complete", true);
+
+        Map<String, JobResult> jobResultsMap =
+                (Map<String, JobResult>) ReflectionTestUtils.getField(taskManager, "jobResults");
+        assertNotNull(jobResultsMap);
+        assertTrue(jobResultsMap.containsKey(oldJobId));
+
+        // Act
+        taskManager.cleanupOldJobs();
+
+        // Assert: nothing was removed locally, and no jobStore.delete was issued.
+        assertTrue(jobResultsMap.containsKey(oldJobId));
+        verify(jobStore, never()).delete(anyString());
+        verify(fileStorage, never()).deleteFile(anyString());
+    }
+
     @Test
     void testShutdown() {
         // This mainly tests that the shutdown method doesn't throw exceptions
@@ -310,4 +351,33 @@ class TaskManagerTest {
         // Assert
         assertFalse(result);
     }
+
+    @Test
+    void testWriteThroughOnUpdate() {
+        // Mutating calls must write through to the injected JobStore.
+        String jobId = "write-through-job";
+        taskManager.createTask(jobId);
+        taskManager.setResult(jobId, "done");
+
+        ArgumentCaptor<JobStoreEntry> captor = ArgumentCaptor.forClass(JobStoreEntry.class);
+        verify(jobStore, atLeast(2)).put(captor.capture(), any());
+
+        JobStoreEntry last = captor.getValue();
+        assertEquals(jobId, last.jobId());
+        assertEquals(JobState.COMPLETE, last.state());
+        assertEquals("test-node", last.owningNodeId());
+    }
+
+    @Test
+    void testFindJobKeyByFileId_FallsBackToJobStore() {
+        // When the file id is not in the local map, TaskManager delegates to JobStore.
+        String fileId = "remote-file-id";
+        String expectedJobKey = "remote-job-key";
+        when(jobStore.findJobIdByFileId(fileId)).thenReturn(Optional.of(expectedJobKey));
+
+        String actual = taskManager.findJobKeyByFileId(fileId);
+
+        assertEquals(expectedJobKey, actual);
+        verify(jobStore).findJobIdByFileId(fileId);
+    }
 }

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

@@ -330,6 +330,24 @@ aiEngine:
   url: http://localhost:5001 # URL of the Python AI engine
   timeoutSeconds: 120 # Timeout in seconds for AI engine requests
 
+# Cluster configuration. NOT YET ENABLED - scaffolding for later work. Leave at defaults.
+cluster:
+  enabled: false # Master switch. 'false' (default) wires the in-process backplane and skips all cluster checks. Single-instance installs do not need to change anything here.
+  backplane: inprocess # Backplane implementation: 'inprocess' (single JVM only) or 'valkey' (multi-node via Valkey/Redis)
+  artifactStore: local # Transient cluster job-artifact backend: 'local' (per-node disk; single-node only) or 's3' (shared object store; required for multi-node). Distinct from 'storage.provider' which controls persistent user uploads - when both are 's3' they share the storage.s3.* credentials block. Multi-node deployments MUST set this to 's3'.
+  valkey:
+    url: "" # Valkey/Redis URL, e.g. 'redis://valkey:6379' or 'rediss://...' for TLS. Required when enabled=true and backplane=valkey.
+    tls:
+      skipCertVerification: false # set to 'true' to skip TLS certificate verification on Valkey connections (dev/test only)
+  node:
+    id: "" # Optional explicit node id. Blank = auto-generated UUID at startup.
+    role: both # 'web' (serves HTTP), 'worker' (runs jobs), or 'both' (default)
+    internalAddress: "" # host:port advertised in the instance registry for peer-to-peer cluster traffic. Blank = derived at startup.
+    scheme: http # 'http' or 'https' - scheme peers use to call this node's /internal/cluster/** endpoints
+    heartbeatIntervalMs: 5000 # Heartbeat publish interval for the instance registry (ms)
+  engine:
+    sharedSecret: "" # Shared secret for Java <-> Python AI engine auth. Generate via 'openssl rand -hex 32' and set BOTH this value and the engine pod's STIRLING_ENGINE_SHARED_SECRET env var to the same string. Blank = dev mode (auth disabled, NEVER in production).
+
 pdfEditor:
   fallback-font: classpath:/static/fonts/NotoSans-Regular.ttf # Override to point at a custom fallback font
   cache:

app/proprietary/src/main/java/stirling/software/proprietary/controller/api/AiEngineController.java

@@ -87,7 +87,7 @@ public class AiEngineController {
     @GetMapping("/health")
     @Operation(
             summary = "AI engine health check",
-            description = "Returns the health status of the AI engine including configured models")
+            description = "Returns the health status of the AI engine.")
     public ResponseEntity<String> health() throws IOException {
         String response = aiEngineClient.get("/health");
         return ResponseEntity.ok().contentType(MediaType.APPLICATION_JSON).body(response);

app/proprietary/src/main/java/stirling/software/proprietary/service/AiEngineClient.java

@@ -1,7 +1,9 @@
 package stirling.software.proprietary.service;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.URI;
+import java.net.UnknownHostException;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
@@ -12,9 +14,15 @@ import java.util.stream.Stream;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
+import org.springframework.security.authentication.AnonymousAuthenticationToken;
+import org.springframework.security.core.Authentication;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.security.core.userdetails.UserDetails;
 import org.springframework.stereotype.Service;
 import org.springframework.web.server.ResponseStatusException;
 
+import jakarta.annotation.PostConstruct;
+
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.model.ApplicationProperties;
@@ -23,6 +31,9 @@ import stirling.software.common.model.ApplicationProperties;
 @Service
 public class AiEngineClient {
 
+    static final String ENGINE_AUTH_HEADER = "X-Engine-Auth";
+    static final String USER_ID_HEADER = "X-User-Id";
+
     private final ApplicationProperties applicationProperties;
     private final HttpClient httpClient;
 
@@ -43,15 +54,113 @@ public class AiEngineClient {
         this.httpClient = httpClient;
     }
 
+    @PostConstruct
+    void warnIfEngineSecretSentOverPlaintextHttp() {
+        String engineSecret = resolveEngineSecret();
+        if (engineSecret == null || engineSecret.isBlank()) {
+            return;
+        }
+        String engineUrl = applicationProperties.getAiEngine().getUrl();
+        if (engineUrl == null || !engineUrl.startsWith("http://")) {
+            return;
+        }
+        URI uri;
+        try {
+            uri = URI.create(engineUrl);
+        } catch (IllegalArgumentException ex) {
+            log.warn(
+                    "Engine URL {} could not be parsed; skipping plaintext-secret check",
+                    engineUrl);
+            return;
+        }
+        String host = uri.getHost();
+        if (host == null || isLoopbackOrPrivateHost(host)) {
+            return;
+        }
+        log.error(
+                "SECURITY: engine shared secret will be sent over plaintext HTTP to a non-loopback /"
+                        + " non-private-network host: {}. Use https:// for the engine URL in"
+                        + " production.",
+                engineUrl);
+    }
+
+    private static boolean isLoopbackOrPrivateHost(String host) {
+        // Cover the literal forms first to avoid a DNS lookup on the common cases.
+        if (host.equalsIgnoreCase("localhost")
+                || host.equals("127.0.0.1")
+                || host.equals("::1")
+                || host.equals("0:0:0:0:0:0:0:1")) {
+            return true;
+        }
+        try {
+            InetAddress addr = InetAddress.getByName(host);
+            if (addr.isLoopbackAddress()
+                    || addr.isSiteLocalAddress()
+                    || addr.isLinkLocalAddress()) {
+                return true;
+            }
+            byte[] bytes = addr.getAddress();
+            if (bytes.length == 4) {
+                int b0 = bytes[0] & 0xFF;
+                int b1 = bytes[1] & 0xFF;
+                // 10/8
+                if (b0 == 10) return true;
+                // 172.16/12
+                if (b0 == 172 && b1 >= 16 && b1 <= 31) return true;
+                // 192.168/16
+                if (b0 == 192 && b1 == 168) return true;
+            }
+            return false;
+        } catch (UnknownHostException ex) {
+            // Cannot resolve - treat as remote so we err on the side of warning.
+            return false;
+        }
+    }
+
+    private HttpRequest.Builder decorate(HttpRequest.Builder builder) {
+        String secret = resolveEngineSecret();
+        if (secret != null && !secret.isBlank()) {
+            builder.header(ENGINE_AUTH_HEADER, secret);
+        }
+        String userId = resolveUserId();
+        if (userId != null && !userId.isBlank()) {
+            builder.header(USER_ID_HEADER, userId);
+        }
+        return builder;
+    }
+
+    private String resolveEngineSecret() {
+        ApplicationProperties.Cluster cluster = applicationProperties.getCluster();
+        if (cluster == null || cluster.getEngine() == null) {
+            return "";
+        }
+        return cluster.getEngine().getSharedSecret();
+    }
+
+    private String resolveUserId() {
+        try {
+            Authentication auth = SecurityContextHolder.getContext().getAuthentication();
+            if (auth == null
+                    || !auth.isAuthenticated()
+                    || auth instanceof AnonymousAuthenticationToken) {
+                return null;
+            }
+            Object principal = auth.getPrincipal();
+            if (principal instanceof UserDetails ud) {
+                return ud.getUsername();
+            }
+            return auth.getName();
+        } catch (RuntimeException ex) {
+            return null;
+        }
+    }
+
     public String post(String path, String jsonBody) throws IOException {
         ApplicationProperties.AiEngine config = applicationProperties.getAiEngine();
         return postWithTimeout(path, jsonBody, Duration.ofSeconds(config.getTimeoutSeconds()));
     }
 
-    /**
-     * POST with an explicit per-call timeout, for heavy operations (e.g. RAG ingestion of a large
-     * document) that legitimately take longer than the default timeout.
-     */
+    /** POST with an explicit per-call timeout for heavy operations (e.g. RAG ingestion). */
     public String postLongRunning(String path, String jsonBody) throws IOException {
         ApplicationProperties.AiEngine config = applicationProperties.getAiEngine();
         return postWithTimeout(
@@ -70,12 +179,13 @@ public class AiEngineClient {
         log.debug("Proxying AI engine request to {} (timeout {}s)", url, timeout.toSeconds());
 
         HttpRequest request =
-                HttpRequest.newBuilder()
-                        .uri(URI.create(url))
-                        .header("Content-Type", "application/json")
-                        .header("Accept", "application/json")
-                        .timeout(timeout)
-                        .POST(HttpRequest.BodyPublishers.ofString(jsonBody))
+                decorate(
+                                HttpRequest.newBuilder()
+                                        .uri(URI.create(url))
+                                        .header("Content-Type", "application/json")
+                                        .header("Accept", "application/json")
+                                        .timeout(timeout)
+                                        .POST(HttpRequest.BodyPublishers.ofString(jsonBody)))
                         .build();
 
         HttpResponse<String> response = sendRequest(request);
@@ -85,15 +195,7 @@ public class AiEngineClient {
         return response.body();
     }
 
-    /**
-     * POST a JSON body and consume the response as a stream of NDJSON lines. Each line is passed to
-     * {@code lineConsumer} in arrival order; the call returns when the engine closes the stream.
-     *
-     * <p>This is the right shape for long-running orchestrator calls that emit incremental
-     * progress. The total HTTP timeout is the long-running timeout (typically 600s+), but in
-     * practice line arrival keeps the connection logically alive: as long as the engine emits
-     * events, the work is progressing. Genuine engine hangs still hit the total timeout.
-     */
+    /** POST a JSON body and consume the response as a stream of NDJSON lines. */
     public void streamPost(String path, String jsonBody, Consumer<String> lineConsumer)
             throws IOException {
         ApplicationProperties.AiEngine config = applicationProperties.getAiEngine();
@@ -110,12 +212,13 @@ public class AiEngineClient {
                 timeout.toSeconds());
 
         HttpRequest request =
-                HttpRequest.newBuilder()
-                        .uri(URI.create(url))
-                        .header("Content-Type", "application/json")
-                        .header("Accept", "application/x-ndjson")
-                        .timeout(timeout)
-                        .POST(HttpRequest.BodyPublishers.ofString(jsonBody))
+                decorate(
+                                HttpRequest.newBuilder()
+                                        .uri(URI.create(url))
+                                        .header("Content-Type", "application/json")
+                                        .header("Accept", "application/x-ndjson")
+                                        .timeout(timeout)
+                                        .POST(HttpRequest.BodyPublishers.ofString(jsonBody)))
                         .build();
 
         HttpResponse<Stream<String>> response;
@@ -160,11 +263,12 @@ public class AiEngineClient {
         log.debug("Proxying AI engine GET request to {}", url);
 
         HttpRequest request =
-                HttpRequest.newBuilder()
-                        .uri(URI.create(url))
-                        .header("Accept", "application/json")
-                        .timeout(Duration.ofSeconds(config.getTimeoutSeconds()))
-                        .GET()
+                decorate(
+                                HttpRequest.newBuilder()
+                                        .uri(URI.create(url))
+                                        .header("Accept", "application/json")
+                                        .timeout(Duration.ofSeconds(config.getTimeoutSeconds()))
+                                        .GET())
                         .build();
 
         HttpResponse<String> response = sendRequest(request);
@@ -180,9 +284,6 @@ public class AiEngineClient {
         } catch (HttpTimeoutException e) {
             throw new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, "AI engine timed out", e);
         } catch (IOException e) {
-            // Connection refused, DNS failure, socket reset, etc. — surface as
-            // SERVICE_UNAVAILABLE so every caller of this client sees a structured
-            // status rather than a raw 500 from an unhandled IOException.
             throw new ResponseStatusException(
                     HttpStatus.SERVICE_UNAVAILABLE, "AI engine unreachable: " + e.getMessage(), e);
         } catch (InterruptedException e) {

app/proprietary/src/test/java/stirling/software/proprietary/service/AiEngineClientTest.java

@@ -1,31 +1,42 @@
 package stirling.software.proprietary.service;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.net.ConnectException;
 import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.net.http.HttpTimeoutException;
+import java.util.List;
+import java.util.stream.Stream;
 
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.springframework.http.HttpStatus;
+import org.springframework.security.authentication.AnonymousAuthenticationToken;
+import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
+import org.springframework.security.core.authority.SimpleGrantedAuthority;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.security.core.userdetails.User;
+import org.springframework.security.core.userdetails.UserDetails;
 import org.springframework.web.server.ResponseStatusException;
 
 import stirling.software.common.model.ApplicationProperties;
 
-/**
- * Verifies that AiEngineClient surfaces network-layer failures as structured HTTP statuses so every
- * AI tool caller sees a consistent, meaningful error rather than a raw 500.
- */
 class AiEngineClientTest {
 
+    private static final String ENGINE_SECRET = "shared-engine-secret";
+
     private ApplicationProperties applicationProperties;
     private HttpClient httpClient;
     private AiEngineClient client;
@@ -38,6 +49,12 @@ class AiEngineClientTest {
         applicationProperties.getAiEngine().setTimeoutSeconds(5);
         httpClient = mock(HttpClient.class);
         client = new AiEngineClient(applicationProperties, httpClient);
+        SecurityContextHolder.clearContext();
+    }
+
+    @AfterEach
+    void tearDown() {
+        SecurityContextHolder.clearContext();
     }
 
     @Test
@@ -84,4 +101,127 @@ class AiEngineClientTest {
 
         assertEquals(HttpStatus.SERVICE_UNAVAILABLE, ex.getStatusCode());
     }
+
+    // --- decorate() contract: X-Engine-Auth + X-User-Id stamping ---------------------------
+
+    @Test
+    void postStampsEngineAuthAndUserIdHeaders() throws Exception {
+        applicationProperties.getCluster().getEngine().setSharedSecret(ENGINE_SECRET);
+        authenticateAs("alice");
+        stubOkResponse();
+
+        client.post("/x", "{}");
+
+        HttpRequest sent = captureSentRequest();
+        assertThat(sent.headers().firstValue("X-Engine-Auth")).contains(ENGINE_SECRET);
+        assertThat(sent.headers().firstValue("X-User-Id")).contains("alice");
+    }
+
+    @Test
+    void getStampsEngineAuthAndUserIdHeaders() throws Exception {
+        applicationProperties.getCluster().getEngine().setSharedSecret(ENGINE_SECRET);
+        authenticateAs("bob");
+        stubOkResponse();
+
+        client.get("/x");
+
+        HttpRequest sent = captureSentRequest();
+        assertThat(sent.headers().firstValue("X-Engine-Auth")).contains(ENGINE_SECRET);
+        assertThat(sent.headers().firstValue("X-User-Id")).contains("bob");
+    }
+
+    @Test
+    void streamPostStampsEngineAuthAndUserIdHeaders() throws Exception {
+        applicationProperties.getCluster().getEngine().setSharedSecret(ENGINE_SECRET);
+        applicationProperties.getAiEngine().setLongRunningTimeoutSeconds(60);
+        authenticateAs("carol");
+        @SuppressWarnings("unchecked")
+        HttpResponse<Stream<String>> response =
+                (HttpResponse<Stream<String>>) mock(HttpResponse.class);
+        when(response.statusCode()).thenReturn(200);
+        when(response.body()).thenReturn(Stream.empty());
+        when(httpClient.send(any(), any(HttpResponse.BodyHandler.class))).thenReturn(response);
+
+        client.streamPost("/stream", "{}", line -> {});
+
+        HttpRequest sent = captureSentRequest();
+        assertThat(sent.headers().firstValue("X-Engine-Auth")).contains(ENGINE_SECRET);
+        assertThat(sent.headers().firstValue("X-User-Id")).contains("carol");
+    }
+
+    @Test
+    void postOmitsEngineAuthWhenSharedSecretBlank() throws Exception {
+        applicationProperties.getCluster().getEngine().setSharedSecret("");
+        authenticateAs("alice");
+        stubOkResponse();
+
+        client.post("/x", "{}");
+
+        HttpRequest sent = captureSentRequest();
+        assertThat(sent.headers().firstValue("X-Engine-Auth"))
+                .as("blank engine secret must NOT produce a header")
+                .isEmpty();
+        assertThat(sent.headers().firstValue("X-User-Id")).contains("alice");
+    }
+
+    @Test
+    void postOmitsUserIdHeaderWhenSecurityContextEmpty() throws Exception {
+        // Unauthenticated context: X-User-Id must be omitted; X-Engine-Auth still applies.
+        applicationProperties.getCluster().getEngine().setSharedSecret(ENGINE_SECRET);
+        SecurityContextHolder.clearContext();
+        stubOkResponse();
+
+        client.post("/x", "{}");
+
+        HttpRequest sent = captureSentRequest();
+        assertThat(sent.headers().firstValue("X-Engine-Auth")).contains(ENGINE_SECRET);
+        assertThat(sent.headers().firstValue("X-User-Id"))
+                .as("anonymous context must NOT stamp an X-User-Id header")
+                .isEmpty();
+    }
+
+    @Test
+    void postOmitsUserIdHeaderForAnonymousAuthenticationToken() throws Exception {
+        // AnonymousAuthenticationToken#isAuthenticated() returns true and getName() ==
+        // "anonymousUser". resolveUserId() must special-case it so the engine never sees
+        // anonymous traffic conflated under the literal identity "anonymousUser".
+        applicationProperties.getCluster().getEngine().setSharedSecret(ENGINE_SECRET);
+        SecurityContextHolder.getContext()
+                .setAuthentication(
+                        new AnonymousAuthenticationToken(
+                                "key",
+                                "anonymousUser",
+                                List.of(new SimpleGrantedAuthority("ROLE_ANONYMOUS"))));
+        stubOkResponse();
+
+        client.post("/x", "{}");
+
+        HttpRequest sent = captureSentRequest();
+        assertThat(sent.headers().firstValue("X-User-Id"))
+                .as("anonymous principal must NOT stamp an X-User-Id header")
+                .isEmpty();
+    }
+
+    private void authenticateAs(String username) {
+        UserDetails principal =
+                User.withUsername(username).password("n/a").authorities("USER").build();
+        SecurityContextHolder.getContext()
+                .setAuthentication(
+                        new UsernamePasswordAuthenticationToken(
+                                principal, "n/a", principal.getAuthorities()));
+    }
+
+    @SuppressWarnings("unchecked")
+    private void stubOkResponse() throws Exception {
+        HttpResponse<String> response = (HttpResponse<String>) mock(HttpResponse.class);
+        when(response.statusCode()).thenReturn(200);
+        when(response.body()).thenReturn("{}");
+        when(httpClient.send(any(), any(HttpResponse.BodyHandler.class))).thenReturn(response);
+    }
+
+    private HttpRequest captureSentRequest() throws Exception {
+        ArgumentCaptor<HttpRequest> captor = ArgumentCaptor.forClass(HttpRequest.class);
+        verify(httpClient).send(captor.capture(), any(HttpResponse.BodyHandler.class));
+        return captor.getValue();
+    }
 }

app/saas/src/main/java/stirling/software/saas/ai/controller/AiCreateController.java

@@ -38,8 +38,8 @@ import stirling.software.proprietary.security.database.repository.UserRepository
 import stirling.software.proprietary.security.model.User;
 import stirling.software.saas.ai.model.AiCreateSession;
 import stirling.software.saas.ai.repository.AiCreateSessionRepository;
-import stirling.software.saas.ai.service.AiCreateProxyService;
 import stirling.software.saas.ai.service.AiCreateSessionService;
+import stirling.software.saas.ai.service.AiProxyService;
 import stirling.software.saas.service.CreditService;
 import stirling.software.saas.service.TeamCreditService;
 import stirling.software.saas.util.AuthenticationUtils;
@@ -53,7 +53,7 @@ import stirling.software.saas.util.CreditHeaderUtils;
 public class AiCreateController {
 
     private final AiCreateSessionService sessionService;
-    private final AiCreateProxyService proxyService;
+    private final AiProxyService proxyService;
     private final ObjectMapper objectMapper = new ObjectMapper();
     private final CreditService creditService;
     private final TeamCreditService teamCreditService;

app/saas/src/main/java/stirling/software/saas/ai/service/AiCreateProxyService.java

@@ -1,145 +0,0 @@
-package stirling.software.saas.ai.service;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.UncheckedIOException;
-import java.net.URI;
-import java.net.http.HttpClient;
-import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
-
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Profile;
-import org.springframework.stereotype.Service;
-
-import jakarta.servlet.http.HttpServletRequest;
-
-import lombok.extern.slf4j.Slf4j;
-
-import stirling.software.proprietary.security.database.repository.UserRepository;
-import stirling.software.proprietary.security.service.UserService;
-
-@Service
-@Profile("saas")
-@Slf4j
-public class AiCreateProxyService {
-
-    private static final String DEFAULT_AI_BASE_URL = "http://localhost:5001";
-
-    private final String aiServiceBaseUrl;
-
-    private final HttpClient httpClient;
-    private final UserRepository userRepository;
-    private final UserService userService;
-
-    public AiCreateProxyService(
-            @Value("${app.ai.service-base-url:" + DEFAULT_AI_BASE_URL + "}")
-                    String aiServiceBaseUrl,
-            UserRepository userRepository,
-            UserService userService) {
-        this.aiServiceBaseUrl = aiServiceBaseUrl;
-        this.httpClient = HttpClient.newBuilder().build();
-        this.userRepository = userRepository;
-        this.userService = userService;
-    }
-
-    public HttpResponse<InputStream> forward(
-            String method, String path, HttpServletRequest request, boolean acceptEventStream)
-            throws IOException, InterruptedException {
-        String targetUrl = buildTargetUrl(path, request.getQueryString());
-        HttpRequest.Builder builder = HttpRequest.newBuilder(URI.create(targetUrl));
-
-        String contentType = request.getContentType();
-        if (contentType != null && !contentType.isBlank()) {
-            builder.header("Content-Type", contentType);
-        }
-
-        String authorization = request.getHeader("Authorization");
-        if (authorization != null && !authorization.isBlank()) {
-            builder.header("Authorization", authorization);
-        }
-
-        // Extract user API key from authenticated user and forward to AI backend
-        String apiKey = request.getHeader("X-API-KEY");
-        if (apiKey == null || apiKey.isBlank()) {
-            apiKey = extractUserApiKey();
-        }
-        if (apiKey != null && !apiKey.isBlank()) {
-            builder.header("X-API-KEY", apiKey);
-            log.debug("Forwarding X-API-KEY header to AI backend");
-        }
-
-        String accept = request.getHeader("Accept");
-        if (acceptEventStream) {
-            builder.header("Accept", "text/event-stream");
-        } else if (accept != null && !accept.isBlank()) {
-            builder.header("Accept", accept);
-        }
-
-        builder.method(method, buildBodyPublisher(method, request));
-        log.debug("Proxying AI create request {} {}", method, targetUrl);
-        return httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofInputStream());
-    }
-
-    private String buildTargetUrl(String path, String queryString) {
-        String baseUrl = aiServiceBaseUrl;
-        if (baseUrl == null || baseUrl.isBlank()) {
-            baseUrl = DEFAULT_AI_BASE_URL;
-        }
-        baseUrl = baseUrl.trim();
-        if (baseUrl.endsWith("/")) {
-            baseUrl = baseUrl.substring(0, baseUrl.length() - 1);
-        }
-        if (!path.startsWith("/")) {
-            path = "/" + path;
-        }
-        String url = baseUrl + path;
-        if (queryString != null && !queryString.isBlank()) {
-            url += "?" + queryString;
-        }
-        return url;
-    }
-
-    private HttpRequest.BodyPublisher buildBodyPublisher(
-            String method, HttpServletRequest request) {
-        if ("GET".equalsIgnoreCase(method) || "DELETE".equalsIgnoreCase(method)) {
-            return HttpRequest.BodyPublishers.noBody();
-        }
-        return HttpRequest.BodyPublishers.ofInputStream(
-                () -> {
-                    try {
-                        return request.getInputStream();
-                    } catch (IOException exc) {
-                        throw new UncheckedIOException(exc);
-                    }
-                });
-    }
-
-    /**
-     * Extract the authenticated user's API key from the database. If the user doesn't have an API
-     * key, one will be created automatically.
-     *
-     * @return The user's API key, or null if not authenticated or key creation fails
-     */
-    private String extractUserApiKey() {
-        try {
-            // Use getCurrentUsername() which handles all auth types including anonymous users
-            String username = userService.getCurrentUsername();
-            if (username == null || username.isBlank()) {
-                log.debug("No authenticated user found for API key extraction");
-                return null;
-            }
-
-            // getApiKeyForUser will create a key if it doesn't exist
-            String apiKey = userService.getApiKeyForUser(username);
-            log.debug("Retrieved API key for user: {}", username);
-            return apiKey;
-        } catch (Exception e) {
-            log.error(
-                    "Failed to extract or create user API key for user: {}",
-                    userService.getCurrentUsername(),
-                    e);
-            return null;
-        }
-    }
-}

app/saas/src/main/java/stirling/software/saas/ai/service/AiProxyService.java

@@ -20,9 +20,20 @@ import jakarta.servlet.http.Part;
 
 import lombok.extern.slf4j.Slf4j;
 
+import stirling.software.common.model.ApplicationProperties;
 import stirling.software.proprietary.security.database.repository.UserRepository;
 import stirling.software.proprietary.security.service.UserService;
 
+/**
+ * Proxies saas-side HTTP requests to the AI engine and enforces the auth-header contract:
+ *
+ * <ul>
+ *   <li>Client-supplied {@code Authorization} and {@code X-API-KEY} are stripped; the server-
+ *       resolved API key is stamped so callers cannot spoof identity downstream.
+ *   <li>{@code X-Engine-Auth} is stamped when the cluster shared secret is configured.
+ *   <li>{@code X-User-Id} is stamped for authenticated principals; omitted for anonymous.
+ * </ul>
+ */
 @Service
 @Profile("saas")
 @Slf4j
@@ -30,21 +41,29 @@ public class AiProxyService {
 
     private static final String DEFAULT_AI_BASE_URL = "http://localhost:5001";
 
+    private static final String ENGINE_AUTH_HEADER = "X-Engine-Auth";
+    private static final String USER_ID_HEADER = "X-User-Id";
+    private static final String API_KEY_HEADER = "X-API-KEY";
+    private static final String AUTHORIZATION_HEADER = "Authorization";
+
     private final String aiServiceBaseUrl;
 
     private final HttpClient httpClient;
     private final UserRepository userRepository;
     private final UserService userService;
+    private final ApplicationProperties applicationProperties;
 
     public AiProxyService(
             @Value("${app.ai.service-base-url:" + DEFAULT_AI_BASE_URL + "}")
                     String aiServiceBaseUrl,
             UserRepository userRepository,
-            UserService userService) {
+            UserService userService,
+            ApplicationProperties applicationProperties) {
         this.aiServiceBaseUrl = aiServiceBaseUrl;
         this.httpClient = HttpClient.newBuilder().build();
         this.userRepository = userRepository;
         this.userService = userService;
+        this.applicationProperties = applicationProperties;
     }
 
     public HttpResponse<InputStream> forward(
@@ -55,21 +74,6 @@ public class AiProxyService {
 
         String contentType = request.getContentType();
 
-        String authorization = request.getHeader("Authorization");
-        if (authorization != null && !authorization.isBlank()) {
-            builder.header("Authorization", authorization);
-        }
-
-        // Extract user API key from authenticated user and forward to AI backend
-        String apiKey = request.getHeader("X-API-KEY");
-        if (apiKey == null || apiKey.isBlank()) {
-            apiKey = extractUserApiKey();
-        }
-        if (apiKey != null && !apiKey.isBlank()) {
-            builder.header("X-API-KEY", apiKey);
-            log.debug("Forwarding X-API-KEY header to AI backend");
-        }
-
         String accept = request.getHeader("Accept");
         if (acceptEventStream) {
             builder.header("Accept", "text/event-stream");
@@ -84,10 +88,68 @@ public class AiProxyService {
             builder.header("Content-Type", contentType);
         }
         builder.method(method, body.publisher);
+
+        // INVARIANT: stamp auth headers LAST. setHeader() overrides any previously-set value,
+        // including any caller-supplied Authorization / X-API-KEY that arrived via this proxy.
+        // Any .header(...) call AFTER this point would defeat the strip-on-output guarantee.
+        stampAuthHeaders(builder);
+
         log.debug("Proxying AI request {} {}", method, targetUrl);
         return httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofInputStream());
     }
 
+    private void stampAuthHeaders(HttpRequest.Builder builder) {
+        // setHeader (not header) with empty string causes the JDK builder to omit the header
+        // entirely in the built request, dropping any caller-supplied value.
+        builder.setHeader(AUTHORIZATION_HEADER, "");
+
+        String apiKey = extractUserApiKey();
+        if (apiKey != null && !apiKey.isBlank()) {
+            builder.setHeader(API_KEY_HEADER, apiKey);
+            log.debug("Attaching server-resolved X-API-KEY for the engine call");
+        } else {
+            builder.setHeader(API_KEY_HEADER, "");
+        }
+
+        String engineSecret = resolveEngineSecret();
+        if (engineSecret != null && !engineSecret.isBlank()) {
+            builder.setHeader(ENGINE_AUTH_HEADER, engineSecret);
+        }
+
+        try {
+            String username = userService.getCurrentUsername();
+            if (username != null && !username.isBlank() && !"anonymousUser".equals(username)) {
+                builder.setHeader(USER_ID_HEADER, username);
+            }
+        } catch (RuntimeException ex) {
+            log.warn("Could not resolve current username for X-User-Id header", ex);
+        }
+    }
+
+    private String resolveEngineSecret() {
+        ApplicationProperties.Cluster cluster = applicationProperties.getCluster();
+        if (cluster == null || cluster.getEngine() == null) {
+            return "";
+        }
+        return cluster.getEngine().getSharedSecret();
+    }
+
+    private String extractUserApiKey() {
+        try {
+            String username = userService.getCurrentUsername();
+            if (username == null || username.isBlank()) {
+                log.debug("No authenticated user found for API key extraction");
+                return null;
+            }
+            String apiKey = userService.getApiKeyForUser(username);
+            log.debug("Retrieved API key for user: {}", username);
+            return apiKey;
+        } catch (Exception e) {
+            log.error("Failed to extract or create user API key", e);
+            return null;
+        }
+    }
+
     private String buildTargetUrl(String path, String queryString) {
         String baseUrl = aiServiceBaseUrl;
         if (baseUrl == null || baseUrl.isBlank()) {
@@ -176,34 +238,6 @@ public class AiProxyService {
         output.write("\r\n".getBytes(StandardCharsets.UTF_8));
     }
 
-    /**
-     * Extract the authenticated user's API key from the database. If the user doesn't have an API
-     * key, one will be created automatically.
-     *
-     * @return The user's API key, or null if not authenticated or key creation fails
-     */
-    private String extractUserApiKey() {
-        try {
-            // Use getCurrentUsername() which handles all auth types including anonymous users
-            String username = userService.getCurrentUsername();
-            if (username == null || username.isBlank()) {
-                log.debug("No authenticated user found for API key extraction");
-                return null;
-            }
-
-            // getApiKeyForUser will create a key if it doesn't exist
-            String apiKey = userService.getApiKeyForUser(username);
-            log.debug("Retrieved API key for user: {}", username);
-            return apiKey;
-        } catch (Exception e) {
-            log.error(
-                    "Failed to extract or create user API key for user: {}",
-                    userService.getCurrentUsername(),
-                    e);
-            return null;
-        }
-    }
-
     private static class BodyPublisherWithContentType {
         private final HttpRequest.BodyPublisher publisher;
         private final String contentType;

app/saas/src/test/java/stirling/software/saas/ai/service/AiProxyServiceTest.java

@@ -0,0 +1,157 @@
+package stirling.software.saas.ai.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.when;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.Optional;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.security.database.repository.UserRepository;
+import stirling.software.proprietary.security.service.UserService;
+
+/** Verifies the auth-header strip-and-stamp contract on the saas AI proxy. */
+@ExtendWith(MockitoExtension.class)
+class AiProxyServiceTest {
+
+    private static final String ENGINE_URL = "http://engine.local:5001";
+    private static final String ENGINE_SECRET = "shared-engine-secret";
+    private static final String SERVER_API_KEY = "server-resolved-api-key";
+    private static final String CURRENT_USERNAME = "alice";
+
+    @Mock UserRepository userRepository;
+    @Mock UserService userService;
+    @Mock HttpClient httpClient;
+    @Mock HttpResponse<InputStream> mockResponse;
+
+    private ApplicationProperties applicationProperties;
+    private AiProxyService proxy;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        applicationProperties = new ApplicationProperties();
+        applicationProperties.getCluster().getEngine().setSharedSecret(ENGINE_SECRET);
+
+        proxy = new AiProxyService(ENGINE_URL, userRepository, userService, applicationProperties);
+        // Replace the internally-constructed HttpClient with our mock so we can capture requests.
+        ReflectionTestUtils.setField(proxy, "httpClient", httpClient);
+
+        lenient().when(mockResponse.body()).thenReturn(new ByteArrayInputStream(new byte[0]));
+        lenient().when(mockResponse.statusCode()).thenReturn(200);
+        when(httpClient.send(any(HttpRequest.class), any(HttpResponse.BodyHandler.class)))
+                .thenReturn(mockResponse);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"/api/v1/chat", "/api/create/sessions"})
+    void forward_dropsClientAuthorizationAndApiKeyHeaders(String path) throws Exception {
+        when(userService.getCurrentUsername()).thenReturn(CURRENT_USERNAME);
+        when(userService.getApiKeyForUser(CURRENT_USERNAME)).thenReturn(SERVER_API_KEY);
+
+        MockHttpServletRequest inbound = new MockHttpServletRequest("POST", path);
+        inbound.addHeader("Authorization", "Bearer client-supplied-bearer");
+        inbound.addHeader("X-API-KEY", "client-supplied-api-key");
+        inbound.setContentType("application/json");
+        inbound.setContent("{}".getBytes());
+
+        proxy.forward("POST", path, inbound, false);
+
+        HttpRequest outbound = captureOutboundRequest();
+
+        // setHeader with empty string causes the JDK builder to omit the header or send it empty.
+        assertThat(outbound.headers().firstValue("Authorization").orElse(""))
+                .as("client Authorization must never reach the engine")
+                .isEmpty();
+        assertThat(outbound.headers().allValues("Authorization"))
+                .as("client Authorization value must not be among the outbound values")
+                .doesNotContain("Bearer client-supplied-bearer");
+
+        Optional<String> apiKey = outbound.headers().firstValue("X-API-KEY");
+        assertThat(apiKey)
+                .as("X-API-KEY must be the server-resolved value, not the client-supplied one")
+                .contains(SERVER_API_KEY);
+        assertThat(apiKey.orElse(""))
+                .as("client-supplied X-API-KEY must not leak through")
+                .isNotEqualTo("client-supplied-api-key");
+        assertThat(outbound.headers().allValues("X-API-KEY"))
+                .as("client X-API-KEY must not be among the outbound values")
+                .doesNotContain("client-supplied-api-key");
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"/api/v1/chat", "/api/create/sessions"})
+    void forward_stampsEngineAuthAndUserIdHeaders(String path) throws Exception {
+        when(userService.getCurrentUsername()).thenReturn(CURRENT_USERNAME);
+        when(userService.getApiKeyForUser(CURRENT_USERNAME)).thenReturn(SERVER_API_KEY);
+
+        MockHttpServletRequest inbound = new MockHttpServletRequest("POST", path);
+        inbound.setContentType("application/json");
+        inbound.setContent("{}".getBytes());
+
+        proxy.forward("POST", path, inbound, false);
+
+        HttpRequest outbound = captureOutboundRequest();
+
+        assertThat(outbound.headers().firstValue("X-Engine-Auth"))
+                .as("engine shared secret must be stamped on outbound requests")
+                .contains(ENGINE_SECRET);
+        assertThat(outbound.headers().firstValue("X-User-Id"))
+                .as("authenticated username must be stamped on outbound requests")
+                .contains(CURRENT_USERNAME);
+    }
+
+    @Test
+    void forward_omitsEngineAuthWhenSecretBlank() throws Exception {
+        applicationProperties.getCluster().getEngine().setSharedSecret("");
+        when(userService.getCurrentUsername()).thenReturn(CURRENT_USERNAME);
+        when(userService.getApiKeyForUser(CURRENT_USERNAME)).thenReturn(SERVER_API_KEY);
+
+        MockHttpServletRequest inbound = new MockHttpServletRequest("GET", "/api/v1/health");
+
+        proxy.forward("GET", "/api/v1/health", inbound, false);
+
+        HttpRequest outbound = captureOutboundRequest();
+        assertThat(outbound.headers().firstValue("X-Engine-Auth"))
+                .as("blank engine secret must not produce a header")
+                .isEmpty();
+    }
+
+    @Test
+    void forward_omitsUserIdForAnonymousPrincipal() throws Exception {
+        when(userService.getCurrentUsername()).thenReturn("anonymousUser");
+
+        MockHttpServletRequest inbound =
+                new MockHttpServletRequest("GET", "/api/create/sessions/abc");
+
+        proxy.forward("GET", "/api/create/sessions/abc", inbound, false);
+
+        HttpRequest outbound = captureOutboundRequest();
+        assertThat(outbound.headers().firstValue("X-User-Id"))
+                .as("anonymous principal must not produce an X-User-Id header")
+                .isEmpty();
+    }
+
+    private HttpRequest captureOutboundRequest() throws Exception {
+        ArgumentCaptor<HttpRequest> captor = ArgumentCaptor.forClass(HttpRequest.class);
+        org.mockito.Mockito.verify(httpClient)
+                .send(captor.capture(), any(HttpResponse.BodyHandler.class));
+        return captor.getValue();
+    }
+}

engine/src/stirling/api/app.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
+import logging
 from contextlib import asynccontextmanager
-from typing import Annotated
 
-from fastapi import Depends, FastAPI
+from fastapi import FastAPI
 from pydantic_ai import Agent
 from pydantic_ai.models.instrumented import InstrumentationSettings
 
@@ -16,7 +16,7 @@ from stirling.agents import (
 )
 from stirling.agents.ledger import MathAuditorAgent
 from stirling.agents.pdf_comment import PdfCommentAgent
-from stirling.api.middleware import UserIdMiddleware
+from stirling.api.middleware import EngineAuthMiddleware, UserIdMiddleware
 from stirling.api.routes import (
     agent_draft_router,
     document_router,
@@ -63,7 +63,21 @@ async def lifespan(fast_api: FastAPI):
 
 
 app = FastAPI(title="Stirling AI Engine", lifespan=lifespan, version="0.1.0")
+
+try:
+    _engine_shared_secret = load_settings().engine_shared_secret or ""
+except (AttributeError, KeyError) as cfg_err:
+    raise RuntimeError(
+        "engine_shared_secret missing from settings; ensure STIRLING_ENGINE_SHARED_SECRET "
+        "is declared in the env (blank value is allowed for dev mode)."
+    ) from cfg_err
+if not _engine_shared_secret:
+    logging.getLogger(__name__).warning(
+        "STIRLING_ENGINE_SHARED_SECRET is blank - running in dev (open) mode."
+    )
+
 app.add_middleware(UserIdMiddleware)
+app.add_middleware(EngineAuthMiddleware, expected_secret=_engine_shared_secret)
 app.include_router(orchestrator_router)
 app.include_router(pdf_edit_router)
 app.include_router(pdf_question_router)
@@ -75,9 +89,5 @@ app.include_router(pdf_comments_router)
 
 
 @app.get("/health", response_model=HealthResponse)
-async def healthcheck(settings: Annotated[AppSettings, Depends(load_settings)]) -> HealthResponse:
-    return HealthResponse(
-        status="ok",
-        smart_model=settings.smart_model_name,
-        fast_model=settings.fast_model_name,
-    )
+async def healthcheck() -> HealthResponse:
+    return HealthResponse(status="ok")

engine/src/stirling/api/middleware.py

@@ -1,16 +1,20 @@
 from __future__ import annotations
 
+import hmac
+
 from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
 from starlette.requests import Request
-from starlette.responses import Response
+from starlette.responses import JSONResponse, Response
 
 from stirling.services.tracking import current_user_id
 
 _USER_ID_HEADER = "X-User-Id"
+_ENGINE_AUTH_HEADER = "X-Engine-Auth"
+_HEALTH_PATHS = {"/health", "/healthz", "/readyz"}
 
 
 class UserIdMiddleware(BaseHTTPMiddleware):
-    """Extract X-User-Id header and set it as the current user for PostHog tracking."""
+    """Set X-User-Id (stamped by the trusted Java proxy) as request context."""
 
     async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
         user_id = request.headers.get(_USER_ID_HEADER)
@@ -21,3 +25,21 @@ class UserIdMiddleware(BaseHTTPMiddleware):
             finally:
                 current_user_id.reset(token)
         return await call_next(request)
+
+
+class EngineAuthMiddleware(BaseHTTPMiddleware):
+    """Validate shared-secret header. Blank secret = dev mode (open); health probes exempt."""
+
+    def __init__(self, app, expected_secret: str) -> None:
+        super().__init__(app)
+        self._expected_secret = expected_secret or ""
+        # Precompute the bytes form once so per-request work is just the constant-time compare.
+        self._expected_secret_bytes = self._expected_secret.encode("utf-8")
+
+    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
+        if request.url.path in _HEALTH_PATHS or not self._expected_secret:
+            return await call_next(request)
+        presented = request.headers.get(_ENGINE_AUTH_HEADER, "")
+        if not hmac.compare_digest(presented.encode("utf-8"), self._expected_secret_bytes):
+            return JSONResponse({"detail": "engine authentication failed"}, status_code=401)
+        return await call_next(request)

engine/src/stirling/config/settings.py

@@ -98,6 +98,9 @@ class AppSettings(BaseSettings):
     posthog_api_key: str = Field(validation_alias="STIRLING_POSTHOG_API_KEY")
     posthog_host: str = Field(validation_alias="STIRLING_POSTHOG_HOST")
 
+    # Engine shared-secret for cluster deployments. Blank = dev mode (open).
+    engine_shared_secret: str = Field(default="", validation_alias="STIRLING_ENGINE_SHARED_SECRET")
+
 
 def _configure_logging(level_name: str, log_file: str, http_debug: bool) -> None:
     """Configure the ``stirling`` logger hierarchy."""

engine/src/stirling/contracts/health.py

@@ -5,5 +5,3 @@ from stirling.models import ApiModel
 
 class HealthResponse(ApiModel):
     status: str
-    smart_model: str
-    fast_model: str

engine/tests/test_engine_auth_middleware.py

@@ -0,0 +1,59 @@
+"""Tests for EngineAuthMiddleware - shared-secret gating and health probe exemption."""
+from __future__ import annotations
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from stirling.api.middleware import EngineAuthMiddleware
+
+
+def _app(expected_secret: str) -> FastAPI:
+    app = FastAPI()
+    app.add_middleware(EngineAuthMiddleware, expected_secret=expected_secret)
+
+    @app.get("/health")
+    async def health():
+        return {"status": "ok"}
+
+    @app.get("/v1/agent")
+    async def agent():
+        return {"ok": True}
+
+    return app
+
+
+def test_correct_secret_allows_request():
+    client = TestClient(_app("s3cret"))
+    res = client.get("/v1/agent", headers={"X-Engine-Auth": "s3cret"})
+    assert res.status_code == 200
+    assert res.json() == {"ok": True}
+
+
+def test_missing_header_rejected_with_401():
+    client = TestClient(_app("s3cret"))
+    res = client.get("/v1/agent")
+    assert res.status_code == 401
+
+
+def test_wrong_header_rejected_with_401():
+    client = TestClient(_app("s3cret"))
+    res = client.get("/v1/agent", headers={"X-Engine-Auth": "wrong"})
+    assert res.status_code == 401
+
+
+def test_blank_secret_dev_mode_allows_unauthenticated():
+    client = TestClient(_app(""))
+    res = client.get("/v1/agent")
+    assert res.status_code == 200
+
+
+def test_health_endpoint_exempt_from_auth():
+    client = TestClient(_app("s3cret"))
+    res = client.get("/health")
+    assert res.status_code == 200
+
+
+def test_health_endpoint_exempt_in_dev_mode():
+    client = TestClient(_app(""))
+    res = client.get("/health")
+    assert res.status_code == 200

engine/tests/test_user_context.py

@@ -0,0 +1,50 @@
+"""Tests for UserIdMiddleware - X-User-Id propagates as context.
+
+X-Tenant-Id is deliberately NOT accepted from the wire: the Java proxy does not stamp
+one and a client-supplied tenant id would be spoofable. If/when tenant scoping arrives
+it must be derived from the server-side security context, not from request headers.
+"""
+from __future__ import annotations
+
+from fastapi import FastAPI, Request
+from fastapi.testclient import TestClient
+
+from stirling.api.middleware import UserIdMiddleware
+from stirling.services.tracking import current_user_id
+
+
+def _app() -> FastAPI:
+    app = FastAPI()
+    app.add_middleware(UserIdMiddleware)
+
+    @app.get("/me")
+    async def me(request: Request):
+        return {
+            "user_id": current_user_id.get(),
+            "tenant_id": getattr(request.state, "tenant_id", None),
+        }
+
+    return app
+
+
+def test_user_id_header_is_propagated():
+    client = TestClient(_app())
+    res = client.get("/me", headers={"X-User-Id": "alice"})
+    assert res.status_code == 200
+    assert res.json()["user_id"] == "alice"
+
+
+def test_tenant_id_header_is_ignored():
+    client = TestClient(_app())
+    res = client.get("/me", headers={"X-User-Id": "alice", "X-Tenant-Id": "acme"})
+    assert res.status_code == 200
+    body = res.json()
+    assert body["user_id"] == "alice"
+    assert body["tenant_id"] is None
+
+
+def test_missing_user_id_returns_no_user_in_context():
+    client = TestClient(_app())
+    res = client.get("/me")
+    assert res.status_code == 200
+    assert res.json()["user_id"] in ("", None)