fix(agents): wire the deployment-wait machinery that only the ZIP importer ever used

docs/changelog.md

@@ -5,6 +5,17 @@
 
 ---
 
+## 🔗 fix(agents): wire the deployment-wait machinery that only the ZIP importer ever used (2026-08-09)
+
+**Repo:** EDDI (`fix/deployment-wait-machinery`)
+
+Wave F of the Agent / Group Agent review. `AgentFactory.getAgent` has always had a branch for "the agent is deploying right now" — `waitForDeploymentCompletion`, which awaits a future from `DeploymentListener`. That future was only ever registered by **one** caller in all of `src/main`: `RestImportService`, the startup ZIP importer. Every ordinary deploy fired `onDeploymentEvent` but never registered, so `getRegisteredDeploymentEvent` returned `null`, the wait had nothing to await, and a caller racing a deployment simply got a null agent. The machinery was dead outside one flow while reading as live.
+
+- `RestAgentAdministration.deploy` now registers **before** starting the deployment. Ordering matters: `agentFactory.deployAgent` is what publishes the IN_PROGRESS placeholder a waiter can observe, so registering afterwards would leave exactly the window this closes.
+- `DeploymentListener.registerAgentDeployment` self-expires. The map was pruned only by an arriving `DeploymentEvent`, so a registration whose event never came — a rejected deployment callable, a process that died mid-deploy — stayed for the lifetime of the JVM. Registrations now carry a `REGISTRATION_TTL` (5 minutes, a leak bound rather than a deployment SLA) and remove themselves on *any* completion. Removal is `remove(key, future)`, not `remove(key)`, so a stale completion cannot evict a live registration for the same agent.
+- `RestImportService`'s `allOf(...).join()` is now tolerant. It could previously only block forever; with self-expiring registrations it can complete exceptionally, and one initial agent that never reports must not hang startup — it logs and continues with the agents that did deploy.
+
+Suites: 347 tests green across `DeploymentListener*`, `RestAgentAdministration*`, `AgentFactory*`, `RestImportService*`; 12 new.
 ## 📚 docs(groups): attachments, protocol defaults, context scopes, and a "not yet supported" section (2026-08-09)
 
 **Repo:** EDDI (`docs/group-agent-accuracy`)

src/main/java/ai/labs/eddi/backup/impl/RestImportService.java

@@ -146,8 +146,22 @@ public List<AgentDeploymentStatus> importInitialAgents() {
                 }
             }
 
-            // Wait for all deployments to complete
-            CompletableFuture.allOf(deploymentFutures.toArray(new CompletableFuture[0])).join();
+            // Wait for all deployments to complete.
+            //
+            // Tolerant of a failure, and bounded: a registration now self-expires
+            // (DeploymentListener.REGISTRATION_TTL), so this can complete
+            // exceptionally where it previously could only block forever. One initial
+            // agent that never reports must not hang startup — log it and carry on
+            // with the ones that did deploy.
+            CompletableFuture.allOf(deploymentFutures.toArray(new CompletableFuture[0]))
+                    .handle((ignored, error) -> {
+                        if (error != null) {
+                            LOGGER.warnf("Not every initial agent reported a deployment event (%s) — continuing with those that did",
+                                    error.getMessage());
+                        }
+                        return null;
+                    })
+                    .join();
 
             LOGGER.info("Imported & Deployed Initial Agents");
             return restAgentAdministration.getDeploymentStatuses(production);

src/main/java/ai/labs/eddi/engine/internal/RestAgentAdministration.java

@@ -214,6 +214,18 @@ private void enforceAgentQuota(Deployment.Environment environment, String agentI
     }
 
     private Future<Void> deploy(final Deployment.Environment environment, final String agentId, final Integer version, final Boolean autoDeploy) {
+        // Register BEFORE the deployment starts, so a concurrent getAgent that finds
+        // the agent IN_PROGRESS has a future to await.
+        //
+        // This is what makes AgentFactory's wait machinery reachable at all. Only
+        // RestImportService ever registered, so for every ordinary deploy
+        // getRegisteredDeploymentEvent returned null, waitForDeploymentCompletion had
+        // nothing to await, and a caller racing a deployment simply got a null agent.
+        // Registration must precede agentFactory.deployAgent: that call is what
+        // publishes the IN_PROGRESS placeholder a waiter can observe, so registering
+        // afterwards would leave exactly the window this closes.
+        deploymentListener.registerAgentDeployment(agentId, version);
+
         Callable<Void> deployAgentCallable = () -> {
             try {
                 if (EnumSet.of(NOT_FOUND, ERROR).contains(checkDeploymentStatus(environment, agentId, version))) {

src/main/java/ai/labs/eddi/engine/runtime/internal/DeploymentListener.java

@@ -5,26 +5,73 @@
 package ai.labs.eddi.engine.runtime.internal;
 
 import ai.labs.eddi.engine.runtime.model.DeploymentEvent;
+import ai.labs.eddi.utils.LogSanitizer;
 import jakarta.enterprise.context.ApplicationScoped;
+import org.jboss.logging.Logger;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import static ai.labs.eddi.engine.model.Deployment.Status.ERROR;
 import static ai.labs.eddi.engine.model.Deployment.Status.READY;
 
 @ApplicationScoped
 public class DeploymentListener implements IDeploymentListener {
+
+    private static final Logger LOGGER = Logger.getLogger(DeploymentListener.class);
+
+    /**
+     * How long a registration may sit unresolved before it self-expires.
+     * <p>
+     * A registration is removed when its {@link DeploymentEvent} arrives, and the
+     * deploy path fires one on both the success and the failure branch — but a
+     * process that dies in between, or a runtime that rejects the deployment
+     * callable outright, leaves an entry no event will ever claim. Nothing swept
+     * this map, so those accumulated for the lifetime of the JVM.
+     * <p>
+     * Generous by design: this bounds a leak, it is not a deployment SLA. A
+     * deployment that takes longer than this has other problems, and a waiter that
+     * wants a tighter bound arms its own (see
+     * {@code AgentFactory#waitForDeploymentCompletion}, which waits 60s).
+     */
+    static final Duration REGISTRATION_TTL = Duration.ofMinutes(5);
+
     private final Map<String, CompletableFuture<Void>> deploymentFutures = new ConcurrentHashMap<>();
 
     @Override
     public CompletableFuture<Void> getRegisteredDeploymentEvent(String agentId, Integer version) {
         return deploymentFutures.get(createKey(agentId, version));
     }
 
+    /**
+     * Registers interest in a deployment, so a concurrent
+     * {@code AgentFactory#getAgent} that finds the agent IN_PROGRESS has something
+     * to await instead of falling straight through.
+     * <p>
+     * The returned future self-expires after {@link #REGISTRATION_TTL} and removes
+     * itself from the map on <em>any</em> completion — normal, exceptional or
+     * timed-out — so the map cannot grow without bound when an event never arrives.
+     */
     public CompletableFuture<Void> registerAgentDeployment(String agentId, Integer version) {
-        return deploymentFutures.computeIfAbsent(createKey(agentId, version), k -> new CompletableFuture<>());
+        return deploymentFutures.computeIfAbsent(createKey(agentId, version), key -> {
+            var future = new CompletableFuture<Void>();
+            // orTimeout returns THIS future, so the timeout arms the entry itself
+            // rather than a derived one nobody holds.
+            future.orTimeout(REGISTRATION_TTL.toSeconds(), TimeUnit.SECONDS);
+            future.whenComplete((result, error) -> {
+                // remove(key, future), not remove(key): a later registration for the
+                // same agent+version must not be evicted by this one's completion.
+                if (deploymentFutures.remove(key, future) && error instanceof TimeoutException) {
+                    LOGGER.warnf("No deployment event arrived for %s within %s — dropping its registration",
+                            LogSanitizer.sanitize(key), REGISTRATION_TTL);
+                }
+            });
+            return future;
+        });
     }
 
     public void onDeploymentEvent(DeploymentEvent event) {

src/test/java/ai/labs/eddi/engine/runtime/internal/DeploymentListenerRegistrationTest.java

@@ -0,0 +1,170 @@
+/*
+ * Copyright EDDI contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package ai.labs.eddi.engine.runtime.internal;
+
+import ai.labs.eddi.engine.model.Deployment;
+import ai.labs.eddi.engine.runtime.model.DeploymentEvent;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Registration lifecycle in {@link DeploymentListener}.
+ * <p>
+ * Two properties matter here. A registration must be <em>discoverable</em>
+ * while the deployment is in flight — that is what makes
+ * {@code AgentFactory#waitForDeploymentCompletion} able to wait at all — and it
+ * must not outlive the deployment it describes: the map was only ever pruned by
+ * an arriving {@link DeploymentEvent}, so a registration whose event never came
+ * (a rejected callable, a process that died mid-deploy) stayed for the lifetime
+ * of the JVM.
+ *
+ * @author ginccc
+ */
+@DisplayName("DeploymentListener — registration lifecycle")
+class DeploymentListenerRegistrationTest {
+
+    private DeploymentListener listener;
+
+    @BeforeEach
+    void setUp() {
+        listener = new DeploymentListener();
+    }
+
+    private static DeploymentEvent event(String agentId, Integer version, Deployment.Status status) {
+        return new DeploymentEvent(agentId, version, Deployment.Environment.production, status);
+    }
+
+    @Test
+    @DisplayName("a registered deployment is discoverable while it is in flight")
+    void registeredDeploymentIsDiscoverable() {
+        CompletableFuture<Void> registered = listener.registerAgentDeployment("agent-1", 1);
+
+        assertNotNull(registered);
+        assertSame(registered, listener.getRegisteredDeploymentEvent("agent-1", 1),
+                "a waiter must find the same future the deployer registered");
+    }
+
+    @Test
+    @DisplayName("an unregistered deployment is not discoverable")
+    void unregisteredDeploymentIsNotDiscoverable() {
+        assertNull(listener.getRegisteredDeploymentEvent("never-registered", 1));
+    }
+
+    @Test
+    @DisplayName("registering twice for the same agent+version hands back the same future")
+    void repeatedRegistrationIsIdempotent() {
+        var first = listener.registerAgentDeployment("agent-1", 1);
+        var second = listener.registerAgentDeployment("agent-1", 1);
+
+        assertSame(first, second);
+    }
+
+    @Test
+    @DisplayName("different versions register independently")
+    void versionsAreIndependent() {
+        var v1 = listener.registerAgentDeployment("agent-1", 1);
+        var v2 = listener.registerAgentDeployment("agent-1", 2);
+
+        assertNotSame(v1, v2);
+    }
+
+    @Test
+    @DisplayName("a READY event completes the registration and removes it")
+    void readyEventCompletesAndRemoves() {
+        var registered = listener.registerAgentDeployment("agent-1", 1);
+
+        listener.onDeploymentEvent(event("agent-1", 1, Deployment.Status.READY));
+
+        assertTrue(registered.isDone());
+        assertNull(listener.getRegisteredDeploymentEvent("agent-1", 1), "a settled registration must not linger");
+    }
+
+    @Test
+    @DisplayName("an ERROR event completes it exceptionally and removes it")
+    void errorEventCompletesExceptionallyAndRemoves() {
+        var registered = listener.registerAgentDeployment("agent-1", 1);
+
+        listener.onDeploymentEvent(event("agent-1", 1, Deployment.Status.ERROR));
+
+        assertTrue(registered.isCompletedExceptionally());
+        assertNull(listener.getRegisteredDeploymentEvent("agent-1", 1));
+    }
+
+    @Test
+    @DisplayName("a registration whose event never arrives expires instead of leaking")
+    void unclaimedRegistrationExpires() {
+        var registered = listener.registerAgentDeployment("agent-1", 1);
+
+        // Stand in for the TTL firing: any exceptional completion must evict the
+        // entry. Asserting the real 5-minute timeout would mean a 5-minute test.
+        registered.completeExceptionally(new TimeoutException("simulated TTL"));
+
+        assertNull(listener.getRegisteredDeploymentEvent("agent-1", 1),
+                "the map was only ever pruned by an arriving event, so an unclaimed registration stayed for the "
+                        + "lifetime of the JVM");
+    }
+
+    @Test
+    @DisplayName("the registration carries a timeout, so it cannot wait forever")
+    void registrationCarriesATimeout() {
+        assertTrue(DeploymentListener.REGISTRATION_TTL.toSeconds() > 0,
+                "an unbounded registration is exactly the leak this bounds");
+    }
+
+    @Test
+    @DisplayName("an expired registration does not evict a later one for the same agent")
+    void expiryDoesNotEvictALaterRegistration() {
+        var first = listener.registerAgentDeployment("agent-1", 1);
+        listener.onDeploymentEvent(event("agent-1", 1, Deployment.Status.READY));
+
+        var second = listener.registerAgentDeployment("agent-1", 1);
+        assertNotSame(first, second, "precondition: the first was evicted, so this is a fresh registration");
+
+        // The first future's completion callback must not remove the second entry.
+        first.complete(null);
+
+        assertSame(second, listener.getRegisteredDeploymentEvent("agent-1", 1),
+                "remove(key, future) — a stale completion must not evict a live registration");
+    }
+
+    @Test
+    @DisplayName("an event for an unregistered deployment is a no-op, not an error")
+    void eventForUnregisteredDeploymentIsANoOp() {
+        listener.onDeploymentEvent(event("ghost", 7, Deployment.Status.READY));
+        listener.onDeploymentEvent(event("ghost", 7, Deployment.Status.ERROR));
+
+        assertNull(listener.getRegisteredDeploymentEvent("ghost", 7));
+    }
+
+    @Test
+    @DisplayName("a completed registration is awaitable by the waiter that holds it")
+    void completedRegistrationIsAwaitable() throws Exception {
+        var registered = listener.registerAgentDeployment("agent-1", 1);
+        listener.onDeploymentEvent(event("agent-1", 1, Deployment.Status.READY));
+
+        registered.get(5, TimeUnit.SECONDS);
+    }
+
+    @Test
+    @DisplayName("a failed registration surfaces the failure to its waiter")
+    void failedRegistrationSurfacesToWaiter() {
+        var registered = listener.registerAgentDeployment("agent-1", 1);
+        listener.onDeploymentEvent(event("agent-1", 1, Deployment.Status.ERROR));
+
+        assertThrows(Exception.class, () -> registered.get(5, TimeUnit.SECONDS));
+    }
+}