fix(deployments): gate READY on workload reachability (AIRCORE-998)

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/backend.py

@@ -608,7 +608,11 @@ async def read_status(self, *, workspace: str, name: str) -> BackendStatusUpdate
             return map_docker_state_to_starting(container_id, state)
 
         if state == "running":
-            host_url = self._primary_host_url(host_ports)
+            # The default (no-declared-probe) reachability check TCP-connects the port, so
+            # it must see only TCP mappings: a UDP-only workload has no TCP listener and
+            # would otherwise be gated STARTING forever. Endpoints still carry every port.
+            tcp_host_ports = self._extract_host_ports(container, protocol="tcp")
+            host_url = self._primary_host_url(tcp_host_ports)
             config = await self._load_config_from_labels(workspace, labels)
             probe = None
             if config is not None and config.containers:
@@ -617,7 +621,7 @@ async def read_status(self, *, workspace: str, name: str) -> BackendStatusUpdate
                 container=container,
                 probe=probe,
                 host_url=host_url,
-                host_ports=host_ports,
+                host_ports=tcp_host_ports,
             )
             if ready and restart_policy == "Always":
                 sidecar_ok, sidecar_reason = await self._sidecars_healthy(workspace, name, config)
@@ -1088,13 +1092,21 @@ async def _load_config_for_deployment_entity(
         except Exception:
             return None
 
-    def _extract_host_ports(self, container: DockerContainer) -> dict[int, int]:
+    def _extract_host_ports(self, container: DockerContainer, *, protocol: str | None = None) -> dict[int, int]:
+        """Map container port -> published host port.
+
+        With *protocol* (e.g. ``"tcp"``) only mappings of that protocol are returned;
+        docker keys the port map as ``"<port>/<proto>"``. Defaults to every protocol.
+        """
         result: dict[int, int] = {}
         ports = container.ports or {}
         for key, bindings in ports.items():
             if not bindings:
                 continue
-            container_port = int(str(key).split("/")[0])
+            key_str = str(key)
+            if protocol is not None and not key_str.endswith(f"/{protocol}"):
+                continue
+            container_port = int(key_str.split("/")[0])
             host_port = bindings[0].get("HostPort")
             if host_port:
                 result[container_port] = int(host_port)

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/docker/probes.py

@@ -16,6 +16,11 @@
 
 logger = logging.getLogger(__name__)
 
+# Timeout for the default reachability probe used when a container declares no
+# readinessProbe. Keep it well under the reconciler poll interval so a probe never
+# stalls the reconcile loop.
+_DEFAULT_TCP_PROBE_TIMEOUT_SECONDS = 2.0
+
 
 async def check_readiness_probe(
     *,
@@ -25,9 +30,17 @@ async def check_readiness_probe(
     host_ports: dict[int, int] | None = None,
     named_ports: dict[str, int] | None = None,
 ) -> tuple[bool, str]:
-    """Return (ready, reason). When no probe is configured, running implies ready."""
+    """Return (ready, reason).
+
+    With a declared probe, evaluate it. With no declared probe, a workload that
+    publishes a port is only ready once that port accepts a connection, so status does
+    not race the process's bind(); a portless workload has no socket to reach, so running
+    implies ready.
+    """
     if probe is None:
-        return True, "no readiness probe configured"
+        if host_url is None or not host_ports:
+            return True, "no readiness probe configured"
+        return await _check_default_tcp(host_url)
 
     if probe.exec_action is not None and probe.exec_action.command:
         return await _check_exec_probe(container, probe)
@@ -151,5 +164,24 @@ def _connect() -> None:
         return False, f"tcp probe failed: {exc}"
 
 
+async def _check_default_tcp(host_url: str) -> tuple[bool, str]:
+    """Default reachability check: TCP-connect the primary published host port."""
+    parsed = urlparse(host_url)
+    host = parsed.hostname or "127.0.0.1"
+    port = parsed.port
+    if port is None:
+        return True, "no host port to probe"
+
+    def _connect() -> None:
+        with socket.create_connection((host, port), timeout=_DEFAULT_TCP_PROBE_TIMEOUT_SECONDS):
+            return
+
+    try:
+        await asyncio.wait_for(asyncio.to_thread(_connect), timeout=_DEFAULT_TCP_PROBE_TIMEOUT_SECONDS)
+        return True, f"default tcp probe connected ({host}:{port})"
+    except Exception as exc:
+        return False, f"default tcp probe not ready ({host}:{port}): {exc}"
+
+
 def host_url_for_port(host: str, host_port: int, *, scheme: str = "http") -> str:
     return f"{scheme}://{host}:{host_port}"

plugins/nemo-deployments/src/nemo_deployments_plugin/backends/openshell/backend.py

@@ -134,6 +134,29 @@ def _delivery_script(path: str, mode: int) -> str:
 """
 _LOG_TAIL_LINES = 20
 
+# Timeout for the default (no-declared-probe) loopback reachability probe. A declared
+# readinessProbe uses its own timeout_seconds instead.
+_DEFAULT_READINESS_TIMEOUT_SECONDS = 3
+
+# Headroom added to a network probe's own timeout when bounding the ExecSandbox RPC, so
+# the RPC outlives the in-sandbox python timeout and captures its (non-zero) exit rather
+# than being cut off first.
+_READINESS_EXEC_TIMEOUT_MARGIN_SECONDS = 5
+
+# Loopback readiness probe programs, run by the sandbox's python. urlopen raises on a
+# refused connection or an HTTP status >= 400 (so a still-starting 503 reads as not
+# ready); create_connection raises until the socket is actually bound. The HTTP probe
+# uses an unverified TLS context so an https readinessProbe against a loopback/self-signed
+# cert is not rejected on verification (matching Kubernetes httpGet HTTPS probe semantics);
+# the context is ignored for plain http, so one program covers both schemes.
+_HTTP_PROBE_PROGRAM = (
+    "import sys, ssl, urllib.request; "
+    "urllib.request.urlopen(sys.argv[1], timeout=float(sys.argv[2]), context=ssl._create_unverified_context())"
+)
+_TCP_PROBE_PROGRAM = (
+    "import sys, socket; socket.create_connection((sys.argv[1], int(sys.argv[2])), timeout=float(sys.argv[3])).close()"
+)
+
 # Cached on first status read; needs the proto enums so it cannot be built at import
 # time (see _ensure_openshell). None until built.
 _PHASE_TO_STATUS: dict[int, DeploymentStatus] | None = None
@@ -544,6 +567,13 @@ async def _advance_provisioning(self, sandbox: Any, sandbox_nm: str, workspace:
         if state == "pending":
             return BackendStatusUpdate(status="STARTING", status_message="Serve launched; awaiting serve pid")
 
+        # A live pid has bound its pidfile, not necessarily its socket. Do not expose
+        # (which reads as READY) until the workload actually accepts a connection, so a
+        # caller trusting READY does not 502 against a process still starting up.
+        pending = await self._readiness_pending(sandbox_id, container)
+        if pending is not None:
+            return pending
+
         # Serve launched but ports not yet exposed: expose them.
         try:
             endpoints = await self._expose_ports(sandbox_nm, container)
@@ -658,16 +688,21 @@ async def _try_get_sandbox(self, sandbox_nm: str) -> Any | None:
         return response.sandbox
 
     async def _exec_detached(
-        self, sandbox_id: str, command: list[str], *, stdin: bytes | None = None
+        self, sandbox_id: str, command: list[str], *, timeout: int | None = None, stdin: bytes | None = None
     ) -> tuple[int | None, str]:
-        """Run a command, draining its event stream. Returns (exit_code, combined output).
+        """Run *command* to completion, returning (exit_code, stdout+stderr merged);
+        ``exit_code`` is None when the stream carried no exit event.
+
+        *timeout* bounds both the RPC and the sandbox-side command, defaulting to the
+        executor's control-plane ``request_timeout_seconds``; readiness probes pass a much
+        shorter bound so a hung probe cannot stall the serial reconcile loop.
 
         When *stdin* is given it is streamed to the command as its standard input. The
         ExecSandboxRequest carries a first-class ``stdin`` bytes field, so config-file
         content is piped verbatim into ``cat``: the bytes never touch the argv nor the
         (single-line-only, size-capped) sandbox environment.
         """
-        timeout = self._executor_config.request_timeout_seconds
+        timeout = timeout if timeout is not None else self._executor_config.request_timeout_seconds
         request = pb.ExecSandboxRequest(sandbox_id=sandbox_id, command=command, timeout_seconds=timeout)
         if stdin is not None:
             request.stdin = stdin
@@ -733,6 +768,33 @@ async def _deliver_config_files(
                 )
         return None
 
+    async def _readiness_pending(self, sandbox_id: str, container: Container) -> BackendStatusUpdate | None:
+        """A STARTING update while the workload is not yet reachable, else None.
+
+        Probed from inside the sandbox against loopback, so readiness does not depend on
+        the gateway route or its TLS. Because a port is exposed only once this passes,
+        the fast path's "endpoints exist -> READY" stays sticky and never re-probes, so a
+        momentarily refusing port cannot flap a serving deployment. A workload that never
+        becomes reachable stays STARTING; the reconciler's starting-timeout is the
+        progress deadline that eventually fails it.
+        """
+        probe_command = _readiness_probe_command(container)
+        if probe_command is None:
+            return None
+        command, description, exec_timeout = probe_command
+        exit_code, _ = await self._exec_detached(sandbox_id, command, timeout=exec_timeout)
+        # Readiness fails closed (liveness fails open): the gate admits only positive
+        # proof of reachability, so a flaky probe never exposes an unready workload.
+        #   - exit 0        -> reachable; expose the port and read READY
+        #   - nonzero       -> not reachable yet; stay STARTING
+        #   - no exit event -> undecidable; stay STARTING and re-probe next poll (the port
+        #                      is not exposed while pending, so this self-heals)
+        # A workload that can never be probed never claims READY -- the contract this gate
+        # keeps. Timeouts and RPC errors surface as UNKNOWN upstream, not as a None exit.
+        if exit_code == 0:
+            return None
+        return BackendStatusUpdate(status="STARTING", status_message=f"Awaiting readiness: {description}")
+
     async def _list_endpoints(self, sandbox_nm: str) -> list[Endpoint]:
         """Return the sandbox's currently exposed services as endpoints."""
         try:
@@ -756,6 +818,87 @@ async def _expose_ports(self, sandbox_nm: str, container: Container) -> list[End
         return endpoints
 
 
+def _resolve_probe_port(port: int | str, container: Container) -> int | None:
+    """Resolve a probe port (a number, or a container-port name) to a number, or None."""
+    if isinstance(port, int):
+        return port
+    for declared in container.ports:
+        if declared.name == port:
+            return declared.container_port
+    return None
+
+
+def _default_probe_port(container: Container) -> int | None:
+    """The first TCP container port, used for the default reachability probe.
+
+    UDP ports are skipped: a TCP connect against a UDP listener never succeeds and would
+    wedge a healthy workload in STARTING until the progress deadline.
+    """
+    for declared in container.ports:
+        if declared.protocol == "TCP":
+            return declared.container_port
+    return None
+
+
+def _loopback_probe_script(program: str, *args: str) -> str:
+    """Wrap a python probe *program* so it runs against the sandbox's own python.
+
+    Selects ``python3`` then ``python`` off PATH and runs *program* with *args*, exiting
+    0 when the probe connects and nonzero when it does not. If neither interpreter is on
+    PATH the probe cannot run, so it exits 0 to preserve the prior expose-on-alive
+    behaviour rather than wedging a workload in STARTING until the progress deadline.
+    """
+    quoted_args = " ".join(shlex.quote(arg) for arg in args)
+    return (
+        "if command -v python3 >/dev/null 2>&1; then _py=python3; "
+        "elif command -v python >/dev/null 2>&1; then _py=python; "
+        "else exit 0; fi; "
+        f'"$_py" -c {shlex.quote(program)} {quoted_args}'
+    )
+
+
+def _readiness_probe_command(container: Container) -> tuple[list[str], str, int] | None:
+    """The in-sandbox probe: (command, description, exec_timeout_seconds), or None.
+
+    Succeeds (exit 0) once the workload is reachable. Honours a declared readinessProbe
+    (exec/httpGet/tcpSocket); with no probe declared, falls back to a TCP connect on the
+    first TCP container port. Returns None when there is nothing to probe (no declared
+    probe and no TCP port), meaning "treat as ready" -- a portless (or UDP-only) workload
+    has no TCP socket a caller could reach anyway.
+
+    ``exec_timeout_seconds`` bounds the ExecSandbox RPC so a hung probe cannot stall the
+    serial reconcile loop: an exec probe is bounded by its own ``timeoutSeconds``; a
+    network probe self-times in python, so the RPC is given that timeout plus headroom.
+    """
+    probe = container.readiness_probe
+    timeout = probe.timeout_seconds if probe is not None else _DEFAULT_READINESS_TIMEOUT_SECONDS
+    network_exec_timeout = timeout + _READINESS_EXEC_TIMEOUT_MARGIN_SECONDS
+
+    if probe is not None and probe.exec_action is not None and probe.exec_action.command:
+        return list(probe.exec_action.command), "exec readiness probe", timeout
+
+    # A declared probe naming a port that does not resolve falls back to the first TCP
+    # port rather than skipping the gate, so a misconfigured probe cannot silently
+    # re-open the bind race.
+    if probe is not None and probe.http_get is not None:
+        port = _resolve_probe_port(probe.http_get.port, container) or _default_probe_port(container)
+        if port is None:
+            return None
+        path = probe.http_get.path if probe.http_get.path.startswith("/") else f"/{probe.http_get.path}"
+        url = f"{probe.http_get.scheme.lower()}://127.0.0.1:{port}{path}"
+        script = _loopback_probe_script(_HTTP_PROBE_PROGRAM, url, str(timeout))
+        return ["/bin/sh", "-c", script], f"httpGet {url}", network_exec_timeout
+
+    if probe is not None and probe.tcp_socket is not None:
+        port = _resolve_probe_port(probe.tcp_socket.port, container) or _default_probe_port(container)
+    else:
+        port = _default_probe_port(container)
+    if port is None:
+        return None
+    script = _loopback_probe_script(_TCP_PROBE_PROGRAM, "127.0.0.1", str(port), str(timeout))
+    return ["/bin/sh", "-c", script], f"tcp 127.0.0.1:{port}", network_exec_timeout
+
+
 def _sandbox_name(workspace: str, name: str) -> str:
     """OpenShell sandbox name, within ``_MAX_ROUTABLE_NAME_LEN``.
 

plugins/nemo-deployments/tests/unit/backends/docker/test_backend_mocked.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import socket
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -614,6 +615,86 @@ async def test_read_status_ready_when_running_without_probe(
     assert update.status == "READY"
 
 
+def _running_container_with_published_port(host_port: int) -> MagicMock:
+    container = MagicMock()
+    container.id = "abc123def456"
+    container.status = "running"
+    container.labels = {
+        "managed-by": MANAGED_BY_LABEL,
+        DEPLOYMENT_WORKSPACE_LABEL: "default",
+        DEPLOYMENT_NAME_LABEL: "srv",
+        RESTART_POLICY_LABEL: "Always",
+        CONFIG_NAME_LABEL: "cfg1",
+        RESOURCE_SCOPE_LABEL: DEFAULT_RESOURCE_SCOPE,
+    }
+    container.ports = {"8000/tcp": [{"HostPort": str(host_port)}]}
+    container.attrs = container_attrs()
+    return container
+
+
+@pytest.mark.asyncio
+async def test_read_status_ready_when_running_port_bound(
+    docker_backend: DockerDeploymentBackend,
+    mock_entities: AsyncMock,
+    mock_docker_client: MagicMock,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # No declared probe, published port accepting connections -> READY.
+    monkeypatch.setenv("NMP_LOOPBACK_ADDRESS", "127.0.0.1")
+    mock_entities.get.return_value = sample_config()
+    with socket.socket() as server:
+        server.bind(("127.0.0.1", 0))
+        server.listen(1)
+        host_port = server.getsockname()[1]
+        mock_docker_client.containers.get.return_value = _running_container_with_published_port(host_port)
+
+        update = await docker_backend.read_status(workspace="default", name="srv")
+
+    assert update.status == "READY"
+
+
+@pytest.mark.asyncio
+async def test_read_status_starting_when_running_port_not_bound(
+    docker_backend: DockerDeploymentBackend,
+    mock_entities: AsyncMock,
+    mock_docker_client: MagicMock,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # No declared probe, nothing yet listening on the published port -> STARTING, so
+    # READY does not race the workload's bind(). Hold the port bound-but-not-listening
+    # for the whole probe so nothing else can bind and listen on it mid-test; a
+    # connect() still gets ECONNREFUSED, the not-yet-bound state under test.
+    monkeypatch.setenv("NMP_LOOPBACK_ADDRESS", "127.0.0.1")
+    mock_entities.get.return_value = sample_config()
+    with socket.socket() as probe_socket:
+        probe_socket.bind(("127.0.0.1", 0))
+        host_port = probe_socket.getsockname()[1]
+        mock_docker_client.containers.get.return_value = _running_container_with_published_port(host_port)
+
+        update = await docker_backend.read_status(workspace="default", name="srv")
+
+    assert update.status == "STARTING"
+    assert "not ready" in update.status_message
+
+
+@pytest.mark.asyncio
+async def test_read_status_ready_when_running_udp_only_port(
+    docker_backend: DockerDeploymentBackend,
+    mock_entities: AsyncMock,
+    mock_docker_client: MagicMock,
+) -> None:
+    # A UDP-only workload has no TCP listener, so the default TCP probe is skipped and
+    # running implies ready rather than wedging STARTING until the progress deadline.
+    mock_entities.get.return_value = sample_config()
+    container = _running_container_with_published_port(0)
+    container.ports = {"9000/udp": [{"HostPort": "34567"}]}
+    mock_docker_client.containers.get.return_value = container
+
+    update = await docker_backend.read_status(workspace="default", name="srv")
+
+    assert update.status == "READY"
+
+
 def _running_server_container() -> MagicMock:
     container = MagicMock()
     container.id = "abc123def456"