fix(gateway): never read a gateway result whose exec transport failed

src/agent_env/providers/env_providers/env_gateway_provider.py

@@ -46,6 +46,8 @@
 
 logger = logging.getLogger(__name__)
 
+_TRAJECTORY_READ_ATTEMPTS = 3
+
 # MCP server health-check budget. Some servers wait on an upstream before they bind; a server
 # that is up passes its first probe, so the long start period only delays failing a broken one.
 MCP_HC_RETRIES = 30
@@ -755,12 +757,7 @@ async def _stand_up(_spec):
 
         # Print gateway logs for debugging
         logger.info("Gateway logs:")
-        gateway_container_id = await self._get_container_id(sandbox, GATEWAY_SERVICE_NAME)
-        if gateway_container_id:
-            exit_code, logs, stderr = await sandbox.exec_with_output( "sudo", "docker", "logs", gateway_container_id)
-            logger.info(f"  stdout:\n{logs}")
-            if stderr:
-                logger.info(f"  stderr:\n{stderr}")
+        await self._log_service(sandbox, GATEWAY_SERVICE_NAME)
         if not gateway_ready:
             raise RuntimeError("Gateway did not become ready in time")
 
@@ -770,13 +767,8 @@ async def _stand_up(_spec):
             if self._needs_local_postgres else []
         )
         for name in sidecar_names:
-            container_id = await self._get_container_id(sandbox, name)
-            if container_id:
-                exit_code, logs, stderr = await sandbox.exec_with_output("sudo", "docker", "logs", container_id)
-                logger.info(f"{name} logs:\n  stdout:\n{logs}")
-                if stderr:
-                    logger.info(f"  stderr:\n{stderr}")
-            else:
+            logger.info(f"{name} logs:")
+            if not await self._log_service(sandbox, name):
                 logger.warning(f"{name} container not found")
 
         gateway_url = sandbox.tunnel_urls.get(gateway_port)
@@ -987,25 +979,58 @@ async def _wait_for_gateway(self, sandbox: VmSandbox, port: int, timeout: int =
         return False
 
     async def read_trajectory(self, sandbox: VmSandbox) -> list[dict]:
-        """Read trajectory JSONL from gateway container."""
+        """Read trajectory JSONL from gateway container. A read whose exec transport failed (exit -1) may
+        hold only part of the history, so it is retried and then raised, never parsed."""
         container_id = await self._get_container_id(sandbox, GATEWAY_SERVICE_NAME)
         if not container_id:
             return []
 
-        exit_code, stdout, stderr = await sandbox.exec_with_output(
-            "sudo", "docker", "exec", container_id, "cat", "/var/log/agentenv/trajectory.jsonl"
-        )
+        for attempt in range(_TRAJECTORY_READ_ATTEMPTS):
+            exit_code, stdout, stderr = await sandbox.exec_with_output(
+                "sudo", "docker", "exec", container_id, "cat", "/var/log/agentenv/trajectory.jsonl"
+            )
+            if exit_code != -1:
+                break
+            logger.warning(f"Trajectory read lost its exec transport (attempt {attempt + 1}): {stderr[-200:]}")
+        else:
+            raise RuntimeError(f"Could not read the gateway trajectory: exec transport failed {_TRAJECTORY_READ_ATTEMPTS} times")
         events = []
         for line in stdout.strip().split("\n"):
             if line:
                 events.append(json.loads(line))
         return events
 
+    async def _log_service(self, sandbox: VmSandbox, compose_service: str) -> bool:
+        """Log a compose service's container output; False when it has no container. Diagnostics only, so
+        a lookup or ``docker logs`` that fails is reported as a warning, never raised into the deploy."""
+        try:
+            container_id = await self._get_container_id(sandbox, compose_service)
+            if not container_id:
+                return False
+            exit_code, logs, stderr = await sandbox.exec_with_output("sudo", "docker", "logs", container_id)
+        except Exception as e:
+            logger.warning(f"Skipping {compose_service} logs: {e}")
+            return True
+        if exit_code != 0:
+            logger.warning(f"Skipping {compose_service} logs: docker logs exited {exit_code}: {stderr.strip()[-500:]}")
+            return True
+        logger.info(f"  stdout:\n{logs}")
+        if stderr:
+            logger.info(f"  stderr:\n{stderr}")
+        return True
+
     async def _get_container_id(self, sandbox: VmSandbox, compose_service: str) -> str | None:
-        """Get container ID for a docker-compose service (including exited containers)."""
-        exit_code, stdout, stderr = await sandbox.exec_with_output(
-            "sudo", "docker", "compose", "-f", DOCKER_COMPOSE_PATH, "ps", "-a", "-q", compose_service
-        )
+        """Get container ID for a docker-compose service (including exited containers). A lookup whose
+        exec transport failed (exit -1) is retried, then raised: it can't tell "no container" apart."""
+        for attempt in range(_TRAJECTORY_READ_ATTEMPTS):
+            exit_code, stdout, stderr = await sandbox.exec_with_output(
+                "sudo", "docker", "compose", "-f", DOCKER_COMPOSE_PATH, "ps", "-a", "-q", compose_service
+            )
+            if exit_code != -1:
+                break
+            logger.warning(f"Container lookup for {compose_service} lost its exec transport (attempt {attempt + 1}): {stderr[-200:]}")
+        else:
+            raise RuntimeError(f"Could not look up the {compose_service} container: exec transport failed {_TRAJECTORY_READ_ATTEMPTS} times")
         container_id = stdout.strip()
         return container_id if container_id else None
 

tst/unit/providers/env_providers/env_gateway_provider_test.py

@@ -548,6 +548,57 @@ async def test_modal_vm_provider_routes_to_vm_path_not_containers():
     gp._deploy_via_containers.assert_not_awaited()
 
 
+@pytest.mark.asyncio
+async def test_read_trajectory_retries_a_read_whose_transport_failed():
+    gp = EnvironmentGatewayProvider()
+    gp._get_container_id = AsyncMock(return_value="gw")
+    sandbox = MagicMock()
+    sandbox.exec_with_output = AsyncMock(side_effect=[(-1, "", "host lost"), (0, '{"a": 1}\n{"b": 2}\n', "")])
+
+    assert await gp.read_trajectory(sandbox) == [{"a": 1}, {"b": 2}]
+    assert sandbox.exec_with_output.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_read_trajectory_raises_rather_than_parse_a_read_that_keeps_failing():
+    gp = EnvironmentGatewayProvider()
+    gp._get_container_id = AsyncMock(return_value="gw")
+    sandbox = MagicMock()
+    sandbox.exec_with_output = AsyncMock(return_value=(-1, '{"a": 1}\n', "host lost"))
+
+    with pytest.raises(RuntimeError, match="exec transport failed 3 times"):
+        await gp.read_trajectory(sandbox)
+
+
+@pytest.mark.asyncio
+async def test_a_container_lookup_whose_transport_keeps_failing_raises_rather_than_read_as_no_container():
+    gp = EnvironmentGatewayProvider()
+    sandbox = MagicMock()
+    sandbox.exec_with_output = AsyncMock(return_value=(-1, "", "host lost"))
+
+    with pytest.raises(RuntimeError, match="Could not look up the .* container: exec transport failed 3 times"):
+        await gp.read_trajectory(sandbox)
+
+
+@pytest.mark.asyncio
+async def test_a_container_lookup_retries_a_failed_transport():
+    gp = EnvironmentGatewayProvider()
+    sandbox = MagicMock()
+    sandbox.exec_with_output = AsyncMock(side_effect=[(-1, "", "host lost"), (0, "gw\n", ""), (0, '{"a": 1}\n', "")])
+
+    assert await gp.read_trajectory(sandbox) == [{"a": 1}]
+
+
+@pytest.mark.asyncio
+async def test_read_trajectory_of_a_gateway_with_no_history_yet_is_empty():
+    gp = EnvironmentGatewayProvider()
+    gp._get_container_id = AsyncMock(return_value="gw")
+    sandbox = MagicMock()
+    sandbox.exec_with_output = AsyncMock(return_value=(1, "", "cat: /var/log/agentenv/trajectory.jsonl: No such file"))
+
+    assert await gp.read_trajectory(sandbox) == []
+
+
 @pytest.mark.asyncio
 async def test_sail_vm_provider_routes_to_vm_path_not_containers():
     """A Sailbox is a Docker-capable VM: the gateway deploys onto it with docker-compose."""
@@ -1643,3 +1694,43 @@ def test_every_mcp_server_gets_the_same_healthcheck_budget():
         assert "      timeout: 5s" in block
         assert "      retries: 30" in block
         assert "      start_period: 60s" in block
+
+
+@pytest.mark.asyncio
+async def test_a_log_lookup_that_keeps_losing_its_transport_never_stops_a_deploy(caplog):
+    gp = EnvironmentGatewayProvider()
+    sandbox = MagicMock()
+    sandbox.exec_with_output = AsyncMock(return_value=(-1, "", "host lost"))
+
+    assert await gp._log_service(sandbox, "gateway") is True
+    assert "Skipping gateway logs" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_a_service_without_a_container_is_reported_as_missing():
+    gp = EnvironmentGatewayProvider()
+    sandbox = MagicMock()
+    sandbox.exec_with_output = AsyncMock(return_value=(0, "", ""))
+
+    assert await gp._log_service(sandbox, "pgweb") is False
+
+
+@pytest.mark.asyncio
+async def test_a_failing_docker_logs_call_never_stops_a_deploy(caplog):
+    gp = EnvironmentGatewayProvider()
+    sandbox = MagicMock()
+    sandbox.exec_with_output = AsyncMock(side_effect=[(0, "gw\n", ""), RuntimeError("logs unavailable")])
+
+    assert await gp._log_service(sandbox, "gateway") is True
+    assert "Skipping gateway logs: logs unavailable" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_a_docker_logs_call_that_exits_nonzero_is_reported_not_printed_as_logs(caplog):
+    gp = EnvironmentGatewayProvider()
+    sandbox = MagicMock()
+    sandbox.exec_with_output = AsyncMock(side_effect=[(0, "gw\n", ""), (1, "", "Error: No such container: gw\n")])
+
+    assert await gp._log_service(sandbox, "gateway") is True
+    assert "Skipping gateway logs: docker logs exited 1: Error: No such container: gw" in caplog.text
+    assert "stdout:" not in caplog.text