fix(svc-ai-agent-broker): let the swarm allocator realize an overlay before joining it

ensure() creates the agent's network and attaches the broker to it in the next
statement. Under swarm that network is swarm scoped: the control plane holds it
the moment create returns, so the GET inside connect answers, while realizing it
on a node is the allocator's asynchronous work and connect is node local. The
attach therefore ran against a network the node did not have yet and took a 404.

Run 36107421422 shows both halves. The broker logged agent_failed with 'POST
/networks/<id>/connect -> 404: network not found' for every owner, and dockerd
on the same node logged 'Failed to find network <id> on node' once per node, for
all three. Every agent request then answered 502 and the drill reported the two
agent services missing. The compose row passed throughout, which fits: a bridge
network is local and usable as soon as it exists.

connect now retries while the engine answers 404, up to realize_timeout, and
re-raises anything else at once. EngineError carries the status for that, set at
the single raise site in call, so the decision reads a number rather than
matching the message text. The wait is scoped to the one condition known to
clear by itself: a 403 from the socket proxy still fails immediately, and a 404
that outlives the window still fails, because then the network really is gone.

Three cases pin it in both directions, and each fails against the previous
engine.py: a 404 that clears is waited out, a 404 that persists raises at the
deadline, a 403 raises on the first attempt.

Not verified: no swarm run has exercised this. realize_timeout is 120 seconds
from a single observed window of at least five, and it converts a structurally
missing network from a fast failure into a slow one.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

roles/svc-ai-agent-broker/files/python/engine.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import http.client
 import json
 import socket
+import time
 import urllib.parse
 
 LABEL_PLATFORM = "infinito.agent.platform"
@@ -10,7 +11,14 @@ LABEL_OWNER = "infinito.agent.owner"
 
 
 class EngineError(RuntimeError):
-    pass
+    """Args:
+    message: what the call was and how it answered.
+    status: HTTP status the engine returned, or None when it never answered.
+    """
+
+    def __init__(self, message, status=None):
+        super().__init__(message)
+        self.status = status
 
 
 class _UnixConnection(http.client.HTTPConnection):
@@ -53,7 +61,7 @@ class Engine:
         finally:
             connection.close()
         if status not in expect:
-            raise EngineError(f"{method} {path} -> {status}: {raw[:400]!r}")
+            raise EngineError(f"{method} {path} -> {status}: {raw[:400]!r}", status)
         if not raw:
             return None
         try:
@@ -99,17 +107,44 @@ class Engine:
         )
         return created["Id"]
 
-    def connect(self, network_id, container, alias):
-        detail = self.call("GET", f"/networks/{network_id}")
-        attached = detail.get("Containers") or {}
-        if any(key.startswith(container) for key in attached):
+    def connect(self, network_id, container, alias, realize_timeout=120, pause=2):
+        """Attach ``container`` to ``network_id`` once the network is usable.
+
+        Exception: a swarm-scoped overlay answers GET from the control plane
+        before its allocator has realized it on any node, while connect is
+        node-local, so connect returns 404 for a network that does exist. A 404
+        that outlives ``realize_timeout`` is raised, because then the network
+        really is gone.
+
+        Args:
+            network_id: id ``ensure_network`` returned.
+            container: container id or name to attach.
+            alias: name the other members of the network reach it by.
+            realize_timeout: seconds to allow the swarm allocator.
+            pause: seconds between attempts.
+        """
+        deadline = time.monotonic() + realize_timeout
+        while True:
+            detail = self.call("GET", f"/networks/{network_id}")
+            attached = detail.get("Containers") or {}
+            if any(key.startswith(container) for key in attached):
+                return
+            try:
+                self.call(
+                    "POST",
+                    f"/networks/{network_id}/connect",
+                    body={
+                        "Container": container,
+                        "EndpointConfig": {"Aliases": [alias]},
+                    },
+                    expect=(200, 204),
+                )
+            except EngineError as error:
+                if error.status != 404 or time.monotonic() >= deadline:
+                    raise
+                time.sleep(pause)
+                continue
             return
-        self.call(
-            "POST",
-            f"/networks/{network_id}/connect",
-            body={"Container": container, "EndpointConfig": {"Aliases": [alias]}},
-            expect=(200, 204),
-        )
 
 
 class ComposeBackend:

tests/unit/python/roles/svc-ai-agent-broker/files/test_engine.py

@@ -107,6 +107,56 @@ class TestUnixTransport(unittest.TestCase):
             engine.Engine(self.path, timeout=5).call("GET", "/containers/x/exec")
 
 
+class _ConnectEngine(engine.Engine):
+    """Engine whose connect answers 404 for the first ``misses`` attempts."""
+
+    def __init__(self, misses):
+        self.misses = misses
+        self.attempts = 0
+        self.slept = 0.0
+
+    def call(self, method, path, body=None, query=None, expect=(200,), timeout=None):
+        if method == "GET":
+            return {"Containers": {}}
+        self.attempts += 1
+        if self.attempts <= self.misses:
+            raise engine.EngineError(f"POST {path} -> 404: not found", 404)
+        return None
+
+
+class TestConnect(unittest.TestCase):
+    def setUp(self):
+        self.real_sleep = engine.time.sleep
+        engine.time.sleep = lambda seconds: None
+
+    def tearDown(self):
+        engine.time.sleep = self.real_sleep
+
+    def test_a_swarm_overlay_is_joined_once_its_allocator_realizes_it(self):
+        fake = _ConnectEngine(misses=3)
+        fake.connect("net-1", "broker", "broker-alias")
+        self.assertEqual(fake.attempts, 4)
+
+    def test_a_network_that_stays_missing_is_not_waited_on_forever(self):
+        fake = _ConnectEngine(misses=10**6)
+        with self.assertRaises(engine.EngineError):
+            fake.connect("net-1", "broker", "broker-alias", realize_timeout=0)
+        self.assertEqual(fake.attempts, 1)
+
+    def test_a_refusal_that_is_not_a_missing_network_is_raised_at_once(self):
+        fake = _ConnectEngine(misses=0)
+
+        def deny(method, path, body=None, query=None, expect=(200,), timeout=None):
+            if method == "GET":
+                return {"Containers": {}}
+            raise engine.EngineError(f"POST {path} -> 403: denied", 403)
+
+        fake.call = deny
+        with self.assertRaises(engine.EngineError) as caught:
+            fake.connect("net-1", "broker", "broker-alias")
+        self.assertEqual(caught.exception.status, 403)
+
+
 class TestSwarmBackend(unittest.TestCase):
     def test_a_service_is_placed_on_the_sandbox_nodes_and_starts_scaled_down(self):
         fake = RecordingEngine(runtime="runsc")