wehub-resource-sync
609 lines
20 KiB
Python
609 lines
20 KiB
Python
"""
|
|
Always-on CLI diagnostics log.
|
|
|
|
Captures exceptions, warnings, and diagnostic info to a per-invocation
|
|
log file under ``~/.omnigent/logs/cli-*.log``. Separate from the
|
|
``--log`` conversation JSON transcript and the ``--debug-events`` SSE
|
|
tape — this layer is always on so crash context is available even when
|
|
the user didn't know to enable debugging ahead of time.
|
|
|
|
**Privacy contract:** At ``INFO`` level, no user prompts, message text,
|
|
tool arguments, or conversation content are logged. Only lifecycle
|
|
events (startup, shutdown, error tracebacks) appear. A redaction
|
|
filter strips obvious secrets (``Authorization`` headers, bearer
|
|
tokens, env vars matching ``*_TOKEN`` / ``*_API_KEY`` / ``*SECRET*``,
|
|
``sk-*``, ``dapi*``). Redaction runs on the fully-formatted output
|
|
(after ``%``-interpolation and traceback rendering) so secrets in
|
|
``logger.info("key=%s", val)`` args and exception frames are covered.
|
|
|
|
Log files are created with ``0o600`` permissions and pruned to keep
|
|
at most :data:`MAX_LOG_FILES` entries. A ``latest-cli.log`` symlink
|
|
is maintained for quick access.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import io
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass
|
|
from logging.handlers import RotatingFileHandler
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
from omnigent_ui_sdk import state_dir
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
#: Subdirectory under :func:`state_dir` for CLI diagnostic logs.
|
|
_LOGS_SUBDIR = "logs"
|
|
|
|
#: Maximum number of ``cli-*.log`` files kept before pruning.
|
|
MAX_LOG_FILES = 20
|
|
|
|
#: Per-file size cap before rotation (bytes).
|
|
MAX_LOG_BYTES = 10 * 1024 * 1024 # 10 MB
|
|
|
|
#: Backup count for the rotating handler (per invocation — rarely
|
|
#: hits this, but guards runaway loops).
|
|
_BACKUP_COUNT = 1
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Result type
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CliLogContext:
|
|
"""
|
|
Returned by :func:`setup_cli_logging` to let callers reference
|
|
the log path (e.g. in error messages, the Ctrl+O debug overlay,
|
|
or ``/report``).
|
|
|
|
:param path: Absolute path to the current invocation's log file,
|
|
e.g. ``~/.omnigent/logs/cli-20260518-143012-12345-a1b2c3.log``.
|
|
:param invocation_id: Short unique id for this CLI run, e.g.
|
|
``"12345-a1b2c3"``.
|
|
"""
|
|
|
|
path: Path
|
|
invocation_id: str
|
|
|
|
|
|
# Module-level holder so ``current_cli_log_path()`` works without
|
|
# threading the context through every call site.
|
|
_current: CliLogContext | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _LoggingStreamSnapshot:
|
|
"""
|
|
Original stream for a logging handler retargeted during TUI stderr capture.
|
|
|
|
:param handler: Stream handler whose output was redirected.
|
|
:param stream: Stream the handler wrote to before redirection.
|
|
"""
|
|
|
|
handler: logging.StreamHandler[io.TextIOBase]
|
|
stream: io.TextIOBase
|
|
|
|
|
|
_redirected_logging_streams: list[_LoggingStreamSnapshot] = []
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Secret redaction filter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
#: Patterns that match values likely to be secrets. Applied to every
|
|
#: log record's formatted message before it hits the file.
|
|
_SECRET_PATTERNS: list[re.Pattern[str]] = [
|
|
# Header values: "Authorization: Bearer xxx" or "bearer xxx"
|
|
re.compile(r"(?i)(authorization\s*[:=]\s*)\S+"),
|
|
re.compile(r"(?i)(bearer\s+)\S+"),
|
|
# Env-var style keys: FOO_TOKEN=xxx, FOO_API_KEY=xxx, ...
|
|
re.compile(r"(?i)(\b\w*(?:token|api_key|secret|password)\s*[:=]\s*)\S+"),
|
|
# Anthropic / OpenAI style keys
|
|
re.compile(r"\bsk-[A-Za-z0-9_-]{10,}\b"),
|
|
# Databricks PATs
|
|
re.compile(r"\bdapi[A-Za-z0-9]{10,}\b"),
|
|
]
|
|
_REDACTED = "[REDACTED]"
|
|
|
|
|
|
def _redact(text: str) -> str:
|
|
"""
|
|
Replace secret-shaped substrings in *text* with :data:`_REDACTED`.
|
|
|
|
:param text: Arbitrary log text (may include tracebacks).
|
|
:returns: Scrubbed text.
|
|
"""
|
|
for pat in _SECRET_PATTERNS:
|
|
text = pat.sub(
|
|
lambda m: m.group(1) + _REDACTED if m.lastindex else _REDACTED,
|
|
text,
|
|
)
|
|
return text
|
|
|
|
|
|
class _RedactingFormatter(logging.Formatter):
|
|
"""
|
|
Formatter that scrubs obvious secrets from the *final* formatted
|
|
output — after ``%``-interpolation of ``record.args`` and after
|
|
traceback rendering.
|
|
|
|
A ``logging.Filter`` on ``record.msg`` would run *before*
|
|
formatting, so secrets passed as ``logger.info("key=%s", secret)``
|
|
or appearing in exception tracebacks would slip through.
|
|
Overriding :meth:`format` is the correct interception point
|
|
because the base class returns the fully-assembled string
|
|
(message + traceback) and nothing downstream mutates it before
|
|
the handler writes.
|
|
"""
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
"""
|
|
Format *record* then redact secrets from the result.
|
|
|
|
:param record: The log record to format.
|
|
:returns: Formatted, redacted string ready for the handler.
|
|
"""
|
|
return _redact(super().format(record))
|
|
|
|
|
|
class _RedactingStderr(io.TextIOBase):
|
|
"""
|
|
Text stream that redirects stderr writes to the CLI log with redaction.
|
|
|
|
:param inner: Open log file handle receiving redirected stderr writes.
|
|
:param original: Original terminal stderr stream to restore after the TUI
|
|
exits.
|
|
"""
|
|
|
|
def __init__(self, inner: io.TextIOWrapper, original: io.TextIOBase) -> None:
|
|
"""
|
|
Create a redacting wrapper around an open log file.
|
|
|
|
:param inner: Log file handle opened in append text mode.
|
|
:param original: Original terminal stderr stream saved for restoration.
|
|
"""
|
|
self._inner = inner
|
|
self._original_stderr = original
|
|
|
|
def write(self, text: str) -> int:
|
|
"""
|
|
Redact and write a stderr text chunk to the log file.
|
|
|
|
:param text: Text sent to ``sys.stderr.write``.
|
|
:returns: The length of the caller's original text.
|
|
"""
|
|
self._inner.write(_redact(text))
|
|
return len(text)
|
|
|
|
def flush(self) -> None:
|
|
"""
|
|
Flush the wrapped log file.
|
|
|
|
:returns: ``None``.
|
|
"""
|
|
self._inner.flush()
|
|
|
|
def close(self) -> None:
|
|
"""
|
|
Close the wrapped log file.
|
|
|
|
:returns: ``None``.
|
|
"""
|
|
if self.closed:
|
|
return
|
|
super().close()
|
|
if not self._inner.closed:
|
|
self._inner.close()
|
|
|
|
def isatty(self) -> bool:
|
|
"""
|
|
Report that redirected stderr is not an interactive terminal.
|
|
|
|
:returns: Always ``False`` because writes are redirected to a file.
|
|
"""
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _log_dir() -> Path:
|
|
"""
|
|
Return the CLI diagnostics log directory.
|
|
|
|
Uses :func:`omnigent_ui_sdk.state_dir` as the shared
|
|
``~/.omnigent`` root so the path is defined in one place.
|
|
|
|
:returns: ``~/.omnigent/logs``.
|
|
"""
|
|
return Path(state_dir()) / _LOGS_SUBDIR
|
|
|
|
|
|
def setup_cli_logging(argv: list[str]) -> CliLogContext:
|
|
"""
|
|
Configure the always-on CLI diagnostics log.
|
|
|
|
Creates the log directory, opens a per-invocation log file,
|
|
installs the redaction filter, wires up the ``omnigent`` and
|
|
``omnigent_ui_sdk`` logger hierarchies, and prunes old log
|
|
files beyond :data:`MAX_LOG_FILES`.
|
|
|
|
Call as early as possible in :func:`omnigent.cli.main` —
|
|
before Click dispatch — so unhandled startup exceptions are
|
|
captured.
|
|
|
|
:param argv: ``sys.argv[1:]`` snapshot, logged as the first line
|
|
for post-mortem context.
|
|
:returns: A :class:`CliLogContext` with the log path and
|
|
invocation id.
|
|
"""
|
|
global _current
|
|
|
|
log_dir = _log_dir()
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
invocation_id = f"{os.getpid():05d}-{_short_id()}"
|
|
timestamp = time.strftime("%Y%m%d-%H%M%S")
|
|
filename = f"cli-{timestamp}-{invocation_id}.log"
|
|
log_path = log_dir / filename
|
|
|
|
# Rotating handler — caps a single invocation at MAX_LOG_BYTES.
|
|
handler = RotatingFileHandler(
|
|
log_path,
|
|
maxBytes=MAX_LOG_BYTES,
|
|
backupCount=_BACKUP_COUNT,
|
|
encoding="utf-8",
|
|
)
|
|
# Best-effort 0600 permissions on the log file.
|
|
with contextlib.suppress(OSError):
|
|
os.chmod(log_path, 0o600)
|
|
|
|
handler.setFormatter(
|
|
_RedactingFormatter(
|
|
fmt="%(asctime)s %(levelname)-5s [%(name)s] %(message)s",
|
|
datefmt="%H:%M:%S",
|
|
)
|
|
)
|
|
|
|
# Wire our two package hierarchies at INFO so their records reach
|
|
# the file handler.
|
|
for name in ("omnigent", "omnigent_ui_sdk"):
|
|
logger = logging.getLogger(name)
|
|
logger.setLevel(logging.INFO)
|
|
logger.addHandler(handler)
|
|
logger.propagate = False
|
|
|
|
# Suppress noisy third-party loggers that are commonly present.
|
|
for name in ("httpx", "httpcore", "asyncio", "urllib3"):
|
|
logging.getLogger(name).setLevel(logging.WARNING)
|
|
|
|
# NOTE: stderr is NOT redirected here — that's scoped to the TUI
|
|
# lifetime via ``redirect_stderr_to_log`` /
|
|
# ``restore_stderr``. Non-TUI subcommands (``server``,
|
|
# ``version``, one-shot ``run -p``) keep stderr on the
|
|
# terminal so Click errors, tracebacks, and Ctrl-C output
|
|
# remain visible.
|
|
|
|
# Symlink latest-cli.log → this file.
|
|
_update_latest_symlink(log_dir, log_path)
|
|
|
|
# Prune old cli-*.log files beyond the cap.
|
|
_prune_old_logs(log_dir)
|
|
|
|
ctx = CliLogContext(path=log_path, invocation_id=invocation_id)
|
|
_current = ctx
|
|
|
|
# First line: the invocation context for post-mortem debugging.
|
|
root = logging.getLogger("omnigent.cli_diagnostics")
|
|
root.info("CLI start — argv=%s pid=%d", argv, os.getpid())
|
|
|
|
return ctx
|
|
|
|
|
|
def current_cli_log_path() -> Path | None:
|
|
"""
|
|
Return the active invocation's log path, or ``None`` if
|
|
:func:`setup_cli_logging` has not been called yet.
|
|
|
|
:returns: Absolute path to the current ``cli-*.log`` file, or
|
|
``None``.
|
|
"""
|
|
return _current.path if _current is not None else None
|
|
|
|
|
|
def install_asyncio_exception_handler(loop: asyncio.AbstractEventLoop) -> None:
|
|
"""
|
|
Install a custom exception handler on *loop* that logs unhandled
|
|
asyncio exceptions (e.g. fire-and-forget tasks that raise) to
|
|
the CLI diagnostics log instead of printing to stderr.
|
|
|
|
:param loop: The running event loop (typically from
|
|
``asyncio.get_running_loop()`` inside the REPL's async
|
|
context).
|
|
"""
|
|
logger = logging.getLogger("omnigent.asyncio")
|
|
|
|
def _handler(
|
|
loop: asyncio.AbstractEventLoop, # noqa: ARG001 — signature mandated by asyncio
|
|
context: dict[str, object],
|
|
) -> None:
|
|
"""
|
|
Log unhandled asyncio exceptions with full traceback.
|
|
|
|
:param loop: The event loop that caught the exception.
|
|
:param context: Exception context dict from asyncio.
|
|
"""
|
|
exc = context.get("exception")
|
|
msg = context.get("message", "Unhandled asyncio exception")
|
|
if isinstance(exc, BaseException):
|
|
logger.error("%s", msg, exc_info=exc)
|
|
else:
|
|
logger.error("asyncio: %s — context=%s", msg, context)
|
|
|
|
loop.set_exception_handler(_handler)
|
|
|
|
|
|
def log_cli_error_hint(exc: BaseException) -> None:
|
|
"""
|
|
Print a one-line pointer to the log file on stderr.
|
|
|
|
Call this in the outermost exception handler (``main()``) so the
|
|
user knows where to find the full traceback. No-op if
|
|
:func:`setup_cli_logging` was never called.
|
|
|
|
:param exc: The exception that triggered the hint.
|
|
"""
|
|
path = current_cli_log_path()
|
|
if path is None:
|
|
return
|
|
# Log the full traceback to the file.
|
|
log_cli_exception(exc, prefix="Fatal CLI error")
|
|
# One quiet line on the real terminal for the user. sys.stderr
|
|
# may have been redirected to the log file, so reach through to
|
|
# the original.
|
|
dest = getattr(sys.stderr, "_original_stderr", sys.stderr)
|
|
print(f"Details logged to {path}", file=dest)
|
|
|
|
|
|
def print_setup_hint() -> None:
|
|
"""
|
|
Print a one-line configuration-recovery hint on stderr.
|
|
|
|
Used by the top-level :func:`omnigent.cli.main` exception
|
|
handlers so any error the CLI surfaces ends with a pointer to
|
|
the model-configuration command. The dominant root cause for CLI
|
|
failures in the wild is a missing or misconfigured model
|
|
credential — a hint that nudges the user toward
|
|
``omnigent setup`` keeps the recovery path obvious without
|
|
requiring per-call classification of "is this auth?".
|
|
|
|
Like :func:`log_cli_error_hint`, the line is written through
|
|
to the original ``stderr`` so it survives any logging-driven
|
|
stderr redirection that may have already happened during the
|
|
failing turn.
|
|
|
|
:returns: ``None``.
|
|
"""
|
|
dest = getattr(sys.stderr, "_original_stderr", sys.stderr)
|
|
print(
|
|
"If this looks like an auth or configuration problem, run "
|
|
"`omnigent setup` to configure a model credential.",
|
|
file=dest,
|
|
)
|
|
|
|
|
|
def log_cli_exception(exc: BaseException, *, prefix: str = "CLI error") -> None:
|
|
"""
|
|
Write a CLI exception and traceback to the active diagnostics log.
|
|
|
|
Use this for expected CLI exception boundaries that should be
|
|
visible in ``cli-*.log`` without necessarily printing the
|
|
user-facing "Details logged..." hint.
|
|
|
|
:param exc: Exception to record, e.g. ``click.ClickException("bad")``.
|
|
:param prefix: Log message prefix, e.g. ``"Fatal CLI error"``.
|
|
:returns: ``None``.
|
|
"""
|
|
if current_cli_log_path() is None:
|
|
return
|
|
logging.getLogger("omnigent.cli_diagnostics").error(
|
|
"%s: %s",
|
|
prefix,
|
|
exc,
|
|
exc_info=(type(exc), exc, exc.__traceback__),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internals
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def redirect_stderr_to_log() -> None:
|
|
"""
|
|
Replace ``sys.stderr`` with a file object appending to the active
|
|
CLI log.
|
|
|
|
Call when the TUI takes over the terminal. Every write that
|
|
previously went to the terminal's stderr now lands in the CLI
|
|
diagnostic log. The original stderr is saved so
|
|
:func:`restore_stderr` can bring it back when the TUI exits.
|
|
|
|
No-op if :func:`setup_cli_logging` has not been called yet or
|
|
if the log file cannot be opened.
|
|
"""
|
|
path = current_cli_log_path()
|
|
if path is None:
|
|
return
|
|
try:
|
|
log_fh = open(path, "a", encoding="utf-8") # noqa: SIM115 — intentionally kept open for TUI lifetime
|
|
except OSError as exc:
|
|
log_cli_exception(exc, prefix="Failed to redirect stderr to CLI log")
|
|
return
|
|
original = cast(io.TextIOBase, sys.stderr)
|
|
redirected = _RedactingStderr(log_fh, original)
|
|
sys.stderr = redirected
|
|
_retarget_stderr_logging_handlers(original, redirected)
|
|
|
|
|
|
def restore_stderr() -> None:
|
|
"""
|
|
Undo :func:`redirect_stderr_to_log` — restore the real terminal
|
|
stderr.
|
|
|
|
Safe to call even if the redirect was never installed.
|
|
"""
|
|
original = getattr(sys.stderr, "_original_stderr", None)
|
|
if original is None:
|
|
return
|
|
redirected = sys.stderr
|
|
sys.stderr = original
|
|
_restore_logging_handlers()
|
|
try:
|
|
redirected.close()
|
|
except OSError as exc:
|
|
log_cli_exception(exc, prefix="Failed to close redirected stderr")
|
|
|
|
|
|
def _retarget_stderr_logging_handlers(
|
|
original: io.TextIOBase,
|
|
redirected: io.TextIOBase,
|
|
) -> None:
|
|
"""
|
|
Point existing stderr-backed logging handlers at redirected stderr.
|
|
|
|
``logging.StreamHandler`` stores a concrete stream object at handler
|
|
construction time. Replacing ``sys.stderr`` later does not affect
|
|
handlers that already captured the old stream, including root handlers
|
|
installed by third-party SDKs. During the TUI lifetime, retarget those
|
|
handlers so their warning/error records land in the diagnostics log
|
|
instead of painting over prompt-toolkit.
|
|
|
|
:param original: Stderr stream that was current before redirection.
|
|
:param redirected: Replacement stream writing to the CLI log.
|
|
:returns: ``None``.
|
|
"""
|
|
global _redirected_logging_streams
|
|
|
|
if _redirected_logging_streams:
|
|
return
|
|
seen_handlers: set[int] = set()
|
|
for logger in _existing_loggers():
|
|
for handler in logger.handlers:
|
|
if id(handler) in seen_handlers:
|
|
continue
|
|
seen_handlers.add(id(handler))
|
|
if not isinstance(handler, logging.StreamHandler):
|
|
continue
|
|
stream = cast(io.TextIOBase, handler.stream)
|
|
if stream is not original:
|
|
continue
|
|
handler.setStream(redirected)
|
|
_redirected_logging_streams.append(
|
|
_LoggingStreamSnapshot(handler=handler, stream=stream)
|
|
)
|
|
|
|
|
|
def _restore_logging_handlers() -> None:
|
|
"""
|
|
Restore logging handlers retargeted by stderr redirection.
|
|
|
|
:returns: ``None``.
|
|
"""
|
|
global _redirected_logging_streams
|
|
|
|
snapshots = _redirected_logging_streams
|
|
_redirected_logging_streams = []
|
|
for snapshot in snapshots:
|
|
snapshot.handler.setStream(snapshot.stream)
|
|
|
|
|
|
def _existing_loggers() -> list[logging.Logger]:
|
|
"""
|
|
Return root plus all instantiated loggers in the logging registry.
|
|
|
|
:returns: Existing loggers that may own stderr-backed handlers.
|
|
"""
|
|
loggers = [logging.getLogger()]
|
|
loggers.extend(
|
|
logger
|
|
for logger in logging.Logger.manager.loggerDict.values()
|
|
if isinstance(logger, logging.Logger)
|
|
)
|
|
return loggers
|
|
|
|
|
|
def _short_id() -> str:
|
|
"""
|
|
Generate a 6-character hex id for the invocation.
|
|
|
|
:returns: A short random hex string, e.g. ``"a1b2c3"``.
|
|
"""
|
|
return os.urandom(3).hex()
|
|
|
|
|
|
def _update_latest_symlink(log_dir: Path, log_path: Path) -> None:
|
|
"""
|
|
Point ``latest-cli.log`` at *log_path*.
|
|
|
|
Best-effort — silently ignored if the filesystem doesn't support
|
|
symlinks (e.g. some Windows configurations).
|
|
|
|
:param log_dir: Parent directory containing the symlink.
|
|
:param log_path: Absolute path to the current log file.
|
|
"""
|
|
link = log_dir / "latest-cli.log"
|
|
try:
|
|
link.unlink(missing_ok=True)
|
|
link.symlink_to(log_path.name)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _safe_mtime(path: Path) -> float:
|
|
"""Return *path*'s mtime, or ``0.0`` if it has vanished.
|
|
|
|
``_prune_old_logs`` runs at the start of every ``omnigent run``, so two
|
|
concurrent launches can glob the same log set then race to delete it. A
|
|
plain ``p.stat()`` in the sort key would then hit a just-removed file and
|
|
raise ``FileNotFoundError``, aborting the whole prune and crashing CLI
|
|
startup. Treat a vanished file as oldest (it's already gone, so the
|
|
suppressed ``unlink`` below is a no-op).
|
|
"""
|
|
try:
|
|
return path.stat().st_mtime
|
|
except OSError:
|
|
return 0.0
|
|
|
|
|
|
def _prune_old_logs(log_dir: Path) -> None:
|
|
"""
|
|
Remove the oldest ``cli-*.log`` files when the count exceeds
|
|
:data:`MAX_LOG_FILES`.
|
|
|
|
Sorts by mtime (oldest first) and removes excess files. Backup
|
|
files from rotation (``cli-*.log.1``) are included in the count.
|
|
|
|
:param log_dir: Directory to prune.
|
|
"""
|
|
pattern = "cli-*.log*"
|
|
logs = sorted(log_dir.glob(pattern), key=_safe_mtime)
|
|
# Keep the newest MAX_LOG_FILES; delete the rest.
|
|
excess = logs[: max(0, len(logs) - MAX_LOG_FILES)]
|
|
for old in excess:
|
|
with contextlib.suppress(OSError):
|
|
old.unlink()
|