refactor: pre-M4 protocol cleanup — ACP path unification, legacy removal, type safety

openspec/changes/m4-multi-config/tasks.md

@@ -119,6 +119,20 @@
 - [ ] 14.5 Implement `AgentPool.from_registry(registry: ConfigRegistry, config_id: str = "default") -> AgentPool` classmethod — retrieves config from registry, initializes infrastructure from HostConfig, constructs HostContext with `config_id`, compiles agents via AgentFactory
 - [ ] 14.6 Verify `async with AgentPool("config.yml")` still works: internally creates ConfigRegistry, registers file with `config_id="default"`, delegates to `from_registry()`
 - [ ] 14.7 Write unit tests: `from_registry` produces correct HostContext with config_id, file-path constructor preserves all behavior, HostContext model_cache is per-Host scoped, HostContext reconstruction preserves config_id/tenant_id
+- [ ] 14.8 Remove pool-level `input_provider` fallback in `NodeContext.get_input_provider()` (step 3 of 4 in `src/agentpool/messaging/context.py:60-61`). This fallback accesses `self.pool._input_provider` (private field on `AgentPool`), which is already deprecated — `BaseAgent._input_provider` warns to "Use SessionState.input_provider instead". The session-level provider (step 2) and ContextVar fallback (step 4) already cover all cases. After removal, `NodeContext.pool` is only used for `prompt_manager` (which `HostContext` already has), enabling the `NodeContext.pool` → `NodeContext.host` migration in task 14.9.
+  - Verify: `grep -n 'pool\._input_provider' src/agentpool/messaging/context.py` returns 0
+  - Verify: `uv run pytest tests/ -x` passes (no test relies on pool-level input_provider fallback)
+- [ ] 14.9 Migrate `NodeContext.pool: AgentPool | None` → `NodeContext.host: HostContext | None` in `src/agentpool/messaging/context.py`:
+  - Update `get_input_provider()` to use `self.host.input_provider` instead of `self.pool._input_provider`
+  - Update `prompt_manager` property to use `self.host.prompt_manager` instead of `self.pool.prompt_manager`
+  - Update `TeamContext(pool=...)` → `TeamContext(host=...)` in `base_team.py:get_context()`
+  - Update `AgentContext(pool=...)` → `AgentContext(host=...)` in `base_agent.py:get_context()`
+  - Verify: `grep -rn 'NodeContext.*pool' src/` returns 0 (field renamed)
+  - Verify: `grep -rn '\.pool\b' src/agentpool/messaging/context.py` returns 0
+- [ ] 14.10 Move `get_skill_instructions_for_node()` from `AgentPool` to `SkillsManager` — update `base_team.py:_load_skill_instructions()` to use `self.host_context.skills_registry.get_skill_instructions_for_node()` instead of `self._agent_pool.get_skill_instructions_for_node()`. Also move `skill_provider` property if needed.
+  - Verify: `grep -rn 'get_skill_instructions_for_node' src/agentpool/delegation/` uses `skills_registry` not `_agent_pool`
+- [ ] 14.11 Audit remaining `_agent_pool` references in protocol servers (`acp_server/acp_agent.py:254,1114`, `opencode_server/state.py:119`, `opencode_server/routes/agent_routes.py:162`) and migrate to `host_context` accessors where possible. References that need the agent registry (for `get_agent()` / `register()`) should use `AgentRegistry` interface from `AgentContext.agent_registry`. Defer any `from_callback(agent_pool=)` refactoring to M5.
+  - Verify: `grep -rn '\._agent_pool' src/agentpool_server/` returns 0 (or only M5-deferred `from_callback` sites)
 
 ## 15. Hot Reload: Triggers and Turn-Level Snapshot
 

src/acp/agent/acp_agent_api.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
 
 from acp.schema import (
     AuthenticateRequest,
@@ -22,7 +22,7 @@
 
 
 if TYPE_CHECKING:
-    from collections.abc import Sequence
+    from collections.abc import AsyncIterator, Sequence
 
     from acp.agent.protocol import Agent
     from acp.schema import (
@@ -35,26 +35,70 @@
         NewSessionResponse,
         PromptResponse,
         ResumeSessionResponse,
+        SessionUpdate,
         SetSessionConfigOptionResponse,
         SetSessionModelResponse,
         SetSessionModeResponse,
     )
     from acp.schema.mcp import McpServer
 
 
+@runtime_checkable
+class _SessionStateProtocol(Protocol):
+    """Protocol for session state objects that ACPAgentAPI can poll for updates."""
+
+    def pop_update(self) -> SessionUpdate | None: ...
+    def clear(self) -> None: ...
+
+
+@runtime_checkable
+class _UpdateEventProtocol(Protocol):
+    """Protocol for update events that ACPAgentAPI can wait on."""
+
+    async def wait_with_timeout(self, timeout: float | None = None) -> bool: ...
+    def clear(self) -> None: ...
+
+
 class ACPAgentAPI:
     """Thin wrapper for client-to-agent ACP interactions.
 
     Avoids manual instantiation of request/notification objects.
+
+    When optional ``state`` and ``update_event`` are provided, the instance
+    also satisfies the :class:`~agentpool.agents.acp_agent.turn.ACPClientProtocol`
+    protocol by implementing :meth:`stream_events` and :meth:`get_messages`.
     """
 
-    def __init__(self, connection: Agent) -> None:
+    def __init__(
+        self,
+        connection: Agent,
+        *,
+        state: _SessionStateProtocol | None = None,
+        update_event: _UpdateEventProtocol | None = None,
+    ) -> None:
         """Initialize agent API helper.
 
         Args:
             connection: The Agent protocol connection (e.g., ClientSideConnection)
+            state: Optional session state for polling updates (enables stream_events)
+            update_event: Optional event signaled when new updates arrive
         """
         self.connection = connection
+        self._state: _SessionStateProtocol | None = state
+        self._update_event: _UpdateEventProtocol | None = update_event
+        self._consumed_updates: list[SessionUpdate] = []
+
+    def _attach_state(
+        self,
+        state: _SessionStateProtocol,
+        update_event: _UpdateEventProtocol,
+    ) -> None:
+        """Attach state and update event after construction.
+
+        Allows deferred wiring when state/event are created after the API.
+        """
+        self._state = state
+        self._update_event = update_event
 
     async def initialize(
         self,
@@ -224,3 +268,45 @@ async def ext_method(self, method: str, params: dict[str, Any]) -> dict[str, Any
     async def ext_notification(self, method: str, params: dict[str, Any]) -> None:
         """Send an extension notification to the agent."""
         await self.connection.ext_notification(method, params)
+
+    async def stream_events(
+        self,
+        response: PromptResponse,
+    ) -> AsyncIterator[SessionUpdate]:
+        """Yield raw ACP session updates from the state queue.
+
+        Polls :meth:`_SessionStateProtocol.pop_update` in a loop, waiting
+        up to 50 ms between drain cycles for new updates to arrive via
+        ``_update_event``.  Once a full drain cycle produces no updates,
+        the iterator ends.
+
+        Updates are also collected in ``_consumed_updates`` so that
+        :meth:`get_messages` can return them after streaming completes.
+
+        Args:
+            response: The prompt response (unused — updates come from state)
+        """
+        self._consumed_updates.clear()
+        if self._state is None or self._update_event is None:
+            return
+        while True:
+            try:
+                await self._update_event.wait_with_timeout(0.05)
+                self._update_event.clear()
+            except TimeoutError:
+                pass
+            drained_any = False
+            while (update := self._state.pop_update()) is not None:
+                self._consumed_updates.append(update)
+                yield update
+                drained_any = True
+            if not drained_any:
+                break
+
+    async def get_messages(self, session_id: str) -> list[SessionUpdate]:
+        """Return all session updates consumed during :meth:`stream_events`.
+
+        Args:
+            session_id: The ACP session ID (unused — updates are already collected)
+        """
+        return list(self._consumed_updates)