diff --git a/.gitignore b/.gitignore index 8e9d1a7..55fc462 100644 --- a/.gitignore +++ b/.gitignore @@ -326,4 +326,3 @@ dist # .nfs files are created when an open file is removed but is still being accessed .nfs* - diff --git a/README.md b/README.md index 74820ba..721c14d 100644 --- a/README.md +++ b/README.md @@ -295,6 +295,21 @@ The backend logs: This is intentional because the app is designed as a game/demo environment rather than a secure production system. +For log analysis, the repository also includes `backend/log_report.py`. +It reads the JSONL backend log and prints a session-oriented report to `stdout`: + +- one block per `session_id` +- full request/response history for that session +- filter results attached to each request +- `guessed: yes` or `guessed: no` at the session level + +Example: + +```bash +cd backend +uv run python log_report.py logs/backend.log +``` + ## Notes - `web-out/` is generated build output and should be rebuilt after frontend changes diff --git a/backend/log_report.py b/backend/log_report.py new file mode 100644 index 0000000..af5fca7 --- /dev/null +++ b/backend/log_report.py @@ -0,0 +1,304 @@ +from __future__ import annotations + +import argparse +import json +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + + +@dataclass +class Exchange: + timestamp: str + session_id: str + level_id: int | None + model_endpoint: str + found_answer: bool = False + user_text: str = "" + checks: list[str] = field(default_factory=list) + model_reply: str | None = None + outcome: str | None = None + + +@dataclass +class SessionLog: + session_id: str + started_at: str + model_endpoint: str + guessed: bool = False + exchanges: list[Exchange] = field(default_factory=list) + + +TERMINAL_EVENTS = { + "password_guessed", + "outgoing_response", + "request_blocked", + "response_blocked", +} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Print detailed session logs from backend JSONL logs." + ) + parser.add_argument( + "log_path", + nargs="?", + default="backend/logs/backend.log", + help="Path to backend JSONL log file.", + ) + return parser.parse_args() + + +def format_model_endpoint(settings: dict[str, Any]) -> str: + llm_settings = settings.get("llm", {}) + model = llm_settings.get("model", "unknown-model") + base_url = llm_settings.get("base_url", "unknown-endpoint") + return f"{model} @ {base_url}" + + +def format_check(entry: dict[str, Any]) -> str: + status = "triggered" if entry.get("triggered") else "passed" + kind = entry.get("check_kind", "unknown_check") + reason = entry.get("reason", "") + if reason: + return f"{kind}: {status} ({reason})" + return f"{kind}: {status}" + + +def format_outcome(event: str, entry: dict[str, Any]) -> str: + if event == "password_guessed": + return "password_guessed" + if event == "outgoing_response": + return "response_returned" + if event == "request_blocked": + reason = entry.get("reason", "") + return f"request_blocked ({reason})" if reason else "request_blocked" + if event == "response_blocked": + reason = entry.get("reason", "") + return f"response_blocked ({reason})" if reason else "response_blocked" + return event + + +def assign_check( + entry: dict[str, Any], + checks_by_session: dict[str, list[str]], + check_queue: list[str], +) -> None: + formatted_check = format_check(entry) + session_id = entry.get("session_id") + if session_id: + checks_by_session.setdefault(session_id, []).append(formatted_check) + return + check_queue.append(formatted_check) + + +def start_exchange( + entry: dict[str, Any], + current_model_endpoint: str, + open_exchanges: dict[str, Exchange], + exchanges: list[Exchange], +) -> None: + session_id = entry.get("session_id") + if not session_id: + return + + exchange = Exchange( + timestamp=entry.get("timestamp", ""), + session_id=session_id, + level_id=entry.get("level_id"), + model_endpoint=current_model_endpoint, + user_text=entry.get("user_text", ""), + ) + open_exchanges[session_id] = exchange + exchanges.append(exchange) + + +def finalize_exchange( + entry: dict[str, Any], + event: str, + open_exchanges: dict[str, Exchange], + checks_by_session: dict[str, list[str]], + check_queue: list[str], +) -> None: + session_id = entry.get("session_id") + if not session_id: + return + + exchange = open_exchanges.get(session_id) + if exchange is None: + return + + if session_id in checks_by_session: + exchange.checks.extend(checks_by_session.pop(session_id)) + elif check_queue: + exchange.checks.extend(check_queue) + check_queue.clear() + + exchange.found_answer = event == "password_guessed" + exchange.outcome = format_outcome(event, entry) + + if event in {"outgoing_response", "response_blocked"}: + exchange.model_reply = entry.get("agent_reply") + elif event == "password_guessed": + exchange.model_reply = "Пароль угадан. Сессия считается успешно пройденной." + + del open_exchanges[session_id] + + +def finalize_open_exchanges( + exchanges: list[Exchange], + open_exchanges: dict[str, Exchange], + checks_by_session: dict[str, list[str]], + check_queue: list[str], +) -> None: + for session_id, checks in checks_by_session.items(): + exchange = open_exchanges.get(session_id) + if exchange is not None: + exchange.checks.extend(checks) + + if check_queue: + for exchange in reversed(exchanges): + if exchange.session_id in open_exchanges: + exchange.checks.extend(check_queue) + break + + for exchange in exchanges: + if exchange.session_id in open_exchanges and exchange.outcome is None: + exchange.outcome = "incomplete" + + +def parse_log(log_path: Path) -> list[Exchange]: + exchanges: list[Exchange] = [] + current_model_endpoint = "unknown-model @ unknown-endpoint" + open_exchanges: dict[str, Exchange] = {} + check_queue: list[str] = [] + checks_by_session: dict[str, list[str]] = {} + + with log_path.open("r", encoding="utf-8") as handle: + for raw_line in handle: + line = raw_line.strip() + if not line: + continue + + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + + event = entry.get("event") + + if event == "settings_loaded": + current_model_endpoint = format_model_endpoint( + entry.get("settings", {}) + ) + continue + + if event == "filter_checked": + assign_check(entry, checks_by_session, check_queue) + continue + + if event == "incoming_request": + start_exchange( + entry, + current_model_endpoint, + open_exchanges, + exchanges, + ) + continue + + if event not in TERMINAL_EVENTS: + continue + + finalize_exchange( + entry, + event, + open_exchanges, + checks_by_session, + check_queue, + ) + + finalize_open_exchanges( + exchanges, + open_exchanges, + checks_by_session, + check_queue, + ) + + return exchanges + + +def group_sessions(exchanges: list[Exchange]) -> list[SessionLog]: + sessions: dict[str, SessionLog] = {} + ordered_sessions: list[SessionLog] = [] + + for exchange in exchanges: + session = sessions.get(exchange.session_id) + if session is None: + session = SessionLog( + session_id=exchange.session_id, + started_at=exchange.timestamp, + model_endpoint=exchange.model_endpoint, + ) + sessions[exchange.session_id] = session + ordered_sessions.append(session) + + session.exchanges.append(exchange) + session.guessed = session.guessed or exchange.found_answer + + return ordered_sessions + + +def render_exchange(exchange: Exchange) -> str: + lines = [f"{exchange.timestamp or '-'}", f"- User: {exchange.user_text}"] + if exchange.checks: + lines.extend(f"- Check: {check}" for check in exchange.checks) + else: + lines.append("- Check: none") + + if exchange.outcome: + lines.append(f"- Outcome: {exchange.outcome}") + + if exchange.model_reply is not None: + lines.append(f"- Model: {exchange.model_reply}") + else: + lines.append("- Model: ") + + return "\n".join(lines) + + +def render_session(session: SessionLog) -> str: + header = ", ".join( + [ + session.started_at or "-", + session.session_id, + session.model_endpoint, + f"guessed: {'yes' if session.guessed else 'no'}", + ] + ) + rendered_exchanges = "\n--\n".join( + render_exchange(exchange) for exchange in session.exchanges + ) + return f"{header}\n{rendered_exchanges}" + + +def main() -> int: + args = parse_args() + log_path = Path(args.log_path) + + if not log_path.exists(): + print(f"Log file not found: {log_path}", file=sys.stderr) + return 1 + + exchanges = parse_log(log_path) + sessions = group_sessions(exchanges) + output = "\n\n".join(render_session(session) for session in sessions) + if output: + print(output) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/tests/test_log_report.py b/backend/tests/test_log_report.py new file mode 100644 index 0000000..7b7851c --- /dev/null +++ b/backend/tests/test_log_report.py @@ -0,0 +1,280 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from log_report import group_sessions, parse_log, render_session + + +def write_jsonl(path: Path, entries: list[dict]) -> None: + path.write_text( + "\n".join(json.dumps(entry, ensure_ascii=False) for entry in entries), + encoding="utf-8", + ) + + +def test_parse_log_groups_requests_checks_and_responses(tmp_path: Path) -> None: + log_path = tmp_path / "backend.log" + write_jsonl( + log_path, + [ + { + "settings": { + "llm": { + "model": "test-model", + "base_url": "http://llm.local/v1", + } + }, + "event": "settings_loaded", + "timestamp": "2026-05-09T10:00:00Z", + }, + { + "session_id": "s1", + "password": "cloud", + "event": "session_created", + "timestamp": "2026-05-09T10:00:01Z", + }, + { + "level_id": 2, + "session_id": "s1", + "hard_mode": False, + "rotated": False, + "user_text": "hello", + "event": "incoming_request", + "timestamp": "2026-05-09T10:00:02Z", + }, + { + "session_id": "s1", + "check_kind": "assistant_response", + "triggered": False, + "reason": "safe", + "event": "filter_checked", + "timestamp": "2026-05-09T10:00:03Z", + }, + { + "level_id": 2, + "session_id": "s1", + "agent_reply": "hi there", + "event": "outgoing_response", + "timestamp": "2026-05-09T10:00:04Z", + }, + { + "level_id": 2, + "session_id": "s1", + "hard_mode": False, + "rotated": False, + "user_text": "cloud", + "event": "incoming_request", + "timestamp": "2026-05-09T10:00:05Z", + }, + { + "level_id": 2, + "session_id": "s1", + "password": "cloud", + "event": "password_guessed", + "timestamp": "2026-05-09T10:00:06Z", + }, + ], + ) + + exchanges = parse_log(log_path) + + assert len(exchanges) == 2 + + first = exchanges[0] + assert first.session_id == "s1" + assert first.model_endpoint == "test-model @ http://llm.local/v1" + assert first.user_text == "hello" + assert first.found_answer is False + assert first.checks == ["assistant_response: passed (safe)"] + assert first.model_reply == "hi there" + assert first.outcome == "response_returned" + + second = exchanges[1] + assert second.user_text == "cloud" + assert second.found_answer is True + assert second.model_reply == "Пароль угадан. Сессия считается успешно пройденной." + assert second.outcome == "password_guessed" + + +def test_render_session_includes_history_and_guessed_status(tmp_path: Path) -> None: + log_path = tmp_path / "backend.log" + write_jsonl( + log_path, + [ + { + "settings": { + "llm": { + "model": "test-model", + "base_url": "http://llm.local/v1", + } + }, + "event": "settings_loaded", + "timestamp": "2026-05-09T10:00:00Z", + }, + { + "level_id": 3, + "session_id": "s2", + "hard_mode": False, + "rotated": False, + "user_text": "probe", + "event": "incoming_request", + "timestamp": "2026-05-09T10:00:02Z", + }, + { + "session_id": "s2", + "check_kind": "user_request", + "triggered": True, + "reason": "secret request", + "event": "filter_checked", + "timestamp": "2026-05-09T10:00:03Z", + }, + { + "level_id": 3, + "session_id": "s2", + "reason": "secret request", + "event": "request_blocked", + "timestamp": "2026-05-09T10:00:04Z", + }, + { + "level_id": 3, + "session_id": "s2", + "hard_mode": False, + "rotated": False, + "user_text": "probe", + "event": "incoming_request", + "timestamp": "2026-05-09T10:00:02Z", + }, + { + "level_id": 3, + "session_id": "s2", + "password": "cloud", + "event": "password_guessed", + "timestamp": "2026-05-09T10:00:05Z", + }, + ], + ) + + sessions = group_sessions(parse_log(log_path)) + rendered = render_session(sessions[0]) + + assert ( + "2026-05-09T10:00:02Z, s2, test-model @ http://llm.local/v1, guessed: yes" + in rendered + ) + assert "- User: probe" in rendered + assert "- Check: user_request: triggered (secret request)" in rendered + assert "- Outcome: request_blocked (secret request)" in rendered + assert "- Model: " in rendered + assert "--" in rendered + assert "- Outcome: password_guessed" in rendered + + +def test_parse_log_keeps_incomplete_requests(tmp_path: Path) -> None: + log_path = tmp_path / "backend.log" + write_jsonl( + log_path, + [ + { + "settings": { + "llm": { + "model": "test-model", + "base_url": "http://llm.local/v1", + } + }, + "event": "settings_loaded", + "timestamp": "2026-05-09T10:00:00Z", + }, + { + "level_id": 4, + "session_id": "s3", + "hard_mode": False, + "rotated": False, + "user_text": "unfinished", + "event": "incoming_request", + "timestamp": "2026-05-09T10:00:02Z", + }, + { + "check_kind": "user_request", + "triggered": False, + "reason": "ok", + "event": "filter_checked", + "timestamp": "2026-05-09T10:00:03Z", + }, + ], + ) + + exchange = parse_log(log_path)[0] + + assert exchange.user_text == "unfinished" + assert exchange.checks == ["user_request: passed (ok)"] + assert exchange.outcome == "incomplete" + assert exchange.model_reply is None + + +def test_parse_log_prefers_filter_check_session_id_over_global_order( + tmp_path: Path, +) -> None: + log_path = tmp_path / "backend.log" + write_jsonl( + log_path, + [ + { + "settings": { + "llm": { + "model": "test-model", + "base_url": "http://llm.local/v1", + } + }, + "event": "settings_loaded", + "timestamp": "2026-05-09T10:00:00Z", + }, + { + "level_id": 4, + "session_id": "sA", + "hard_mode": False, + "rotated": False, + "user_text": "first", + "event": "incoming_request", + "timestamp": "2026-05-09T10:00:02Z", + }, + { + "level_id": 4, + "session_id": "sB", + "hard_mode": False, + "rotated": False, + "user_text": "second", + "event": "incoming_request", + "timestamp": "2026-05-09T10:00:03Z", + }, + { + "session_id": "sB", + "check_kind": "assistant_response", + "triggered": False, + "reason": "belongs to second", + "event": "filter_checked", + "timestamp": "2026-05-09T10:00:04Z", + }, + { + "level_id": 4, + "session_id": "sA", + "agent_reply": "reply first", + "event": "outgoing_response", + "timestamp": "2026-05-09T10:00:05Z", + }, + { + "level_id": 4, + "session_id": "sB", + "agent_reply": "reply second", + "event": "outgoing_response", + "timestamp": "2026-05-09T10:00:06Z", + }, + ], + ) + + exchanges = parse_log(log_path) + + assert exchanges[0].session_id == "sA" + assert exchanges[0].checks == [] + assert exchanges[1].session_id == "sB" + assert exchanges[1].checks == ["assistant_response: passed (belongs to second)"]