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 1/5] 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) From e0cc8f8ed04a7b22e076f82267a1e6925951afd7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sergio=20B=C3=A9jar?= Date: Sat, 29 Aug 2026 01:39:54 -0600 Subject: [PATCH 2/5] fix(windows): address trace locking review feedback --- src/telemetry/reasoning_traces.py | 27 +++++++++++++++------------ 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/src/telemetry/reasoning_traces.py b/src/telemetry/reasoning_traces.py index a97b33cf..e27268b5 100644 --- a/src/telemetry/reasoning_traces.py +++ b/src/telemetry/reasoning_traces.py @@ -7,21 +7,23 @@ This module provides structured JSONL logging of LLM inputs/outputs. import contextlib import json import os +import sys import time -from collections.abc import Iterator +from collections.abc import Generator from contextlib import contextmanager from pathlib import Path from typing import IO, Any -try: # POSIX - import fcntl -except ImportError: # pragma: no cover - platform dependent - fcntl = None # type: ignore[assignment] +fcntl: Any = None +msvcrt: Any = None +if sys.platform == "win32": + import msvcrt as _msvcrt -try: # Windows - import msvcrt -except ImportError: # pragma: no cover - platform dependent - msvcrt = None # type: ignore[assignment] + msvcrt = _msvcrt +else: + import fcntl as _fcntl + + fcntl = _fcntl from pydantic import BaseModel @@ -34,7 +36,7 @@ from src.config import ( @contextmanager -def _locked(f: IO[str]) -> Iterator[None]: +def _locked(f: IO[str]) -> Generator[None, None, 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 @@ -42,13 +44,13 @@ def _locked(f: IO[str]) -> Iterator[None]: 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: + if sys.platform != "win32": fcntl.flock(f.fileno(), fcntl.LOCK_EX) try: yield finally: fcntl.flock(f.fileno(), fcntl.LOCK_UN) - elif msvcrt is not None: + elif sys.platform == "win32": f.seek(0, os.SEEK_END) try: msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1) @@ -144,3 +146,4 @@ def log_reasoning_trace( # Use file locking to handle concurrent writes from multiple processes with open(traces_file, "a") as f, _locked(f): f.write(json.dumps(trace_entry) + "\n") + f.flush() From d2ebf6e94155be28b562e6af34c9581aba5f5d87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sergio=20B=C3=A9jar?= Date: Sat, 29 Aug 2026 02:00:25 -0600 Subject: [PATCH 3/5] fix(windows): coordinate trace file locking --- src/telemetry/reasoning_traces.py | 30 ++++++++++++------------------ 1 file changed, 12 insertions(+), 18 deletions(-) diff --git a/src/telemetry/reasoning_traces.py b/src/telemetry/reasoning_traces.py index e27268b5..a4a0e5fa 100644 --- a/src/telemetry/reasoning_traces.py +++ b/src/telemetry/reasoning_traces.py @@ -6,26 +6,14 @@ This module provides structured JSONL logging of LLM inputs/outputs. import contextlib import json -import os import sys import time from collections.abc import Generator from contextlib import contextmanager +from importlib import import_module from pathlib import Path from typing import IO, Any -fcntl: Any = None -msvcrt: Any = None -if sys.platform == "win32": - import msvcrt as _msvcrt - - msvcrt = _msvcrt -else: - import fcntl as _fcntl - - fcntl = _fcntl - - from pydantic import BaseModel from src.config import ( @@ -34,6 +22,8 @@ from src.config import ( settings, ) +locking_module: Any = import_module("msvcrt" if sys.platform == "win32" else "fcntl") + @contextmanager def _locked(f: IO[str]) -> Generator[None, None, None]: @@ -45,23 +35,27 @@ def _locked(f: IO[str]) -> Generator[None, None, None]: write still proceeds unlocked rather than losing the trace. """ if sys.platform != "win32": - fcntl.flock(f.fileno(), fcntl.LOCK_EX) + locking_module.flock(f.fileno(), locking_module.LOCK_EX) try: yield finally: - fcntl.flock(f.fileno(), fcntl.LOCK_UN) + locking_module.flock(f.fileno(), locking_module.LOCK_UN) elif sys.platform == "win32": - f.seek(0, os.SEEK_END) + # Every writer must coordinate on the same byte range. Keep the lock + # offset so it can be restored before LK_UNLCK after the append. + lock_offset = 0 + f.seek(lock_offset) try: - msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1) + locking_module.locking(f.fileno(), locking_module.LK_LOCK, 1) except OSError: # lock unavailable after retries — do not drop the trace yield return try: yield finally: + f.seek(lock_offset) with contextlib.suppress(OSError): - msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1) + locking_module.locking(f.fileno(), locking_module.LK_UNLCK, 1) else: # pragma: no cover - no locking primitive available yield From 0985a2cbbfa5c9749cae5f2472fdd06512e66925 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sergio=20B=C3=A9jar?= Date: Sat, 29 Aug 2026 03:25:31 -0600 Subject: [PATCH 4/5] fix(windows): retry trace lock explicitly, never append unlocked --- src/telemetry/reasoning_traces.py | 61 +++++++++++++++++++++---------- 1 file changed, 42 insertions(+), 19 deletions(-) diff --git a/src/telemetry/reasoning_traces.py b/src/telemetry/reasoning_traces.py index a4a0e5fa..ec7db3d7 100644 --- a/src/telemetry/reasoning_traces.py +++ b/src/telemetry/reasoning_traces.py @@ -6,6 +6,7 @@ This module provides structured JSONL logging of LLM inputs/outputs. import contextlib import json +import logging import sys import time from collections.abc import Generator @@ -22,42 +23,63 @@ from src.config import ( settings, ) +logger = logging.getLogger(__name__) + locking_module: Any = import_module("msvcrt" if sys.platform == "win32" else "fcntl") +# Windows has no blocking whole-file lock, so acquisition is retried explicitly +# rather than relying on msvcrt's implicit LK_LOCK retry policy. +_LOCK_RETRIES = 10 +_LOCK_RETRY_DELAY_SECONDS = 0.1 + @contextmanager -def _locked(f: IO[str]) -> Generator[None, None, None]: +def _locked(f: IO[str]) -> Generator[bool, None, 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. + writes must be serialized. POSIX uses fcntl.flock; Windows uses msvcrt.locking + on a fixed byte range, retried under an explicit policy. + + Yields True when the lock is held. If Windows cannot acquire it within the + retry budget the block is entered with False and the caller must skip the + write: an unlocked append can interleave with another process and corrupt the + JSONL file, so a dropped trace is preferable. Tracing is an opt-in debugging + aid, so failure is logged rather than raised into the LLM call path. """ if sys.platform != "win32": locking_module.flock(f.fileno(), locking_module.LOCK_EX) try: - yield + yield True finally: locking_module.flock(f.fileno(), locking_module.LOCK_UN) - elif sys.platform == "win32": - # Every writer must coordinate on the same byte range. Keep the lock - # offset so it can be restored before LK_UNLCK after the append. - lock_offset = 0 + return + + # Every writer must coordinate on the same byte range. Keep the lock offset + # so it can be restored before LK_UNLCK after the append. + lock_offset = 0 + for attempt in range(_LOCK_RETRIES): f.seek(lock_offset) try: - locking_module.locking(f.fileno(), locking_module.LK_LOCK, 1) - except OSError: # lock unavailable after retries — do not drop the trace - yield - return + locking_module.locking(f.fileno(), locking_module.LK_NBLCK, 1) + except OSError: + if attempt < _LOCK_RETRIES - 1: + time.sleep(_LOCK_RETRY_DELAY_SECONDS) + continue try: - yield + yield True finally: f.seek(lock_offset) with contextlib.suppress(OSError): locking_module.locking(f.fileno(), locking_module.LK_UNLCK, 1) - else: # pragma: no cover - no locking primitive available - yield + return + + logger.warning( + "Could not lock reasoning traces file after %d attempts; dropping trace " + "rather than appending without a lock.", + _LOCK_RETRIES, + ) + yield False def get_reasoning_traces_file_path() -> Path | None: @@ -138,6 +160,7 @@ 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, _locked(f): - f.write(json.dumps(trace_entry) + "\n") - f.flush() + with open(traces_file, "a") as f, _locked(f) as acquired: + if acquired: + f.write(json.dumps(trace_entry) + "\n") + f.flush() From 3b7c6d1dfee27d2515b4cb310f5d72e4e719e328 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sergio=20B=C3=A9jar?= Date: Sat, 29 Aug 2026 04:16:39 -0600 Subject: [PATCH 5/5] docs(telemetry): use Google-style Yields section for _locked --- src/telemetry/reasoning_traces.py | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/src/telemetry/reasoning_traces.py b/src/telemetry/reasoning_traces.py index ec7db3d7..b6a18384 100644 --- a/src/telemetry/reasoning_traces.py +++ b/src/telemetry/reasoning_traces.py @@ -41,11 +41,17 @@ def _locked(f: IO[str]) -> Generator[bool, None, None]: writes must be serialized. POSIX uses fcntl.flock; Windows uses msvcrt.locking on a fixed byte range, retried under an explicit policy. - Yields True when the lock is held. If Windows cannot acquire it within the - retry budget the block is entered with False and the caller must skip the - write: an unlocked append can interleave with another process and corrupt the - JSONL file, so a dropped trace is preferable. Tracing is an opt-in debugging - aid, so failure is logged rather than raised into the LLM call path. + Acquisition failure is logged rather than raised: tracing is an opt-in + debugging aid, and callers invoke it from the LLM call path. + + Args: + f: Open file handle to lock for the duration of the block. + + Yields: + True while the lock is held, False when Windows cannot acquire it + within the retry budget. Callers must skip the write when False, + because an unlocked append can interleave with another process + and corrupt the JSONL file. """ if sys.platform != "win32": locking_module.flock(f.fileno(), locking_module.LOCK_EX)