From db77d69e3af978f4c9bcaeab679de54dd25c50dd Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 30 Jun 2026 16:51:28 +0000 Subject: [PATCH] =?UTF-8?q?add:=20solvent=20logs=20=E2=80=94=20tail=20and?= =?UTF-8?q?=20filter=20the=20structured=20event=20log?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - solvent logs show last 20 log lines (human-readable) - solvent logs -n 50 show last 50 lines - solvent logs -f live-follow like tail -f (0.5s poll) - solvent logs --job filter by job ID prefix - solvent logs --stage filter by pipeline stage name - solvent logs --json raw JSON output for scripting - solvent logs --path print the log file path Co-Authored-By: Claude Sonnet 4.6 Claude-Session: https://claude.ai/code/session_01SGjM79GJpSksy9wSSBa7e6 --- solvent/__main__.py | 5 + solvent/logs.py | 207 ++++++++++++++++++++++++++++++++++++ tests/test_logs.py | 251 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 463 insertions(+) create mode 100644 solvent/logs.py create mode 100644 tests/test_logs.py diff --git a/solvent/__main__.py b/solvent/__main__.py index 06c0a5c..12a3ae4 100644 --- a/solvent/__main__.py +++ b/solvent/__main__.py @@ -14,6 +14,7 @@ status live summary: balance, jobs, API key presence; --watch to auto-refresh upgrade check for newer version on PyPI; --check exits 1 if outdated jobs list/show/retry/cancel jobs (jobs --help for sub-commands) + logs tail the structured event log; -f to follow, --job/--stage to filter serve webhooks + job API + hosted briefs worker resume incomplete jobs, process the queue telegram long-poll the Telegram bot @@ -60,6 +61,10 @@ def main(): sys.argv.pop(1) from .upgrade import main as upgrade_main upgrade_main() + elif len(sys.argv) > 1 and sys.argv[1] == "logs": + sys.argv.pop(1) + from .logs import main as logs_main + logs_main() elif len(sys.argv) > 1 and sys.argv[1] == "serve": sys.argv.pop(1) from .server import main as serve_main diff --git a/solvent/logs.py b/solvent/logs.py new file mode 100644 index 0000000..4ea8795 --- /dev/null +++ b/solvent/logs.py @@ -0,0 +1,207 @@ +""" +logs.py — tail and filter the SOLVENT structured event log. + +Usage: + solvent logs show last 20 log lines + solvent logs -n 50 show last 50 lines + solvent logs --follow live-follow like tail -f + solvent logs --job filter to a specific job ID (prefix match) + solvent logs --stage filter by stage name + solvent logs --json emit raw JSON lines (default: human-readable) +""" + +from __future__ import annotations + +import json +import sys +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterator, Optional + +from .paths import data_dir + +_LOG_FIELDS_ORDER = ("ts", "job_id", "stage", "stripe_ref", "margin_actual", "duration_ms") + + +def log_path() -> Path: + return data_dir() / "solvent.log" + + +def _fmt_ts(ts: float) -> str: + dt = datetime.fromtimestamp(ts, tz=timezone.utc).astimezone() + return dt.strftime("%H:%M:%S") + + +def _fmt_duration(ms: float) -> str: + if ms < 1000: + return f"{ms:.0f}ms" + return f"{ms / 1000:.1f}s" + + +def _fmt_margin(m: float) -> str: + return f"{m * 100:.1f}%" + + +def _human_line(record: dict) -> str: + ts = _fmt_ts(record.get("ts", time.time())) + job = str(record.get("job_id", ""))[:8] + stage = record.get("stage", "?") + parts = [f"{ts} [{job}] {stage}"] + dur = record.get("duration_ms") + if dur is not None: + parts.append(_fmt_duration(dur)) + margin = record.get("margin_actual") + if margin is not None: + parts.append(f"margin={_fmt_margin(margin)}") + stripe = record.get("stripe_ref") + if stripe: + parts.append(f"stripe={stripe[:12]}") + extra_skip = {"ts", "job_id", "stage", "duration_ms", "margin_actual", "stripe_ref", + "margin_est", "simulated"} + for k, v in record.items(): + if k not in extra_skip: + parts.append(f"{k}={v}") + return " ".join(parts) + + +def _iter_lines(path: Path) -> Iterator[str]: + """Yield non-empty lines from the log file.""" + if not path.is_file(): + return + with path.open(encoding="utf-8", errors="replace") as f: + for line in f: + line = line.strip() + if line: + yield line + + +def _tail_lines(path: Path, n: int) -> list[str]: + """Return the last n lines of the log file efficiently.""" + if not path.is_file(): + return [] + with path.open("rb") as f: + f.seek(0, 2) + size = f.tell() + if size == 0: + return [] + chunk = min(size, n * 200) + f.seek(max(0, size - chunk)) + raw = f.read() + lines = raw.decode(errors="replace").splitlines() + lines = [l for l in lines if l.strip()] + return lines[-n:] + + +def _parse(line: str) -> dict | None: + try: + return json.loads(line) + except json.JSONDecodeError: + return None + + +def _matches(record: dict, *, job: Optional[str], stage: Optional[str]) -> bool: + if job: + if not str(record.get("job_id", "")).startswith(job): + return False + if stage: + if record.get("stage", "") != stage: + return False + return True + + +def show_logs( + *, + n: int = 20, + job: Optional[str] = None, + stage: Optional[str] = None, + as_json: bool = False, + follow: bool = False, +) -> None: + path = log_path() + + if not follow: + lines = _tail_lines(path, max(n * 4, 200)) # over-fetch then filter + records = [] + for line in lines: + r = _parse(line) + if r and _matches(r, job=job, stage=stage): + records.append(r) + records = records[-n:] + if not records: + print("(no log entries)" if not path.is_file() else "(no matching entries)") + return + for r in records: + print(json.dumps(r) if as_json else _human_line(r)) + return + + # --follow: print matching tail lines then watch for new ones + if path.is_file(): + lines = _tail_lines(path, n) + for line in lines: + r = _parse(line) + if r and _matches(r, job=job, stage=stage): + print(json.dumps(r) if as_json else _human_line(r)) + + pos = path.stat().st_size if path.is_file() else 0 + try: + while True: + time.sleep(0.5) + if not path.is_file(): + continue + size = path.stat().st_size + if size < pos: + pos = 0 # log rotated + if size == pos: + continue + with path.open(encoding="utf-8", errors="replace") as f: + f.seek(pos) + new = f.read() + pos = f.tell() + for line in new.splitlines(): + line = line.strip() + if not line: + continue + r = _parse(line) + if r and _matches(r, job=job, stage=stage): + print(json.dumps(r) if as_json else _human_line(r), flush=True) + except KeyboardInterrupt: + pass + + +def main() -> None: + import argparse + + p = argparse.ArgumentParser( + description="Tail the SOLVENT structured event log.", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + p.add_argument("-n", "--lines", type=int, default=20, metavar="N", + help="number of recent lines to show (default 20)") + p.add_argument("-f", "--follow", action="store_true", + help="follow the log live (like tail -f)") + p.add_argument("--job", metavar="ID", + help="filter to job ID (prefix match)") + p.add_argument("--stage", metavar="STAGE", + help="filter to a specific pipeline stage") + p.add_argument("--json", dest="as_json", action="store_true", + help="emit raw JSON lines") + p.add_argument("--path", action="store_true", + help="print the log file path and exit") + args = p.parse_args() + + if args.path: + print(log_path()) + sys.exit(0) + + show_logs( + n=args.lines, + job=args.job, + stage=args.stage, + as_json=args.as_json, + follow=args.follow, + ) + + +if __name__ == "__main__": + main() diff --git a/tests/test_logs.py b/tests/test_logs.py new file mode 100644 index 0000000..b76716d --- /dev/null +++ b/tests/test_logs.py @@ -0,0 +1,251 @@ +"""Tests for solvent/logs.py.""" + +from __future__ import annotations + +import json +import sys +import time +from io import StringIO +from pathlib import Path +from unittest import mock + +import pytest + +from solvent.logs import ( + _fmt_duration, + _fmt_margin, + _fmt_ts, + _human_line, + _matches, + _parse, + _tail_lines, + show_logs, +) + + +# --------------------------------------------------------------------------- +# Formatting helpers +# --------------------------------------------------------------------------- + +def test_fmt_ts_returns_string(): + result = _fmt_ts(time.time()) + assert ":" in result and len(result) == 8 # HH:MM:SS + + +def test_fmt_duration_ms(): + assert "ms" in _fmt_duration(500) + + +def test_fmt_duration_seconds(): + assert "s" in _fmt_duration(2500) + assert "ms" not in _fmt_duration(2500) + + +def test_fmt_margin(): + assert "50.0%" == _fmt_margin(0.5) + + +# --------------------------------------------------------------------------- +# _parse +# --------------------------------------------------------------------------- + +def test_parse_valid_json(): + r = _parse('{"ts": 1.0, "stage": "fulfill"}') + assert r == {"ts": 1.0, "stage": "fulfill"} + + +def test_parse_invalid_returns_none(): + assert _parse("not json") is None + + +def test_parse_empty_string(): + assert _parse("") is None + + +# --------------------------------------------------------------------------- +# _matches +# --------------------------------------------------------------------------- + +def test_matches_no_filter(): + assert _matches({"job_id": "abc", "stage": "x"}, job=None, stage=None) + + +def test_matches_job_prefix(): + assert _matches({"job_id": "abc123"}, job="abc", stage=None) + assert not _matches({"job_id": "xyz"}, job="abc", stage=None) + + +def test_matches_stage(): + assert _matches({"stage": "fulfill"}, job=None, stage="fulfill") + assert not _matches({"stage": "quote"}, job=None, stage="fulfill") + + +def test_matches_both_filters(): + r = {"job_id": "abc", "stage": "fulfill"} + assert _matches(r, job="abc", stage="fulfill") + assert not _matches(r, job="abc", stage="quote") + + +# --------------------------------------------------------------------------- +# _human_line +# --------------------------------------------------------------------------- + +def test_human_line_basic(): + r = {"ts": time.time(), "job_id": "deadbeef", "stage": "quote"} + line = _human_line(r) + assert "deadbee" in line + assert "quote" in line + + +def test_human_line_with_duration(): + r = {"ts": time.time(), "job_id": "abc", "stage": "fulfill", "duration_ms": 1500} + line = _human_line(r) + assert "1.5s" in line + + +def test_human_line_with_margin(): + r = {"ts": time.time(), "job_id": "abc", "stage": "fulfill", "margin_actual": 0.3} + line = _human_line(r) + assert "30.0%" in line + + +def test_human_line_with_stripe(): + r = {"ts": time.time(), "job_id": "abc", "stage": "charge", "stripe_ref": "pi_123456789012345"} + line = _human_line(r) + assert "pi_123456789" in line + + +def test_human_line_extra_fields(): + r = {"ts": time.time(), "job_id": "abc", "stage": "x", "customer": "test@x.com"} + line = _human_line(r) + assert "customer=test@x.com" in line + + +# --------------------------------------------------------------------------- +# _tail_lines +# --------------------------------------------------------------------------- + +def test_tail_lines_nonexistent(tmp_path): + assert _tail_lines(tmp_path / "missing.log", 10) == [] + + +def test_tail_lines_empty(tmp_path): + p = tmp_path / "log" + p.write_text("") + assert _tail_lines(p, 10) == [] + + +def test_tail_lines_returns_last_n(tmp_path): + p = tmp_path / "log" + p.write_text("\n".join(f"line{i}" for i in range(50)) + "\n") + result = _tail_lines(p, 5) + assert result == [f"line{i}" for i in range(45, 50)] + + +def test_tail_lines_fewer_than_n(tmp_path): + p = tmp_path / "log" + p.write_text("line1\nline2\n") + result = _tail_lines(p, 10) + assert result == ["line1", "line2"] + + +# --------------------------------------------------------------------------- +# show_logs +# --------------------------------------------------------------------------- + +def _make_log(tmp_path: Path, records: list[dict]) -> Path: + p = tmp_path / "solvent.log" + with p.open("w") as f: + for r in records: + f.write(json.dumps(r) + "\n") + return p + + +def _capture_show(tmp_path, **kwargs) -> str: + with mock.patch("solvent.logs.log_path", return_value=tmp_path / "solvent.log"): + buf = StringIO() + with mock.patch("sys.stdout", buf): + show_logs(**kwargs) + return buf.getvalue() + + +def test_show_logs_empty_file(tmp_path): + _make_log(tmp_path, []) + out = _capture_show(tmp_path) + assert "no" in out.lower() + + +def test_show_logs_no_file(tmp_path): + out = _capture_show(tmp_path) + assert "no" in out.lower() + + +def test_show_logs_human_readable(tmp_path): + _make_log(tmp_path, [{"ts": time.time(), "job_id": "abc123", "stage": "fulfill"}]) + out = _capture_show(tmp_path, n=5) + assert "abc123" in out + assert "fulfill" in out + + +def test_show_logs_json_mode(tmp_path): + record = {"ts": time.time(), "job_id": "abc", "stage": "quote"} + _make_log(tmp_path, [record]) + out = _capture_show(tmp_path, n=5, as_json=True) + parsed = json.loads(out.strip()) + assert parsed["stage"] == "quote" + + +def test_show_logs_job_filter(tmp_path): + records = [ + {"ts": time.time(), "job_id": "abc", "stage": "quote"}, + {"ts": time.time(), "job_id": "xyz", "stage": "fulfill"}, + ] + _make_log(tmp_path, records) + out = _capture_show(tmp_path, n=10, job="abc") + assert "abc" in out + assert "xyz" not in out + + +def test_show_logs_stage_filter(tmp_path): + records = [ + {"ts": time.time(), "job_id": "abc", "stage": "quote"}, + {"ts": time.time(), "job_id": "abc", "stage": "fulfill"}, + ] + _make_log(tmp_path, records) + out = _capture_show(tmp_path, n=10, stage="quote") + assert "quote" in out + assert "fulfill" not in out + + +def test_show_logs_n_limits_output(tmp_path): + records = [{"ts": time.time(), "job_id": f"j{i}", "stage": "x"} for i in range(10)] + _make_log(tmp_path, records) + out = _capture_show(tmp_path, n=3) + lines = [l for l in out.splitlines() if l.strip()] + assert len(lines) == 3 + + +# --------------------------------------------------------------------------- +# CLI main +# --------------------------------------------------------------------------- + +def test_main_path_flag(tmp_path, capsys): + with mock.patch("solvent.logs.log_path", return_value=tmp_path / "solvent.log"): + with mock.patch("sys.argv", ["solvent", "--path"]): + with pytest.raises(SystemExit) as exc: + from solvent.logs import main + main() + assert exc.value.code == 0 + captured = capsys.readouterr() + assert "solvent.log" in captured.out + + +def test_main_default_runs(tmp_path): + _make_log(tmp_path, [{"ts": time.time(), "job_id": "a", "stage": "b"}]) + with mock.patch("solvent.logs.log_path", return_value=tmp_path / "solvent.log"): + with mock.patch("sys.argv", ["solvent"]): + buf = StringIO() + with mock.patch("sys.stdout", buf): + from solvent.logs import main + main() + assert "b" in buf.getvalue()