fix(core): fail deploy when controllers stay unhealthy

packages/cli/canyonos/deploy.py

@@ -141,9 +141,7 @@ def feed(self, line):
         return None, None, False
 
     def agents_ready_message(self):
-        """(message, all_ready). `_wait_for_healthy` gives up after its timeout and
-        lets the controller start anyway, so the workflow can come up short.
-        """
+        """Summarize observed readiness, including incomplete older runtimes."""
         ready = len(self.replicas_ready)
         if not self.replicas_total:
             return "Workflow ready", True
@@ -351,8 +349,12 @@ def _interrupted():
 
 def _tail_verbose(stream, state, api_port, config_path, serve):
     """Every log line, verbatim, until the workflow is up -- what `-v` restores."""
-    for line in stream:
+    tracker = PhaseTracker()
+    for line in _drain(_queued_lines(stream), state):
         print(line, end="")
+        _, _, is_error = tracker.feed(line)
+        if is_error:
+            return None
         # Logged exactly once, right after the workflow finishes coming up.
         if "Global controller started, polling every" in line:
             return _deploy_summary(state, api_port, config_path, serve)

packages/cli/tests/test_deploy_progress.py

@@ -120,9 +120,7 @@ def test_a_re_read_ready_line_does_not_double_count():
 
 
 def test_coming_up_short_of_the_announced_replicas_is_not_reported_as_success():
-    """`_wait_for_healthy` gives up after its timeout and the controller starts
-    anyway, so the up-marker can arrive with agents still unhealthy.
-    """
+    """Keep partial-readiness output safe for logs from older runtimes."""
     tracker, _, _, _ = drive(
         [
             "INFO:canyonos_core.controller.global_controller:Waiting for 3 replica(s) to become healthy (timeout=300s)...\n",
@@ -174,6 +172,20 @@ def test_benign_warnings_do_not_trip_the_error_path(line):
     assert not errored
 
 
+def test_verbose_tail_stops_on_a_fatal_line(monkeypatch, capsys):
+    lines = [
+        "INFO:canyonos_core:Deploying from config: config.yaml\n",
+        "CRITICAL:canyonos_core.controller.global_controller:Controller readiness timed out\n",
+    ]
+    monkeypatch.setattr(deploy_cmd, "_queued_lines", lambda stream: stream)
+    monkeypatch.setattr(deploy_cmd, "_drain", lambda stream, state: iter(stream))
+
+    result = deploy_cmd._tail_verbose(lines, {}, None, "config.yaml", serve=False)
+
+    assert result is None
+    assert "Controller readiness timed out" in capsys.readouterr().out
+
+
 def test_the_deploy_is_only_declared_dead_after_two_consecutive_checks(monkeypatch):
     """One dropped request shouldn't end a deploy that is merely busy."""
     replies = iter([None, {"running": True}, None, None])

packages/core/canyonos_core/controller/global_controller.py

@@ -575,15 +575,26 @@ def _wait_for_healthy(self, timeout=30, interval=2):
             if pending:
                 time.sleep(interval)
 
-        if pending:
-            for instance in pending:
-                logger.warning(
-                    "Controller %s (%s:%s) not ready after %ds.",
-                    instance["agent_name"],
-                    instance["host"],
-                    instance["host_port"],
-                    timeout,
-                )
+        not_ready = []
+        for instance in pending:
+            name = instance["agent_name"]
+            host = instance["host"]
+            port = instance["host_port"]
+            node_redis = self._get_node_redis_for(host)
+            endpoint = self.instance_manager._routing_endpoint_for(instance)
+            status = node_redis.get(f"controller:{endpoint}:status")
+            if status == "healthy":
+                logger.info("Controller %s (%s:%s) is ready.", name, host, port)
+                self._last_status[(host, port)] = "healthy"
+            else:
+                not_ready.append(f"{name} ({host}:{port})={status or 'unknown'}")
+
+        if not_ready:
+            message = f"Controller readiness timed out after {timeout}s: " + ", ".join(
+                not_ready
+            )
+            logger.critical(message)
+            raise RuntimeError(message)
 
     # ------------------------------------------------------------------ #
     #  Polling loop                                                       #

packages/core/tests/test_global_controller_readiness.py

@@ -0,0 +1,72 @@
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from canyonos_core.controller.global_controller import GlobalController
+
+
+class _FakeRedis:
+    def __init__(self, statuses):
+        self.statuses = statuses
+
+    def get(self, key):
+        return self.statuses.get(key)
+
+
+class _FakeInstanceManager:
+    def __init__(self, instances):
+        self.instances = instances
+
+    def list_instances(self):
+        return list(self.instances)
+
+    def _routing_endpoint_for(self, instance):
+        return instance["endpoint"]
+
+
+def _controller(statuses):
+    instances = [
+        {
+            "agent_name": "ResearchAgent",
+            "host": "localhost",
+            "host_port": "8000",
+            "endpoint": "localhost:8000",
+        }
+    ]
+    controller = GlobalController.__new__(GlobalController)
+    controller.redis = _FakeRedis(statuses)
+    controller.node_redis = {}
+    controller.instance_manager = _FakeInstanceManager(instances)
+    controller._last_status = {}
+    return controller
+
+
+class GlobalControllerReadinessTests(unittest.TestCase):
+    def test_healthy_controller_completes_readiness(self):
+        controller = _controller({"controller:localhost:8000:status": "healthy"})
+
+        controller._wait_for_healthy(timeout=0, interval=0)
+
+        self.assertEqual(controller._last_status[("localhost", "8000")], "healthy")
+
+    def test_failed_controller_fails_deploy_after_deadline(self):
+        controller = _controller({"controller:localhost:8000:status": "failed"})
+
+        with self.assertRaisesRegex(
+            RuntimeError,
+            r"Controller readiness timed out after 0s: ResearchAgent "
+            r"\(localhost:8000\)=failed",
+        ):
+            controller._wait_for_healthy(timeout=0, interval=0)
+
+    def test_missing_status_fails_deploy_after_deadline(self):
+        controller = _controller({})
+
+        with self.assertRaisesRegex(RuntimeError, r"ResearchAgent .*unknown"):
+            controller._wait_for_healthy(timeout=0, interval=0)
+
+
+if __name__ == "__main__":
+    unittest.main()