add generic gateway startup readiness checks

tests/gateway/test_gateway_shutdown.py

@@ -125,6 +125,25 @@ async def test_gateway_stop_service_restart_sets_named_exit_code():
     assert runner._exit_code == GATEWAY_SERVICE_RESTART_EXIT_CODE
 
 
+@pytest.mark.asyncio
+async def test_gateway_stop_emits_shutdown_hook_after_drain(monkeypatch):
+    runner, adapter = make_restart_runner()
+    adapter.disconnect = AsyncMock()
+    runner.hooks.emit = AsyncMock()
+
+    with patch("gateway.status.remove_pid_file"), patch("gateway.status.write_runtime_status"):
+        await runner.stop(restart=True, service_restart=True)
+
+    runner.hooks.emit.assert_awaited_once_with(
+        "gateway:shutdown",
+        {
+            "restart": True,
+            "service_restart": True,
+            "detached_restart": False,
+        },
+    )
+
+
 @pytest.mark.asyncio
 async def test_drain_active_agents_throttles_status_updates():
     runner, _adapter = make_restart_runner()

tests/gateway/test_hooks.py

@@ -9,7 +9,7 @@ import pytest
 from gateway.hooks import HookRegistry
 
 
-def _create_hook(hooks_dir, hook_name, events, handler_code):
+def _create_hook(hooks_dir, hook_name, events, handler_code, *, manifest_extra=""):
     """Helper to create a hook directory with HOOK.yaml and handler.py."""
     hook_dir = hooks_dir / hook_name
     hook_dir.mkdir(parents=True)
@@ -17,6 +17,7 @@ def _create_hook(hooks_dir, hook_name, events, handler_code):
         f"name: {hook_name}\n"
         f"description: Test hook\n"
         f"events: {events}\n"
+        f"{manifest_extra}"
     )
     (hook_dir / "handler.py").write_text(handler_code)
     return hook_dir
@@ -112,6 +113,24 @@ class TestDiscoverAndLoad:
 
         assert len(reg.loaded_hooks) == 2
 
+    def test_preserves_optional_startup_readiness_metadata(self, tmp_path):
+        _create_hook(
+            tmp_path,
+            "ready-hook",
+            '["gateway:startup"]',
+            "def handle(e, c): pass\n",
+            manifest_extra="startup_readiness:\n  id: beam-runtime\n  required: false\n",
+        )
+
+        reg = HookRegistry()
+        with patch("gateway.hooks.HOOKS_DIR", tmp_path), _patch_no_builtins(reg):
+            reg.discover_and_load()
+
+        assert reg.loaded_hooks[0]["startup_readiness"] == {
+            "id": "beam-runtime",
+            "required": False,
+        }
+
 
 class TestEmit:
     @pytest.mark.asyncio

tests/gateway/test_runner_startup_failures.py

@@ -132,6 +132,68 @@ async def test_runner_records_connected_platform_state_on_success(monkeypatch, t
     assert state["platforms"]["discord"]["error_message"] is None
 
 
+@pytest.mark.asyncio
+async def test_runner_discovers_plugins_before_loading_hooks(monkeypatch, tmp_path):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    config = GatewayConfig(
+        platforms={
+            Platform.DISCORD: PlatformConfig(enabled=True, token="***")
+        },
+        sessions_dir=tmp_path / "sessions",
+    )
+    runner = GatewayRunner(config)
+    order: list[str] = []
+
+    monkeypatch.setattr(runner, "_create_adapter", lambda platform, platform_config: _SuccessfulAdapter())
+    monkeypatch.setattr("hermes_cli.plugins.discover_plugins", lambda: order.append("plugins"))
+    monkeypatch.setattr(runner.hooks, "discover_and_load", lambda: order.append("hooks"))
+    monkeypatch.setattr(runner.hooks, "emit", AsyncMock())
+
+    ok = await runner.start()
+
+    assert ok is True
+    assert order == ["plugins", "hooks"]
+
+
+@pytest.mark.asyncio
+async def test_runner_initializes_startup_checks_before_gateway_startup_emit(monkeypatch, tmp_path):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    config = GatewayConfig(
+        platforms={
+            Platform.DISCORD: PlatformConfig(enabled=True, token="***")
+        },
+        sessions_dir=tmp_path / "sessions",
+    )
+    runner = GatewayRunner(config)
+
+    runner.hooks._loaded_hooks = [
+        {
+            "name": "beam-runtime",
+            "events": ["gateway:startup"],
+            "path": str(tmp_path / "hook"),
+            "startup_readiness": {
+                "id": "beam-runtime",
+                "required": True,
+            },
+        }
+    ]
+    monkeypatch.setattr(runner, "_create_adapter", lambda platform, platform_config: _SuccessfulAdapter())
+    monkeypatch.setattr("hermes_cli.plugins.discover_plugins", lambda: None)
+    monkeypatch.setattr(runner.hooks, "discover_and_load", lambda: None)
+
+    async def _assert_checks(event_type, context):
+        state = read_runtime_status()
+        assert event_type == "gateway:startup"
+        assert state["startup_checks"]["beam-runtime"]["state"] == "pending"
+        assert state["startup_checks"]["beam-runtime"]["required"] is True
+
+    monkeypatch.setattr(runner.hooks, "emit", _assert_checks)
+
+    ok = await runner.start()
+
+    assert ok is True
+
+
 @pytest.mark.asyncio
 async def test_start_gateway_verbosity_imports_redacting_formatter(monkeypatch, tmp_path):
     """Verbosity != None must not crash with NameError on RedactingFormatter (#8044)."""

tests/gateway/test_status.py

@@ -132,6 +132,72 @@ class TestGatewayRuntimeStatus:
         assert payload["platforms"]["discord"]["error_code"] is None
         assert payload["platforms"]["discord"]["error_message"] is None
 
+    def test_reset_startup_checks_replaces_previous_run_entries(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+        status.write_runtime_status(
+            gateway_state="running",
+            startup_checks={
+                "old-check": {
+                    "state": "ready",
+                    "required": True,
+                    "source": "old-hook",
+                    "detail": None,
+                }
+            },
+        )
+
+        status.reset_startup_checks([
+            {
+                "name": "new-hook",
+                "startup_readiness": {
+                    "id": "new-check",
+                    "required": False,
+                },
+            }
+        ])
+
+        payload = status.read_runtime_status()
+        assert set(payload["startup_checks"]) == {"new-check"}
+        assert payload["startup_checks"]["new-check"]["state"] == "pending"
+        assert payload["startup_checks"]["new-check"]["required"] is False
+        assert payload["startup_checks"]["new-check"]["source"] == "new-hook"
+
+    def test_mark_startup_check_ready_persists_detail(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+        status.reset_startup_checks([
+            {
+                "name": "beam",
+                "startup_readiness": {
+                    "id": "beam-runtime",
+                    "required": True,
+                },
+            }
+        ])
+
+        status.mark_startup_check_ready("beam-runtime", detail="ready for RPC")
+
+        payload = status.read_runtime_status()
+        assert payload["startup_checks"]["beam-runtime"]["state"] == "ready"
+        assert payload["startup_checks"]["beam-runtime"]["detail"] == "ready for RPC"
+
+    def test_mark_startup_check_failed_creates_missing_entry(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+        status.mark_startup_check_failed(
+            "late-hook",
+            detail="startup hook crashed",
+            required=False,
+            source="late-hook",
+        )
+
+        payload = status.read_runtime_status()
+        assert payload["startup_checks"]["late-hook"]["state"] == "failed"
+        assert payload["startup_checks"]["late-hook"]["required"] is False
+        assert payload["startup_checks"]["late-hook"]["source"] == "late-hook"
+        assert payload["startup_checks"]["late-hook"]["detail"] == "startup hook crashed"
+
 
 class TestTerminatePid:
     def test_force_uses_taskkill_on_windows(self, monkeypatch):