diff --git a/hindsight-embed/hindsight_embed/cli.py b/hindsight-embed/hindsight_embed/cli.py index 3724f19775..17b6090cc2 100644 --- a/hindsight-embed/hindsight_embed/cli.py +++ b/hindsight-embed/hindsight_embed/cli.py @@ -785,7 +785,8 @@ def do_ui(args, config: dict, logger): status_text = Text() status_text.append("UI is running\n\n", style="green bold") status_text.append(" URL: ", style="dim") - status_text.append(f"http://127.0.0.1:{effective_port}\n", style="cyan") + # localhost, not a loopback literal: the UI can be bound to ::1 only. + status_text.append(f"http://localhost:{effective_port}\n", style="cyan") status_text.append(" Logs: ", style="dim") status_text.append(f"{paths.ui_log}", style="") diff --git a/hindsight-embed/hindsight_embed/daemon_embed_manager.py b/hindsight-embed/hindsight_embed/daemon_embed_manager.py index 65397fde00..f7ec94e925 100644 --- a/hindsight-embed/hindsight_embed/daemon_embed_manager.py +++ b/hindsight-embed/hindsight_embed/daemon_embed_manager.py @@ -26,7 +26,7 @@ from rich.text import Text from .embed_manager import EmbedManager -from .profile_manager import ProfileManager, lock_file, unlock_file +from .profile_manager import ProfileLockTimeout, ProfileManager, lock_file, unlock_file logger = logging.getLogger(__name__) console = Console(stderr=True) @@ -87,6 +87,50 @@ def _parse_non_negative_int(value: str | None, default: int, name: str) -> int: _parse_float_env("HINDSIGHT_EMBED_PORT_HEALTH_CHECK_INTERVAL", 0.5), 0.5, ) +# Two probe budgets, split by what a wrong answer costs. +# +# The daemon serves /health on the same asyncio event loop that runs LLM calls +# and embeddings, so a slow provider call can stall the response for tens of +# seconds. Where a false negative makes us *kill* the listener — the reclaim +# decision in _port_health_ok — 2s was not enough to tell "busy" from "dead" +# (issue #3099), so that probe waits as long as the worker-side liveness +# threshold does. +HEALTH_PROBE_TIMEOUT = _safe_positive_float( + _parse_float_env("HINDSIGHT_EMBED_HEALTH_PROBE_TIMEOUT", 10.0), + 10.0, +) +# Everywhere else the question is only "is it up?", asked on hot paths (every +# _ensure_started, profile delete, CLI status) and often on ports where nothing +# is listening at all. A false negative there is cheap — the caller re-runs +# ensure_running, which consults the long probe above before touching anything — +# so these stay short. They were briefly raised to HEALTH_PROBE_TIMEOUT, which +# pushed the control center's delete handler (one daemon probe + one UI probe +# per loopback family) past the 5s default client timeout on Windows. +LIVENESS_PROBE_TIMEOUT = 2.0 +# Connecting is not the slow part for a busy daemon — being answered is. Cap +# connect separately so an unreachable address cannot spend the whole read +# budget, which is what turns two loopback families into double the wait. +PROBE_CONNECT_TIMEOUT = 1.0 + +# Both loopback families are probed: Next.js binds ::1 only when started with +# `--hostname localhost`, so an IPv4-only health check reported a perfectly +# healthy control plane as down (issue #3527). +LOOPBACK_HOSTS = ("127.0.0.1", "::1") + +# Before signalling a PID we require its command line to identify it as one of +# ours. Selecting the victim purely by "who holds the port" killed unrelated +# services that happened to be on the same port (issue #3520). +DAEMON_PROCESS_MARKERS = ("hindsight-api", "hindsight_api") +# The control plane runs as `npx @vectorize-io/hindsight-control-plane` (or +# `node .../hindsight-control-plane/bin/cli.js` in the monorepo), but Next.js +# rewrites argv to "next-server (vX.Y.Z)" once it is serving, so the package +# name is not always still visible in the command line. +UI_PROCESS_MARKERS = ("hindsight-control-plane", "next-server", "next start") + + +def _probe_timeout(read: float) -> httpx.Timeout: + """Timeout with a short connect and a caller-chosen read budget.""" + return httpx.Timeout(read, connect=min(read, PROBE_CONNECT_TIMEOUT)) def _detach_popen_kwargs(log_handle: IO[bytes]) -> dict: @@ -192,10 +236,15 @@ def get_url(self, profile: str) -> str: return f"http://127.0.0.1:{paths.port}" def is_running(self, profile: str) -> bool: - """Check if daemon is running and responsive.""" + """Check if daemon is running and responsive. + + Uses the short liveness budget, not HEALTH_PROBE_TIMEOUT: this runs on + hot paths and a false negative only costs a re-run of ensure_running, + which consults the long probe before deciding anything destructive. + """ daemon_url = self.get_url(profile) try: - with httpx.Client(timeout=2) as client: + with httpx.Client(timeout=_probe_timeout(LIVENESS_PROBE_TIMEOUT)) as client: response = client.get(f"{daemon_url}/health") return response.status_code == 200 except Exception: @@ -317,45 +366,165 @@ def _find_api_command(self, api_version: str, env: Mapping[str, str] | None = No @staticmethod def _is_port_in_use(port: int) -> bool: - """Check if a port is in use using a socket connection (cross-platform).""" + """Check if a port is in use using a socket connection (cross-platform). + + Probes both loopback families: a server bound to ::1 only (issue #3527) + is invisible to an IPv4-only connect, which would make the caller treat + an occupied port as free. + """ import socket - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.settimeout(1) - return sock.connect_ex(("127.0.0.1", port)) == 0 + for host in LOOPBACK_HOSTS: + family = socket.AF_INET6 if ":" in host else socket.AF_INET + try: + with socket.socket(family, socket.SOCK_STREAM) as sock: + sock.settimeout(1) + if sock.connect_ex((host, port)) == 0: + return True + except OSError: + # The family isn't available on this host (e.g. IPv6 disabled). + continue + return False @staticmethod - def _find_pid_on_port(port: int) -> int | None: - """Find the PID of the process listening on a port.""" - import platform - + def _run_probe(cmd: list[str]) -> str | None: + """Run a short read-only probe command, returning stdout or None.""" try: - if platform.system() == "Windows": - # Use netstat on Windows - create_no_window = getattr(subprocess, "CREATE_NO_WINDOW", 0) - result = subprocess.run( - ["netstat", "-ano", "-p", "TCP"], - capture_output=True, - text=True, - timeout=5, - creationflags=create_no_window, - ) - if result.returncode == 0: - for line in result.stdout.splitlines(): - if f"127.0.0.1:{port}" in line and "LISTENING" in line: - return int(line.strip().split()[-1]) - else: - # Use lsof on macOS/Linux - result = subprocess.run( - ["lsof", "-ti", f":{port}", "-sTCP:LISTEN"], - capture_output=True, - text=True, - timeout=5, + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=5, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + except (subprocess.TimeoutExpired, OSError): + return None + return result.stdout if result.returncode == 0 else None + + @staticmethod + def _windows_listening_pids(port: int) -> list[int]: + """PIDs listening on `port` according to netstat.""" + output = DaemonEmbedManager._run_probe(["netstat", "-ano", "-p", "TCP"]) + if output is None: + return [] + pids: list[int] = [] + for line in output.splitlines(): + fields = line.split() + # proto, local address, foreign address, state, pid + if len(fields) < 5 or fields[-2] != "LISTENING": + continue + # Match on the local address' port only: the daemon binds 127.0.0.1 + # but the UI can bind 0.0.0.0 or [::1] (issue #3527), and an + # IPv4-literal match would miss those listeners entirely. + local_address = fields[1] + if local_address.rsplit(":", 1)[-1] != str(port): + continue + try: + pids.append(int(fields[-1])) + except ValueError: + continue + return pids + + @staticmethod + def _posix_listening_pids(port: int) -> list[int]: + """PIDs listening on `port`, via lsof, falling back to ss. + + `lsof` is the default on macOS but is absent from minimal containers and + several Linux distributions, where it used to leave the caller with no + PID at all (issue #3517). `ss` (iproute2) covers those hosts. + """ + output = DaemonEmbedManager._run_probe(["lsof", "-ti", f":{port}", "-sTCP:LISTEN"]) + pids: list[int] = [] + if output: + for token in output.split(): + try: + pids.append(int(token)) + except ValueError: + continue + if pids: + return pids + + # Lines look like: + # LISTEN 0 4096 127.0.0.1:9177 0.0.0.0:* users:(("hindsight-api",pid=15774,fd=19)) + output = DaemonEmbedManager._run_probe(["ss", "-tlnp", f"sport = :{port}"]) + if output: + for line in output.splitlines(): + if "users:" not in line: + continue + pids.extend(int(match) for match in re.findall(r"pid=(\d+)", line)) + return pids + + @staticmethod + def _listening_pids(port: int) -> list[int]: + """All PIDs listening on a local port, de-duplicated, in discovery order.""" + if platform.system() == "Windows": + found = DaemonEmbedManager._windows_listening_pids(port) + else: + found = DaemonEmbedManager._posix_listening_pids(port) + return list(dict.fromkeys(found)) + + @staticmethod + def _process_command_line(pid: int) -> str | None: + """The command line of `pid`, or None when it cannot be determined.""" + if platform.system() == "Windows": + output = DaemonEmbedManager._run_probe( + ["wmic", "process", "where", f"ProcessId={pid}", "get", "CommandLine", "/format:list"] + ) + if output is None: + output = DaemonEmbedManager._run_probe( + [ + "powershell", + "-NoProfile", + "-NonInteractive", + "-Command", + f"(Get-CimInstance Win32_Process -Filter 'ProcessId={pid}').CommandLine", + ] ) - if result.returncode == 0 and result.stdout.strip(): - return int(result.stdout.strip().split()[0]) - except (subprocess.TimeoutExpired, ValueError, OSError, FileNotFoundError): - pass + if output is None: + return None + text = output.replace("CommandLine=", " ").strip() + return text or None + + proc_cmdline = Path(f"/proc/{pid}/cmdline") + try: + raw = proc_cmdline.read_bytes() + except OSError: + raw = None + if raw: + return raw.replace(b"\x00", b" ").decode("utf-8", "replace").strip() + + # macOS (and Linux hosts without procfs). + output = DaemonEmbedManager._run_probe(["ps", "-p", str(pid), "-o", "args="]) + return output.strip() if output and output.strip() else None + + @staticmethod + def _process_matches(pid: int, markers: tuple[str, ...]) -> bool: + """True when `pid`'s command line identifies it as one of our processes.""" + cmdline = DaemonEmbedManager._process_command_line(pid) + if cmdline is None: + return False + lowered = cmdline.lower() + return any(marker in lowered for marker in markers) + + @staticmethod + def _owned_pid_on_port(port: int, markers: tuple[str, ...], description: str) -> int | None: + """PID listening on `port` that we can positively identify as ours. + + Returns None — refusing to signal anything — when the listener cannot be + identified. Failing to reclaim a port is recoverable; SIGTERMing an + unrelated service that merely holds the port is not (issue #3520). + """ + pids = DaemonEmbedManager._listening_pids(port) + if not pids: + logger.warning(f"Could not find PID for port {port}") + return None + for pid in pids: + if DaemonEmbedManager._process_matches(pid, markers): + return pid + logger.warning( + f"Port {port} is held by PID(s) {', '.join(str(p) for p in pids)}, which do not look like " + f"a {description}; refusing to signal them" + ) return None @staticmethod @@ -379,7 +548,7 @@ def _kill_process(pid: int) -> bool: def _port_health_ok(port: int) -> bool: """Return True when the listener on port responds like initialized Hindsight.""" try: - with httpx.Client(timeout=2) as client: + with httpx.Client(timeout=_probe_timeout(HEALTH_PROBE_TIMEOUT)) as client: response = client.get(f"http://127.0.0.1:{port}/health") if response.status_code != 200: return False @@ -392,7 +561,13 @@ def _port_health_ok(port: int) -> bool: return False def _wait_for_port_health(self, port: int, timeout: float | None = None) -> bool: - """Wait briefly for a just-bound daemon port to become healthy.""" + """Wait briefly for a just-bound daemon port to become healthy. + + `timeout` bounds when the last probe may *start*, not when it returns, so + a listener that accepts and hangs can push the wall clock to roughly + timeout + HEALTH_PROBE_TIMEOUT. That slack is deliberate: it is what lets + a daemon stalled on a slow LLM call answer before we call it stale. + """ timeout = _safe_non_negative_float( PORT_HEALTH_GRACE_TIMEOUT if timeout is None else timeout, 0.0, @@ -420,11 +595,12 @@ def _clear_port(self, port: int) -> bool: Killing it would race concurrent starts (one process kills the other's freshly-started daemon, both rush to rebind the port). * Port occupied but /health is unreachable, non-200, or does not return - Hindsight's initialized health payload → treat as a stale daemon (or - foreign process) and attempt to reclaim by killing the PID listening - on the port. This preserves the original intent of clearing stale - daemons from version upgrades. - * Kill failed, or non-hindsight process occupying the port → False. + Hindsight's initialized health payload → treat as a stale daemon and + attempt to reclaim it, but only after the listener's command line + confirms it is a Hindsight daemon. This preserves the original intent + of clearing stale daemons from version upgrades without SIGTERMing a + foreign service that merely holds the port (issue #3520). + * Kill failed, or unidentifiable/foreign process occupying the port → False. """ if not self._is_port_in_use(port): return True @@ -440,7 +616,7 @@ def _clear_port(self, port: int) -> bool: if not self._is_port_in_use(port): return True - pid = self._find_pid_on_port(port) + pid = self._owned_pid_on_port(port, DAEMON_PROCESS_MARKERS, "hindsight daemon") if pid is None: logger.warning(f"Port {port} is in use by another process") return False @@ -461,15 +637,24 @@ def _start_daemon(self, config: dict, profile: str, extra_args: list[str] | None time cannot race into `_clear_port` and kill each other's daemons. Inside the lock we re-check `is_running()`; the second caller sees the first caller's daemon and short-circuits. + + The lock wait is bounded (see HINDSIGHT_EMBED_LOCK_TIMEOUT): a caller + that cannot get in reports a startup failure instead of raising an + opaque OS error (issue #3100). """ paths = self._profile_manager.resolve_profile_paths(profile) paths.lock.parent.mkdir(parents=True, exist_ok=True) # Hold the per-profile start lock for the full startup sequence. - # lock_file() blocks until the lock is acquired on Unix (flock) and - # Windows (msvcrt), so concurrent callers serialize here. + # lock_file() retries until the lock is acquired or the wait budget + # expires, on both Unix (flock) and Windows (msvcrt), so concurrent + # callers serialize here. with open(paths.lock, "w") as lock_fd: - lock_file(lock_fd) + try: + lock_file(lock_fd) + except ProfileLockTimeout as exc: + logger.error(f"Cannot start daemon for profile '{profile}': {exc}") + return False try: if self.is_running(profile): logger.debug(f"Daemon for profile '{profile}' came up while waiting for start lock") @@ -774,24 +959,37 @@ def _find_ui_command(self, cp_version: str) -> list[str]: return ["npx", "-y", f"@vectorize-io/hindsight-control-plane@{cp_version}"] return [npx_path, "-y", f"@vectorize-io/hindsight-control-plane@{cp_version}"] - def get_ui_url(self, profile: str, ui_port: int | None = None, hostname: str | None = None) -> str: - """Get the URL for the UI serving this profile.""" + def get_ui_url(self, profile: str, ui_port: int | None = None) -> str: + """Get the URL for the UI serving this profile (callers render it for display).""" if ui_port is None: paths = self._profile_manager.resolve_profile_paths(profile) ui_port = paths.ui_port - host = hostname or "0.0.0.0" - return f"http://{host}:{ui_port}" + return f"http://0.0.0.0:{ui_port}" def is_ui_running(self, profile: str, ui_port: int | None = None) -> bool: """Check if the UI is running and responsive.""" - # Always health-check on 127.0.0.1 regardless of bind hostname - ui_url = self.get_ui_url(profile, ui_port, hostname="127.0.0.1") - try: - with httpx.Client(timeout=2) as client: - response = client.get(f"{ui_url}/api/health") - return response.status_code == 200 - except Exception: - return False + return self._reachable_ui_host(profile, ui_port) is not None + + def _reachable_ui_host(self, profile: str, ui_port: int | None = None) -> str | None: + """Loopback host on which the UI answers /api/health, or None. + + The UI is probed on every loopback family rather than on 127.0.0.1 + alone: Next.js started with `--hostname localhost` binds ::1 only, so + the IPv4 probe refused the connection and `ui start`/`ui status` + reported a healthy control plane as down (issue #3527). + """ + if ui_port is None: + ui_port = self._profile_manager.resolve_profile_paths(profile).ui_port + for host in LOOPBACK_HOSTS: + # IPv6 literals have to be bracketed in a URL authority. + base = f"http://[{host}]:{ui_port}" if ":" in host else f"http://{host}:{ui_port}" + try: + with httpx.Client(timeout=_probe_timeout(LIVENESS_PROBE_TIMEOUT)) as client: + if client.get(f"{base}/api/health").status_code == 200: + return host + except Exception: + continue + return None @staticmethod def _ui_port_file(paths) -> Path: @@ -873,9 +1071,12 @@ def start_ui(self, profile: str, ui_port: int | None = None, hostname: str = "0. live.refresh() while time.time() - start_time < 30: - if self.is_ui_running(profile, ui_port): + if self._reachable_ui_host(profile, ui_port) is not None: self._record_ui_port(paths, ui_port) - log_lines.append(f"✓ UI started at http://127.0.0.1:{ui_port}") + # "localhost" rather than a loopback literal: the UI may + # have bound ::1 only (issue #3527), and name resolution + # covers both families for whoever opens the link. + log_lines.append(f"✓ UI started at http://localhost:{ui_port}") log_lines.append(f"Logs: {ui_log}") content = Text("\n".join(log_lines), style="dim") success_title = ( @@ -961,7 +1162,7 @@ def stop_ui(self, profile: str, ui_port: int | None = None) -> bool: targets.add(recorded) for port in targets: - pid = self._find_pid_on_port(port) + pid = self._owned_pid_on_port(port, UI_PROCESS_MARKERS, "control plane UI") if pid is not None: logger.debug(f"Found UI PID {pid} on port {port}") self._kill_process(pid) @@ -1019,9 +1220,9 @@ def stop(self, profile: str) -> bool: logger.debug(f"Daemon not running for profile '{profile}'") return True - pid = self._find_pid_on_port(port) + pid = self._owned_pid_on_port(port, DAEMON_PROCESS_MARKERS, "hindsight daemon") if pid is None: - logger.warning(f"Port {port} is bound but no PID could be found") + logger.warning(f"Port {port} is bound but no hindsight daemon could be identified on it") return False logger.debug(f"Found daemon PID {pid} on port {port}") diff --git a/hindsight-embed/hindsight_embed/profile_manager.py b/hindsight-embed/hindsight_embed/profile_manager.py index d309d86de3..f358bc71ad 100644 --- a/hindsight-embed/hindsight_embed/profile_manager.py +++ b/hindsight-embed/hindsight_embed/profile_manager.py @@ -7,12 +7,14 @@ import hashlib import json import logging +import math import os import sys +import time from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path -from typing import Optional +from typing import IO, Optional logger = logging.getLogger(__name__) @@ -21,34 +23,67 @@ # ============================================================================== # Why not use a library like portalocker or fasteners? # -# 1. Minimal dependency: Our use case is extremely simple - only basic -# exclusive file locking for metadata persistence. Adding a new dependency -# (even a small one) for such a narrow use case is unnecessary. +# 1. Minimal dependency: Our use case is narrow — exclusive locking for +# metadata persistence and daemon-start serialization. Adding a new +# dependency for that is unnecessary. # # 2. Portability: We only need to support the two major platforms (Unix and -# Windows), both of which have well-understood file locking mechanisms -# that can be implemented in ~10 lines of code each. +# Windows), both of which expose a non-blocking exclusive lock primitive +# that we drive from one shared retry loop below. # # 3. Maintainability: The code is straightforward and has no external # dependencies to track or update. The locking logic is localized here, # making it easy to understand and modify if needed. # -# 4. Feature scope: Libraries like portalocker provide many features we don't -# need (timeout handling, shared locks, lock files, etc.), which would add -# unnecessary complexity to our simple use case. -# -# If our locking requirements become more complex in the future (e.g., needing -# timeouts, better error handling, or supporting more edge cases), reconsider -# using a dedicated library like portalocker. +# Locks are acquired with a bounded wait, never an unbounded blocking call: +# the Windows `msvcrt.LK_LOCK` mode retries internally exactly 10 times and +# then raises, which made concurrent daemon starts for the same profile fail +# non-deterministically (issue #3100). Both platforms now take the +# non-blocking mode in a retry loop with exponential backoff, and time out +# with an error naming the lock file and its recorded holder. # ============================================================================== +ENV_LOCK_TIMEOUT = "HINDSIGHT_EMBED_LOCK_TIMEOUT" +# The daemon-start path holds the profile lock for the whole startup sequence +# (up to HINDSIGHT_EMBED_DAEMON_STARTUP_TIMEOUT, 180s by default), so the wait +# budget has to comfortably exceed that or a legitimate concurrent start would +# time out waiting for the winner. +DEFAULT_LOCK_TIMEOUT = 300.0 +_LOCK_RETRY_INITIAL = 0.01 +_LOCK_RETRY_MAX = 0.25 + + +class ProfileLockTimeout(TimeoutError): + """Raised when an exclusive lock could not be acquired within the timeout.""" + + +def _default_lock_timeout() -> float: + """Lock wait budget in seconds, overridable via HINDSIGHT_EMBED_LOCK_TIMEOUT.""" + raw = os.getenv(ENV_LOCK_TIMEOUT) + if raw is None: + return DEFAULT_LOCK_TIMEOUT + try: + value = float(raw) + except ValueError: + value = float("nan") + # Reject nan/inf/non-positive: an unbounded wait is the bug being fixed. + if not math.isfinite(value) or value <= 0: + logger.warning("Invalid %s=%r; using %s", ENV_LOCK_TIMEOUT, raw, DEFAULT_LOCK_TIMEOUT) + return DEFAULT_LOCK_TIMEOUT + return value + + if sys.platform != "win32": import fcntl - def lock_file(file_obj): - fcntl.flock(file_obj.fileno(), fcntl.LOCK_EX) + def _try_lock(file_obj: IO[str]) -> bool: + try: + fcntl.flock(file_obj.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + return True + except OSError: + return False - def unlock_file(file_obj): + def _release_lock(file_obj: IO[str]) -> None: fcntl.flock(file_obj.fileno(), fcntl.LOCK_UN) else: import msvcrt @@ -60,15 +95,93 @@ def unlock_file(file_obj): # position), then unlock — at which point the unlock request targets a # byte past the data and Windows returns EACCES. Seek to 0 on both sides # so lock and unlock always act on byte 0. - def lock_file(file_obj): + def _try_lock(file_obj: IO[str]) -> bool: file_obj.seek(0) - msvcrt.locking(file_obj.fileno(), msvcrt.LK_LOCK, 1) + try: + msvcrt.locking(file_obj.fileno(), msvcrt.LK_NBLCK, 1) + return True + except OSError: + return False - def unlock_file(file_obj): + def _release_lock(file_obj: IO[str]) -> None: file_obj.seek(0) msvcrt.locking(file_obj.fileno(), msvcrt.LK_UNLCK, 1) +def _owner_path(file_obj: IO[str]) -> Optional[Path]: + """Sidecar file recording the PID currently holding `file_obj`'s lock. + + The lock file itself is opened with mode "w" by callers, so a waiter + truncates it before it ever tries to lock — the holder's identity cannot + live there. The sidecar is only ever written by the lock holder. + """ + name = getattr(file_obj, "name", None) + if not isinstance(name, str): + return None + return Path(name + ".owner") + + +def _record_lock_owner(file_obj: IO[str]) -> None: + path = _owner_path(file_obj) + if path is None: + return + try: + path.write_text(str(os.getpid())) + except OSError: + logger.debug("Could not record lock owner for %s", path, exc_info=True) + + +def _clear_lock_owner(file_obj: IO[str]) -> None: + path = _owner_path(file_obj) + if path is None: + return + try: + path.unlink(missing_ok=True) + except OSError: + logger.debug("Could not clear lock owner for %s", path, exc_info=True) + + +def _describe_lock_holder(file_obj: IO[str]) -> str: + path = _owner_path(file_obj) + if path is None: + return "" + try: + holder = path.read_text().strip() + except OSError: + return "" + return f" (held by PID {holder})" if holder else "" + + +def lock_file(file_obj: IO[str], timeout: Optional[float] = None) -> None: + """Acquire an exclusive lock on `file_obj`, waiting up to `timeout` seconds. + + Raises: + ProfileLockTimeout: if the lock is still held when the budget expires. + """ + budget = _default_lock_timeout() if timeout is None else timeout + deadline = time.monotonic() + max(budget, 0.0) + delay = _LOCK_RETRY_INITIAL + while True: + if _try_lock(file_obj): + _record_lock_owner(file_obj) + return + remaining = deadline - time.monotonic() + if remaining <= 0: + name = getattr(file_obj, "name", "") + raise ProfileLockTimeout( + f"Timed out after {budget:g}s waiting for the lock on {name}{_describe_lock_holder(file_obj)}. " + f"Set {ENV_LOCK_TIMEOUT} to wait longer, or remove the lock file if the holder is gone." + ) + time.sleep(min(delay, remaining)) + delay = min(delay * 2, _LOCK_RETRY_MAX) + + +def unlock_file(file_obj: IO[str]) -> None: + """Release a lock acquired with lock_file().""" + _clear_lock_owner(file_obj) + _release_lock(file_obj) + + import httpx # Configuration paths @@ -335,10 +448,11 @@ def delete_profile(self, name: str): if config_path.exists(): config_path.unlink() - # Remove lock file + # Remove the lock file and the sidecar recording its holder (a crash + # while holding the lock leaves the sidecar behind). lock_path = self._get_profiles_dir() / f"{name}.lock" - if lock_path.exists(): - lock_path.unlink() + lock_path.unlink(missing_ok=True) + lock_path.with_name(f"{lock_path.name}.owner").unlink(missing_ok=True) # Remove the active log and any retained rotation backups. A log that # cannot be removed (still held open on Windows, say) must not abort the diff --git a/hindsight-embed/tests/test_daemon_client.py b/hindsight-embed/tests/test_daemon_client.py index 88c44f7779..d3a48384b4 100644 --- a/hindsight-embed/tests/test_daemon_client.py +++ b/hindsight-embed/tests/test_daemon_client.py @@ -8,6 +8,11 @@ from hindsight_embed import daemon_client from hindsight_embed.daemon_embed_manager import DaemonEmbedManager, _parse_non_negative_int +# What /proc//cmdline reports for a daemon this manager would have started. +# Ownership is decided on the listener's command line, so tests that expect a +# kill have to present one that identifies the process as ours (#3520). +_DAEMON_CMDLINE = "/home/u/.venv/bin/hindsight-api --port 9555" + @pytest.fixture def config(): @@ -201,7 +206,7 @@ def test_port_occupied_by_healthy_hindsight_is_reused(self): with ( patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), patch("httpx.Client") as mock_httpx_cls, - patch.object(DaemonEmbedManager, "_find_pid_on_port") as mock_find_pid, + patch.object(DaemonEmbedManager, "_listening_pids") as mock_find_pid, patch.object(DaemonEmbedManager, "_kill_process") as mock_kill, ): mock_client = MagicMock() @@ -255,7 +260,7 @@ def test_unhealthy_daemon_pid_not_found_returns_false(self): with ( patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), patch("httpx.Client") as mock_httpx_cls, - patch.object(DaemonEmbedManager, "_find_pid_on_port", return_value=None), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[]), patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 0.0), ): mock_client = MagicMock() @@ -272,7 +277,8 @@ def test_unhealthy_daemon_kill_fails_returns_false(self): with ( patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), patch("httpx.Client") as mock_httpx_cls, - patch.object(DaemonEmbedManager, "_find_pid_on_port", return_value=12345), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[12345]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE), patch.object(DaemonEmbedManager, "_kill_process", return_value=False), patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 0.0), ): @@ -290,7 +296,8 @@ def test_unhealthy_daemon_kill_succeeds_returns_true(self): with ( patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), patch("httpx.Client") as mock_httpx_cls, - patch.object(DaemonEmbedManager, "_find_pid_on_port", return_value=12345), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[12345]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE), patch.object(DaemonEmbedManager, "_kill_process", return_value=True), patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 0.0), ): @@ -308,7 +315,7 @@ def test_port_occupied_by_warming_hindsight_is_reused(self): with ( patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), patch("httpx.Client") as mock_httpx_cls, - patch.object(DaemonEmbedManager, "_find_pid_on_port") as mock_find_pid, + patch.object(DaemonEmbedManager, "_listening_pids") as mock_find_pid, patch.object(DaemonEmbedManager, "_kill_process") as mock_kill, patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 1.0), patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_CHECK_INTERVAL", 0.01), @@ -337,7 +344,7 @@ def test_port_occupied_by_foreign_health_200_returns_false(self): with ( patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), patch("httpx.Client") as mock_httpx_cls, - patch.object(DaemonEmbedManager, "_find_pid_on_port", return_value=None) as mock_find_pid, + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[]) as mock_find_pid, patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 0.0), ): mock_client = MagicMock() @@ -358,7 +365,7 @@ def test_port_cleared_during_grace_returns_true(self): with ( patch.object(DaemonEmbedManager, "_is_port_in_use", side_effect=[True, False, False]), patch("httpx.Client") as mock_httpx_cls, - patch.object(DaemonEmbedManager, "_find_pid_on_port") as mock_find_pid, + patch.object(DaemonEmbedManager, "_listening_pids") as mock_find_pid, patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_GRACE_TIMEOUT", 1.0), patch("hindsight_embed.daemon_embed_manager.PORT_HEALTH_CHECK_INTERVAL", 0.01), ): @@ -562,7 +569,8 @@ def test_busy_daemon_is_terminated(self, tmp_path): "_port_health_ok", side_effect=AssertionError("stop() must not consult /health"), ), - patch.object(DaemonEmbedManager, "_find_pid_on_port", return_value=4242), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE), patch.object(DaemonEmbedManager, "_kill_process", return_value=True) as mock_kill, ): assert manager.stop("default") is True @@ -571,10 +579,11 @@ def test_busy_daemon_is_terminated(self, tmp_path): def test_unresponsive_listener_is_reclaimed_like_clear_port(self, tmp_path): """stop() reclaims an occupied, unhealthy port the way _clear_port() does. - Without an ownership receipt "busy" and "foreign" are the same - observable state, so the start path already kills the listener holding - the profile's port. Refusing here instead would leave a wedged daemon - unstoppable, which is the #3169 symptom. + Responsiveness cannot distinguish "busy" from "foreign", so both paths + reclaim the profile's port from an unhealthy listener — but only once + the listener's command line identifies it as a hindsight daemon + (#3520). Refusing outright would leave a wedged daemon unstoppable, + which is the #3169 symptom. """ manager = DaemonEmbedManager() with ( @@ -585,7 +594,8 @@ def test_unresponsive_listener_is_reclaimed_like_clear_port(self, tmp_path): ), patch.object(DaemonEmbedManager, "_is_port_in_use", side_effect=[True, False]), patch.object(DaemonEmbedManager, "_port_health_ok", return_value=False), - patch.object(DaemonEmbedManager, "_find_pid_on_port", return_value=9999), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[9999]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE), patch.object(DaemonEmbedManager, "_kill_process", return_value=True) as mock_kill, ): assert manager.stop("default") is True @@ -594,7 +604,8 @@ def test_unresponsive_listener_is_reclaimed_like_clear_port(self, tmp_path): with ( patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), patch.object(DaemonEmbedManager, "_wait_for_port_health", return_value=False), - patch.object(DaemonEmbedManager, "_find_pid_on_port", return_value=9999), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[9999]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE), patch.object(DaemonEmbedManager, "_kill_process", return_value=True) as mock_kill, ): assert manager._clear_port(9700) is True @@ -610,7 +621,8 @@ def test_failed_termination_returns_false(self, tmp_path): return_value=self._paths(tmp_path), ), patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), - patch.object(DaemonEmbedManager, "_find_pid_on_port", return_value=4242), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE), patch.object(DaemonEmbedManager, "_kill_process", return_value=False), ): assert manager.stop("default") is False @@ -625,7 +637,7 @@ def test_bound_port_without_pid_returns_false(self, tmp_path): return_value=self._paths(tmp_path), ), patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), - patch.object(DaemonEmbedManager, "_find_pid_on_port", return_value=None), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[]), patch.object(DaemonEmbedManager, "_kill_process") as mock_kill, ): assert manager.stop("default") is False @@ -641,7 +653,7 @@ def test_unbound_port_reports_already_stopped(self, tmp_path): return_value=self._paths(tmp_path), ), patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=False), - patch.object(DaemonEmbedManager, "_find_pid_on_port") as mock_find, + patch.object(DaemonEmbedManager, "_listening_pids") as mock_find, ): assert manager.stop("default") is True mock_find.assert_not_called() @@ -656,8 +668,185 @@ def test_lingering_listener_after_kill_returns_false(self, tmp_path): return_value=self._paths(tmp_path), ), patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), - patch.object(DaemonEmbedManager, "_find_pid_on_port", return_value=4242), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value=_DAEMON_CMDLINE), patch.object(DaemonEmbedManager, "_kill_process", return_value=True), patch("hindsight_embed.daemon_embed_manager.time.sleep"), ): assert manager.stop("default") is False + + +class TestListeningPidDiscovery: + """Regression coverage for #3517: PID lookup must not depend on lsof alone.""" + + def test_posix_falls_back_to_ss_when_lsof_is_missing(self, monkeypatch): + """Hosts without lsof (minimal containers, Arch-based distros) still resolve a PID.""" + commands = [] + + def fake_run(cmd, **kwargs): + commands.append(cmd) + if cmd[0] == "lsof": + raise FileNotFoundError("lsof") + return Mock( + returncode=0, + stdout=('LISTEN 0 4096 127.0.0.1:9177 0.0.0.0:* users:(("hindsight-api",pid=15774,fd=19))\n'), + ) + + monkeypatch.setattr("hindsight_embed.daemon_embed_manager.platform.system", lambda: "Linux") + monkeypatch.setattr("hindsight_embed.daemon_embed_manager.subprocess.run", fake_run) + + assert DaemonEmbedManager._listening_pids(9177) == [15774] + assert [cmd[0] for cmd in commands] == ["lsof", "ss"] + + def test_posix_prefers_lsof_and_returns_every_listener(self, monkeypatch): + """lsof output can name several PIDs; all are candidates, not just the first.""" + monkeypatch.setattr("hindsight_embed.daemon_embed_manager.platform.system", lambda: "Linux") + monkeypatch.setattr( + "hindsight_embed.daemon_embed_manager.subprocess.run", + lambda cmd, **kwargs: Mock(returncode=0, stdout="111\n222\n"), + ) + + assert DaemonEmbedManager._listening_pids(9177) == [111, 222] + + def test_windows_matches_non_ipv4_local_addresses(self, monkeypatch): + """The UI can bind 0.0.0.0 or [::1]; a 127.0.0.1-literal match missed it.""" + netstat = ( + " TCP 0.0.0.0:19177 0.0.0.0:0 LISTENING 4321\n" + " TCP [::1]:19177 [::]:0 LISTENING 4322\n" + " TCP 127.0.0.1:29177 0.0.0.0:0 LISTENING 9999\n" + ) + monkeypatch.setattr("hindsight_embed.daemon_embed_manager.platform.system", lambda: "Windows") + monkeypatch.setattr( + "hindsight_embed.daemon_embed_manager.subprocess.run", + lambda cmd, **kwargs: Mock(returncode=0, stdout=netstat), + ) + + assert DaemonEmbedManager._listening_pids(19177) == [4321, 4322] + + +class TestProcessOwnership: + """Regression coverage for #3520: never signal a process we cannot identify.""" + + def test_foreign_listener_is_not_selected(self): + """A PID whose command line isn't ours is refused, not returned for killing.""" + with ( + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value="/usr/bin/postgres -D /data"), + ): + assert DaemonEmbedManager._owned_pid_on_port(9555, ("hindsight-api",), "hindsight daemon") is None + + def test_unknown_command_line_is_refused(self): + """When the command line cannot be read at all, we refuse rather than guess.""" + with ( + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value=None), + ): + assert DaemonEmbedManager._owned_pid_on_port(9555, ("hindsight-api",), "hindsight daemon") is None + + def test_our_daemon_is_selected_among_several_listeners(self): + """With more than one socket on the port, pick ours instead of the first PID.""" + with ( + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[111, 222]), + patch.object( + DaemonEmbedManager, + "_process_command_line", + side_effect=lambda pid: "/usr/bin/nginx" if pid == 111 else _DAEMON_CMDLINE, + ), + ): + assert DaemonEmbedManager._owned_pid_on_port(9555, ("hindsight-api",), "hindsight daemon") == 222 + + def test_clear_port_leaves_foreign_process_alive(self): + """The #3520 scenario: an unrelated service on the port must survive.""" + manager = DaemonEmbedManager() + with ( + patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), + patch.object(DaemonEmbedManager, "_wait_for_port_health", return_value=False), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value="/usr/sbin/sshd -D"), + patch.object(DaemonEmbedManager, "_kill_process") as mock_kill, + ): + assert manager._clear_port(9555) is False + mock_kill.assert_not_called() + + def test_stop_leaves_foreign_process_alive(self, tmp_path): + """stop() reclaims the port the same way, so it needs the same guard.""" + from hindsight_embed.profile_manager import ProfilePaths + + manager = DaemonEmbedManager() + paths = ProfilePaths( + config=tmp_path / "embed", + lock=tmp_path / "daemon.lock", + log=tmp_path / "daemon.log", + port=9700, + ) + with ( + patch.object(manager._profile_manager, "resolve_profile_paths", return_value=paths), + patch.object(DaemonEmbedManager, "_is_port_in_use", return_value=True), + patch.object(DaemonEmbedManager, "_listening_pids", return_value=[4242]), + patch.object(DaemonEmbedManager, "_process_command_line", return_value="/usr/sbin/sshd -D"), + patch.object(DaemonEmbedManager, "_kill_process") as mock_kill, + ): + assert manager.stop("default") is False + mock_kill.assert_not_called() + + def test_dev_mode_and_windows_daemon_command_lines_are_recognized(self): + """The daemon is spawned several ways; each must still be identifiable.""" + for cmdline in ( + "/home/u/.venv/bin/hindsight-api", + "C:\\venv\\Scripts\\pythonw.exe -m hindsight_api.main", + "/home/u/.cache/uv/archive/bin/hindsight-api", + ): + with patch.object(DaemonEmbedManager, "_process_command_line", return_value=cmdline): + from hindsight_embed.daemon_embed_manager import DAEMON_PROCESS_MARKERS + + assert DaemonEmbedManager._process_matches(1, DAEMON_PROCESS_MARKERS) is True + + +class TestStartDaemonLockTimeout: + """A start that cannot get the profile lock reports failure, not an OSError.""" + + def test_lock_timeout_returns_false(self, tmp_path, monkeypatch): + from hindsight_embed.profile_manager import ProfileLockTimeout + + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + manager = DaemonEmbedManager() + + def boom(file_obj, timeout=None): + raise ProfileLockTimeout("held by PID 999") + + with ( + patch("hindsight_embed.daemon_embed_manager.lock_file", boom), + patch.object(DaemonEmbedManager, "_start_daemon_locked") as mock_locked, + patch.object(DaemonEmbedManager, "is_running", return_value=False), + ): + assert manager._start_daemon({}, "codex") is False + mock_locked.assert_not_called() + + +class TestProcessCommandLine: + """`_process_command_line` decides every kill, so it needs direct coverage. + + The ownership tests above patch it out; this exercises the real lookup + against a process whose command line we already know. + """ + + def test_reads_the_command_line_of_a_live_process(self): + import os + import sys + + cmdline = DaemonEmbedManager._process_command_line(os.getpid()) + assert cmdline is not None + # argv[0] is the interpreter running pytest. + assert os.path.basename(sys.executable).split(".")[0] in cmdline.lower() + + def test_returns_none_for_a_pid_that_does_not_exist(self): + # Above the default pid_max on Linux and unused elsewhere, so no live + # process can answer and both the procfs and `ps` lookups must fail. + assert DaemonEmbedManager._process_command_line(4_294_967_000) is None + + def test_unknown_pid_is_never_treated_as_ours(self): + """The refusal path depends on an unreadable command line meaning "not ours".""" + from hindsight_embed.daemon_embed_manager import DAEMON_PROCESS_MARKERS + + assert DaemonEmbedManager._process_matches(4_294_967_000, DAEMON_PROCESS_MARKERS) is False diff --git a/hindsight-embed/tests/test_embed_manager.py b/hindsight-embed/tests/test_embed_manager.py index 0be1034961..754aa4289d 100644 --- a/hindsight-embed/tests/test_embed_manager.py +++ b/hindsight-embed/tests/test_embed_manager.py @@ -365,7 +365,7 @@ def test_find_api_command_windows_prefers_scripts_dir_pythonw_for_wrappers(tmp_p assert manager._find_api_command("0.0.0") == [str(pythonw), "-m", "hindsight_api.main"] -def test_find_pid_on_port_windows_hides_netstat_console(monkeypatch): +def test_listening_pids_windows_hides_netstat_console(monkeypatch): """Windows netstat probes must not flash a console window.""" calls = [] @@ -380,7 +380,7 @@ def fake_run(*args, **kwargs): monkeypatch.setattr("hindsight_embed.daemon_embed_manager.subprocess.CREATE_NO_WINDOW", 0x08000000, raising=False) monkeypatch.setattr("hindsight_embed.daemon_embed_manager.subprocess.run", fake_run) - assert DaemonEmbedManager._find_pid_on_port(9177) == 4321 + assert DaemonEmbedManager._listening_pids(9177) == [4321] assert calls[0][1]["creationflags"] == 0x08000000 @@ -396,7 +396,16 @@ def test_stop_ui_kills_recorded_and_configured_ports(tmp_path, monkeypatch): assert manager._ui_port_file(paths).exists() killed = [] - monkeypatch.setattr(manager, "_find_pid_on_port", lambda port: {9000: 111, 9001: 222}.get(port)) + monkeypatch.setattr( + DaemonEmbedManager, + "_listening_pids", + staticmethod(lambda port: {9000: [111], 9001: [222]}.get(port, [])), + ) + monkeypatch.setattr( + DaemonEmbedManager, + "_process_command_line", + staticmethod(lambda pid: "next-server (v16.2.11)"), + ) monkeypatch.setattr(DaemonEmbedManager, "_kill_process", staticmethod(lambda pid: killed.append(pid) or True)) monkeypatch.setattr(manager, "_is_port_in_use", lambda port: False) @@ -449,3 +458,159 @@ def test_component_version_resolution(tmp_path, monkeypatch): "p", {"HINDSIGHT_API_LLM_PROVIDER": "openai", "HINDSIGHT_EMBED_CP_VERSION": "1.2.3"} ) assert manager._component_version("p", "HINDSIGHT_EMBED_CP_VERSION") == "1.2.3" + + +def test_is_ui_running_detects_ipv6_only_ui(tmp_path, monkeypatch): + """Regression for #3527: `--hostname localhost` binds ::1 only. + + The old IPv4-only probe got ECONNREFUSED and reported a healthy control + plane as down, so `ui start` always timed out and `ui status` lied. + """ + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + manager = DaemonEmbedManager() + + requested = [] + + class FakeClient: + def __init__(self, *args, **kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def get(self, url): + requested.append(url) + if url.startswith("http://[::1]:"): + return MagicMock(status_code=200) + raise OSError("Connection refused") + + monkeypatch.setattr("hindsight_embed.daemon_embed_manager.httpx.Client", FakeClient) + + assert manager.is_ui_running("hermes", 19177) is True + assert requested == [ + "http://127.0.0.1:19177/api/health", + "http://[::1]:19177/api/health", + ] + + +def test_is_ui_running_false_when_no_loopback_answers(tmp_path, monkeypatch): + """Both families refused — the UI really is down.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + manager = DaemonEmbedManager() + + class FakeClient: + def __init__(self, *args, **kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def get(self, url): + raise OSError("Connection refused") + + monkeypatch.setattr("hindsight_embed.daemon_embed_manager.httpx.Client", FakeClient) + + assert manager.is_ui_running("hermes", 19177) is False + + +def test_is_port_in_use_checks_both_loopback_families(monkeypatch): + """An ::1-only listener occupies the port even though IPv4 refuses.""" + import socket + + attempted = [] + + class FakeSocket: + def __init__(self, family, type_): + self.family = family + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def settimeout(self, value): + pass + + def connect_ex(self, address): + attempted.append(address[0]) + return 0 if self.family == socket.AF_INET6 else 1 + + monkeypatch.setattr(socket, "socket", FakeSocket) + + assert DaemonEmbedManager._is_port_in_use(19177) is True + assert attempted == ["127.0.0.1", "::1"] + + +class _RecordingClient: + """httpx.Client stand-in that records the timeout each probe was given.""" + + timeouts: list = [] + + def __init__(self, *args, **kwargs): + _RecordingClient.timeouts.append(kwargs.get("timeout")) + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def get(self, url): + return MagicMock(status_code=200) + + +def test_reclaim_probe_waits_long_enough_for_a_busy_daemon(monkeypatch): + """Regression for #3099: a busy event loop must not read as a dead daemon. + + _port_health_ok is the probe whose false negative gets the listener killed, + so it is the one that has to allow for a stalled loop. + """ + from hindsight_embed import daemon_embed_manager + + assert daemon_embed_manager.HEALTH_PROBE_TIMEOUT >= 10.0 + + _RecordingClient.timeouts = [] + monkeypatch.setattr("hindsight_embed.daemon_embed_manager.httpx.Client", _RecordingClient) + monkeypatch.setattr(daemon_embed_manager, "HEALTH_PROBE_TIMEOUT", 25.0) + + DaemonEmbedManager._port_health_ok(9177) + assert [t.read for t in _RecordingClient.timeouts] == [25.0] + + +def test_liveness_probes_stay_short(tmp_path, monkeypatch): + """The "is it up?" probes must not inherit the reclaim budget. + + The control center's delete handler asks is_running once and the UI probe + once per loopback family. At the 10s reclaim budget that path exceeded the + 5s default client timeout on Windows and the request never came back. + """ + from hindsight_embed import daemon_embed_manager + + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + manager = DaemonEmbedManager() + + _RecordingClient.timeouts = [] + monkeypatch.setattr("hindsight_embed.daemon_embed_manager.httpx.Client", _RecordingClient) + + assert manager.is_running("hermes") is True + assert manager.is_ui_running("hermes", 19177) is True + + assert [t.read for t in _RecordingClient.timeouts] == [2.0, 2.0] + assert all(t.connect == daemon_embed_manager.PROBE_CONNECT_TIMEOUT for t in _RecordingClient.timeouts) + + # An address that swallows the SYN hangs in connect, not in read, so the + # connect cap is what bounds the delete handler's three serial probes + # (daemon + one per loopback family). At 1s each that is 3s, below the 5s + # default client timeout — and below the 4s the two uncapped 2s probes + # could reach before this change. + assert daemon_embed_manager.PROBE_CONNECT_TIMEOUT * 3 < 5.0 diff --git a/hindsight-embed/tests/test_profile_lock.py b/hindsight-embed/tests/test_profile_lock.py new file mode 100644 index 0000000000..37c21667ac --- /dev/null +++ b/hindsight-embed/tests/test_profile_lock.py @@ -0,0 +1,122 @@ +"""Tests for the cross-platform profile lock (regression coverage for #3100). + +The Windows branch used to call `msvcrt.locking(..., LK_LOCK, 1)`, a blocking +lock that retries exactly 10 times internally and then raises. Two processes +starting the same profile concurrently made the loser fail with an opaque +OSError. Both platforms now drive the non-blocking primitive from one bounded +retry loop. +""" + +import os +import threading +import time + +import pytest + +from hindsight_embed import profile_manager +from hindsight_embed.profile_manager import ( + ENV_LOCK_TIMEOUT, + ProfileLockTimeout, + lock_file, + unlock_file, +) + + +def test_lock_is_exclusive_and_reacquirable(tmp_path): + """A released lock can be taken by the next caller.""" + lock_path = tmp_path / "daemon.lock" + + with open(lock_path, "w") as first: + lock_file(first, timeout=1) + unlock_file(first) + + with open(lock_path, "w") as second: + lock_file(second, timeout=1) + unlock_file(second) + + +def test_contended_lock_times_out_instead_of_blocking(tmp_path): + """The loser gets a bounded, explicit timeout — not an unbounded block.""" + lock_path = tmp_path / "daemon.lock" + + with open(lock_path, "w") as holder: + lock_file(holder, timeout=1) + try: + started = time.monotonic() + with open(lock_path, "w") as waiter: + with pytest.raises(ProfileLockTimeout) as excinfo: + lock_file(waiter, timeout=0.2) + elapsed = time.monotonic() - started + assert elapsed < 5 # bounded wait, not a hang + message = str(excinfo.value) + assert str(lock_path) in message # names the lock file + assert f"PID {os.getpid()}" in message # names the holder + assert ENV_LOCK_TIMEOUT in message # says how to wait longer + finally: + unlock_file(holder) + + +def test_waiter_acquires_once_the_holder_releases(tmp_path): + """A concurrent start that waits its turn succeeds instead of erroring out.""" + lock_path = tmp_path / "daemon.lock" + acquired = threading.Event() + + with open(lock_path, "w") as holder: + lock_file(holder, timeout=1) + + def waiter(): + with open(lock_path, "w") as f: + lock_file(f, timeout=10) + acquired.set() + unlock_file(f) + + thread = threading.Thread(target=waiter) + thread.start() + assert not acquired.wait(timeout=0.3) # still blocked by the holder + unlock_file(holder) + thread.join(timeout=10) + + assert acquired.is_set() + + +def test_owner_sidecar_is_cleaned_up_on_release(tmp_path): + """The holder receipt must not outlive the lock it describes.""" + lock_path = tmp_path / "daemon.lock" + owner_path = tmp_path / "daemon.lock.owner" + + with open(lock_path, "w") as holder: + lock_file(holder, timeout=1) + assert owner_path.read_text() == str(os.getpid()) + unlock_file(holder) + + assert not owner_path.exists() + + +def test_invalid_timeout_env_falls_back_to_default(monkeypatch): + """A malformed HINDSIGHT_EMBED_LOCK_TIMEOUT must not disable the bound.""" + for bad in ("not-a-number", "0", "-5", "inf"): + monkeypatch.setenv(ENV_LOCK_TIMEOUT, bad) + assert profile_manager._default_lock_timeout() == profile_manager.DEFAULT_LOCK_TIMEOUT + + monkeypatch.setenv(ENV_LOCK_TIMEOUT, "12.5") + assert profile_manager._default_lock_timeout() == 12.5 + + +def test_delete_profile_removes_the_lock_owner_sidecar(tmp_path, monkeypatch): + """A crash while holding the lock leaves a sidecar; delete must not orphan it.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + + manager = profile_manager.ProfileManager() + manager.create_profile("doomed", {"HINDSIGHT_API_LLM_PROVIDER": "openai"}) + + profiles_dir = tmp_path / ".hindsight" / "profiles" + lock_path = profiles_dir / "doomed.lock" + owner_path = profiles_dir / "doomed.lock.owner" + lock_path.write_text("") + owner_path.write_text("4242") # left behind by a crashed holder + + manager.delete_profile("doomed") + + assert not lock_path.exists() + assert not owner_path.exists()