refactor(agent): Flag streamer follow-ups and cover 404 cleanup path

- Add TODO(#11223) notes on CPU/Memory api_impl paths: after sysfs-first
  lands, the Docker stream on these plugins is only needed for network
  and blkio (which neither consumes), so the stream should migrate to
  a network/IO consumer.
- Add TODO(#11232) at the two per-plugin DockerStatsStreamer instantiation
  sites flagging the shared-streamer refactor.
- Add a test covering the DockerError 404 branch of _read_stream: reader
  exits cleanly on container-gone without spinning retries.

Refs #11219
Refs #11223
Refs #11224
Refs #11232

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rapsealk

src/ai/backend/agent/docker/intrinsic.py

@@ -356,6 +356,7 @@ class CPUPlugin(AbstractComputePlugin):
 
     async def init(self, context: Any | None = None) -> None:
         self._docker = Docker()
+        # TODO(#11232): Consolidate per-plugin streamer into a single shared instance owned by the agent.
         self._stats_streamer = DockerStatsStreamer(self._docker)
 
     async def cleanup(self) -> None:
@@ -466,6 +467,7 @@ async def sysfs_impl(container_id: str) -> float | None:
                 return None
             return cpu_used
 
+        # TODO(#11223): After sysfs-first lands, CPU is sourced from sysfs and this stream becomes redundant here; migrate to a network/IO consumer.
         async def api_impl(container_id: str) -> float | None:
             ret = self._stats_streamer.get_latest(container_id)
             if ret is None:
@@ -687,6 +689,7 @@ class MemoryPlugin(AbstractComputePlugin):
 
     async def init(self, context: Any | None = None) -> None:
         self._docker = Docker()
+        # TODO(#11232): Consolidate per-plugin streamer into a single shared instance owned by the agent.
         self._stats_streamer = DockerStatsStreamer(self._docker)
 
     async def cleanup(self) -> None:
@@ -958,6 +961,7 @@ async def sysfs_impl(
                 scratch_sz,
             )
 
+        # TODO(#11223): After sysfs-first lands, memory is sourced from sysfs and this stream becomes redundant here; migrate to a network/IO consumer.
         async def api_impl(
             container_id: str,
         ) -> tuple[int, int, int, int, int, int, int] | None:

tests/unit/agent/test_docker_intrinsic.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 from collections.abc import Generator
 from contextlib import contextmanager
 from dataclasses import dataclass
@@ -10,6 +11,7 @@
 
 import aiohttp
 import pytest
+from aiodocker.exceptions import DockerError
 
 from ai.backend.agent.agent import AbstractAgent
 from ai.backend.agent.docker.intrinsic import (
@@ -820,6 +822,61 @@ def fake_container_cls(docker: Any, id: str) -> _FakeDockerContainer:
             assert task.done()
             assert task.cancelled() or task.exception() is None
 
+    async def test_reader_exits_cleanly_on_container_gone_404(
+        self,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """DockerError 404 from the stats stream means the container is gone:
+        the reader must exit cleanly (no retry spin, no warning-level log)."""
+
+        call_count = 0
+
+        async def frames(_cid: str) -> Any:
+            nonlocal call_count
+            call_count += 1
+            # Raise on the very first iteration to simulate an already-gone
+            # container. A no-op ``yield`` in an unreachable branch keeps this
+            # function an async generator without tripping ``unreachable`` mypy.
+            if call_count < 0:
+                yield {}
+            raise DockerError(404, {"message": "No such container"})
+
+        def fake_container_cls(docker: Any, id: str) -> _FakeDockerContainer:
+            return _FakeDockerContainer(frames, id)
+
+        with (
+            patch(
+                "ai.backend.agent.docker.intrinsic.DockerContainer",
+                side_effect=fake_container_cls,
+            ),
+            caplog.at_level(logging.DEBUG, logger="ai.backend.agent.docker.intrinsic"),
+        ):
+            streamer = DockerStatsStreamer(AsyncMock())
+            streamer.start("cid_gone")
+            task = streamer._tasks["cid_gone"]
+            # The reader must exit on its own; 404 must NOT spin in the retry loop.
+            await asyncio.wait_for(task, timeout=2.0)
+            assert task.done()
+            assert not task.cancelled()
+            assert task.exception() is None
+            # Only one stream-open attempt; no reconnect/retry on 404.
+            assert call_count == 1
+            # ``get_latest`` may re-spawn a reader as a safety net, but the
+            # cached sample was dropped in the reader's finally block and the
+            # respawned reader will hit the same 404 path, so the return value
+            # is still ``None``.
+            assert streamer._latest.get("cid_gone") is None
+            await streamer.close()
+            # 404 is a debug-level path; container-gone is not an error.
+            intrinsic_records = [
+                r for r in caplog.records if r.name == "ai.backend.agent.docker.intrinsic"
+            ]
+            for record in intrinsic_records:
+                assert record.levelno < logging.WARNING, (
+                    f"unexpected warning-or-above log: {record.levelname} {record.getMessage()}"
+                )
+            assert streamer.get_latest("cid_gone") is None
+
 
 class _AgentStub:
     """Minimal stand-in exposing only the attributes the notify helpers read."""