diff --git a/src/iai_mcp/_filelock.py b/src/iai_mcp/_filelock.py new file mode 100644 index 00000000..db1d11a4 --- /dev/null +++ b/src/iai_mcp/_filelock.py @@ -0,0 +1,100 @@ +"""Platform-agnostic file locking shim. + +On POSIX: thin wrapper around fcntl.flock. +On Windows: msvcrt.locking with errno normalisation so callers checking +errno.EWOULDBLOCK / errno.EAGAIN on non-blocking failures work unchanged. +The file offset is saved/restored around each call (msvcrt locks relative to +the file position; fcntl.flock does not move it) and the blocking path polls +so it waits indefinitely like POSIX rather than giving up after msvcrt's ~10 s. + +Known divergence — shared locks are not truly shared on Windows. +``msvcrt.locking`` only offers exclusive byte-range locks, so LOCK_SH is +serviced as an exclusive lock: a second concurrent reader blocks where POSIX +would let both in. This is a throughput limitation, not a correctness one, and +it is deliberately NOT fixed with Win32 ``LockFileEx`` (which does support +shared locks) because callers in ``hippo/_db.py`` rely on fcntl.flock's atomic +lock *conversion* — downgrading EXCLUSIVE->SHARED and escalating SHARED-> +EXCLUSIVE in place on the same fd. ``LockFileEx`` has no atomic conversion +(you must Unlock then re-Lock, racing other waiters), so swapping it in would +trade a throughput limit for a correctness hazard on the conversion paths. +A faithful port would need those call sites reworked to a conversion-free +protocol first. +""" +from __future__ import annotations + +import os +import platform + +if platform.system() == "Windows": + import errno as _errno + import msvcrt as _msvcrt + import time as _time + + LOCK_SH = 1 + LOCK_EX = 2 + LOCK_NB = 4 + LOCK_UN = 8 + + _LOCK_BYTES = 2**30 + # Poll interval when emulating POSIX's block-until-acquired behaviour. + _BLOCK_POLL_SECONDS = 0.05 + + def flock(fd: int, operation: int) -> None: + if not isinstance(fd, int): + fd = fd.fileno() + # msvcrt.locking locks bytes starting from the current file position, so + # we must seek to 0 to lock a consistent byte range across callers. + # fcntl.flock leaves the file offset untouched, however, so save the + # caller's offset and restore it afterwards to match POSIX semantics. + try: + saved_offset: int | None = os.lseek(fd, 0, os.SEEK_CUR) + except OSError: + saved_offset = None + os.lseek(fd, 0, os.SEEK_SET) + try: + if operation & LOCK_UN: + try: + _msvcrt.locking(fd, _msvcrt.LK_UNLCK, _LOCK_BYTES) + except OSError: + pass + elif operation & (LOCK_EX | LOCK_SH): + if operation & LOCK_NB: + try: + _msvcrt.locking(fd, _msvcrt.LK_NBLCK, _LOCK_BYTES) + except OSError: + raise OSError( + _errno.EWOULDBLOCK, "resource temporarily unavailable" + ) + else: + # POSIX flock blocks until the lock is acquired, but msvcrt + # has no infinite-block mode (LK_LOCK gives up after ~10 s + # and raises). Poll LK_NBLCK so a blocking acquire matches + # POSIX semantics instead of spuriously failing under long + # contention (e.g. while the consolidator holds the lock). + while True: + try: + _msvcrt.locking(fd, _msvcrt.LK_NBLCK, _LOCK_BYTES) + break + except OSError: + os.lseek(fd, 0, os.SEEK_SET) + _time.sleep(_BLOCK_POLL_SECONDS) + finally: + if saved_offset is not None: + try: + os.lseek(fd, saved_offset, os.SEEK_SET) + except OSError: + pass + +else: + import fcntl as _fcntl + + LOCK_SH = _fcntl.LOCK_SH + LOCK_EX = _fcntl.LOCK_EX + LOCK_NB = _fcntl.LOCK_NB + LOCK_UN = _fcntl.LOCK_UN + + def flock(fd: int, operation: int) -> None: + _fcntl.flock(fd, operation) + + +__all__ = ["flock", "LOCK_EX", "LOCK_NB", "LOCK_SH", "LOCK_UN"] diff --git a/src/iai_mcp/capture.py b/src/iai_mcp/capture.py index 6b4a74d5..a0ddbe83 100644 --- a/src/iai_mcp/capture.py +++ b/src/iai_mcp/capture.py @@ -819,8 +819,11 @@ def write_deferred_event( fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) # O_CREAT mode applies only on create; enforce 0600 on every open so a # pre-existing 0644 file is tightened too (idempotent, best-effort). + # os.fchmod is POSIX-only (absent on Windows, where ACLs govern access); + # guard with hasattr, matching crypto.py / memory_bank.py. try: - os.fchmod(fd, 0o600) + if hasattr(os, "fchmod"): + os.fchmod(fd, 0o600) except OSError: pass try: diff --git a/src/iai_mcp/capture_queue.py b/src/iai_mcp/capture_queue.py index 3c601afa..cf2a5425 100644 --- a/src/iai_mcp/capture_queue.py +++ b/src/iai_mcp/capture_queue.py @@ -1,7 +1,6 @@ from __future__ import annotations import errno -import fcntl import json import os import secrets @@ -11,6 +10,8 @@ from datetime import datetime, timezone from pathlib import Path +from iai_mcp import _filelock as fcntl + DEFAULT_QUEUE_DIR: Path = Path.home() / ".iai-mcp" / "pending" """Production location for the persistent queue.""" diff --git a/src/iai_mcp/cli/__init__.py b/src/iai_mcp/cli/__init__.py index 046a91c5..614d62a0 100644 --- a/src/iai_mcp/cli/__init__.py +++ b/src/iai_mcp/cli/__init__.py @@ -23,6 +23,7 @@ DAEMON_LABEL: str = "com.iai-mcp.daemon" SERVICE_NAME: str = "iai-mcp-daemon.service" +SCHTASKS_TASK_NAME: str = "iai-mcp-daemon" CONSENT_BANNER: str = """\ ============================================================================== @@ -56,6 +57,10 @@ def _is_linux() -> bool: return platform.system() == "Linux" +def _is_windows() -> bool: + return platform.system() == "Windows" + + def _ensure_crypto_key_present(): if os.environ.get("IAI_MCP_CRYPTO_PASSPHRASE"): return None diff --git a/src/iai_mcp/cli/_crypto.py b/src/iai_mcp/cli/_crypto.py index 34e9925f..2e02a234 100644 --- a/src/iai_mcp/cli/_crypto.py +++ b/src/iai_mcp/cli/_crypto.py @@ -33,10 +33,19 @@ def cmd_crypto_status(args: argparse.Namespace) -> int: st = path.stat() mode_octal = f"0o{st.st_mode & 0o777:03o}" length = st.st_size + # POSIX mode bits and euid ownership are the security model on + # macOS/Linux; on Windows os.geteuid() does not exist and st_mode/st_uid + # are synthetic (access is governed by ACLs, which crypto.py sets via + # icacls). Report the POSIX-specific assessments only where they mean + # something, mirroring the os.name guard in crypto.py's read path. Key + # insertion order (mode, mode_secure, uid, uid_matches_process) is kept + # identical to the pre-Windows code so the POSIX JSON output is unchanged. status["mode"] = mode_octal - status["mode_secure"] = (st.st_mode & 0o077 == 0) + status["mode_secure"] = (st.st_mode & 0o077 == 0) if _os.name != "nt" else None status["uid"] = st.st_uid - status["uid_matches_process"] = (st.st_uid == _os.geteuid()) + status["uid_matches_process"] = ( + (st.st_uid == _os.geteuid()) if _os.name != "nt" else None + ) status["length_bytes"] = length status["length_valid"] = (length == KEY_BYTES) status["passphrase_fallback_set"] = bool( diff --git a/src/iai_mcp/cli/_daemon.py b/src/iai_mcp/cli/_daemon.py index 58e8f640..c1168d41 100644 --- a/src/iai_mcp/cli/_daemon.py +++ b/src/iai_mcp/cli/_daemon.py @@ -62,6 +62,65 @@ def _render_systemd_unit() -> str: return text +def _find_pythonw() -> str: + # Prefer pythonw.exe so the daemon runs with no console window; fall back to + # the current interpreter if the windowless variant is not alongside it. + exe = Path(sys.executable) + pythonw = exe.parent / "pythonw.exe" + if pythonw.exists(): + return str(pythonw) + return sys.executable + + +def _render_schtasks_xml() -> str: + from xml.sax.saxutils import escape as _xml_escape + # Escape the interpolated values: a Windows account name or interpreter path + # may legally contain '&' (and, defensively, '<'/'>'), which would otherwise + # produce malformed XML that schtasks /Create rejects. + pythonw = _xml_escape(_find_pythonw()) + username = _xml_escape(os.environ.get("USERNAME", "")) + # No : the Task Scheduler engine rejects a working + # directory set via XML when the path contains spaces (e.g. the default + # %APPDATA% under "C:\\Users\\First Last\\..."), failing the launch with + # 0x8007010B "The directory name is invalid" — even though the path exists + # and CreateProcess accepts it fine outside the scheduler. The daemon never + # depends on cwd (all state lives under ~/.iai-mcp via absolute paths), so + # we omit it and let the task default to %windir%\\system32. + return f"""\ + + + + iai-mcp sleep daemon — background memory consolidation for Claude Code + + + + true + {username} + + + + + {username} + InteractiveToken + LeastPrivilege + + + + IgnoreNew + false + false + PT0S + true + + + + {pythonw} + -m iai_mcp.daemon + + +""" + + def _prompt_consent(stream_out=None) -> bool: from iai_mcp import cli as _cli if stream_out is None: @@ -124,15 +183,61 @@ def cmd_daemon_install(args: argparse.Namespace) -> int: elif _cli._is_linux(): content = _render_systemd_unit() target = _cli.SYSTEMD_TARGET + elif _cli._is_windows(): + content = _render_schtasks_xml() + target = None else: print(f"Unsupported OS: {platform.system()}", file=sys.stderr) return 1 if dry_run: - print(f"# Would install to: {target}") + if target is not None: + print(f"# Would install to: {target}") + else: + print(f"# Would create scheduled task: {_cli.SCHTASKS_TASK_NAME}") print(content) return 0 + if _cli._is_windows(): + _cli._ensure_crypto_key_present() + + import subprocess as _sp + import tempfile as _tmpmod + + # schtasks /Create ingests the task definition as a UTF-16 XML file. + fd, xml_path = _tmpmod.mkstemp(suffix=".xml", prefix="iai-mcp-task-") + try: + with os.fdopen(fd, "w", encoding="utf-16") as f: + f.write(content) + result = _sp.run( + [ + "schtasks", "/Create", + "/TN", _cli.SCHTASKS_TASK_NAME, + "/XML", xml_path, + "/F", + ], + check=False, capture_output=True, text=True, + ) + if result.returncode != 0: + print( + f"schtasks /Create failed ({result.returncode}): " + f"{result.stderr.strip()}", + file=sys.stderr, + ) + return 1 + finally: + try: + os.unlink(xml_path) + except OSError: + pass + + _sp.run( + ["schtasks", "/Run", "/TN", _cli.SCHTASKS_TASK_NAME], + check=False, capture_output=True, + ) + print(f"Installed scheduled task: {_cli.SCHTASKS_TASK_NAME}") + return 0 + target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content) try: @@ -142,7 +247,7 @@ def cmd_daemon_install(args: argparse.Namespace) -> int: _cli._ensure_crypto_key_present() - uid = os.getuid() + uid = os.getuid() if hasattr(os, "getuid") else 0 if _cli._is_macos(): _cli.subprocess.run( ["launchctl", "bootout", f"gui/{uid}", str(target)], @@ -211,7 +316,7 @@ def cmd_daemon_uninstall(args: argparse.Namespace) -> int: print("Uninstall cancelled.", file=sys.stderr) return 1 - uid = os.getuid() + uid = os.getuid() if hasattr(os, "getuid") else 0 if _cli._is_macos(): if _cli.LAUNCHD_TARGET.exists(): _cli.subprocess.run( @@ -236,6 +341,16 @@ def cmd_daemon_uninstall(args: argparse.Namespace) -> int: ["systemctl", "--user", "daemon-reload"], check=False, capture_output=True, ) + elif _cli._is_windows(): + import subprocess as _sp + _sp.run( + ["schtasks", "/End", "/TN", _cli.SCHTASKS_TASK_NAME], + check=False, capture_output=True, + ) + _sp.run( + ["schtasks", "/Delete", "/TN", _cli.SCHTASKS_TASK_NAME, "/F"], + check=False, capture_output=True, + ) _remove_state_files() print("Daemon uninstalled. State files removed.") @@ -244,7 +359,7 @@ def cmd_daemon_uninstall(args: argparse.Namespace) -> int: def cmd_daemon_start(args: argparse.Namespace) -> int: from iai_mcp import cli as _cli - uid = os.getuid() + uid = os.getuid() if hasattr(os, "getuid") else 0 if _cli._is_macos(): target = _cli.LAUNCHD_TARGET _cli.subprocess.run( @@ -264,6 +379,12 @@ def cmd_daemon_start(args: argparse.Namespace) -> int: ["systemctl", "--user", "start", _cli.SERVICE_NAME], check=False, ) + elif _cli._is_windows(): + import subprocess as _sp + _sp.run( + ["schtasks", "/Run", "/TN", _cli.SCHTASKS_TASK_NAME], + check=False, capture_output=True, + ) else: print(f"Unsupported OS: {platform.system()}", file=sys.stderr) return 1 diff --git a/src/iai_mcp/daemon/__init__.py b/src/iai_mcp/daemon/__init__.py index 3336d12c..dd45f4ca 100644 --- a/src/iai_mcp/daemon/__init__.py +++ b/src/iai_mcp/daemon/__init__.py @@ -6,7 +6,6 @@ import json import logging import os -import resource import signal import sys import threading @@ -127,6 +126,11 @@ def _hippo_health_check_on_boot(store) -> dict[str, int | str]: def _raise_fd_limit() -> None: + if sys.platform == "win32": + return + + import resource + try: floor = int( os.environ.get("IAI_MCP_DAEMON_NOFILE_FLOOR", _DAEMON_NOFILE_FLOOR_DEFAULT) @@ -1054,7 +1058,7 @@ def _capture_handler(record: dict) -> None: shutdown = asyncio.Event() loop = asyncio.get_running_loop() - for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP): + for sig in filter(None, (signal.SIGTERM, signal.SIGINT, getattr(signal, "SIGHUP", None))): try: loop.add_signal_handler(sig, shutdown.set) except (NotImplementedError, RuntimeError): diff --git a/src/iai_mcp/doctor/_lifecycle_checks.py b/src/iai_mcp/doctor/_lifecycle_checks.py index dd4e25df..88dc4dd7 100644 --- a/src/iai_mcp/doctor/_lifecycle_checks.py +++ b/src/iai_mcp/doctor/_lifecycle_checks.py @@ -165,7 +165,8 @@ def check_b_socket_fresh() -> CheckResult: def check_c_lock_healthy() -> CheckResult: import errno as _errno - import fcntl as _fcntl + + from iai_mcp import _filelock as _fcntl lock_path = _resolve_hippo_db_path().parent / ".lock" if not lock_path.exists(): diff --git a/src/iai_mcp/lifecycle.py b/src/iai_mcp/lifecycle.py index e668d362..77129ae0 100644 --- a/src/iai_mcp/lifecycle.py +++ b/src/iai_mcp/lifecycle.py @@ -2,7 +2,6 @@ import asyncio import errno -import fcntl import os from contextlib import contextmanager from datetime import datetime, timezone @@ -10,6 +9,7 @@ from pathlib import Path from typing import Any, Iterator +from iai_mcp import _filelock as fcntl from iai_mcp.lifecycle_event_log import LifecycleEventLog from iai_mcp.lifecycle_state import ( LIFECYCLE_STATE_PATH, diff --git a/src/iai_mcp/lifecycle_event_log.py b/src/iai_mcp/lifecycle_event_log.py index 1ee02fd5..979e32b0 100644 --- a/src/iai_mcp/lifecycle_event_log.py +++ b/src/iai_mcp/lifecycle_event_log.py @@ -1,7 +1,6 @@ from __future__ import annotations import errno -import fcntl import gzip import json import os @@ -10,6 +9,8 @@ from pathlib import Path from typing import Any +from iai_mcp import _filelock as fcntl + DEFAULT_LOG_DIR: Path = Path.home() / ".iai-mcp" / "logs" KNOWN_EVENT_KINDS: frozenset[str] = frozenset( diff --git a/src/iai_mcp/lillibrain/io.py b/src/iai_mcp/lillibrain/io.py index 44ebcea1..e998a786 100644 --- a/src/iai_mcp/lillibrain/io.py +++ b/src/iai_mcp/lillibrain/io.py @@ -5,7 +5,6 @@ """ from __future__ import annotations -import fcntl import os import sys from pathlib import Path @@ -16,6 +15,13 @@ _DARWIN: bool = sys.platform == "darwin" +# F_FULLFSYNC is a macOS-only fcntl. Only import fcntl there — other platforms +# (Windows, Linux) do not need it and Windows does not ship it in the stdlib. +if _DARWIN: + import fcntl # noqa: F401 (used inside the _DARWIN-gated branches below) +else: + fcntl = None # type: ignore[assignment] + def _full_flush_disabled() -> bool: """True when the caller opted out of the macOS full drive-cache flush. diff --git a/src/iai_mcp/migrate/_to_lilli_verify.py b/src/iai_mcp/migrate/_to_lilli_verify.py index fa30e057..92b4df60 100644 --- a/src/iai_mcp/migrate/_to_lilli_verify.py +++ b/src/iai_mcp/migrate/_to_lilli_verify.py @@ -436,7 +436,10 @@ def _plant_keys_for_recall( path.parent.mkdir(parents=True, exist_ok=True) fd = os.open(str(path), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) try: - os.fchmod(fd, 0o600) + # os.fchmod is POSIX-only; on Windows the O_CREAT mode is a no-op and + # access is governed by ACLs, so guard the tightening call. + if hasattr(os, "fchmod"): + os.fchmod(fd, 0o600) os.write(fd, crypto_key) finally: os.close(fd) diff --git a/src/iai_mcp/socket_server.py b/src/iai_mcp/socket_server.py index c2a67f44..4c99c514 100644 --- a/src/iai_mcp/socket_server.py +++ b/src/iai_mcp/socket_server.py @@ -255,9 +255,6 @@ async def serve(self, socket_path: Path | None = None) -> None: env_path = os.environ.get("IAI_DAEMON_SOCKET_PATH") socket_path = Path(env_path) if env_path else SOCKET_PATH - sig = inspect.signature(asyncio.start_unix_server) - supports_cleanup_socket = "cleanup_socket" in sig.parameters - # One JSON-RPC request is one line; a relayed document upload # (base64, ≤25 MB raw) must fit the StreamReader line buffer. _line_limit = 64 * 1024 * 1024 @@ -274,6 +271,10 @@ async def serve(self, socket_path: Path | None = None) -> None: finally: shutdown_ipc() return + + sig = inspect.signature(asyncio.start_unix_server) + supports_cleanup_socket = "cleanup_socket" in sig.parameters + inherited = _inherit_activated_socket() if inherited is not None: server = await asyncio.start_unix_server(