From a693e88565cff14036f98560791285cb8ebc07e7 Mon Sep 17 00:00:00 2001 From: Jun Zhou Date: Wed, 15 Jul 2026 10:34:42 -0700 Subject: [PATCH 1/7] =?UTF-8?q?feat(daemon):=20idle=20timeout=20=E2=80=94?= =?UTF-8?q?=20exit=20after=20inactivity,=20configurable=20via=20settings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The daemon now tracks client activity (every accepted connection, plus each handler task's completion so long streaming index runs count to their end) and exits after daemon.idle_timeout_minutes (default 180, 0 = never) without it, instead of holding the embedding model in RAM forever. The idle exit goes through the existing graceful shutdown path, so it leaves a last_exit marker (reason "idle_timeout") and clients transparently and silently restart the daemon on the next request. - settings: new DaemonSettings dataclass with idle_timeout_minutes, parsed from an optional daemon: section in global_settings.yml (absent section and missing file both fall back to the default) - protocol: HeartbeatRequest/HeartbeatResponse (daemon-side dispatch only; the MCP client loop comes separately) and idle observability fields on DaemonStatusResponse (idle_seconds, idle_timeout_minutes, last_heartbeat_seconds) - daemon: IdleReaper with a pure should_exit predicate — exit requires a positive timeout, no external supervisor (COCOINDEX_CODE_DAEMON_SUPERVISED), no live handler task, no project indexing (new ProjectRegistry.any_indexing), and idle time past the timeout. A periodic asyncio task (60 s interval, injectable for tests along with a seconds-scale timeout on run_daemon) triggers the shutdown. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JaFKwEvJXo7WQhZvrW5Qoe --- src/cocoindex_code/daemon.py | 143 +++++++++++++++++-- src/cocoindex_code/protocol.py | 22 +++ src/cocoindex_code/settings.py | 18 ++- tests/test_daemon_idle.py | 245 +++++++++++++++++++++++++++++++++ tests/test_protocol.py | 28 +++- tests/test_settings.py | 57 ++++++++ 6 files changed, 497 insertions(+), 16 deletions(-) create mode 100644 tests/test_daemon_idle.py diff --git a/src/cocoindex_code/daemon.py b/src/cocoindex_code/daemon.py index 1a339b9..068947b 100644 --- a/src/cocoindex_code/daemon.py +++ b/src/cocoindex_code/daemon.py @@ -42,6 +42,8 @@ ErrorResponse, HandshakeRequest, HandshakeResponse, + HeartbeatRequest, + HeartbeatResponse, IndexRequest, IndexStreamResponse, IndexWaitingNotice, @@ -60,6 +62,7 @@ ) from .settings import ( ChunkerMapping, + DaemonSettings, UserSettings, format_path_for_display, get_host_path_mappings, @@ -196,6 +199,57 @@ def list_projects(self) -> list[DaemonProjectInfo]: for root, project in self._projects.items() ] + def any_indexing(self) -> bool: + """True when any loaded project currently holds its index lock.""" + return any(project._index_lock.locked() for project in self._projects.values()) + + +# --------------------------------------------------------------------------- +# Idle reaper +# --------------------------------------------------------------------------- + + +class IdleReaper: + """Tracks client activity and decides when the daemon should idle-exit. + + ``last_activity`` is refreshed on every accepted connection and again when + each handler task finishes, so a long streaming request (e.g. an index run) + counts as activity up to its end. ``last_heartbeat`` additionally records + when the most recent ``HeartbeatRequest`` was dispatched — for status + reporting only; heartbeat connections refresh ``last_activity`` like any + other connection. + """ + + def __init__(self, timeout_s: float, *, supervised: bool) -> None: + self.timeout_s = timeout_s + self.supervised = supervised + self.last_activity = time.monotonic() + self.last_heartbeat: float | None = None + + def record_activity(self) -> None: + self.last_activity = time.monotonic() + + def record_heartbeat(self) -> None: + self.last_heartbeat = time.monotonic() + + def idle_seconds(self, now: float | None = None) -> float: + return (time.monotonic() if now is None else now) - self.last_activity + + def should_exit(self, *, now: float, active_handlers: int, indexing: bool) -> bool: + """Pure exit predicate — testable without an event loop. + + Exit only when a timeout is configured, no external supervisor owns + the daemon lifecycle, no handler task is live, no project is indexing, + and the idle period exceeds the timeout. + """ + if self.timeout_s <= 0: + return False + if self.supervised: + return False + if active_handlers > 0 or indexing: + return False + return now - self.last_activity > self.timeout_s + # --------------------------------------------------------------------------- # Connection handler @@ -210,6 +264,7 @@ async def handle_connection( settings_mtime_us: int | None, settings_env_names: list[str], handshake_warnings: list[str], + reaper: IdleReaper, ) -> None: """Handle a single client connection (per-request model). @@ -247,7 +302,7 @@ async def handle_connection( data = await loop.run_in_executor(None, conn.recv_bytes) req = decode_request(data) - result = await _dispatch(req, registry, start_time, on_shutdown, settings_env_names) + result = await _dispatch(req, registry, start_time, on_shutdown, settings_env_names, reaper) if isinstance(result, AsyncIterator): try: async for resp in result: @@ -463,6 +518,7 @@ async def _dispatch( start_time: float, on_shutdown: Callable[[], None], settings_env_names: list[str], + reaper: IdleReaper, ) -> ( Response | AsyncIterator[IndexStreamResponse] @@ -506,10 +562,16 @@ async def _dispatch( return project.get_status() if isinstance(req, DaemonStatusRequest): + now = time.monotonic() return DaemonStatusResponse( version=__version__, - uptime_seconds=time.monotonic() - start_time, + uptime_seconds=now - start_time, projects=registry.list_projects(), + idle_seconds=reaper.idle_seconds(now), + idle_timeout_minutes=round(reaper.timeout_s / 60), + last_heartbeat_seconds=( + None if reaper.last_heartbeat is None else now - reaper.last_heartbeat + ), ) if isinstance(req, RemoveProjectRequest): @@ -520,6 +582,10 @@ async def _dispatch( on_shutdown() return StopResponse(ok=True) + if isinstance(req, HeartbeatRequest): + reaper.record_heartbeat() + return HeartbeatResponse(ok=True) + if isinstance(req, DaemonEnvRequest): from .protocol import DbPathMappingEntry from .settings import get_db_path_mappings @@ -551,12 +617,21 @@ async def _dispatch( # --------------------------------------------------------------------------- -def run_daemon() -> None: +def run_daemon( + *, + idle_timeout_s: float | None = None, + idle_check_interval_s: float = 60.0, +) -> None: """Main entry point for the daemon process (blocking). Sets up the listener, runs the asyncio event loop (``loop.run_forever``) to serve connections, and performs cleanup when shutdown is requested via - ``StopRequest`` or a signal (SIGTERM / SIGINT). + ``StopRequest``, a signal (SIGTERM / SIGINT), or the idle timeout. + + ``idle_timeout_s`` and ``idle_check_interval_s`` exist so tests can run a + seconds-scale idle timeout in-process; production leaves them at their + defaults and the timeout comes from ``daemon.idle_timeout_minutes`` in + ``global_settings.yml`` (the dataclass default when the file is missing). """ daemon_runtime_dir().mkdir(parents=True, exist_ok=True) @@ -570,8 +645,10 @@ def run_daemon() -> None: indexing_params: dict[str, Any] = {} query_params: dict[str, Any] = {} handshake_warnings: list[str] = [] + daemon_settings = DaemonSettings() if user_settings_path().is_file(): user_settings = load_user_settings() + daemon_settings = user_settings.daemon settings_env_keys = list(user_settings.envs.keys()) for key, value in user_settings.envs.items(): os.environ[key] = value @@ -630,10 +707,19 @@ def run_daemon() -> None: loop = asyncio.new_event_loop() tasks: set[asyncio.Task[Any]] = set() + reaper = IdleReaper( + timeout_s=( + idle_timeout_s + if idle_timeout_s is not None + else daemon_settings.idle_timeout_minutes * 60.0 + ), + supervised=os.environ.get("COCOINDEX_CODE_DAEMON_SUPERVISED") == "1", + ) + shutdown_reason: str | None = None def _request_shutdown(reason: str) -> None: - """Trigger daemon shutdown — called by StopRequest or signal handler. + """Trigger daemon shutdown — called by StopRequest, signal, or idle reaper. Records *reason* (first one wins) for the graceful-exit marker written during cleanup. @@ -644,6 +730,7 @@ def _request_shutdown(reason: str) -> None: loop.stop() def _spawn_handler(conn: Connection) -> None: + reaper.record_activity() task = loop.create_task( handle_connection( conn, @@ -653,10 +740,37 @@ def _spawn_handler(conn: Connection) -> None: settings_mtime_us, settings_env_keys, handshake_warnings, + reaper, ) ) tasks.add(task) - task.add_done_callback(tasks.discard) + + def _on_done(t: asyncio.Task[Any]) -> None: + tasks.discard(t) + # A finished handler is activity too — a long streaming request + # (e.g. an index run) counts up to its end, not just its start. + reaper.record_activity() + + task.add_done_callback(_on_done) + + async def _idle_reaper_loop() -> None: + """Periodically check the idle predicate; trigger shutdown when it fires.""" + while True: + await asyncio.sleep(idle_check_interval_s) + if reaper.should_exit( + now=time.monotonic(), + active_handlers=len(tasks), + indexing=registry.any_indexing(), + ): + logger.info( + "Idle for %.1fm (timeout %.1fm), shutting down", + reaper.idle_seconds() / 60, + reaper.timeout_s / 60, + ) + _request_shutdown("idle_timeout") + return + + reaper_task = loop.create_task(_idle_reaper_loop()) # Handle signals for graceful shutdown try: @@ -683,20 +797,21 @@ def _accept_loop() -> None: try: loop.run_forever() finally: - # 0. Record the graceful exit (StopRequest, signal). Written first so - # a client that races the rest of the cleanup already sees it. A - # crashed daemon never gets here — the marker's absence is how the - # client detects a crash. + # 0. Record the graceful exit (idle timeout, StopRequest, signal). + # Written first so a client that races the rest of the cleanup + # already sees it. A crashed daemon never gets here — the marker's + # absence is how the client detects a crash. write_last_exit_marker(pid=os.getpid(), reason=shutdown_reason or "unknown") # 1. Stop accepting new connections. listener.close() - # 2. Cancel handler tasks (they may be blocked in run_in_executor). - for task in tasks: + # 2. Cancel handler tasks (they may be blocked in run_in_executor) + # and the idle-reaper task. + pending = [*tasks, reaper_task] + for task in pending: task.cancel() - if tasks: - loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True)) + loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) # 3. Release project resources. registry.close_all() diff --git a/src/cocoindex_code/protocol.py b/src/cocoindex_code/protocol.py index 6bcde27..8c54c3d 100644 --- a/src/cocoindex_code/protocol.py +++ b/src/cocoindex_code/protocol.py @@ -50,6 +50,16 @@ class DaemonEnvRequest(_msgspec.Struct, tag="daemon_env"): pass +class HeartbeatRequest(_msgspec.Struct, tag="heartbeat"): + """Counts as client activity for the daemon's idle timeout. + + Sent periodically by long-lived MCP servers so the daemon does not + idle-exit under an active session. + """ + + pass + + Request = ( HandshakeRequest | IndexRequest @@ -60,6 +70,7 @@ class DaemonEnvRequest(_msgspec.Struct, tag="daemon_env"): | StopRequest | DoctorRequest | DaemonEnvRequest + | HeartbeatRequest ) # --------------------------------------------------------------------------- @@ -144,6 +155,12 @@ class DaemonStatusResponse(_msgspec.Struct, tag="daemon_status"): version: str uptime_seconds: float projects: list[DaemonProjectInfo] + # Idle-timeout observability: seconds since the last client activity, the + # configured timeout (0 = never exit), and seconds since the last MCP + # heartbeat (None = never received one). + idle_seconds: float + idle_timeout_minutes: int + last_heartbeat_seconds: float | None = None class RemoveProjectResponse(_msgspec.Struct, tag="remove_project"): @@ -154,6 +171,10 @@ class StopResponse(_msgspec.Struct, tag="stop"): ok: bool +class HeartbeatResponse(_msgspec.Struct, tag="heartbeat"): + ok: bool + + class DoctorCheckResult(_msgspec.Struct): name: str ok: bool @@ -198,6 +219,7 @@ class ErrorResponse(_msgspec.Struct, tag="error"): | DaemonStatusResponse | RemoveProjectResponse | StopResponse + | HeartbeatResponse | DoctorResponse | DaemonEnvResponse | ErrorResponse diff --git a/src/cocoindex_code/settings.py b/src/cocoindex_code/settings.py index 57d117f..e228839 100644 --- a/src/cocoindex_code/settings.py +++ b/src/cocoindex_code/settings.py @@ -105,10 +105,18 @@ class EmbeddingSettings: query_params: dict[str, Any] | None = None +@dataclass +class DaemonSettings: + # Minutes without client activity before the daemon exits (0 = never exit). + # Clients auto-restart the daemon on the next request, so exiting is cheap. + idle_timeout_minutes: int = 180 + + @dataclass class UserSettings: embedding: EmbeddingSettings envs: dict[str, str] = field(default_factory=dict) + daemon: DaemonSettings = field(default_factory=DaemonSettings) @dataclass @@ -433,6 +441,8 @@ def _user_settings_to_dict(settings: UserSettings) -> dict[str, Any]: d: dict[str, Any] = {"embedding": _embedding_settings_to_dict(settings.embedding)} if settings.envs: d["envs"] = dict(settings.envs) + if settings.daemon != DaemonSettings(): + d["daemon"] = {"idle_timeout_minutes": settings.daemon.idle_timeout_minutes} return d @@ -457,7 +467,13 @@ def _user_settings_from_dict(d: dict[str, Any]) -> UserSettings: emb_kwargs["query_params"] = dict(emb_dict["query_params"] or {}) embedding = EmbeddingSettings(**emb_kwargs) envs = d.get("envs", {}) - return UserSettings(embedding=embedding, envs=envs) + # `daemon:` section is optional — missing keys use the dataclass defaults. + daemon_dict = d.get("daemon") or {} + daemon_kwargs: dict[str, Any] = {} + if "idle_timeout_minutes" in daemon_dict: + daemon_kwargs["idle_timeout_minutes"] = int(daemon_dict["idle_timeout_minutes"]) + daemon = DaemonSettings(**daemon_kwargs) + return UserSettings(embedding=embedding, envs=envs, daemon=daemon) def _project_settings_to_dict(settings: ProjectSettings) -> dict[str, Any]: diff --git a/tests/test_daemon_idle.py b/tests/test_daemon_idle.py new file mode 100644 index 0000000..fce83f0 --- /dev/null +++ b/tests/test_daemon_idle.py @@ -0,0 +1,245 @@ +"""Idle-timeout tests: IdleReaper predicate units + in-process daemon E2E. + +The E2E tests start the daemon in-process (a background thread) with a +seconds-scale idle timeout injected via ``run_daemon(idle_timeout_s=..., +idle_check_interval_s=...)`` and verify it exits — or refuses to exit — at +the right times, and that the socket and PID file are cleaned up. +""" + +from __future__ import annotations + +import os +import tempfile +import threading +import time +from multiprocessing.connection import Client, Connection +from pathlib import Path + +import pytest +from conftest import make_test_user_settings + +import cocoindex_code.daemon as dm +from cocoindex_code._daemon_paths import ( + connection_family, + daemon_pid_path, + read_last_exit_marker, +) +from cocoindex_code._version import __version__ +from cocoindex_code.daemon import IdleReaper +from cocoindex_code.protocol import ( + HandshakeRequest, + IndexProgressUpdate, + IndexRequest, + IndexResponse, + IndexWaitingNotice, + StopRequest, + decode_response, + encode_request, +) +from cocoindex_code.settings import ( + default_project_settings, + save_project_settings, + save_user_settings, +) + +# --------------------------------------------------------------------------- +# IdleReaper predicate (pure, no event loop) +# --------------------------------------------------------------------------- + + +def test_reaper_exits_after_timeout_elapsed() -> None: + reaper = IdleReaper(timeout_s=60.0, supervised=False) + now = reaper.last_activity + 61.0 + assert reaper.should_exit(now=now, active_handlers=0, indexing=False) is True + + +def test_reaper_stays_before_timeout() -> None: + reaper = IdleReaper(timeout_s=60.0, supervised=False) + now = reaper.last_activity + 59.0 + assert reaper.should_exit(now=now, active_handlers=0, indexing=False) is False + + +def test_reaper_stays_with_live_handler() -> None: + reaper = IdleReaper(timeout_s=60.0, supervised=False) + now = reaper.last_activity + 61.0 + assert reaper.should_exit(now=now, active_handlers=1, indexing=False) is False + + +def test_reaper_stays_while_indexing() -> None: + reaper = IdleReaper(timeout_s=60.0, supervised=False) + now = reaper.last_activity + 61.0 + assert reaper.should_exit(now=now, active_handlers=0, indexing=True) is False + + +def test_reaper_timeout_zero_never_exits() -> None: + reaper = IdleReaper(timeout_s=0.0, supervised=False) + now = reaper.last_activity + 1_000_000.0 + assert reaper.should_exit(now=now, active_handlers=0, indexing=False) is False + + +def test_reaper_supervised_never_exits() -> None: + reaper = IdleReaper(timeout_s=60.0, supervised=True) + now = reaper.last_activity + 1_000_000.0 + assert reaper.should_exit(now=now, active_handlers=0, indexing=False) is False + + +def test_reaper_activity_resets_idle_clock() -> None: + reaper = IdleReaper(timeout_s=60.0, supervised=False) + reaper.last_activity -= 120.0 # long idle... + reaper.record_activity() # ...then a connection arrives + now = reaper.last_activity + 1.0 + assert reaper.should_exit(now=now, active_handlers=0, indexing=False) is False + assert reaper.idle_seconds(now) == pytest.approx(1.0) + + +def test_reaper_heartbeat_tracked_separately() -> None: + reaper = IdleReaper(timeout_s=60.0, supervised=False) + assert reaper.last_heartbeat is None + reaper.record_heartbeat() + assert reaper.last_heartbeat is not None + + +# --------------------------------------------------------------------------- +# In-process daemon E2E +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def idle_env(monkeypatch: pytest.MonkeyPatch) -> Path: + """Point COCOINDEX_CODE_DIR at a fresh short temp dir (AF_UNIX path limit).""" + base_dir = Path(tempfile.mkdtemp(prefix="ccc_idle_")) + monkeypatch.setenv("COCOINDEX_CODE_DIR", str(base_dir)) + return base_dir + + +def _start_daemon_thread( + *, idle_timeout_s: float, idle_check_interval_s: float +) -> tuple[threading.Thread, str]: + """Run the daemon in-process and wait for its socket to appear.""" + thread = threading.Thread( + target=lambda: dm.run_daemon( + idle_timeout_s=idle_timeout_s, + idle_check_interval_s=idle_check_interval_s, + ), + daemon=True, + ) + thread.start() + + sock_path = dm.daemon_socket_path() + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + if os.path.exists(sock_path): + return thread, sock_path + if not thread.is_alive(): + raise RuntimeError("Daemon thread exited before its socket appeared") + time.sleep(0.05) + raise TimeoutError("In-process daemon did not start") + + +def _stop_daemon_thread(thread: threading.Thread, sock_path: str) -> None: + """Best-effort graceful stop for tests where the daemon is still running.""" + if not thread.is_alive(): + return + try: + conn = Client(sock_path, family=connection_family()) + conn.send_bytes(encode_request(HandshakeRequest(version=__version__))) + conn.recv_bytes() + conn.send_bytes(encode_request(StopRequest())) + conn.recv_bytes() + conn.close() + except Exception: + pass + thread.join(timeout=10) + + +def _assert_cleaned_up(sock_path: str) -> None: + assert not os.path.exists(sock_path), "socket file not cleaned up" + assert not daemon_pid_path().exists(), "PID file not cleaned up" + + +def _recv_index_stream(conn: Connection) -> IndexResponse: + while True: + resp = decode_response(conn.recv_bytes()) + if isinstance(resp, IndexProgressUpdate | IndexWaitingNotice): + continue + if isinstance(resp, IndexResponse): + return resp + raise AssertionError(f"Unexpected response during indexing: {type(resp).__name__}") + + +def test_daemon_exits_when_idle(idle_env: Path) -> None: + """With no client activity the daemon exits after the timeout, and the + socket + PID file are cleaned up. Runs in no-settings mode (no embedder) + to prove an unconfigured daemon still idle-exits. + """ + thread, sock_path = _start_daemon_thread(idle_timeout_s=1.0, idle_check_interval_s=0.2) + try: + thread.join(timeout=15) + assert not thread.is_alive(), "daemon did not idle-exit" + _assert_cleaned_up(sock_path) + + # The graceful exit left a marker recording the idle timeout as the + # reason (the in-process daemon shares pytest's pid). + marker = read_last_exit_marker() + assert marker is not None + assert marker.reason == "idle_timeout" + assert marker.pid == os.getpid() + finally: + _stop_daemon_thread(thread, sock_path) + + +def test_daemon_does_not_exit_with_live_connection(idle_env: Path) -> None: + """A live handler task (open connection awaiting its request) blocks the + idle exit; once the connection closes the daemon exits on the next timeout. + """ + thread, sock_path = _start_daemon_thread(idle_timeout_s=0.5, idle_check_interval_s=0.1) + try: + conn = Client(sock_path, family=connection_family()) + conn.send_bytes(encode_request(HandshakeRequest(version=__version__))) + conn.recv_bytes() + # Hold the connection open well past the timeout without sending a request. + time.sleep(1.5) + assert thread.is_alive(), "daemon idle-exited despite a live connection" + + conn.close() + thread.join(timeout=15) + assert not thread.is_alive(), "daemon did not exit after the connection closed" + _assert_cleaned_up(sock_path) + finally: + _stop_daemon_thread(thread, sock_path) + + +def test_daemon_does_not_exit_during_index_run(idle_env: Path) -> None: + """An in-flight index run that outlives the idle timeout is not killed: + the stream completes successfully, then the daemon idle-exits afterwards. + """ + save_user_settings(make_test_user_settings()) + project = idle_env / "proj" + project.mkdir() + save_project_settings(project, default_project_settings()) + for i in range(60): + (project / f"module_{i}.py").write_text( + f'"""Module {i}."""\n\ndef func_{i}(x: int) -> int:\n' + f' """Compute something for module {i}."""\n' + f" return x * {i} + {i}\n" + ) + + thread, sock_path = _start_daemon_thread(idle_timeout_s=1.0, idle_check_interval_s=0.2) + try: + conn = Client(sock_path, family=connection_family()) + conn.send_bytes(encode_request(HandshakeRequest(version=__version__))) + conn.recv_bytes() + conn.send_bytes(encode_request(IndexRequest(project_root=str(project)))) + # If the daemon idle-exited mid-run, the stream would break with + # EOFError instead of delivering a successful IndexResponse. + final = _recv_index_stream(conn) + assert final.success is True + conn.close() + + # After the run finishes and the connection closes, the idle clock + # restarts and the daemon exits on the next timeout. + thread.join(timeout=30) + assert not thread.is_alive(), "daemon did not idle-exit after the index run" + _assert_cleaned_up(sock_path) + finally: + _stop_daemon_thread(thread, sock_path) diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 765cdad..c82b035 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -14,6 +14,8 @@ ErrorResponse, HandshakeRequest, HandshakeResponse, + HeartbeatRequest, + HeartbeatResponse, IndexingProgress, IndexProgressUpdate, IndexRequest, @@ -124,6 +126,8 @@ def test_encode_decode_daemon_status_response() -> None: projects=[ DaemonProjectInfo(project_root="/tmp/proj", indexing=False), ], + idle_seconds=12.5, + idle_timeout_minutes=180, ) data = encode_response(resp) decoded = decode_response(data) @@ -133,6 +137,20 @@ def test_encode_decode_daemon_status_response() -> None: assert len(decoded.projects) == 1 assert decoded.projects[0].project_root == "/tmp/proj" assert decoded.projects[0].indexing is False + assert decoded.idle_seconds == 12.5 + assert decoded.idle_timeout_minutes == 180 + assert decoded.last_heartbeat_seconds is None + + +def test_encode_decode_heartbeat_round_trip() -> None: + req = HeartbeatRequest() + decoded_req = decode_request(encode_request(req)) + assert isinstance(decoded_req, HeartbeatRequest) + + resp = HeartbeatResponse(ok=True) + decoded_resp = decode_response(encode_response(resp)) + assert isinstance(decoded_resp, HeartbeatResponse) + assert decoded_resp.ok is True def test_tagged_union_dispatch() -> None: @@ -212,6 +230,7 @@ def test_all_request_types_round_trip() -> None: StopRequest(), DoctorRequest(project_root="/tmp"), DaemonEnvRequest(), + HeartbeatRequest(), ] for req in requests: data = encode_request(req) @@ -300,9 +319,16 @@ def test_all_response_types_round_trip() -> None: IndexWaitingNotice(), SearchResponse(success=True), ProjectStatusResponse(indexing=False, total_chunks=0, total_files=0, languages={}), - DaemonStatusResponse(version="1.0.0", uptime_seconds=0.0, projects=[]), + DaemonStatusResponse( + version="1.0.0", + uptime_seconds=0.0, + projects=[], + idle_seconds=0.0, + idle_timeout_minutes=180, + ), RemoveProjectResponse(ok=True), StopResponse(ok=True), + HeartbeatResponse(ok=True), DoctorResponse( result=DoctorCheckResult(name="test", ok=True, details=[], errors=[]), ), diff --git a/tests/test_settings.py b/tests/test_settings.py index b50bd07..f9be64b 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -14,6 +14,7 @@ DEFAULT_EXCLUDED_PATTERNS, DEFAULT_INCLUDED_PATTERNS, ChunkerMapping, + DaemonSettings, EmbeddingSettings, LanguageOverride, ProjectSettings, @@ -473,6 +474,62 @@ def test_invalid_env_raises(self, monkeypatch: pytest.MonkeyPatch) -> None: get_host_path_mappings() +# --------------------------------------------------------------------------- +# daemon settings (idle timeout) +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("_patch_user_dir") +def test_daemon_settings_absent_section_uses_default(tmp_path: Path) -> None: + path = tmp_path / ".cocoindex_code" / "global_settings.yml" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("embedding:\n provider: litellm\n model: m\n") + loaded = load_user_settings() + assert loaded.daemon.idle_timeout_minutes == 180 + + +@pytest.mark.usefixtures("_patch_user_dir") +def test_daemon_settings_parses_idle_timeout(tmp_path: Path) -> None: + path = tmp_path / ".cocoindex_code" / "global_settings.yml" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + "embedding:\n provider: litellm\n model: m\ndaemon:\n idle_timeout_minutes: 30\n" + ) + loaded = load_user_settings() + assert loaded.daemon.idle_timeout_minutes == 30 + + +@pytest.mark.usefixtures("_patch_user_dir") +def test_daemon_settings_explicit_zero_means_never(tmp_path: Path) -> None: + path = tmp_path / ".cocoindex_code" / "global_settings.yml" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + "embedding:\n provider: litellm\n model: m\ndaemon:\n idle_timeout_minutes: 0\n" + ) + loaded = load_user_settings() + assert loaded.daemon.idle_timeout_minutes == 0 + + +@pytest.mark.usefixtures("_patch_user_dir") +def test_daemon_settings_round_trip() -> None: + settings = UserSettings( + embedding=EmbeddingSettings(provider="litellm", model="m"), + daemon=DaemonSettings(idle_timeout_minutes=45), + ) + save_user_settings(settings) + loaded = load_user_settings() + assert loaded.daemon.idle_timeout_minutes == 45 + + +@pytest.mark.usefixtures("_patch_user_dir") +def test_daemon_settings_default_omitted_from_yaml() -> None: + from cocoindex_code.settings import user_settings_path + + settings = UserSettings(embedding=EmbeddingSettings(provider="litellm", model="m")) + save_user_settings(settings) + assert "daemon" not in user_settings_path().read_text() + + # --------------------------------------------------------------------------- # find_parent_with_marker — global-only should not match # --------------------------------------------------------------------------- From 88c848e63f3368eb2972c769101182d515b06449 Mon Sep 17 00:00:00 2001 From: Jun Zhou Date: Wed, 15 Jul 2026 10:41:56 -0700 Subject: [PATCH 2/7] feat(mcp): heartbeat loop keeps the daemon warm under a live MCP session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both MCP entry points (ccc mcp and the legacy cocoindex-code serve) now run a background heartbeat loop alongside the initial background index. Each heartbeat is one short-lived daemon connection that counts as activity for the idle reaper, so the daemon never idle-exits while an MCP session is live; when the MCP process dies (even SIGKILL) the heartbeats stop and the daemon exits on the normal timeout. - client.send_heartbeat(): uses the raw non-auto-starting connect path and returns False when no compatible daemon answers — a heartbeat must never start or restart a daemon, or `ccc daemon stop` mid-session would fight the loop - server.run_heartbeat_loop(): interval = timeout/3 clamped to [30 s, 300 s], reading the timeout from global_settings.yml (default when missing); a 0 timeout disables the loop entirely Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JaFKwEvJXo7WQhZvrW5Qoe --- src/cocoindex_code/cli.py | 13 +++++-- src/cocoindex_code/client.py | 26 ++++++++++++++ src/cocoindex_code/server.py | 52 +++++++++++++++++++++++++-- tests/test_daemon_idle.py | 68 ++++++++++++++++++++++++++++++++++++ 4 files changed, 154 insertions(+), 5 deletions(-) diff --git a/src/cocoindex_code/cli.py b/src/cocoindex_code/cli.py index 0b0b38c..3bced3b 100644 --- a/src/cocoindex_code/cli.py +++ b/src/cocoindex_code/cli.py @@ -1003,11 +1003,18 @@ def mcp() -> None: project_root = str(require_project_root()) async def _run_mcp() -> None: - from .server import create_mcp_server + from .server import create_mcp_server, run_heartbeat_loop mcp_server = create_mcp_server(project_root) - asyncio.create_task(_bg_index(project_root)) - await mcp_server.run_stdio_async() + background_tasks = { + asyncio.create_task(_bg_index(project_root)), + asyncio.create_task(run_heartbeat_loop()), + } + try: + await mcp_server.run_stdio_async() + finally: + for task in background_tasks: + task.cancel() asyncio.run(_run_mcp()) diff --git a/src/cocoindex_code/client.py b/src/cocoindex_code/client.py index 25f3c28..272eea8 100644 --- a/src/cocoindex_code/client.py +++ b/src/cocoindex_code/client.py @@ -37,6 +37,8 @@ ErrorResponse, HandshakeRequest, HandshakeResponse, + HeartbeatRequest, + HeartbeatResponse, IndexingProgress, IndexProgressUpdate, IndexRequest, @@ -414,6 +416,30 @@ def stop() -> StopResponse: return _send(StopRequest()) # type: ignore[return-value] +def send_heartbeat() -> bool: + """Send one heartbeat so a running daemon counts an active MCP session as activity. + + Uses the raw (non-auto-starting) connect path and returns False when no + compatible daemon answers: a heartbeat keeps an existing daemon warm but + must NEVER start or restart one — otherwise `ccc daemon stop` during an + MCP session would fight the heartbeat loop. The daemon still auto-starts + lazily on the next real search/index request. + """ + try: + conn, _resp = _raw_connect_and_handshake() + except (ConnectionRefusedError, DaemonVersionError, OSError): + return False + try: + conn.send_bytes(encode_request(HeartbeatRequest())) + data = conn.recv_bytes() + except (EOFError, OSError): + return False + finally: + conn.close() + resp = decode_response(data) + return isinstance(resp, HeartbeatResponse) and resp.ok + + def daemon_env() -> DaemonEnvResponse: """Get environment variable names from the daemon.""" return _send(DaemonEnvRequest()) # type: ignore[return-value] diff --git a/src/cocoindex_code/server.py b/src/cocoindex_code/server.py index 2708c86..8b1ffaf 100644 --- a/src/cocoindex_code/server.py +++ b/src/cocoindex_code/server.py @@ -17,6 +17,8 @@ from mcp.server.fastmcp import FastMCP from pydantic import BaseModel, Field +from .settings import DaemonSettings, load_user_settings + _MCP_INSTRUCTIONS = ( "Code search and codebase understanding tools." "\n" @@ -159,6 +161,45 @@ async def search( return mcp +# === Daemon heartbeat loop === + + +def heartbeat_interval_s(idle_timeout_minutes: int) -> int: + """Heartbeat period for a given idle timeout: a third of the timeout, + clamped to [30 s, 300 s] so several heartbeats always fit in one idle + window without hammering the daemon on very long timeouts. + """ + return max(30, min(300, idle_timeout_minutes * 60 // 3)) + + +async def run_heartbeat_loop() -> None: + """Periodically heartbeat the daemon so it never idle-exits under this + live MCP session; when this process dies (even SIGKILL) the heartbeats + stop and the daemon exits on its normal idle timeout. + + Reads the timeout from the same ``global_settings.yml`` the daemon reads + (dataclass default when the file is missing or invalid). Returns + immediately when the timeout is 0 (daemon never idle-exits). The + heartbeat itself never starts or restarts a daemon — see + ``client.send_heartbeat``. + """ + from .client import send_heartbeat + + try: + timeout_minutes = load_user_settings().daemon.idle_timeout_minutes + except (FileNotFoundError, ValueError): + timeout_minutes = DaemonSettings().idle_timeout_minutes + if timeout_minutes <= 0: + return + + interval = heartbeat_interval_s(timeout_minutes) + loop = asyncio.get_event_loop() + while True: + await asyncio.sleep(interval) + # Client I/O is blocking — run off the event loop thread. + await loop.run_in_executor(None, send_heartbeat) + + # Keep the old `mcp` global for backward compatibility in __init__.py mcp: FastMCP | None = None @@ -326,7 +367,14 @@ def _on_progress(progress: IndexingProgress) -> None: async def _serve() -> None: from .cli import _bg_index - asyncio.create_task(_bg_index(str(project_root))) - await mcp_server.run_stdio_async() + background_tasks = { + asyncio.create_task(_bg_index(str(project_root))), + asyncio.create_task(run_heartbeat_loop()), + } + try: + await mcp_server.run_stdio_async() + finally: + for task in background_tasks: + task.cancel() asyncio.run(_serve()) diff --git a/tests/test_daemon_idle.py b/tests/test_daemon_idle.py index fce83f0..ce2ea76 100644 --- a/tests/test_daemon_idle.py +++ b/tests/test_daemon_idle.py @@ -27,6 +27,8 @@ from cocoindex_code._version import __version__ from cocoindex_code.daemon import IdleReaper from cocoindex_code.protocol import ( + DaemonStatusRequest, + DaemonStatusResponse, HandshakeRequest, IndexProgressUpdate, IndexRequest, @@ -99,6 +101,19 @@ def test_reaper_heartbeat_tracked_separately() -> None: assert reaper.last_heartbeat is not None +# --------------------------------------------------------------------------- +# MCP heartbeat interval +# --------------------------------------------------------------------------- + + +def test_heartbeat_interval_is_a_third_of_the_timeout_clamped() -> None: + from cocoindex_code.server import heartbeat_interval_s + + assert heartbeat_interval_s(1) == 30 # 20 s → clamped up to 30 s + assert heartbeat_interval_s(10) == 200 # 600 s / 3 + assert heartbeat_interval_s(180) == 300 # 3600 s → clamped down to 300 s + + # --------------------------------------------------------------------------- # In-process daemon E2E # --------------------------------------------------------------------------- @@ -209,6 +224,59 @@ def test_daemon_does_not_exit_with_live_connection(idle_env: Path) -> None: _stop_daemon_thread(thread, sock_path) +def test_daemon_does_not_exit_while_heartbeats_arrive(idle_env: Path) -> None: + """Periodic heartbeats keep the daemon alive past its idle timeout; once + they stop, it exits on the normal timeout. Mirrors a live MCP session + dying (heartbeats stop) without any explicit shutdown. + """ + from cocoindex_code.client import send_heartbeat + + thread, sock_path = _start_daemon_thread(idle_timeout_s=1.0, idle_check_interval_s=0.2) + try: + # Heartbeat every 0.4 s for ~3 s — several idle timeouts' worth. + deadline = time.monotonic() + 3.0 + while time.monotonic() < deadline: + assert send_heartbeat() is True + time.sleep(0.4) + assert thread.is_alive(), "daemon idle-exited despite heartbeats" + + # The status endpoint reports the heartbeat age. + conn = Client(sock_path, family=connection_family()) + conn.send_bytes(encode_request(HandshakeRequest(version=__version__))) + conn.recv_bytes() + conn.send_bytes(encode_request(DaemonStatusRequest())) + status = decode_response(conn.recv_bytes()) + conn.close() + assert isinstance(status, DaemonStatusResponse) + assert status.last_heartbeat_seconds is not None + assert status.last_heartbeat_seconds < 2.0 + + # Heartbeats stopped — the daemon exits on the next idle timeout. + thread.join(timeout=15) + assert not thread.is_alive(), "daemon did not exit after heartbeats stopped" + _assert_cleaned_up(sock_path) + finally: + _stop_daemon_thread(thread, sock_path) + + +def test_send_heartbeat_does_not_spawn_a_daemon(idle_env: Path) -> None: + """Against a stopped daemon, send_heartbeat returns False and never starts + one — otherwise `ccc daemon stop` during an MCP session would fight the + heartbeat loop. + """ + from cocoindex_code.client import send_heartbeat + + sock_path = dm.daemon_socket_path() + assert not os.path.exists(sock_path) + + assert send_heartbeat() is False + + # Give a hypothetical spawned daemon time to appear, then confirm nothing did. + time.sleep(1.0) + assert not os.path.exists(sock_path) + assert not daemon_pid_path().exists() + + def test_daemon_does_not_exit_during_index_run(idle_env: Path) -> None: """An in-flight index run that outlives the idle timeout is not killed: the stream completes successfully, then the daemon idle-exits afterwards. From 85a52b540db6af11217fa3d175a559364ba6e80a Mon Sep 17 00:00:00 2001 From: Jun Zhou Date: Wed, 15 Jul 2026 10:44:56 -0700 Subject: [PATCH 3/7] feat(cli): show idle timeout and heartbeat age in daemon status; document idle-exit - ccc daemon status now prints idle duration with the configured timeout (or "disabled" when 0) and the age of the last MCP heartbeat (or "never") - README: document daemon.idle_timeout_minutes in the global settings reference - CLAUDE.md: note the idle-exit + heartbeat + crash-marker behavior in the process model section Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JaFKwEvJXo7WQhZvrW5Qoe --- CLAUDE.md | 2 ++ README.md | 5 +++++ src/cocoindex_code/cli.py | 6 ++++++ 3 files changed, 13 insertions(+) diff --git a/CLAUDE.md b/CLAUDE.md index e902766..be1572f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -31,6 +31,8 @@ The tool runs as a **long-lived background daemon** (`daemon.py`) that accepts c The daemon is auto-started by `client.py` when not running. It is restarted automatically when the `global_settings.yml` mtime changes (version bump or settings edit), detected via the `HandshakeResponse.global_settings_mtime_us` field. +The daemon idle-exits after `daemon.idle_timeout_minutes` (default 180, 0 = never) without client activity, unless a live MCP session is sending heartbeats; clients transparently restart it on the next request. Graceful exits leave a `last_exit` marker file so the client can tell an expected exit from a crash (which triggers a loud restart, capped after consecutive crashes). + ### Key Layers **`protocol.py`** — All IPC message types as `msgspec.Struct` tagged unions, serialized as msgpack. Every request type, response type, and streaming wrapper lives here. The `Request` and `Response` type aliases are the union types used by the decoder. diff --git a/README.md b/README.md index c0fbf14..4b7c5de 100644 --- a/README.md +++ b/README.md @@ -515,10 +515,15 @@ embedding: envs: # extra environment variables for the daemon OPENAI_API_KEY: your-key # only needed if not already in your shell environment + +daemon: + idle_timeout_minutes: 180 # optional: exit the daemon after this long without client activity (default 180, 0 = never) ``` > **Note:** The daemon inherits your shell environment. If an API key (e.g. `OPENAI_API_KEY`) is already set as an environment variable, you don't need to duplicate it in `envs`. The `envs` field is only for values that aren't in your environment. +> **Idle timeout:** the background daemon holds the embedding model in RAM, so it exits after `daemon.idle_timeout_minutes` without client activity and is restarted automatically on your next `ccc` command or MCP search. A live MCP session sends periodic heartbeats, so the daemon never idles out while your coding agent is connected. Set `0` to keep the daemon running forever. + > **Custom location:** set `COCOINDEX_CODE_DIR` to place `global_settings.yml` somewhere other than `~/.cocoindex_code/` — useful if you want the file to live alongside your projects (e.g. on a synced folder). #### `indexing_params` / `query_params` diff --git a/src/cocoindex_code/cli.py b/src/cocoindex_code/cli.py index 3bced3b..bf2e0f6 100644 --- a/src/cocoindex_code/cli.py +++ b/src/cocoindex_code/cli.py @@ -1044,6 +1044,12 @@ def daemon_status() -> None: resp = _client.daemon_status() _typer.echo(f"Daemon version: {resp.version}") _typer.echo(f"Uptime: {resp.uptime_seconds:.1f}s") + timeout_desc = f"{resp.idle_timeout_minutes}m" if resp.idle_timeout_minutes > 0 else "disabled" + _typer.echo(f"Idle: {resp.idle_seconds:.1f}s (timeout: {timeout_desc})") + if resp.last_heartbeat_seconds is None: + _typer.echo("Last heartbeat: never") + else: + _typer.echo(f"Last heartbeat: {resp.last_heartbeat_seconds:.1f}s ago") if resp.projects: _typer.echo("Projects:") for p in resp.projects: From 6bf123f8e61b66972a4834ccd94047986077b101 Mon Sep 17 00:00:00 2001 From: Jun Zhou Date: Fri, 17 Jul 2026 10:50:50 -0700 Subject: [PATCH 4/7] fix(daemon): release Windows named pipe on shutdown by unblocking the accept thread PipeListener.close() only closes the queued next-instance handle; it cannot cancel an in-flight ConnectNamedPipe wait, so a blocked accept() kept a pipe instance open after the daemon exited. The named pipe therefore still existed when the in-process idle tests asserted cleanup, and the test's own os.path.exists() probe completed the pending connection, making the accept thread call call_soon_threadsafe on a closed loop (the PytestUnhandledThreadExceptionWarning noise in CI). Shutdown now sets a shutting_down flag, wakes a blocked accept() on Windows with a dummy client connection, and joins the accept thread before returning, so the last pipe-instance handle is closed before run_daemon exits. POSIX behavior is unchanged (listener.close() already unblocks accept() there). Co-Authored-By: Claude Fable 5 --- src/cocoindex_code/daemon.py | 34 +++++++++++++++++++++++++++++----- 1 file changed, 29 insertions(+), 5 deletions(-) diff --git a/src/cocoindex_code/daemon.py b/src/cocoindex_code/daemon.py index 068947b..c2ad8aa 100644 --- a/src/cocoindex_code/daemon.py +++ b/src/cocoindex_code/daemon.py @@ -12,7 +12,7 @@ import time import traceback from collections.abc import AsyncIterator, Callable -from multiprocessing.connection import Connection, Listener +from multiprocessing.connection import Client, Connection, Listener from pathlib import Path from typing import Any @@ -780,15 +780,28 @@ async def _idle_reaper_loop() -> None: pass # Not in main thread, or not supported on this platform (e.g. Windows) # Accept loop runs in a background thread; new connections are dispatched - # to the event loop via call_soon_threadsafe. The loop exits when - # listener.close() (called during shutdown) causes accept() to raise. + # to the event loop via call_soon_threadsafe. On POSIX the loop exits when + # listener.close() (called during shutdown) causes accept() to raise. On + # Windows, PipeListener.close() does not cancel an in-flight + # ConnectNamedPipe wait, so shutdown additionally sets `shutting_down` and + # wakes accept() with a dummy client connection; the accepted connection + # (dummy or real) is dropped and the thread exits. + shutting_down = threading.Event() + def _accept_loop() -> None: while True: try: conn = listener.accept() - loop.call_soon_threadsafe(_spawn_handler, conn) except OSError: break + if shutting_down.is_set(): + conn.close() + break + try: + loop.call_soon_threadsafe(_spawn_handler, conn) + except RuntimeError: # loop already closed — shutdown race + conn.close() + break accept_thread = threading.Thread(target=_accept_loop, daemon=True) accept_thread.start() @@ -803,8 +816,19 @@ def _accept_loop() -> None: # absence is how the client detects a crash. write_last_exit_marker(pid=os.getpid(), reason=shutdown_reason or "unknown") - # 1. Stop accepting new connections. + # 1. Stop accepting new connections. On Windows a blocked accept() + # holds an open pipe instance that listener.close() can't release + # (it only closes the queued next-instance handle), which would keep + # the named pipe alive after exit — wake it with a dummy connection + # so the accept thread closes it and exits. + shutting_down.set() + if sys.platform == "win32": + try: + Client(sock_path, family=connection_family()).close() + except OSError: + pass listener.close() + accept_thread.join(timeout=5) # 2. Cancel handler tasks (they may be blocked in run_in_executor) # and the idle-reaper task. From 48c981483ad5964137590c6c99831be051c668d8 Mon Sep 17 00:00:00 2001 From: Jun Zhou Date: Fri, 17 Jul 2026 12:51:42 -0700 Subject: [PATCH 5/7] refactor(daemon): type the idle timeout as timedelta Review feedback on #215: a bare float-with-suffix-naming carries the unit only by convention. IdleReaper and run_daemon now take timedelta, converting to seconds only at the comparison against monotonic timestamps and on the wire. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01EQpbhCAsYRJuwL7X83fzDu --- src/cocoindex_code/daemon.py | 29 +++++++++++++++-------------- tests/test_daemon_idle.py | 25 +++++++++++++------------ 2 files changed, 28 insertions(+), 26 deletions(-) diff --git a/src/cocoindex_code/daemon.py b/src/cocoindex_code/daemon.py index c2ad8aa..e3579e5 100644 --- a/src/cocoindex_code/daemon.py +++ b/src/cocoindex_code/daemon.py @@ -12,6 +12,7 @@ import time import traceback from collections.abc import AsyncIterator, Callable +from datetime import timedelta from multiprocessing.connection import Client, Connection, Listener from pathlib import Path from typing import Any @@ -220,8 +221,8 @@ class IdleReaper: other connection. """ - def __init__(self, timeout_s: float, *, supervised: bool) -> None: - self.timeout_s = timeout_s + def __init__(self, timeout: timedelta, *, supervised: bool) -> None: + self.timeout = timeout self.supervised = supervised self.last_activity = time.monotonic() self.last_heartbeat: float | None = None @@ -242,13 +243,13 @@ def should_exit(self, *, now: float, active_handlers: int, indexing: bool) -> bo the daemon lifecycle, no handler task is live, no project is indexing, and the idle period exceeds the timeout. """ - if self.timeout_s <= 0: + if self.timeout <= timedelta(0): return False if self.supervised: return False if active_handlers > 0 or indexing: return False - return now - self.last_activity > self.timeout_s + return now - self.last_activity > self.timeout.total_seconds() # --------------------------------------------------------------------------- @@ -568,7 +569,7 @@ async def _dispatch( uptime_seconds=now - start_time, projects=registry.list_projects(), idle_seconds=reaper.idle_seconds(now), - idle_timeout_minutes=round(reaper.timeout_s / 60), + idle_timeout_minutes=round(reaper.timeout / timedelta(minutes=1)), last_heartbeat_seconds=( None if reaper.last_heartbeat is None else now - reaper.last_heartbeat ), @@ -619,8 +620,8 @@ async def _dispatch( def run_daemon( *, - idle_timeout_s: float | None = None, - idle_check_interval_s: float = 60.0, + idle_timeout: timedelta | None = None, + idle_check_interval: timedelta = timedelta(minutes=1), ) -> None: """Main entry point for the daemon process (blocking). @@ -628,7 +629,7 @@ def run_daemon( to serve connections, and performs cleanup when shutdown is requested via ``StopRequest``, a signal (SIGTERM / SIGINT), or the idle timeout. - ``idle_timeout_s`` and ``idle_check_interval_s`` exist so tests can run a + ``idle_timeout`` and ``idle_check_interval`` exist so tests can run a seconds-scale idle timeout in-process; production leaves them at their defaults and the timeout comes from ``daemon.idle_timeout_minutes`` in ``global_settings.yml`` (the dataclass default when the file is missing). @@ -708,10 +709,10 @@ def run_daemon( tasks: set[asyncio.Task[Any]] = set() reaper = IdleReaper( - timeout_s=( - idle_timeout_s - if idle_timeout_s is not None - else daemon_settings.idle_timeout_minutes * 60.0 + timeout=( + idle_timeout + if idle_timeout is not None + else timedelta(minutes=daemon_settings.idle_timeout_minutes) ), supervised=os.environ.get("COCOINDEX_CODE_DAEMON_SUPERVISED") == "1", ) @@ -756,7 +757,7 @@ def _on_done(t: asyncio.Task[Any]) -> None: async def _idle_reaper_loop() -> None: """Periodically check the idle predicate; trigger shutdown when it fires.""" while True: - await asyncio.sleep(idle_check_interval_s) + await asyncio.sleep(idle_check_interval.total_seconds()) if reaper.should_exit( now=time.monotonic(), active_handlers=len(tasks), @@ -765,7 +766,7 @@ async def _idle_reaper_loop() -> None: logger.info( "Idle for %.1fm (timeout %.1fm), shutting down", reaper.idle_seconds() / 60, - reaper.timeout_s / 60, + reaper.timeout / timedelta(minutes=1), ) _request_shutdown("idle_timeout") return diff --git a/tests/test_daemon_idle.py b/tests/test_daemon_idle.py index ce2ea76..445e6d5 100644 --- a/tests/test_daemon_idle.py +++ b/tests/test_daemon_idle.py @@ -1,8 +1,8 @@ """Idle-timeout tests: IdleReaper predicate units + in-process daemon E2E. The E2E tests start the daemon in-process (a background thread) with a -seconds-scale idle timeout injected via ``run_daemon(idle_timeout_s=..., -idle_check_interval_s=...)`` and verify it exits — or refuses to exit — at +seconds-scale idle timeout injected via ``run_daemon(idle_timeout=..., +idle_check_interval=...)`` and verify it exits — or refuses to exit — at the right times, and that the socket and PID file are cleaned up. """ @@ -12,6 +12,7 @@ import tempfile import threading import time +from datetime import timedelta from multiprocessing.connection import Client, Connection from pathlib import Path @@ -50,43 +51,43 @@ def test_reaper_exits_after_timeout_elapsed() -> None: - reaper = IdleReaper(timeout_s=60.0, supervised=False) + reaper = IdleReaper(timeout=timedelta(seconds=60), supervised=False) now = reaper.last_activity + 61.0 assert reaper.should_exit(now=now, active_handlers=0, indexing=False) is True def test_reaper_stays_before_timeout() -> None: - reaper = IdleReaper(timeout_s=60.0, supervised=False) + reaper = IdleReaper(timeout=timedelta(seconds=60), supervised=False) now = reaper.last_activity + 59.0 assert reaper.should_exit(now=now, active_handlers=0, indexing=False) is False def test_reaper_stays_with_live_handler() -> None: - reaper = IdleReaper(timeout_s=60.0, supervised=False) + reaper = IdleReaper(timeout=timedelta(seconds=60), supervised=False) now = reaper.last_activity + 61.0 assert reaper.should_exit(now=now, active_handlers=1, indexing=False) is False def test_reaper_stays_while_indexing() -> None: - reaper = IdleReaper(timeout_s=60.0, supervised=False) + reaper = IdleReaper(timeout=timedelta(seconds=60), supervised=False) now = reaper.last_activity + 61.0 assert reaper.should_exit(now=now, active_handlers=0, indexing=True) is False def test_reaper_timeout_zero_never_exits() -> None: - reaper = IdleReaper(timeout_s=0.0, supervised=False) + reaper = IdleReaper(timeout=timedelta(0), supervised=False) now = reaper.last_activity + 1_000_000.0 assert reaper.should_exit(now=now, active_handlers=0, indexing=False) is False def test_reaper_supervised_never_exits() -> None: - reaper = IdleReaper(timeout_s=60.0, supervised=True) + reaper = IdleReaper(timeout=timedelta(seconds=60), supervised=True) now = reaper.last_activity + 1_000_000.0 assert reaper.should_exit(now=now, active_handlers=0, indexing=False) is False def test_reaper_activity_resets_idle_clock() -> None: - reaper = IdleReaper(timeout_s=60.0, supervised=False) + reaper = IdleReaper(timeout=timedelta(seconds=60), supervised=False) reaper.last_activity -= 120.0 # long idle... reaper.record_activity() # ...then a connection arrives now = reaper.last_activity + 1.0 @@ -95,7 +96,7 @@ def test_reaper_activity_resets_idle_clock() -> None: def test_reaper_heartbeat_tracked_separately() -> None: - reaper = IdleReaper(timeout_s=60.0, supervised=False) + reaper = IdleReaper(timeout=timedelta(seconds=60), supervised=False) assert reaper.last_heartbeat is None reaper.record_heartbeat() assert reaper.last_heartbeat is not None @@ -133,8 +134,8 @@ def _start_daemon_thread( """Run the daemon in-process and wait for its socket to appear.""" thread = threading.Thread( target=lambda: dm.run_daemon( - idle_timeout_s=idle_timeout_s, - idle_check_interval_s=idle_check_interval_s, + idle_timeout=timedelta(seconds=idle_timeout_s), + idle_check_interval=timedelta(seconds=idle_check_interval_s), ), daemon=True, ) From 3ce7785c1d4dcf38ad13b669cd5bff51a59e45cc Mon Sep 17 00:00:00 2001 From: Jun Zhou Date: Fri, 17 Jul 2026 12:54:55 -0700 Subject: [PATCH 6/7] =?UTF-8?q?refactor(daemon):=20drop=20last=5Fheartbeat?= =?UTF-8?q?=20tracking=20=E2=80=94=20heartbeats=20are=20just=20activity?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback on #215: last_heartbeat is MCP-specific state that the daemon has no use for — it played no part in idle reaping and existed only so 'ccc daemon status' could echo it back. A heartbeat's real effect is the connection itself, which the accept path already records as activity. Remove the field from IdleReaper, the wire protocol, and the status output. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01EQpbhCAsYRJuwL7X83fzDu --- src/cocoindex_code/cli.py | 4 ---- src/cocoindex_code/daemon.py | 15 +++------------ src/cocoindex_code/protocol.py | 6 ++---- tests/test_daemon_idle.py | 20 -------------------- tests/test_protocol.py | 1 - 5 files changed, 5 insertions(+), 41 deletions(-) diff --git a/src/cocoindex_code/cli.py b/src/cocoindex_code/cli.py index bf2e0f6..0621948 100644 --- a/src/cocoindex_code/cli.py +++ b/src/cocoindex_code/cli.py @@ -1046,10 +1046,6 @@ def daemon_status() -> None: _typer.echo(f"Uptime: {resp.uptime_seconds:.1f}s") timeout_desc = f"{resp.idle_timeout_minutes}m" if resp.idle_timeout_minutes > 0 else "disabled" _typer.echo(f"Idle: {resp.idle_seconds:.1f}s (timeout: {timeout_desc})") - if resp.last_heartbeat_seconds is None: - _typer.echo("Last heartbeat: never") - else: - _typer.echo(f"Last heartbeat: {resp.last_heartbeat_seconds:.1f}s ago") if resp.projects: _typer.echo("Projects:") for p in resp.projects: diff --git a/src/cocoindex_code/daemon.py b/src/cocoindex_code/daemon.py index e3579e5..64023e1 100644 --- a/src/cocoindex_code/daemon.py +++ b/src/cocoindex_code/daemon.py @@ -215,24 +215,17 @@ class IdleReaper: ``last_activity`` is refreshed on every accepted connection and again when each handler task finishes, so a long streaming request (e.g. an index run) - counts as activity up to its end. ``last_heartbeat`` additionally records - when the most recent ``HeartbeatRequest`` was dispatched — for status - reporting only; heartbeat connections refresh ``last_activity`` like any - other connection. + counts as activity up to its end. """ def __init__(self, timeout: timedelta, *, supervised: bool) -> None: self.timeout = timeout self.supervised = supervised self.last_activity = time.monotonic() - self.last_heartbeat: float | None = None def record_activity(self) -> None: self.last_activity = time.monotonic() - def record_heartbeat(self) -> None: - self.last_heartbeat = time.monotonic() - def idle_seconds(self, now: float | None = None) -> float: return (time.monotonic() if now is None else now) - self.last_activity @@ -570,9 +563,6 @@ async def _dispatch( projects=registry.list_projects(), idle_seconds=reaper.idle_seconds(now), idle_timeout_minutes=round(reaper.timeout / timedelta(minutes=1)), - last_heartbeat_seconds=( - None if reaper.last_heartbeat is None else now - reaper.last_heartbeat - ), ) if isinstance(req, RemoveProjectRequest): @@ -584,7 +574,8 @@ async def _dispatch( return StopResponse(ok=True) if isinstance(req, HeartbeatRequest): - reaper.record_heartbeat() + # A heartbeat's only effect is the connection itself, which the + # accept path already recorded as activity. return HeartbeatResponse(ok=True) if isinstance(req, DaemonEnvRequest): diff --git a/src/cocoindex_code/protocol.py b/src/cocoindex_code/protocol.py index 8c54c3d..164dedf 100644 --- a/src/cocoindex_code/protocol.py +++ b/src/cocoindex_code/protocol.py @@ -155,12 +155,10 @@ class DaemonStatusResponse(_msgspec.Struct, tag="daemon_status"): version: str uptime_seconds: float projects: list[DaemonProjectInfo] - # Idle-timeout observability: seconds since the last client activity, the - # configured timeout (0 = never exit), and seconds since the last MCP - # heartbeat (None = never received one). + # Idle-timeout observability: seconds since the last client activity and + # the configured timeout (0 = never exit). idle_seconds: float idle_timeout_minutes: int - last_heartbeat_seconds: float | None = None class RemoveProjectResponse(_msgspec.Struct, tag="remove_project"): diff --git a/tests/test_daemon_idle.py b/tests/test_daemon_idle.py index 445e6d5..e7c4f25 100644 --- a/tests/test_daemon_idle.py +++ b/tests/test_daemon_idle.py @@ -28,8 +28,6 @@ from cocoindex_code._version import __version__ from cocoindex_code.daemon import IdleReaper from cocoindex_code.protocol import ( - DaemonStatusRequest, - DaemonStatusResponse, HandshakeRequest, IndexProgressUpdate, IndexRequest, @@ -95,13 +93,6 @@ def test_reaper_activity_resets_idle_clock() -> None: assert reaper.idle_seconds(now) == pytest.approx(1.0) -def test_reaper_heartbeat_tracked_separately() -> None: - reaper = IdleReaper(timeout=timedelta(seconds=60), supervised=False) - assert reaper.last_heartbeat is None - reaper.record_heartbeat() - assert reaper.last_heartbeat is not None - - # --------------------------------------------------------------------------- # MCP heartbeat interval # --------------------------------------------------------------------------- @@ -241,17 +232,6 @@ def test_daemon_does_not_exit_while_heartbeats_arrive(idle_env: Path) -> None: time.sleep(0.4) assert thread.is_alive(), "daemon idle-exited despite heartbeats" - # The status endpoint reports the heartbeat age. - conn = Client(sock_path, family=connection_family()) - conn.send_bytes(encode_request(HandshakeRequest(version=__version__))) - conn.recv_bytes() - conn.send_bytes(encode_request(DaemonStatusRequest())) - status = decode_response(conn.recv_bytes()) - conn.close() - assert isinstance(status, DaemonStatusResponse) - assert status.last_heartbeat_seconds is not None - assert status.last_heartbeat_seconds < 2.0 - # Heartbeats stopped — the daemon exits on the next idle timeout. thread.join(timeout=15) assert not thread.is_alive(), "daemon did not exit after heartbeats stopped" diff --git a/tests/test_protocol.py b/tests/test_protocol.py index c82b035..f026764 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -139,7 +139,6 @@ def test_encode_decode_daemon_status_response() -> None: assert decoded.projects[0].indexing is False assert decoded.idle_seconds == 12.5 assert decoded.idle_timeout_minutes == 180 - assert decoded.last_heartbeat_seconds is None def test_encode_decode_heartbeat_round_trip() -> None: From 2f58c30c735559429ea9e88c93d2de7e902383c3 Mon Sep 17 00:00:00 2001 From: Jun Zhou Date: Fri, 17 Jul 2026 19:12:00 -0700 Subject: [PATCH 7/7] refactor(daemon): send the shutdown wake-up connection unconditionally POSIX doesn't need it (listener.close() makes accept() raise), but it's harmless there and keeps the path exercised on every platform instead of only on Windows CI. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Vgj6XtKW6dHST75F5VtavC --- src/cocoindex_code/daemon.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/src/cocoindex_code/daemon.py b/src/cocoindex_code/daemon.py index 64023e1..d2b38ab 100644 --- a/src/cocoindex_code/daemon.py +++ b/src/cocoindex_code/daemon.py @@ -812,13 +812,15 @@ def _accept_loop() -> None: # holds an open pipe instance that listener.close() can't release # (it only closes the queued next-instance handle), which would keep # the named pipe alive after exit — wake it with a dummy connection - # so the accept thread closes it and exits. + # so the accept thread closes it and exits. POSIX doesn't need the + # wake-up (listener.close() makes accept() raise), but it's harmless + # there and keeps this path exercised on every platform rather than + # only on Windows CI. shutting_down.set() - if sys.platform == "win32": - try: - Client(sock_path, family=connection_family()).close() - except OSError: - pass + try: + Client(sock_path, family=connection_family()).close() + except OSError: + pass listener.close() accept_thread.join(timeout=5)