diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index 493cddfa..96b3d2e3 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 @@ -213,14 +214,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..b6a18384 100644 --- a/src/telemetry/reasoning_traces.py +++ b/src/telemetry/reasoning_traces.py @@ -4,11 +4,16 @@ Utility for logging traces from LLM calls. This module provides structured JSONL logging of LLM inputs/outputs. """ -import fcntl +import contextlib import json +import logging +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 Any +from typing import IO, Any from pydantic import BaseModel @@ -18,6 +23,70 @@ 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[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 + on a fixed byte range, retried under an explicit policy. + + 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) + try: + yield True + finally: + locking_module.flock(f.fileno(), locking_module.LOCK_UN) + 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_NBLCK, 1) + except OSError: + if attempt < _LOCK_RETRIES - 1: + time.sleep(_LOCK_RETRY_DELAY_SECONDS) + continue + try: + yield True + finally: + f.seek(lock_offset) + with contextlib.suppress(OSError): + locking_module.locking(f.fileno(), locking_module.LK_UNLCK, 1) + 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: """Get the traces file path from settings.""" @@ -97,7 +166,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: - fcntl.flock(f.fileno(), fcntl.LOCK_EX) - f.write(json.dumps(trace_entry) + "\n") - fcntl.flock(f.fileno(), fcntl.LOCK_UN) + with open(traces_file, "a") as f, _locked(f) as acquired: + if acquired: + f.write(json.dumps(trace_entry) + "\n") + f.flush()