Use sliding window

src/exo/main.py

@@ -24,7 +24,6 @@ from exo.routing.router import Router, get_node_id_keypair
 from exo.shared.constants import EXO_DEFAULT_MODELS_DIR, EXO_LOG, EXO_PID_FILE
 from exo.shared.election import Election, ElectionResult
 from exo.shared.logging import logger_cleanup, logger_setup
-from exo.shared.telemetry import TelemetryService
 from exo.shared.types.common import NodeId, SessionId
 from exo.utils import STDIO_FDS
 from exo.utils.channels import Receiver, channel
@@ -43,7 +42,6 @@ class Node:
     election_result_receiver: Receiver[ElectionResult]
     master: Master | None
     api: API | None
-    telemetry: TelemetryService
 
     node_id: NodeId
     offline: bool
@@ -72,7 +70,6 @@ class Node:
             external_outbound=router.sender(topics.LOCAL_EVENTS),
             external_inbound=router.receiver(topics.GLOBAL_EVENTS),
         )
-        telemetry = TelemetryService.create(telemetry_disabled=not args.telemetry)
 
         logger.info(f"Starting node {node_id}")
 
@@ -111,7 +108,6 @@ class Node:
                 command_sender=router.sender(topics.COMMANDS),
                 download_command_sender=router.sender(topics.DOWNLOAD_COMMANDS),
                 api_port=args.api_port,
-                telemetry_sink=telemetry.sink(),
             )
         else:
             worker = None
@@ -150,7 +146,6 @@ class Node:
             er_recv,
             master,
             api,
-            telemetry,
             node_id,
             args.offline,
             args.api_port,
@@ -162,7 +157,6 @@ class Node:
             signal.signal(signal.SIGTERM, lambda _, __: self.shutdown())
             tg.start_soon(self.router.run)
             tg.start_soon(self.event_router.run)
-            tg.start_soon(self.telemetry.run)
             tg.start_soon(self.election.run)
             if self.download_coordinator:
                 tg.start_soon(self.download_coordinator.run)
@@ -270,7 +264,6 @@ class Node:
                                 topics.DOWNLOAD_COMMANDS
                             ),
                             api_port=self._api_port,
-                            telemetry_sink=self.telemetry.sink(),
                         )
                         self._tg.start_soon(self.worker.run)
                     if self.api:
@@ -391,7 +384,6 @@ class Args(FrozenModel):
     no_downloads: bool = False
     offline: bool = os.getenv("EXO_OFFLINE", "false").lower() == "true"
     no_batch: bool = False
-    telemetry: bool = False
     fast_synch: bool | None = None  # None = auto, True = force on, False = force off
     legacy_daemon: bool = False
     bootstrap_peers: list[str] = []
@@ -453,11 +445,6 @@ class Args(FrozenModel):
             action="store_true",
             help="Disable continuous batching, use sequential generation",
         )
-        parser.add_argument(
-            "--telemetry",
-            action="store_true",
-            help="Enable telemetry uploads. Disabled by default; disabled mode keeps telemetry in dry-run.",
-        )
         parser.add_argument(
             "--legacy-daemon",
             action="store_true",

src/exo/shared/constants.py

@@ -26,11 +26,6 @@ EXO_CONFIG_HOME = _get_xdg_dir("XDG_CONFIG_HOME", ".config")
 EXO_DATA_HOME = _get_xdg_dir("XDG_DATA_HOME", ".local/share")
 EXO_CACHE_HOME = _get_xdg_dir("XDG_CACHE_HOME", ".cache")
 
-# Exo website API endpoints
-EXO_TELEMETRY_API_URL = os.environ.get(
-    "EXO_TELEMETRY_API_URL", "https://telemetry.exolabs.net/"
-)
-
 # Default models directory (always included as first entry in writable dirs)
 _EXO_DEFAULT_MODELS_DIR_ENV = os.environ.get("EXO_DEFAULT_MODELS_DIR", None)
 EXO_DEFAULT_MODELS_DIR = (
@@ -74,6 +69,8 @@ DASHBOARD_DIR = (
 EXO_LOG_DIR = EXO_CACHE_HOME / "exo_log"
 EXO_LOG = EXO_LOG_DIR / "exo.log"
 EXO_RUNNER_LOG_DIR = EXO_LOG_DIR / "runner_log"
+EXO_RUNNER_STDOUT_LOG = EXO_RUNNER_LOG_DIR / "stdout.log"
+EXO_RUNNER_STDERR_LOG = EXO_RUNNER_LOG_DIR / "stderr.log"
 
 EXO_TEST_LOG = EXO_CACHE_HOME / "exo_test.log"
 EXO_PID_FILE = EXO_CACHE_HOME / "exo.pid"

src/exo/shared/telemetry.py

@@ -1,176 +0,0 @@
-import contextlib
-import hashlib
-from dataclasses import dataclass, field
-from pathlib import Path
-from typing import Self
-from urllib.parse import urlparse
-
-import httpx
-from anyio import BrokenResourceError, ClosedResourceError, WouldBlock, to_thread
-from loguru import logger
-
-from exo.shared.constants import EXO_TELEMETRY_API_URL
-from exo.utils.channels import Receiver, Sender, channel
-from exo.utils.pydantic_ext import FrozenModel, TaggedModel
-from exo.utils.task_group import TaskGroup
-
-CHANNEL_BOUND_SIZE = 64
-TELEMETRY_HTTP_TIMEOUT_SECONDS = 10.0
-
-
-class BaseTelemetrySubmission(TaggedModel):
-    pass
-
-
-class TestSubmission(BaseTelemetrySubmission):
-    pass
-
-
-class RunnerStderrSubmission(BaseTelemetrySubmission):
-    path: Path
-
-
-TelemetrySubmission = TestSubmission | RunnerStderrSubmission
-
-
-class TelemetryPresignResponse(FrozenModel):
-    key: str
-    upload_url: str
-    expires_in: int
-    max_size: int
-
-
-@dataclass(eq=False)
-class TelemetrySink:
-    """
-    A non-blocking non-throwing bounded wrapper around sender/receiver channels
-    to ensure telemetry never blocks or has adverse side-effects, since telemetry
-    is an optional diagnostic feature and hence should never break the main app.
-    """
-
-    _send: Sender[TelemetrySubmission]
-
-    @classmethod
-    def pair(cls) -> tuple[Self, Receiver[TelemetrySubmission]]:
-        send, recv = channel[TelemetrySubmission](CHANNEL_BOUND_SIZE)
-        return cls(_send=send), recv
-
-    def submit(self, submission: TelemetrySubmission):
-        try:
-            self._send.send_nowait(submission)
-        except WouldBlock:
-            logger.debug("Telemetry submission would block. why so many submissions??")
-        except (BrokenResourceError, ClosedResourceError):
-            logger.debug("Telemetry submission receivers are broken or closed. why??")
-
-    def clone(self) -> "TelemetrySink":
-        return TelemetrySink(_send=self._send.clone())
-
-    def close(self):
-        with contextlib.suppress(BrokenResourceError, ClosedResourceError):
-            self._send.close()
-
-
-@dataclass(eq=False)
-class TelemetryService:
-    telemetry_disabled: bool
-    api_url: str
-    _send: Sender[TelemetrySubmission]
-    _recv: Receiver[TelemetrySubmission]
-    _http_transport: httpx.AsyncBaseTransport | None
-    _tg: TaskGroup = field(default_factory=TaskGroup, init=False)
-
-    @classmethod
-    def create(
-        cls,
-        telemetry_disabled: bool,
-        api_url: str = EXO_TELEMETRY_API_URL,
-        http_transport: httpx.AsyncBaseTransport | None = None,
-    ) -> Self:
-        api_url = urlparse(api_url).geturl().rstrip("/")
-
-        send, recv = channel[TelemetrySubmission](CHANNEL_BOUND_SIZE)
-
-        return cls(
-            telemetry_disabled=telemetry_disabled,
-            api_url=api_url,
-            _send=send,
-            _recv=recv,
-            _http_transport=http_transport,
-        )
-
-    @classmethod
-    def dummy(cls) -> Self:
-        return cls.create(True)
-
-    async def run(self):
-        try:
-            async with self._tg as tg:
-                tg.start_soon(self._process)
-        finally:
-            self._send.close()
-            self._recv.close()
-
-    async def _process(self):
-        with self._recv as submissions:
-            async for submission in submissions:
-                if not self.telemetry_disabled:
-                    try:
-                        await self._process_submission(submission)
-                    except Exception as e:
-                        logger.opt(exception=e).warning(
-                            "Exception when processing telemetry submission"
-                        )
-
-    async def _process_submission(self, submission: TelemetrySubmission):
-        match submission:
-            case TestSubmission():
-                pass
-            case RunnerStderrSubmission(path=path):
-                await self._submit_runner_stderr(path)
-
-    async def _submit_runner_stderr(self, path: Path):
-        data = await to_thread.run_sync(path.read_bytes)
-        if not data:
-            logger.debug(f"Skipping empty runner stderr telemetry file: {path}")
-            return
-
-        sha256 = hashlib.sha256(data).hexdigest()
-
-        async with httpx.AsyncClient(
-            timeout=TELEMETRY_HTTP_TIMEOUT_SECONDS,
-            transport=self._http_transport,
-        ) as client:
-            presign_response = await client.post(
-                f"{self.api_url}/telemetry/runner-log/presign",
-                json={
-                    "sha256": sha256,
-                    "size": len(data),
-                },
-            )
-            presign_response.raise_for_status()
-            presign = TelemetryPresignResponse.model_validate_json(
-                presign_response.text,
-            )
-
-            upload_response = await client.put(
-                presign.upload_url,
-                content=data,
-            )
-            upload_response.raise_for_status()
-
-    def sink(self) -> TelemetrySink:
-        sink, recv = TelemetrySink.pair()
-        if self._tg.is_running():
-            self._tg.start_soon(self._ingest, recv)
-        else:
-            self._tg.queue(self._ingest, recv)
-        return sink
-
-    async def _ingest(self, recv: Receiver[TelemetrySubmission]):
-        try:
-            with recv as submissions:
-                async for submission in submissions:
-                    await self._send.send(submission)
-        except ClosedResourceError:
-            pass

src/exo/shared/tests/test_telemetry.py

@@ -1,95 +0,0 @@
-import hashlib
-import json
-from dataclasses import dataclass
-from pathlib import Path
-
-import httpx
-import pytest
-
-from exo.shared.telemetry import RunnerStderrSubmission, TelemetryService
-
-
-@dataclass(frozen=True)
-class RecordedRequest:
-    method: str
-    url: str
-    content: bytes
-
-
-def _queue_submission(
-    service: TelemetryService,
-    submission: RunnerStderrSubmission,
-) -> None:
-    service._send.send_nowait(submission)  # pyright: ignore[reportPrivateUsage]
-    service._send.close()  # pyright: ignore[reportPrivateUsage]
-
-
-@pytest.mark.anyio
-async def test_runner_stderr_upload_hashes_and_uploads_file_bytes(tmp_path: Path):
-    log_bytes = b"runner stderr\nsecond line\n"
-    log_path = tmp_path / "runner.stderr.log"
-    log_path.write_bytes(log_bytes)
-    requests: list[RecordedRequest] = []
-
-    async def handler(request: httpx.Request) -> httpx.Response:
-        requests.append(
-            RecordedRequest(
-                method=request.method,
-                url=str(request.url),
-                content=await request.aread(),
-            )
-        )
-        if request.method == "POST":
-            return httpx.Response(
-                200,
-                json={
-                    "key": "runner_log/test.stderr.log",
-                    "uploadUrl": "https://uploads.example/runner.stderr.log",
-                    "expiresIn": 300,
-                    "maxSize": 52428800,
-                },
-            )
-        if request.method == "PUT":
-            return httpx.Response(200)
-        return httpx.Response(404)
-
-    service = TelemetryService.create(
-        telemetry_disabled=False,
-        api_url="https://telemetry.example/",
-        http_transport=httpx.MockTransport(handler),
-    )
-
-    await service._process_submission(  # pyright: ignore[reportPrivateUsage]
-        RunnerStderrSubmission(path=log_path)
-    )
-
-    assert [r.method for r in requests] == ["POST", "PUT"]
-    assert requests[0].url == "https://telemetry.example/telemetry/runner-log/presign"
-    assert json.loads(requests[0].content) == {
-        "sha256": hashlib.sha256(log_bytes).hexdigest(),
-        "size": len(log_bytes),
-    }
-    assert requests[1].url == "https://uploads.example/runner.stderr.log"
-    assert requests[1].content == log_bytes
-
-
-@pytest.mark.anyio
-async def test_runner_stderr_upload_failure_is_swallowed(tmp_path: Path):
-    log_path = tmp_path / "runner.stderr.log"
-    log_path.write_text("runner stderr\n")
-    requests: list[httpx.Request] = []
-
-    async def handler(request: httpx.Request) -> httpx.Response:
-        requests.append(request)
-        return httpx.Response(500)
-
-    service = TelemetryService.create(
-        telemetry_disabled=False,
-        api_url="https://telemetry.example",
-        http_transport=httpx.MockTransport(handler),
-    )
-    _queue_submission(service, RunnerStderrSubmission(path=log_path))
-
-    await service._process()  # pyright: ignore[reportPrivateUsage]
-
-    assert len(requests) == 1

src/exo/worker/engines/mlx/cache.py

@@ -229,6 +229,47 @@ def has_non_kv_caches(cache: KVCacheType) -> bool:
     return any(is_non_trimmable_cache_entry(c) for c in cache)
 
 
+# Max snapshots retained per cache entry. Each CacheSnapshot pins detached GPU
+# copies of every non-trimmable (SSM/ArraysCache, RotatingKVCache) layer, so
+# retaining one per ~4096-token prefill chunk makes snapshot memory grow linearly
+# with context — the dominant residual cost when a single entry is grown to long
+# contexts on hybrid models (~56 MB/snapshot on Qwen3.5-122B, so a full 256K
+# context = 64 snapshots ≈ 3.6 GB). A sliding window of the most-recent N caps
+# this at N×per-snapshot (~0.9 GB here) while preserving the restore points
+# in-place grows actually use (they always extend from the tip).
+_MAX_RETAINED_SNAPSHOTS = 16
+
+
+def _bounded_snapshots(snapshots: list[CacheSnapshot]) -> list[CacheSnapshot]:
+    """Deduplicate snapshots by token position and bound the retained count.
+
+    Returned list is sorted ascending by ``token_count``.
+    """
+    # Deduplicate by position, keeping the most-recently-appended snapshot per
+    # position. Repeated in-place grows re-snapshot positions the kept old
+    # snapshots already cover, which would otherwise grow `_snapshots`
+    # unbounded even at constant context.
+    # TODO: keying on token_count alone is safe only while a position uniquely
+    # identifies the prefix within an entry (grows are strict prefix-extensions).
+    # If edit-and-regenerate, sliding-window/prefix trimming, cross-entry
+    # snapshot sharing, per-request adapter/LoRA swap, or branchy decoding
+    # (beam/parallel/speculative) is added, enrich the key to
+    # (token_count, prefix_hash[, media/adapter id]) — else a stale snapshot
+    # could be restored for a different prefix (silent wrong output).
+    by_position: dict[int, CacheSnapshot] = {}
+    for snapshot in snapshots:
+        by_position[snapshot.token_count] = snapshot
+    deduped = [by_position[pos] for pos in sorted(by_position)]
+
+    # Sliding window: keep only the most-recent N positions. In-place grows
+    # always extend from the tip, so the newest snapshots are the ones future
+    # grows restore from — dropping the oldest is never incorrect: a later hit on
+    # a prefix older than the window finds no snapshot <= target, so get_kv_cache
+    # returns a fresh cache (matched_index=None) and the request takes a full cold
+    # prefill — correct, just slower than a partial-hit reuse for that one request.
+    return deduped[-_MAX_RETAINED_SNAPSHOTS:]
+
+
 class KVPrefixCache:
     def __init__(self, group: mx.distributed.Group | None):
         self.prompts: list[mx.array] = []  # mx array of tokens (ints)
@@ -261,7 +302,9 @@ class KVPrefixCache:
         self._evict_if_needed()
         self.prompts.append(prompt_tokens)
         self.caches.append(deepcopy(cache))
-        self._snapshots.append(ssm_snapshots)
+        self._snapshots.append(
+            _bounded_snapshots(ssm_snapshots) if ssm_snapshots else None
+        )
         self._media_regions.append(media_regions or [])
         self.prefill_tps.append(prefill_tps)
         self._access_counter += 1
@@ -288,7 +331,7 @@ class KVPrefixCache:
 
         self.prompts[index] = prompt_tokens
         self.caches[index] = deepcopy(cache)
-        self._snapshots[index] = merged or None
+        self._snapshots[index] = _bounded_snapshots(merged) or None
         self._media_regions[index] = media_regions or []
         self.prefill_tps[index] = prefill_tps
         self._access_counter += 1

src/exo/worker/main.py

@@ -15,7 +15,6 @@ from exo.routing.event_router import (
 from exo.shared.apply import apply
 from exo.shared.constants import EXO_MAX_INSTANCE_RETRIES
 from exo.shared.models.model_cards import ModelId, card_cache
-from exo.shared.telemetry import TelemetrySink
 from exo.shared.types.chunks import InputImageChunk
 from exo.shared.types.commands import (
     DeleteInstance,
@@ -75,7 +74,6 @@ class Worker:
         command_sender: Sender[ForwarderCommand],
         download_command_sender: Sender[ForwarderDownloadCommand],
         api_port: int,
-        telemetry_sink: TelemetrySink,
     ):
         self.node_id: NodeId = node_id
         self.event_receiver = event_receiver
@@ -83,7 +81,6 @@ class Worker:
         self.command_sender = command_sender
         self.download_command_sender = download_command_sender
         self.api_port = api_port
-        self.telemetry_sink = telemetry_sink
 
         self.state: State = State()
         self.runners: dict[RunnerId, RunnerSupervisor] = {}
@@ -125,7 +122,6 @@ class Worker:
             self.event_sender.close()
             self.command_sender.close()
             self.download_command_sender.close()
-            self.telemetry_sink.close()
             for runner in self.runners.values():
                 runner.shutdown()
             self._stopped.set()
@@ -385,7 +381,6 @@ class Worker:
         runner = await RunnerSupervisor.create(
             bound_instance=task.bound_instance,
             event_sender=self.event_sender.clone(),
-            telemetry_sink=self.telemetry_sink.clone(),
         )
         self.runners[task.bound_instance.bound_runner_id] = runner
         self._tg.start_soon(runner.run)

src/exo/worker/runner/supervisor.py

@@ -2,9 +2,8 @@ import codecs
 import contextlib
 import signal
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
-from pathlib import Path
-from typing import Self
+from os import PathLike
+from typing import Callable, Self
 
 import anyio
 from anyio import (
@@ -13,13 +12,9 @@ from anyio import (
     CancelScope,
     ClosedResourceError,
 )
-from anyio.streams.text import TextReceiveStream
 from loguru import logger
 
-from exo.shared.constants import (
-    EXO_RUNNER_LOG_DIR,
-)
-from exo.shared.telemetry import RunnerStderrSubmission, TelemetrySink
+from exo.shared.constants import EXO_RUNNER_STDERR_LOG, EXO_RUNNER_STDOUT_LOG
 from exo.shared.types.chunks import ErrorChunk
 from exo.shared.types.events import (
     ChunkGenerated,
@@ -65,12 +60,10 @@ DECODE_TIMEOUT_SECONDS = 5
 
 @dataclass(eq=False)
 class RunnerStdioHandler:
-    _bound_instance: BoundInstance
     _stdout_rx: Receiver[bytes]
     _stderr_rx: Receiver[bytes]
-    _stderr_log_path: Path
+    _stdout_log: AsyncFile[str]
     _stderr_log: AsyncFile[str]
-    _telemetry: TelemetrySink
     diagnostics: RunnerDiagnosticCollector = field(
         default_factory=RunnerDiagnosticCollector
     )
@@ -81,68 +74,58 @@ class RunnerStdioHandler:
     async def create(
         cls,
         *,
-        bound_instance: BoundInstance,
         stdout_rx: Receiver[bytes],
         stderr_rx: Receiver[bytes],
-        telemetry_sink: TelemetrySink,
-        runner_log_dir: Path = EXO_RUNNER_LOG_DIR,
+        stdout_log_path: PathLike[str] = EXO_RUNNER_STDOUT_LOG,
+        stderr_log_path: PathLike[str] = EXO_RUNNER_STDERR_LOG,
     ) -> Self:
-        # create file in <log_dir>/<instanceID>/<runnerID>/<timestamp>.stderr.log
-        now = datetime.now(timezone.utc).strftime("%Y-%m-%d_%H-%M-%S_%fZ")
-        stderr_log_path = (
-            runner_log_dir
-            / bound_instance.instance.instance_id
-            / bound_instance.bound_runner_id
-            / f"{now}.stderr.log"
-        )
+        # these are append only logs used to gather data for log template mining
+        #
+        # TODO: in the future use [Drain3](https://github.com/logpai/Drain3)
+        #       to mine these logs
+        ensure_parent_directory_exists(stdout_log_path)
         ensure_parent_directory_exists(stderr_log_path)
-        stderr_log = await anyio.open_file(stderr_log_path, "w")
+        stdout_log = await anyio.open_file(stdout_log_path, "a")
+        stderr_log = await anyio.open_file(stderr_log_path, "a")
 
+        # instantiate and return
         self = cls(
-            _bound_instance=bound_instance,
             _stdout_rx=stdout_rx,
             _stderr_rx=stderr_rx,
-            _stderr_log_path=stderr_log_path,
+            _stdout_log=stdout_log,
             _stderr_log=stderr_log,
-            _telemetry=telemetry_sink,
         )
         return self
 
     async def run(self):
         try:
             async with self._tg as tg:
-                tg.start_soon(self._handle_stdout)
-                tg.start_soon(self._handle_stderr)
+                tg.start_soon(  # pyright: ignore[reportUnknownArgumentType]
+                    self._handle_runner_output,
+                    self._stdout_rx,
+                    self._stdout_log,
+                    lambda line: logger.info(f"Runner stdout: {line}"),  # pyright: ignore[reportUnknownLambdaType]
+                    lambda _: None,  # pyright: ignore[reportUnknownLambdaType]
+                )
+                tg.start_soon(  # pyright: ignore[reportUnknownArgumentType]
+                    self._handle_runner_output,
+                    self._stderr_rx,
+                    self._stderr_log,
+                    lambda line: logger.warning(f"Runner stderr: {line}"),  # pyright: ignore[reportUnknownLambdaType]
+                    self.diagnostics.record_line,
+                )
         finally:
             with CancelScope(shield=True):
+                await self._stdout_log.aclose()
                 await self._stderr_log.aclose()
 
-                # send off telemetry submission when runner stdio dies;
-                # it may have been for entirely innocuous reasons or
-                # the log may have nothing in it, but its submitted regardless
-                self._telemetry.submit(
-                    RunnerStderrSubmission(
-                        path=self._stderr_log_path,
-                    )
-                )
-                self._telemetry.close()
-
-    def shutdown(self):
-        self._tg.cancel_tasks()
-
-    async def _handle_stdout(self):
-        # We don't expect anything in stdout so even reading this at all is going
-        # to be quite weird; hence handle it by logging error and the received chunk
-
-        rx = TextReceiveStream(self._stdout_rx, encoding="utf-8", errors="replace")
-        try:
-            async with rx:
-                async for chunk in rx:
-                    logger.warning(f"Unexpected runner stdout chunk: {chunk}")
-        except (ClosedResourceError, BrokenResourceError):
-            logger.warning("Runner stdio stream closed before clean EOF")
-
-    async def _handle_stderr(self):
+    async def _handle_runner_output(
+        self,
+        rx: Receiver[bytes],
+        logfile: AsyncFile[str],
+        log_line: Callable[[str], None],
+        record_diagnostic_line: Callable[[str], None],
+    ):
         # The diagnostic collector is deliberately line-level for now. It records
         # bounded stderr context and known failure anchors; the supervisor
         # correlates those hints with the runner exit status before surfacing an
@@ -159,8 +142,8 @@ class RunnerStdioHandler:
                 return
 
             # Send to logger & error recovery task
-            logger.warning(f"Runner stderr: {line}")
-            self.diagnostics.record_line(line)
+            log_line(line)
+            record_diagnostic_line(line)
 
         async def handle_text(text: str):
             nonlocal pending_line
@@ -168,8 +151,8 @@ class RunnerStdioHandler:
             if not text:
                 return
 
-            await self._stderr_log.write(text)
-            await self._stderr_log.flush()
+            await logfile.write(text)
+            await logfile.flush()
 
             # newline buffering
             pending_line += text
@@ -180,15 +163,15 @@ class RunnerStdioHandler:
                 await handle_line(line)
 
         try:
-            with self._stderr_rx:
-                async for chunk in self._stderr_rx:
+            with rx:
+                async for chunk in rx:
                     await handle_text(decoder.decode(chunk, final=False))
         except (ClosedResourceError, BrokenResourceError):
             logger.warning("Runner stdio stream closed before clean EOF")
         finally:
             with CancelScope(shield=True):
                 await handle_text(decoder.decode(b"", final=True))
-                await self._stderr_log.flush()
+                await logfile.flush()
 
                 if pending_line:
                     await handle_line(pending_line)
@@ -222,7 +205,6 @@ class RunnerSupervisor:
         *,
         bound_instance: BoundInstance,
         event_sender: Sender[Event],
-        telemetry_sink: TelemetrySink,
         initialize_timeout: float = 400,
     ) -> Self:
         ev_send, ev_recv = mp_channel[Event | RunnerTerminationError]()
@@ -241,10 +223,7 @@ class RunnerSupervisor:
             daemon=True,
         )
         runner_stdio_handler = await RunnerStdioHandler.create(
-            bound_instance=bound_instance,
-            stdout_rx=runner_process.stdout,
-            stderr_rx=runner_process.stderr,
-            telemetry_sink=telemetry_sink,
+            stdout_rx=runner_process.stdout, stderr_rx=runner_process.stderr
         )
 
         shard_metadata = bound_instance.bound_shard

src/exo/worker/tests/unittests/test_mlx/test_kv_prefix_cache.py

@@ -11,6 +11,7 @@ from mlx_lm.sample_utils import make_sampler
 from exo.shared.types.common import ModelId
 from exo.shared.types.text_generation import InputMessage, TextGenerationTaskParams
 from exo.worker.engines.mlx.cache import (
+    CacheSnapshot,
     KVPrefixCache,
     cache_length,
     encode_prompt,
@@ -77,6 +78,74 @@ class TestGetPrefixLength:
         assert get_prefix_length(a, b) == 0
 
 
+class TestSnapshotAccumulation:
+    """Locks in the fix for the actual per-grow Metal leak on hybrid (SSM)
+    models: `update_kv_cache` must not let `_snapshots` grow without bound when
+    the same entry is grown in place many times."""
+
+    def test_repeated_update_does_not_accumulate_snapshots(self):
+        with patch(
+            "exo.worker.engines.mlx.cache.get_memory_used_percentage",
+            return_value=0.0,
+        ):
+            kv_prefix_cache = KVPrefixCache(None)
+            initial = [
+                CacheSnapshot(states=[None], token_count=4096),
+                CacheSnapshot(states=[None], token_count=8192),
+            ]
+            kv_prefix_cache.add_kv_cache(
+                mx.arange(10000), [KVCache()], ssm_snapshots=initial
+            )
+
+            # Each in-place grow re-prefills from restore_pos and produces a
+            # fresh snapshot at a position the retained old snapshots already
+            # cover. Pre-fix this appended one snapshot per grow forever.
+            for _ in range(50):
+                fresh = [CacheSnapshot(states=[None], token_count=8192)]
+                kv_prefix_cache.update_kv_cache(
+                    0, mx.arange(10000), [KVCache()], fresh, restore_pos=8192
+                )
+
+            stored = kv_prefix_cache._snapshots[0]
+            assert stored is not None
+            # Bounded by the number of distinct snapshot positions (here 2),
+            # not by the 50 grows.
+            assert len(stored) == 2
+            assert sorted(s.token_count for s in stored) == [4096, 8192]
+            # The kept 8192 snapshot must be the most recently supplied one.
+            assert stored[1] is fresh[0]
+
+    def test_extension_caps_snapshots_to_sliding_window(self):
+        """Extending a single entry to a long context (one snapshot per ~4096
+        tokens) must cap retained snapshots to a sliding window of the most-recent
+        N, not keep all of them — that linear-in-context retention was the
+        residual OOM cause."""
+        from exo.worker.engines.mlx.cache import _MAX_RETAINED_SNAPSHOTS
+
+        with patch(
+            "exo.worker.engines.mlx.cache.get_memory_used_percentage",
+            return_value=0.0,
+        ):
+            kv_prefix_cache = KVPrefixCache(None)
+            # 64 distinct positions = a 262144-token context at 4096/chunk.
+            num_positions = 64
+            snaps = [
+                CacheSnapshot(states=[None], token_count=4096 * (i + 1))
+                for i in range(num_positions)
+            ]
+            kv_prefix_cache.add_kv_cache(
+                mx.arange(10), [KVCache()], ssm_snapshots=snaps
+            )
+
+            stored = kv_prefix_cache._snapshots[0]
+            assert stored is not None
+            # Capped at the window; the most-recent N positions are retained
+            # (in-place grows extend from the tip, so these are what get used).
+            assert len(stored) == _MAX_RETAINED_SNAPSHOTS
+            assert stored == snaps[-_MAX_RETAINED_SNAPSHOTS:]
+            assert stored[-1] is snaps[-1]  # tip always kept
+
+
 class TestKVPrefix:
     @pytest.fixture
     def mock_tokenizer(self):

src/exo/worker/tests/unittests/test_runner/test_runner_supervisor.py

@@ -1,11 +1,9 @@
-from pathlib import Path
 from typing import cast
 
 import anyio
 import pytest
 
 from exo.shared.models.model_cards import ModelId
-from exo.shared.telemetry import TelemetryService
 from exo.shared.types.chunks import ErrorChunk
 from exo.shared.types.common import CommandId, NodeId
 from exo.shared.types.events import ChunkGenerated, Event, RunnerStatusUpdated
@@ -38,9 +36,7 @@ class _DeadProcess:
 
 
 @pytest.mark.anyio
-async def test_check_runner_emits_error_chunk_for_inflight_text_generation(
-    tmp_path: Path,
-) -> None:
+async def test_check_runner_emits_error_chunk_for_inflight_text_generation() -> None:
     event_sender, event_receiver = channel[Event]()
     task_sender, _ = mp_channel[Task]()
     cancel_sender, _ = mp_channel[TaskId]()
@@ -54,13 +50,8 @@ async def test_check_runner_emits_error_chunk_for_inflight_text_generation(
     )
 
     proc = cast(AsyncProcess, cast(object, _DeadProcess()))
-    telemetry = TelemetryService.dummy()
     handler = await RunnerStdioHandler.create(
-        bound_instance=bound_instance,
-        stdout_rx=proc.stdout,
-        stderr_rx=proc.stderr,
-        telemetry_sink=telemetry.sink(),
-        runner_log_dir=tmp_path,
+        stdout_rx=proc.stdout, stderr_rx=proc.stderr
     )
     supervisor = RunnerSupervisor(
         shard_metadata=bound_instance.bound_shard,