Feature/reload ws polish 0902

src/octop/api/routers/chat/notify_ws.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import contextlib
 import json
 import logging
@@ -46,20 +45,13 @@ async def dashboard_notifications_ws(
     connection_id = uuid.uuid4().hex
     await websocket.accept()
 
-    # Harness/gateway workers run on the server loop; this handler may run on
-    # another (e.g. Starlette TestClient). Marshal frames onto the socket loop.
-    ws_loop = asyncio.get_running_loop()
-
-    async def _emit_frame(frame: dict[str, Any]) -> None:
+    async def send_frame(frame: dict[str, Any]) -> None:
         if websocket.application_state != WebSocketState.CONNECTED:
             return
         await websocket.send_text(
             json.dumps(frame, ensure_ascii=False, default=json_chunk_default),
         )
 
-    async def send_frame(frame: dict[str, Any]) -> None:
-        await asyncio.wrap_future(asyncio.run_coroutine_threadsafe(_emit_frame(frame), ws_loop))
-
     hub.register(connection_id, send_frame, user_id=user.id)
 
     try:

src/octop/api/routers/chat/ws.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import contextlib
 import json
 import logging
@@ -68,22 +67,13 @@ async def dashboard_chat_ws(
     connection_id = uuid.uuid4().hex
     await websocket.accept()
 
-    # The harness/gateway workers run on the server event loop, but this
-    # handler may run on a different loop (e.g. starlette's TestClient portal).
-    # Marshal outbound frames onto the loop that owns this WebSocket so
-    # cross-loop sends don't deadlock.
-    ws_loop = asyncio.get_running_loop()
-
-    async def _emit_frame(frame: dict[str, Any]) -> None:
+    async def send_frame(frame: dict[str, Any]) -> None:
         if websocket.application_state != WebSocketState.CONNECTED:
             return
         await websocket.send_text(
             json.dumps(frame, ensure_ascii=False, default=json_chunk_default),
         )
 
-    async def send_frame(frame: dict[str, Any]) -> None:
-        await asyncio.wrap_future(asyncio.run_coroutine_threadsafe(_emit_frame(frame), ws_loop))
-
     hub.register(connection_id, send_frame)
 
     try:

src/octop/infra/agents/manager.py

@@ -53,6 +53,7 @@
 )
 from octop.infra.connectors.builder import (
     build_mcp_server_configs_for_user,
+    gateway_mcp_server_names,
     inject_missing_gateway_tools,
 )
 from octop.infra.connectors.service import ConnectorService
@@ -71,6 +72,7 @@
     from octop.infra.cron.manager import CronManager
     from octop.infra.db.repos.agents import AgentRow
     from octop.infra.db.services import RepoBundle
+    from octop.infra.proactive.scheduler import ProactiveCareScheduler
     from octop.infra.utils.paths import PathLayout
 
 logger = logging.getLogger(__name__)
@@ -333,6 +335,7 @@ def __init__(
         self._expert_catalog = expert_catalog
         self._plugin_manager = plugin_manager
         self._cron_manager: CronManager | None = None
+        self._proactive_scheduler: ProactiveCareScheduler | None = None
         self._team_processor: Any | None = None
         self._harness_manager: HarnessAgentManager | None = None
         self._lock = asyncio.Lock()
@@ -403,6 +406,10 @@ def set_cron_manager(self, cron_manager: CronManager) -> None:
         """Attach the process-wide CronManager (must be set before boot())."""
         self._cron_manager = cron_manager
 
+    def set_proactive_scheduler(self, scheduler: ProactiveCareScheduler) -> None:
+        """Attach the process-wide proactive-care scheduler (optional; used after create/delete)."""
+        self._proactive_scheduler = scheduler
+
     def set_team_processor(self, team_processor: Any | None) -> None:
         """Attach harness TeamProcessor (GlobalProcessor); required before boot()."""
         self._team_processor = team_processor
@@ -566,6 +573,8 @@ async def create(self, spec: AgentCreateSpec, *, defer_bootstrap: bool = False)
             self._repos.audit_repo.write(
                 actor=ACTOR_SYSTEM, action="agent.create", target=agent_id, payload=spec.name
             )
+            if self._proactive_scheduler is not None:
+                self._proactive_scheduler.ensure_scheduled(agent_id)
             return row
 
     def _preserve_system_files_path(self, agent_id: str, cfg: dict[str, Any]) -> dict[str, Any]:
@@ -665,6 +674,8 @@ async def delete(self, agent_id: str) -> None:
         except OSError:
             logger.exception("rmtree failed for %s; agent removed from DB anyway", workspace_dir)
         self._repos.agent_repo.delete(agent_id)
+        if self._proactive_scheduler is not None:
+            self._proactive_scheduler.cancel(agent_id)
         self._repos.audit_repo.write(actor=ACTOR_SYSTEM, action="agent.delete", target=agent_id)
 
     async def start(self, agent_id: str) -> None:
@@ -1339,6 +1350,23 @@ async def prepare_chat_mcp(
                         continue
                 agent.config.mcp_server_configs[name] = dict(spec)
 
+        gateway_missing: list[str] = []
+        if builtin_missing and uid is not None:
+            gateway_names = gateway_mcp_server_names(
+                connector_repo=self._repos.connector_repo,
+                user_id=uid,
+            )
+            gateway_missing = [n for n in builtin_missing if n in gateway_names]
+            builtin_missing = [n for n in builtin_missing if n not in gateway_names]
+
+        if gateway_missing and uid is not None:
+            await self._attach_gateway_tools(
+                agent,
+                agent_id=agent_id,
+                user_id=uid,
+                server_names=gateway_missing,
+            )
+
         if builtin_missing:
             logger.info(
                 "Reloading agent %s MCP tools (builtin_missing=%s)",
@@ -1421,6 +1449,43 @@ async def prepare_chat_mcp(
             )
         return still_missing
 
+    async def _attach_gateway_tools(
+        self,
+        agent: HarnessAgent,
+        *,
+        agent_id: str,
+        user_id: int,
+        server_names: list[str],
+    ) -> None:
+        """Add gateway connector tools to the running agent without rebuilding it.
+
+        A rebuild would drop the harness instance (and with it the checkpointer
+        pool an in-flight turn still writes to) only to run the very same
+        in-process injection at the end of ``_post_start_agent``.
+        """
+        repo = self._repos.connector_repo
+        for inst in repo.list_by_user(user_id):
+            if inst.status != "active" or inst.mcp_server_name not in server_names:
+                continue
+            try:
+                await self._connector_svc.ensure_fresh_credentials(inst.instance_id, inst.kind)
+            except Exception:
+                logger.exception(
+                    "prepare_chat_mcp agent=%s: credential refresh failed for %s",
+                    agent_id,
+                    inst.mcp_server_name,
+                )
+        inject_missing_gateway_tools(
+            agent,
+            svc=self._connector_svc,
+            connector_repo=repo,
+            user_id=user_id,
+            agent_id=agent_id,
+            mcp_server_configs=agent.config.mcp_server_configs,
+        )
+        for name in server_names:
+            agent.config.mcp_server_configs.setdefault(name, {})
+
     # ------------------------------------------------------------------
     # Settings persistence — push global policy into harness runtime
     # ------------------------------------------------------------------

src/octop/infra/connectors/builder.py

@@ -529,6 +529,23 @@ def build_mcp_server_configs_for_user(
     return configs
 
 
+def gateway_mcp_server_names(*, connector_repo: Any, user_id: int) -> set[str]:
+    """MCP server names of *user_id*'s active gateway-mode connector instances.
+
+    Gateway connectors carry no HTTP transport: their tools are built in-process
+    from stored credentials, so callers can attach them to a live agent instead
+    of rebuilding it.
+    """
+    names: set[str] = set()
+    for inst in connector_repo.list_by_user(user_id):
+        if inst.status != "active":
+            continue
+        entry = get_catalog_entry(inst.kind)
+        if entry is not None and entry.mcp_mode == "gateway":
+            names.add(inst.mcp_server_name)
+    return names
+
+
 def inject_missing_gateway_tools(
     agent: Any,
     *,

src/octop/infra/db/repos/proactive_care_config.py

@@ -92,25 +92,36 @@ def upsert(self, config: ProactiveCareConfig) -> None:
             )
 
     def list_enabled(self) -> list[ProactiveCareConfig]:
-        """List the configurations of all agents that have proactive care push enabled.
+        """List proactive-care configs for every enabled agent that should be scheduled.
+
+        Agents without a ``proactive_care_config`` row use the same defaults as
+        ``get()`` (enabled=True). An explicit ``enabled=0`` row opts out.
+        Disabled agents (``agents.enabled = 0``) are excluded.
 
         Returns:
             The list of enabled ProactiveCareConfig.
         """
         with self._db.connect() as conn:
             rows = conn.execute(
-                "SELECT agent_id, enabled, active_hours_start, active_hours_end, "
-                "min_interval_hours, max_interval_hours "
-                "FROM proactive_care_config WHERE enabled = 1",
+                "SELECT a.agent_id, c.enabled, c.active_hours_start, "
+                "c.active_hours_end, c.min_interval_hours, c.max_interval_hours "
+                "FROM agents a "
+                "LEFT JOIN proactive_care_config c ON c.agent_id = a.agent_id "
+                "WHERE a.enabled = 1 AND (c.agent_id IS NULL OR c.enabled = 1)",
             ).fetchall()
-        return [
-            ProactiveCareConfig(
-                agent_id=row[0],
-                enabled=bool(row[1]),
-                active_hours_start=row[2],
-                active_hours_end=row[3],
-                min_interval_hours=row[4],
-                max_interval_hours=row[5],
-            )
-            for row in rows
-        ]
+        configs: list[ProactiveCareConfig] = []
+        for row in rows:
+            if row[1] is None:
+                configs.append(ProactiveCareConfig(agent_id=row[0]))
+            else:
+                configs.append(
+                    ProactiveCareConfig(
+                        agent_id=row[0],
+                        enabled=bool(row[1]),
+                        active_hours_start=row[2],
+                        active_hours_end=row[3],
+                        min_interval_hours=row[4],
+                        max_interval_hours=row[5],
+                    )
+                )
+        return configs

src/octop/infra/proactive/scheduler.py

@@ -133,6 +133,7 @@ def __init__(
         self._session_repo = session_repo
         # agent_id -> asyncio.Task
         self._tasks: dict[str, asyncio.Task[None]] = {}
+        self._suspended = False
 
     def replace_persistence(
         self,
@@ -147,11 +148,32 @@ def replace_persistence(
         self._care_service.replace_care_push_repo(care_push_repo)
 
     async def start_all(self) -> None:
-        """At system startup, register random scheduling tasks for all agents with enabled=true."""
+        """At system startup, register random scheduling tasks for all agents with enabled=true.
+
+        Includes agents that never saved a config row: proactive care defaults to on.
+        """
         configs = self._config_repo.list_enabled()
         logger.info("ProactiveCareScheduler: started, found %d enabled agents", len(configs))
         for cfg in configs:
-            self._schedule(cfg.agent_id)
+            self.ensure_scheduled(cfg.agent_id)
+
+    def suspend(self) -> None:
+        """Stop creating new loops. Tests call this so leftover sleeps do not hang pytest."""
+        self._suspended = True
+
+    def ensure_scheduled(self, agent_id: str) -> None:
+        """Start the loop if this agent is enabled and not already scheduled."""
+        if self._suspended:
+            return
+        if agent_id in self._tasks:
+            return
+        cfg = self._config_repo.get(agent_id)
+        if cfg.enabled:
+            self._schedule(agent_id)
+            logger.info(
+                "ProactiveCareScheduler: agent=%s scheduled (default enabled unless opted out)",
+                agent_id,
+            )
 
     def reschedule(self, agent_id: str) -> None:
         """Cancel the current schedule and re-arrange the next trigger time with new config.
@@ -182,14 +204,24 @@ def cancel(self, agent_id: str) -> None:
             task.cancel()
 
     async def shutdown(self) -> None:
-        """Shut down all scheduling tasks."""
-        agent_ids = list(self._tasks.keys())
-        for agent_id in agent_ids:
-            self.cancel(agent_id)
+        """Cancel every loop and wait so a long sleep cannot outlive the process."""
+        pending: list[asyncio.Task[None]] = []
+        for agent_id in list(self._tasks.keys()):
+            task = self._tasks.pop(agent_id, None)
+            if task and not task.done():
+                task.cancel()
+                pending.append(task)
+        if pending:
+            await asyncio.gather(*pending, return_exceptions=True)
         logger.info("ProactiveCareScheduler: all scheduled tasks shut down")
 
     def _schedule(self, agent_id: str) -> None:
         """Create a scheduling task for an agent."""
+        if self._suspended:
+            return
+        existing = self._tasks.get(agent_id)
+        if existing is not None and not existing.done():
+            existing.cancel()
         task = asyncio.create_task(
             self._run_loop(agent_id),
             name=f"proactive_care_{agent_id}",
@@ -199,7 +231,8 @@ def _schedule(self, agent_id: str) -> None:
 
     def _on_task_done(self, agent_id: str, task: asyncio.Task[None]) -> None:
         """Task-completion callback that handles exceptions."""
-        self._tasks.pop(agent_id, None)
+        if self._tasks.get(agent_id) is task:
+            self._tasks.pop(agent_id, None)
         if task.cancelled():
             return
         exc = task.exception()

src/octop/infra/server.py

@@ -274,6 +274,7 @@ async def _boot_runtime(self, config: OctopConfig) -> None:
             config_repo=self.services.repos.proactive_care_config_repo,
             session_repo=self.services.repos.session_repo,
         )
+        registry.set_proactive_scheduler(proactive_scheduler)
 
         await registry.boot()
         await gateway.refresh_media_backends()

tests/conftest.py

@@ -4,6 +4,7 @@
 
 from collections.abc import Iterator
 from pathlib import Path
+from typing import Any
 
 import pytest
 
@@ -43,6 +44,31 @@ def repo_root() -> Path:
     return _REPO_ROOT
 
 
+# Creating an agent now starts a multi-hour asyncio.sleep (proactive care
+# defaults to on). pytest-asyncio waits for leftover tasks before fixture
+# teardown, so any test that boots OctopServer without going through
+# ``octop_client`` would hang the suite. Scheduler unit tests opt out.
+_PROACTIVE_SCHEDULER_TESTS = "tests/unit/proactive/test_scheduler.py"
+
+
+@pytest.fixture(autouse=True)
+def _suspend_proactive_care_loops(
+    request: pytest.FixtureRequest,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    if _module_path(request.node.nodeid) == _PROACTIVE_SCHEDULER_TESTS:
+        return
+    from octop.infra.proactive.scheduler import ProactiveCareScheduler
+
+    monkeypatch.setattr(ProactiveCareScheduler, "ensure_scheduled", lambda self, _id: None)
+
+    async def _start_all(self: Any) -> None:
+        return None
+
+    monkeypatch.setattr(ProactiveCareScheduler, "start_all", _start_all)
+    monkeypatch.setattr(ProactiveCareScheduler, "_schedule", lambda self, _id: None)
+
+
 @pytest.fixture(autouse=True)
 def _isolated_user_home(
     tmp_path_factory: pytest.TempPathFactory,