From fcf0be76f8be85b4cb28f63e58ebaa7320aacd39 Mon Sep 17 00:00:00 2001 From: DesarrolloProsis <38117134+DesarrolloProsis@users.noreply.github.com> Date: Tue, 25 Aug 2026 22:06:53 -0600 Subject: [PATCH] fix(windows): guard Unix-only APIs so the server and deriver start on native Windows Honcho runs fine on Linux/macOS and under WSL/Docker, but cannot start on a native Windows host. Two Unix-only APIs are used unconditionally. 1. src/telemetry/reasoning_traces.py imported fcntl at module scope. fcntl does not exist on Windows, so importing it raises ModuleNotFoundError and takes down the whole app - even though its only use is advisory locking around *optional* trace logging (REASONING_TRACES_FILE, off by default). Rather than skipping the lock on Windows (which would let the API server and deriver interleave writes into the shared traces file), locking is now done through a small _locked() context manager: fcntl.flock on POSIX, msvcrt.locking on Windows, unlocked fallback only if neither exists. 2. src/deriver/queue_manager.py called loop.add_signal_handler(), which raises NotImplementedError on Windows. That exception's str() is empty, so the failure surfaced only as 'Error in main: ' with no message. Rather than skipping signal handling on Windows (which would leave no path to set shutdown_event, stop the schedulers, or drain active_tasks), Windows now registers the same shutdown via signal.signal() - which Windows does support for SIGINT/SIGTERM - bridged onto the loop with call_soon_threadsafe. Unix behaviour is unchanged: both platforms take their native primitive. Verified on Windows 11 / Python 3.11.16 / PostgreSQL 18.6 / pgvector 0.8.6: API serves, deriver drains the queue and generates conclusions, dialectic chat returns at reasoning_level=medium. Concurrent-write test of _locked() with 6 processes x 40 writes produced 240/240 valid JSONL records, 0 malformed. ruff check and ruff format --check both pass. --- src/deriver/queue_manager.py | 29 +++++++++++++---- src/telemetry/reasoning_traces.py | 53 ++++++++++++++++++++++++++++--- 2 files changed, 71 insertions(+), 11 deletions(-) diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index 1c19c131..310e2e17 100644 --- a/src/deriver/queue_manager.py +++ b/src/deriver/queue_manager.py @@ -2,6 +2,7 @@ import asyncio import contextlib import random import signal +import sys import time from asyncio import Task from collections.abc import Iterable, Sequence @@ -200,14 +201,30 @@ class QueueManager: """Setup signal handlers, initialize client, and start the main polling loop""" logger.debug(f"Initializing QueueManager with {self.workers} workers") - # Set up signal handlers + # Set up signal handlers. + # loop.add_signal_handler() is not implemented on Windows (it raises + # NotImplementedError, whose str() is empty). Fall back to signal.signal(), + # which Windows does support for SIGINT/SIGTERM, so Ctrl+C and a service + # stop still trigger the same graceful shutdown path. loop = asyncio.get_running_loop() signals = (signal.SIGTERM, signal.SIGINT) - for sig in signals: - loop.add_signal_handler( - sig, lambda s=sig: asyncio.create_task(self.shutdown(s)) - ) - logger.debug("Signal handlers registered") + if sys.platform != "win32": + for sig in signals: + loop.add_signal_handler( + sig, lambda s=sig: asyncio.create_task(self.shutdown(s)) + ) + logger.debug("Signal handlers registered") + else: + + def _win_signal_handler(signum: int, _frame: object) -> None: + # Runs in the main thread outside the loop; hand off thread-safely. + loop.call_soon_threadsafe( + lambda: asyncio.create_task(self.shutdown(signal.Signals(signum))) + ) + + for sig in signals: + signal.signal(sig, _win_signal_handler) + logger.debug("Signal handlers registered via signal.signal (Windows)") # Start the reconciler scheduler try: diff --git a/src/telemetry/reasoning_traces.py b/src/telemetry/reasoning_traces.py index 1f4f03e3..a97b33cf 100644 --- a/src/telemetry/reasoning_traces.py +++ b/src/telemetry/reasoning_traces.py @@ -4,11 +4,25 @@ Utility for logging traces from LLM calls. This module provides structured JSONL logging of LLM inputs/outputs. """ -import fcntl +import contextlib import json +import os import time +from collections.abc import Iterator +from contextlib import contextmanager from pathlib import Path -from typing import Any +from typing import IO, Any + +try: # POSIX + import fcntl +except ImportError: # pragma: no cover - platform dependent + fcntl = None # type: ignore[assignment] + +try: # Windows + import msvcrt +except ImportError: # pragma: no cover - platform dependent + msvcrt = None # type: ignore[assignment] + from pydantic import BaseModel @@ -19,6 +33,37 @@ from src.config import ( ) +@contextmanager +def _locked(f: IO[str]) -> Iterator[None]: + """Exclusively lock an open file for the duration of the block. + + Multiple processes (API server and deriver) append to the same traces file, so + writes must be serialized. POSIX uses fcntl.flock; Windows uses msvcrt.locking, + which locks a byte range from the current offset. If neither is available the + write still proceeds unlocked rather than losing the trace. + """ + if fcntl is not None: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + elif msvcrt is not None: + f.seek(0, os.SEEK_END) + try: + msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1) + except OSError: # lock unavailable after retries — do not drop the trace + yield + return + try: + yield + finally: + with contextlib.suppress(OSError): + msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1) + else: # pragma: no cover - no locking primitive available + yield + + def get_reasoning_traces_file_path() -> Path | None: """Get the traces file path from settings.""" if settings.REASONING_TRACES_FILE: @@ -97,7 +142,5 @@ def log_reasoning_trace( trace_entry["output"]["tool_calls"] = response.tool_calls_made # Use file locking to handle concurrent writes from multiple processes - with open(traces_file, "a") as f: - fcntl.flock(f.fileno(), fcntl.LOCK_EX) + with open(traces_file, "a") as f, _locked(f): f.write(json.dumps(trace_entry) + "\n") - fcntl.flock(f.fileno(), fcntl.LOCK_UN)