diff --git a/README.md b/README.md index db5f468..90133c8 100644 --- a/README.md +++ b/README.md @@ -704,6 +704,16 @@ Example output: --- +### Process diagnostics (2.0.7) + +In addition to the existing business logs, `console.log` mirrors Python stdout/stderr +(including Uvicorn errors and tracebacks), `process-events.jsonl` records process +startup, exit codes, signals and unhandled exceptions, and `crash.log` captures +Python fatal fault stacks. These files are written alongside `--log-file`. +The console and process-event files rotate by size; see +[process diagnostics](docs/process-diagnostics-2.0.7.md) for details. +**Console logs may contain sensitive data: review before sharing.** + ### Logging Logs include: diff --git a/README_zh.md b/README_zh.md index 2c66451..8bcf07e 100644 --- a/README_zh.md +++ b/README_zh.md @@ -769,6 +769,15 @@ grep projection_applied "$HOME/.workbuddy2api/codebuddy-proxy.jsonl" | jq . --- +### 进程诊断（2.0.7） + +除原有业务日志外，`console.log` 留存 Python stdout/stderr（包括 Uvicorn 异常和 +traceback），`process-events.jsonl` 记录进程启动、退出码、信号及未处理异常， +`crash.log` 用于 Python 底层致命错误堆栈。文件均存放在 `--log-file` 所在目录。 +控制台和进程事件日志按大小轮转。详细说明见 +[进程诊断指南](docs/process-diagnostics-2.0.7.md)。 +**控制台日志可能包含敏感信息，公开前请先脱敏。** + ### 日志 日志包含： diff --git a/docs/process-diagnostics-2.0.7.md b/docs/process-diagnostics-2.0.7.md new file mode 100644 index 0000000..5a1b4d5 --- /dev/null +++ b/docs/process-diagnostics-2.0.7.md @@ -0,0 +1,70 @@ +# Process diagnostics (2.0.7) + +This release adds process-level evidence for unattended proxy failures without +changing the HTTP/LLM business logic. Files use the directory of `--log-file` +(default `~/.workbuddy2api/`). + +## Files + +| File | Content | Retention | +| --- | --- | --- | +| `console.log` | Python-level stdout/stderr, including ordinary tracebacks and Uvicorn error logs, while still displaying them in the terminal | 20 MiB per file, 4 rotations (up to ~100 MiB) | +| `process-events.jsonl` | Structured process start, exit code/reason, caught signal, fatal exception class, memory measurements, unhandled asyncio/thread exceptions and diagnostic log failures | 2 MiB per file, 3 rotations | +| `crash.log` | `faulthandler` native/Python thread stacks for fatal Python runtime signals | Append-only; inspect/truncate manually after incidents | +| `proxy.log`, `codebuddy-proxy.jsonl` | Existing request, proxy and operational logs (no semantic changes) | Existing daily retention | + +The new files are created on startup, and on POSIX new files use mode `0600`. +On Windows their ACL is inherited from the containing directory; protect that +directory if multiple users share the machine. + +**Warning:** Unlike the structured process events, console and crash outputs +can include exception content, upstream fragments, tokens or other private data +printed by dependencies. Do not attach raw files to public issues without +reviewing and redacting them. + +## How to inspect an incident + +1. Find the last `process_start` in `process-events.jsonl` and its `run_id`. +2. Check for `process_fatal_exception`, `process_signal_received`, + `asyncio_unhandled_exception`, `thread_unhandled_exception`, or + `diagnostic_log_failure` for that run. +3. Match with its `process_exit` and `exit_code`. A normal graceful + `uvicorn.run()` return records code `0`. A caught uncaught exception + records `1`, and SIGTERM can record `143` (128+15). +4. Read `console.log` and `crash.log` for the traceback, without relying on + exception messages in the JSONL journal. +5. If the process terminates abruptly (e.g. Windows TerminateProcess, kill -9, + host shutdown), Python cannot run an exit handler. On its **next** launch, + `previous_run_exit_unobserved` records the previous PID/run ID if the + most recent start has no matching exit. This is not proof of a crash; + simultaneous proxy instances can also result in an unmatched run. + +## Signal and output capture limitations + +Signals handled by Uvicorn are observed with a delegating wrapper so its +original signal behavior runs first. SIGTERM is recorded even if Uvicorn +gracefully stops and replays the signal. Windows force termination bypasses +Python handlers, so it is detectable only retrospectively. The Python tee +captures `sys.stdout` and `sys.stderr`; writes directly to native file +descriptors, subprocess handles or the original binary stream are not fully +mirrored. Use a terminal/supervisor-level redirection if those must also be +preserved. + +Logs are best-effort by design: a failed console or JSONL write is reported +in `process-events.jsonl` if possible and to the original stderr without +crashing the proxy. If the entire log volume becomes inaccessible, diagnostics +may be incomplete. + +## Version + +`pyproject.toml` and FastAPI metadata are `2.0.7`. From a source checkout, +run `uv sync` before starting to refresh installed distribution metadata +used by the runtime startup event. The current running proxy is not restarted +by implementing this change. + +## Tests + +`test_process_diagnostics.py` exercises console mirroring, rotation, normal +and abnormal exits, SIGTERM, forced child-process termination, unobserved exit +detection, traceback retention, asyncio/thread exception forwarding, memory +snapshots, log failure fallback, and the version contract. diff --git a/pyproject.toml b/pyproject.toml index b774d00..6692d38 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "workbuddy2api" -version = "2.0.6" +version = "2.0.7" description = "Local OpenAI/Responses/Anthropic compatible proxy for CodeBuddy" readme = "README.md" requires-python = ">=3.10" diff --git a/src/codebuddy_proxy/__main__.py b/src/codebuddy_proxy/__main__.py index f4e61a3..ed3fe35 100755 --- a/src/codebuddy_proxy/__main__.py +++ b/src/codebuddy_proxy/__main__.py @@ -50,6 +50,7 @@ from codebuddy_proxy.billing_proxy import forward_billing from codebuddy_proxy.upstream_request import ChatRequestPreparer from codebuddy_proxy.protocol_response import format_nonstream_response +from codebuddy_proxy.process_diagnostics import ProcessDiagnostics # 尝试导入高级功能模块（可选） try: @@ -93,6 +94,17 @@ def anthropic_to_chat(body): # 日志配置 # ============================================================================ +class ObservedTimedRotatingFileHandler(logging.handlers.TimedRotatingFileHandler): + """Make otherwise-suppressed Python logging handler failures observable.""" + + def handleError(self, record: logging.LogRecord) -> None: + monitor = globals().get("process_diagnostics") + exc = sys.exc_info()[1] + if monitor is not None and exc is not None: + monitor.log_failure("file_logger", exc) + super().handleError(record) + + def setup_logging(log_dir: pathlib.Path) -> logging.Logger: """配置滚动日志：按天分片，保留30天。""" log_dir.mkdir(parents=True, exist_ok=True) @@ -103,7 +115,7 @@ def setup_logging(log_dir: pathlib.Path) -> logging.Logger: logger.propagate = False logger.handlers.clear() - handler = logging.handlers.TimedRotatingFileHandler( + handler = ObservedTimedRotatingFileHandler( log_file, when='midnight', interval=1, backupCount=30, encoding='utf-8' ) handler.suffix = "%Y-%m-%d" @@ -124,7 +136,7 @@ def setup_json_logging(log_file: pathlib.Path) -> logging.Logger: logger.propagate = False logger.handlers.clear() - handler = logging.handlers.TimedRotatingFileHandler( + handler = ObservedTimedRotatingFileHandler( log_file, when="midnight", interval=1, backupCount=30, encoding="utf-8" ) handler.suffix = "%Y-%m-%d" @@ -211,8 +223,10 @@ def write_log(self, event: str, **kwargs) -> None: **kwargs, } self.json_logger.info(json.dumps(record, ensure_ascii=False)) - except Exception: - pass + except Exception as exc: + monitor = globals().get("process_diagnostics") + if monitor is not None: + monitor.log_failure("jsonl_write", exc) def write_body_log(self, event: str, body: bytes, **kwargs) -> None: if self.json_logger is None: @@ -228,18 +242,31 @@ def write_body_log(self, event: str, body: bytes, **kwargs) -> None: **kwargs } self.json_logger.info(json.dumps(record, ensure_ascii=False)) - except Exception: - pass + except Exception as exc: + monitor = globals().get("process_diagnostics") + if monitor is not None: + monitor.log_failure("jsonl_body_write", exc) # ============================================================================ # FastAPI 应用 # ============================================================================ -app = FastAPI(title="CodeBuddy Proxy (FastAPI)", version="2.0") +app = FastAPI(title="CodeBuddy Proxy (FastAPI)", version="2.0.7") # 全局状态（在 main() 中初始化） proxy_state: ProxyState | None = None +process_diagnostics: ProcessDiagnostics | None = None + + +async def install_process_asyncio_diagnostics() -> None: + """Capture unhandled background task errors in the active Uvicorn loop.""" + if process_diagnostics is not None: + process_diagnostics.install_asyncio(asyncio.get_running_loop()) + process_diagnostics.install_signal_observers() + + +app.router.add_event_handler("startup", install_process_asyncio_diagnostics) def get_state() -> ProxyState: @@ -1990,10 +2017,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: return args -def main(): +def _run_proxy(args: argparse.Namespace) -> int: global proxy_state - - args = parse_args() try: resolved_backend = resolve_backend_from_args(args) client = CodeBuddyClient(backend=resolved_backend) @@ -2068,6 +2093,19 @@ def main(): # 启动 uvicorn run_http_server(host=args.host, port=args.port, windows_loop=args.windows_loop) + return 0 + + +def main() -> int: + global process_diagnostics + args = parse_args() + monitor = ProcessDiagnostics(args.log_file.parent) + process_diagnostics = monitor + try: + return monitor.run(lambda: _run_proxy(args)) + finally: + process_diagnostics = None + if __name__ == "__main__": raise SystemExit(main()) diff --git a/src/codebuddy_proxy/process_diagnostics.py b/src/codebuddy_proxy/process_diagnostics.py new file mode 100644 index 0000000..014336c --- /dev/null +++ b/src/codebuddy_proxy/process_diagnostics.py @@ -0,0 +1,445 @@ +"""Best-effort process-level diagnostics, independent of proxy request handling. + +Console stdout/stderr is mirrored verbatim to a bounded file. Lifecycle events +are recorded in separate bounded JSONL. Neither should break the proxy when +logging fails. Forced termination (TerminateProcess/SIGKILL) cannot execute any +Python exit handler; an unpaired startup is only a diagnostic clue. +""" +from __future__ import annotations + +import asyncio +import ctypes +import faulthandler +import json +import logging +import logging.handlers +import os +import signal +from pathlib import Path +import sys +import threading +import time +import traceback +import uuid +from typing import Any, Callable, TextIO + + +def process_memory() -> dict[str, int]: + """Get best-effort process memory sizes in bytes, without dependencies.""" + try: + if sys.platform == "win32": + from ctypes import wintypes + + class MemoryCounters(ctypes.Structure): + _fields_ = [ + ("cb", wintypes.DWORD), + ("PageFaultCount", wintypes.DWORD), + ("PeakWorkingSetSize", ctypes.c_size_t), + ("WorkingSetSize", ctypes.c_size_t), + ("QuotaPeakPagedPoolUsage", ctypes.c_size_t), + ("QuotaPagedPoolUsage", ctypes.c_size_t), + ("QuotaPeakNonPagedPoolUsage", ctypes.c_size_t), + ("QuotaNonPagedPoolUsage", ctypes.c_size_t), + ("PagefileUsage", ctypes.c_size_t), + ("PeakPagefileUsage", ctypes.c_size_t), + ] + + counters = MemoryCounters() + counters.cb = ctypes.sizeof(MemoryCounters) + handle = ctypes.windll.kernel32.GetCurrentProcess() + if ctypes.windll.psapi.GetProcessMemoryInfo( + handle, ctypes.byref(counters), counters.cb + ): + return { + "rss_bytes": int(counters.WorkingSetSize), + "peak_rss_bytes": int(counters.PeakWorkingSetSize), + "private_commit_bytes": int(counters.PagefileUsage), + } + else: + import resource + peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + # Linux returns KiB; macOS returns bytes. + return {"peak_rss_bytes": int(peak * (1 if sys.platform == "darwin" else 1024))} + except (Exception, OSError): + pass + return {} + + +def _secure_touch(path: Path) -> None: + """Apply POSIX owner-only permissions when creating a sensitive log.""" + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) + os.close(fd) + + +class EventJournal: + def __init__(self, path: Path, fallback: TextIO): + self.fallback = fallback + self.lock = threading.RLock() + self.handler: logging.handlers.RotatingFileHandler | None = None + self.failed = False + try: + _secure_touch(path) + self.handler = logging.handlers.RotatingFileHandler( + path, maxBytes=2 * 1024 * 1024, backupCount=3, encoding="utf-8" + ) + self.handler.setFormatter(logging.Formatter("%(message)s")) + except OSError as exc: + self._warn(exc) + + def _warn(self, exc: BaseException) -> None: + if not self.failed: + self.failed = True + try: + self.fallback.write( + "[process diagnostics] lifecycle log unavailable: " + + type(exc).__name__ + "\n" + ) + self.fallback.flush() + except Exception: + pass + + def emit(self, event: str, **fields: Any) -> None: + if self.handler is None: + return + record = { + "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"), + "event": event, + **fields, + } + with self.lock: + try: + # Direct stream writes allow us to detect file write failures, + # which standard logging.Handler.emit may suppress. + line = json.dumps(record, ensure_ascii=False, default=str) + with self.handler.lock: + entry = logging.LogRecord( + "workbuddy.process", logging.INFO, "", 0, line, (), None + ) + if self.handler.shouldRollover(entry): + self.handler.doRollover() + self.handler.stream.write(line + "\n") + self.handler.flush() + except Exception as exc: + self._warn(exc) + + def close(self) -> None: + if self.handler is not None: + try: + self.handler.close() + except (OSError, ValueError) as exc: + self._warn(exc) + + +class ConsoleSink: + """Byte-for-byte text mirroring with bounded file rotation.""" + + def __init__(self, path: Path, error_reporter: Callable[[str, BaseException], None]): + self.lock = threading.RLock() + self.report = error_reporter + self.handler: logging.handlers.RotatingFileHandler | None = None + self.failed = False + try: + _secure_touch(path) + self.handler = logging.handlers.RotatingFileHandler( + path, maxBytes=20 * 1024 * 1024, backupCount=4, encoding="utf-8" + ) + except OSError as exc: + self.report("console_open", exc) + + def write(self, text: str) -> None: + if not text or self.handler is None or self.failed: + return + with self.lock: + try: + with self.handler.lock: + stream = self.handler.stream + if stream is None: + return + if stream.tell() + len(text.encode("utf-8", "replace")) > self.handler.maxBytes: + self.handler.doRollover() + stream = self.handler.stream + stream.write(text) + stream.flush() + except Exception as exc: + self.failed = True + self.report("console_write", exc) + + def flush(self) -> None: + if self.handler is not None and not self.failed: + try: + self.handler.flush() + except Exception as exc: + self.failed = True + self.report("console_flush", exc) + + def close(self) -> None: + if self.handler is not None: + try: + self.handler.close() + except (OSError, ValueError) as exc: + self.report("console_close", exc) + + +class TeeStream: + def __init__(self, original: TextIO, sink: ConsoleSink): + self.original = original + self.sink = sink + + def write(self, data: str) -> int: + # Preserve original console semantics, even if mirroring fails. + count = self.original.write(data) + self.sink.write(data) + return count + + def flush(self) -> None: + try: + self.original.flush() + finally: + self.sink.flush() + + def isatty(self) -> bool: + return self.original.isatty() + + def fileno(self) -> int: + return self.original.fileno() + + def __getattr__(self, name: str) -> Any: + return getattr(self.original, name) + + +class ProcessDiagnostics: + def __init__(self, directory: Path): + self.directory = directory + self.run_id = uuid.uuid4().hex + self.started_at = time.time() + self.original_out: TextIO = sys.stdout + self.original_err: TextIO = sys.stderr + self.journal: EventJournal | None = None + self.sink: ConsoleSink | None = None + self.crash_file: Any = None + self.previous_thread_hook: Any = None + self.previous_unraisable_hook: Any = None + self.exit_recorded = False + self.last_signal: int | None = None + self.signal_handlers: dict[int, tuple[Any, Any]] = {} + self.signal_observers: dict[int, Any] = {} + self.previous_sigterm_handler: Any = None + + def event(self, name: str, **fields: Any) -> None: + if self.journal is not None: + self.journal.emit(name, pid=os.getpid(), run_id=self.run_id, **fields) + + def log_failure(self, where: str, exc: BaseException) -> None: + # Never include actual message/body/token in the diagnostic event. + self.event( + "diagnostic_log_failure", target=where, + exception_type=type(exc).__name__, **process_memory(), + ) + try: + self.original_err.write( + f"[process diagnostics] {where} failed: {type(exc).__name__}\n" + ) + self.original_err.flush() + except Exception: + pass + + def _thread_exception(self, args: Any) -> None: + self.event( + "thread_unhandled_exception", + thread_name=getattr(args.thread, "name", "unknown"), + exception_type=getattr(args.exc_type, "__name__", "unknown"), + **process_memory(), + ) + if self.previous_thread_hook is not None: + self.previous_thread_hook(args) + + def _unraisable_exception(self, args: Any) -> None: + self.event( + "unraisable_exception", + exception_type=getattr(args.exc_type, "__name__", "unknown"), + **process_memory(), + ) + if self.previous_unraisable_hook is not None: + self.previous_unraisable_hook(args) + + def _install_sigterm_exit_handler(self) -> None: + """Preserve an observable SIGTERM exit after Uvicorn replays signals.""" + if threading.current_thread() is not threading.main_thread(): + return + self.previous_sigterm_handler = signal.getsignal(signal.SIGTERM) + + def handle_sigterm(sig: int, frame: Any) -> None: + self.last_signal = sig + self.event("process_signal_received", signal=signal.Signals(sig).name, + **process_memory()) + raise SystemExit(128 + sig) + + try: + signal.signal(signal.SIGTERM, handle_sigterm) + self.signal_handlers[signal.SIGTERM] = (self.previous_sigterm_handler, handle_sigterm) + except (ValueError, OSError): + pass + + def install_signal_observers(self) -> None: + """Wrap existing Uvicorn handlers; delegate signals without changing shutdown.""" + if threading.current_thread() is not threading.main_thread(): + return + for signum in (signal.SIGINT, signal.SIGTERM): + previous = signal.getsignal(signum) + if signum in self.signal_handlers and previous is self.signal_handlers[signum][1]: + continue + if not callable(previous): + continue + + def observer(sig: int, frame: Any, original: Any = previous) -> None: + self.last_signal = sig + self.event("process_signal_received", signal=signal.Signals(sig).name, + **process_memory()) + original(sig, frame) + + try: + signal.signal(signum, observer) + self.signal_observers[signum] = observer + if signum not in self.signal_handlers: + self.signal_handlers[signum] = (previous, observer) + except (OSError, ValueError): + pass + + def install_asyncio(self, loop: asyncio.AbstractEventLoop) -> None: + old_handler = loop.get_exception_handler() + + def handler(current_loop: asyncio.AbstractEventLoop, context: dict[str, Any]) -> None: + exc = context.get("exception") + self.event( + "asyncio_unhandled_exception", + exception_type=type(exc).__name__ if exc is not None else None, + **process_memory(), + ) + if old_handler is not None: + old_handler(current_loop, context) + else: + current_loop.default_exception_handler(context) + + loop.set_exception_handler(handler) + + def __enter__(self) -> ProcessDiagnostics: + try: + self.directory.mkdir(parents=True, exist_ok=True) + except OSError as exc: + self.log_failure("create_log_directory", exc) + return self + self.journal = EventJournal(self.directory / "process-events.jsonl", self.original_err) + self.sink = ConsoleSink(self.directory / "console.log", self.log_failure) + sys.stdout = TeeStream(self.original_out, self.sink) + sys.stderr = TeeStream(self.original_err, self.sink) + self.previous_thread_hook = threading.excepthook + self.previous_unraisable_hook = sys.unraisablehook + threading.excepthook = self._thread_exception + sys.unraisablehook = self._unraisable_exception + try: + _secure_touch(self.directory / "crash.log") + self.crash_file = (self.directory / "crash.log").open("a", encoding="utf-8") + faulthandler.enable(file=self.crash_file, all_threads=True) + except (OSError, RuntimeError, ValueError) as exc: + self.log_failure("faulthandler_enable", exc) + self._install_sigterm_exit_handler() + self._report_incomplete_prior_run() + self.event( + "process_start", ppid=os.getppid(), python=sys.version.split()[0], + **process_memory(), + ) + return self + + def _report_incomplete_prior_run(self) -> None: + """An unmatched previous start suggests an unobserved exit, not proven crash.""" + path = self.directory / "process-events.jsonl" + try: + with path.open("rb") as source: + source.seek(0, os.SEEK_END) + size = source.tell() + source.seek(max(0, size - 256 * 1024)) + events = source.read().decode("utf-8", "replace").splitlines() + last_start = None + for line in events: + try: + item = json.loads(line) + except ValueError: + continue + if item.get("event") == "process_start": + last_start = item + elif (last_start is not None and item.get("event") == "process_exit" + and item.get("run_id") == last_start.get("run_id")): + last_start = None + if last_start is not None: + self.event("previous_run_exit_unobserved", + previous_pid=last_start.get("pid"), + previous_run_id=last_start.get("run_id"), + note="May indicate a force-kill, process crash, or concurrent run") + except (OSError, ValueError): + pass + + def finish(self, code: int, reason: str) -> None: + if self.exit_recorded: + return + self.exit_recorded = True + self.event( + "process_exit", exit_code=code, reason=reason, + uptime_seconds=round(time.time() - self.started_at, 3), + signal=signal.Signals(self.last_signal).name if self.last_signal else None, + **process_memory(), + ) + + def run(self, fn: Callable[[], int | None]) -> int: + with self: + try: + result = fn() + code = result if isinstance(result, int) else 0 + self.finish(code, "signal_shutdown" if self.last_signal else + ("normal" if code == 0 else "nonzero_return")) + return code + except KeyboardInterrupt: + self.event("process_interrupt", signal="SIGINT", **process_memory()) + self.finish(130, "keyboard_interrupt") + return 130 + except SystemExit as exc: + code = exc.code if isinstance(exc.code, int) else (0 if exc.code is None else 1) + self.event("process_system_exit", exit_code=code, **process_memory()) + self.finish(code, "signal_termination" if self.last_signal else "system_exit") + if exc.code is not None and not isinstance(exc.code, int): + print(str(exc.code), file=sys.stderr) + return code + except BaseException as exc: + self.event( + "process_fatal_exception", + exception_type=type(exc).__name__, **process_memory(), + ) + traceback.print_exception(type(exc), exc, exc.__traceback__, file=sys.stderr) + self.finish(1, "uncaught_exception") + return 1 + + def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None: + if not self.exit_recorded: + self.finish(1 if exc_type is not None else 0, "context_exit") + if self.previous_thread_hook is not None: + threading.excepthook = self.previous_thread_hook + if self.previous_unraisable_hook is not None: + sys.unraisablehook = self.previous_unraisable_hook + if threading.current_thread() is threading.main_thread(): + for signum, (previous, observer) in self.signal_handlers.items(): + current = signal.getsignal(signum) + if current is observer or current is self.signal_observers.get(signum): + signal.signal(signum, previous) + self.signal_handlers.clear() + self.signal_observers.clear() + sys.stdout = self.original_out + sys.stderr = self.original_err + if self.crash_file is not None: + try: + faulthandler.disable() + self.crash_file.close() + except Exception: + pass + if self.sink is not None: + self.sink.close() + if self.journal is not None: + self.journal.close() diff --git a/test_process_diagnostics.py b/test_process_diagnostics.py new file mode 100644 index 0000000..0af38b1 --- /dev/null +++ b/test_process_diagnostics.py @@ -0,0 +1,326 @@ +"""Process diagnostics and version contract tests (no running proxy required).""" +from __future__ import annotations + +import asyncio +import json +import os +import signal +import time +from pathlib import Path +import subprocess +import sys +import threading +from types import SimpleNamespace + +import pytest + +from codebuddy_proxy.process_diagnostics import ( + ConsoleSink, + ProcessDiagnostics, + TeeStream, + process_memory, +) + + +def _events(directory: Path): + file = directory / "process-events.jsonl" + return [json.loads(s) for s in file.read_text(encoding="utf-8").splitlines()] + + +def test_successful_run_captures_console_and_lifecycle(tmp_path, capsys): + monitor = ProcessDiagnostics(tmp_path) + + def command(): + print("stdout greeting") + print("stderr greeting", file=sys.stderr) + return 0 + + assert monitor.run(command) == 0 + stdout, stderr = capsys.readouterr() + assert "stdout greeting" in stdout + assert "stderr greeting" in stderr + assert sys.stdout is monitor.original_out + assert sys.stderr is monitor.original_err + recorded = (tmp_path / "console.log").read_text(encoding="utf-8") + assert "stdout greeting\n" in recorded + assert "stderr greeting\n" in recorded + assert [r["event"] for r in _events(tmp_path)] == ["process_start", "process_exit"] + exit_record = _events(tmp_path)[-1] + assert exit_record["exit_code"] == 0 + assert exit_record["reason"] == "normal" + assert exit_record["run_id"] == _events(tmp_path)[0]["run_id"] + assert "uptime_seconds" in exit_record + + +def test_unhandled_exception_traceback_and_nonzero_exit(tmp_path, capsys): + def crash(): + raise RuntimeError("intentional failure for test") + + assert ProcessDiagnostics(tmp_path).run(crash) == 1 + stderr = capsys.readouterr().err + assert "Traceback (most recent call last)" in stderr + captured = (tmp_path / "console.log").read_text(encoding="utf-8") + assert "RuntimeError: intentional failure for test" in captured + names = [r["event"] for r in _events(tmp_path)] + assert names == ["process_start", "process_fatal_exception", "process_exit"] + fatal = _events(tmp_path)[1] + assert fatal["exception_type"] == "RuntimeError" + assert _events(tmp_path)[-1]["reason"] == "uncaught_exception" + assert _events(tmp_path)[-1]["exit_code"] == 1 + + +@pytest.mark.parametrize("fn,code,reason", [ + (lambda: 7, 7, "nonzero_return"), + (lambda: (_ for _ in ()).throw(KeyboardInterrupt()), 130, "keyboard_interrupt"), + (lambda: (_ for _ in ()).throw(SystemExit(4)), 4, "system_exit"), + (lambda: (_ for _ in ()).throw(SystemExit(None)), 0, "system_exit"), +]) +def test_distinct_exit_reasons(tmp_path, fn, code, reason): + assert ProcessDiagnostics(tmp_path).run(fn) == code + assert _events(tmp_path)[-1]["exit_code"] == code + assert _events(tmp_path)[-1]["reason"] == reason + + +def test_next_start_records_unobserved_prior_exit_without_assuming_crash(tmp_path): + def stop_without_exit(): + journal = tmp_path / "process-events.jsonl" + journal.write_text( + json.dumps({"event": "process_start", "pid": 1234, "run_id": "old-id"}) + + "\n", encoding="utf-8", + ) + + stop_without_exit() + assert ProcessDiagnostics(tmp_path).run(lambda: 0) == 0 + events = _events(tmp_path) + assert [r["event"] for r in events] == [ + "process_start", "previous_run_exit_unobserved", "process_start", "process_exit", + ] + assert events[1]["previous_pid"] == 1234 + assert events[1]["previous_run_id"] == "old-id" + + +def test_completed_previous_run_does_not_raise_false_alarm(tmp_path): + assert ProcessDiagnostics(tmp_path).run(lambda: 0) == 0 + assert ProcessDiagnostics(tmp_path).run(lambda: 0) == 0 + assert not any(x["event"] == "previous_run_exit_unobserved" for x in _events(tmp_path)) + + +def test_asyncio_exceptions_are_recorded_and_delegated(tmp_path): + loop = asyncio.new_event_loop() + forwarded = [] + loop.set_exception_handler(lambda loop, context: forwarded.append(context)) + try: + with ProcessDiagnostics(tmp_path) as monitor: + monitor.install_asyncio(loop) + context = {"message": "Background task failed", "exception": ValueError("bad")} + loop.call_exception_handler(context) + assert forwarded == [context] + events = _events(tmp_path) + assert events[1]["event"] == "asyncio_unhandled_exception" + assert events[1]["exception_type"] == "ValueError" + finally: + loop.close() + + +def test_thread_and_unraisable_exceptions_are_recorded_and_forwarded(tmp_path, monkeypatch): + calls = [] + monkeypatch.setattr(threading, "excepthook", lambda args: calls.append("thread")) + monkeypatch.setattr(sys, "unraisablehook", lambda args: calls.append("unraisable")) + with ProcessDiagnostics(tmp_path) as monitor: + thread_args = SimpleNamespace( + exc_type=ValueError, thread=SimpleNamespace(name="worker"), + ) + monitor._thread_exception(thread_args) + unraisable_args = SimpleNamespace(exc_type=RuntimeError) + monitor._unraisable_exception(unraisable_args) + assert calls == ["thread", "unraisable"] + assert {x["event"] for x in _events(tmp_path)} >= { + "thread_unhandled_exception", "unraisable_exception", + } + + +def test_console_rotation_is_bounded_and_mirrors_original(tmp_path): + text = [] + + class Original: + def write(self, value): + text.append(value) + return len(value) + + def flush(self): + pass + + def isatty(self): + return False + + sink = ConsoleSink(tmp_path / "console.log", lambda *args: None) + assert sink.handler is not None + sink.handler.maxBytes = 16 + sink.handler.backupCount = 2 + tee = TeeStream(Original(), sink) + for i in range(8): + tee.write(f"line-{i}\n") + sink.close() + assert "".join(text) == "".join(f"line-{i}\n" for i in range(8)) + assert (tmp_path / "console.log").exists() + assert len(list(tmp_path.glob("console.log*"))) <= 3 + + +def test_console_file_error_does_not_suppress_original(capsys, tmp_path): + errors = [] + sink = ConsoleSink(tmp_path / "console.log", lambda name, exc: errors.append(name)) + assert sink.handler is not None + sink.handler.stream.close() + tee = TeeStream(sys.stdout, sink) + assert tee.write("still visible\n") == len("still visible\n") + assert capsys.readouterr().out == "still visible\n" + assert errors == ["console_write"] + sink.close() + + +def test_log_failure_records_exception_type_not_sensitive_text(tmp_path): + with ProcessDiagnostics(tmp_path) as monitor: + monitor.log_failure("jsonl_write", OSError("SECRET TEST LOG CONTENT")) + output = _events(tmp_path) + failure = next(x for x in output if x["event"] == "diagnostic_log_failure") + assert failure["target"] == "jsonl_write" + assert failure["exception_type"] == "OSError" + assert "SECRET TEST LOG CONTENT" not in (tmp_path / "process-events.jsonl").read_text() + + +def test_sigterm_handler_records_signal_and_exit_code(tmp_path): + monitor = ProcessDiagnostics(tmp_path) + + def simulate_sigterm(): + handler = signal.getsignal(signal.SIGTERM) + assert callable(handler) + handler(signal.SIGTERM, None) + + assert monitor.run(simulate_sigterm) == 128 + signal.SIGTERM + events = _events(tmp_path) + assert any(x['event'] == 'process_signal_received' for x in events) + assert events[-1]['reason'] == 'signal_termination' + assert events[-1]['exit_code'] == 128 + signal.SIGTERM + + +def test_force_kill_is_detectable_on_next_start(tmp_path): + script = ''' +from pathlib import Path +from codebuddy_proxy.process_diagnostics import ProcessDiagnostics +import sys, time +ProcessDiagnostics(Path(sys.argv[1])).run(lambda: time.sleep(40)) +''' + proc = subprocess.Popen([sys.executable, '-c', script, str(tmp_path)], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + try: + for _ in range(150): + if (tmp_path / 'process-events.jsonl').exists(): + if any(x['event'] == 'process_start' for x in _events(tmp_path)): + break + time.sleep(0.03) + else: + pytest.fail('child did not start within deadline') + proc.kill() + proc.wait(timeout=8) + assert ProcessDiagnostics(tmp_path).run(lambda: 0) == 0 + assert any(x['event'] == 'previous_run_exit_unobserved' for x in _events(tmp_path)) + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=8) + + +def test_main_wiring_captures_runtime_and_keeps_cli_result(tmp_path, monkeypatch): + import codebuddy_proxy.__main__ as proxy + + opts = SimpleNamespace(log_file=tmp_path / 'codebuddy-proxy.jsonl') + monkeypatch.setattr(proxy, 'parse_args', lambda: opts) + + def fake_runtime(args): + assert args is opts + print('uvicorn runtime output', file=sys.stderr) + return 0 + + monkeypatch.setattr(proxy, '_run_proxy', fake_runtime) + assert proxy.main() == 0 + assert 'uvicorn runtime output' in (tmp_path / 'console.log').read_text(encoding='utf-8') + assert _events(tmp_path)[-1]['exit_code'] == 0 + assert proxy.process_diagnostics is None + + +def test_file_handler_error_hook_reports_to_process_events(tmp_path, monkeypatch): + import codebuddy_proxy.__main__ as proxy + import logging + + handler = proxy.ObservedTimedRotatingFileHandler(tmp_path / 'proxy.log', when='midnight') + try: + with ProcessDiagnostics(tmp_path) as monitor: + monkeypatch.setattr(proxy, 'process_diagnostics', monitor) + try: + raise OSError('failed file write test') + except OSError: + handler.handleError(logging.LogRecord('test', 40, '', 1, 'test', (), None)) + monkeypatch.setattr(proxy, 'process_diagnostics', None) + finally: + handler.close() + assert any(e['event'] == 'diagnostic_log_failure' and e['target'] == 'file_logger' + for e in _events(tmp_path)) + + +def test_proxy_jsonl_logging_failure_is_not_silenced(tmp_path, monkeypatch): + import codebuddy_proxy.__main__ as proxy + + class FailingLogger: + def info(self, text): + raise OSError('secret payload must not appear in structured diagnostics') + + state = proxy.ProxyState.__new__(proxy.ProxyState) + state.runtime_info = {'app_version': '2.0.7'} + state.json_logger = FailingLogger() + with ProcessDiagnostics(tmp_path) as monitor: + monkeypatch.setattr(proxy, 'process_diagnostics', monitor) + state.write_log('client_request_summary', status=200) + state.write_body_log('upstream_response', b'secret request body') + monkeypatch.setattr(proxy, 'process_diagnostics', None) + failures = [e for e in _events(tmp_path) if e['event'] == 'diagnostic_log_failure'] + assert [e['target'] for e in failures] == ['jsonl_write', 'jsonl_body_write'] + assert all(e['exception_type'] == 'OSError' for e in failures) + assert 'secret' not in (tmp_path / 'process-events.jsonl').read_text(encoding='utf-8') + + +def test_process_memory_is_safe_and_nonnegative(): + result = process_memory() + assert isinstance(result, dict) + assert all(isinstance(value, int) and value >= 0 for value in result.values()) + + +def test_subprocess_captures_real_traceback_without_parent_hook_interference(tmp_path): + script = """ +from pathlib import Path +from codebuddy_proxy.process_diagnostics import ProcessDiagnostics +def boom(): + print("child stdout", flush=True) + raise RuntimeError("child error") +raise SystemExit(ProcessDiagnostics(Path(__import__('sys').argv[1])).run(boom)) +""" + proc = subprocess.run( + [sys.executable, "-c", script, str(tmp_path)], capture_output=True, + text=True, timeout=15, + ) + assert proc.returncode == 1 + assert "child stdout" in proc.stdout + assert "RuntimeError: child error" in proc.stderr + log = (tmp_path / "console.log").read_text(encoding="utf-8") + assert "child stdout" in log + assert "RuntimeError: child error" in log + + +def test_api_version_is_207(): + import codebuddy_proxy.__main__ as proxy + assert proxy.app.version == '2.0.7' + + +def test_project_version_is_207_in_pyproject(): + pyproject = Path(__file__).resolve().parent / "pyproject.toml" + content = pyproject.read_text(encoding="utf-8") + assert 'version = "2.0.7"' in content