"""Oturum bazli yapilandirilmis log altyapisi. Tasarim ------- * Her ajan calismasi yeni bir oturumdur ve `logs/session-YYYYMMDD-HHMMSS.jsonl` dosyasina yazilir. Her satir tek bir olaydir (JSONL). * Olay tipleri: - "info" / "warn" / "error" : serbest mesajlar - "command" : SSH uzerinden calistirilan komutun tam sonucu - "model" : modelden donen icerik (thought / response) - "tool_call" : ajanin cagirdigi tool ve argumanlari - "session_start" / "session_end" * Konsola insan-okur formatta (kisaltilmis), dosyaya tam icerik gider. * Thread-safe (basit Lock); ileride paralel komut calistirma istersek hazir. Kullanim -------- log = SessionLogger.create() log.info("ajan basladi") log.command(result) # CommandResult dogrudan loglanir log.tool_call("run_cmd", {"cmd": "nmap -sV ..."}) log.model("nmap baslatiyorum...") log.close() veya context manager: with SessionLogger.create() as log: ... """ from __future__ import annotations import json import sys import threading from dataclasses import asdict, is_dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional from config import LOGS_DIR # ANSI renk kodlari (PowerShell + Windows Terminal modern surumlerde calisir) _C_RESET = "\033[0m" _C_DIM = "\033[2m" _C_RED = "\033[31m" _C_GREEN = "\033[32m" _C_YELLOW = "\033[33m" _C_BLUE = "\033[34m" _C_MAGENTA = "\033[35m" _C_CYAN = "\033[36m" _LEVEL_COLORS: dict[str, str] = { "info": _C_CYAN, "warn": _C_YELLOW, "error": _C_RED, "command": _C_GREEN, "model": _C_MAGENTA, "tool_call": _C_BLUE, "session_start": _C_DIM, "session_end": _C_DIM, } def _utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat(timespec="milliseconds") def _short(text: str, n: int = 240) -> str: """Konsol icin tek satira indir + kirp.""" if not text: return "" flat = text.replace("\n", " | ").strip() return flat if len(flat) <= n else flat[: n - 3] + "..." def _safe_json(value: Any) -> Any: """JSON'a cevrilebilir hale getir (dataclass / Path / set destekler).""" if value is None or isinstance(value, (str, int, float, bool)): return value if is_dataclass(value): return _safe_json(asdict(value)) if isinstance(value, dict): return {str(k): _safe_json(v) for k, v in value.items()} if isinstance(value, (list, tuple, set)): return [_safe_json(v) for v in value] if isinstance(value, Path): return str(value) return repr(value) class SessionLogger: """Tek oturumluk JSONL + konsol logger'i.""" def __init__( self, path: Path, session_id: str, *, echo_to_console: bool = True, ) -> None: self.path = path self.session_id = session_id self.echo = echo_to_console self._lock = threading.Lock() # line buffering ki kill edilse bile son olaylar diske yazilmis olsun self._fp = path.open("a", encoding="utf-8", buffering=1) self.event( "session_start", session_id=session_id, log_path=str(path), ) # --- yapilandirma -------------------------------------------------- @classmethod def create( cls, *, log_dir: Optional[Path] = None, echo_to_console: bool = True, ) -> "SessionLogger": log_dir = log_dir or LOGS_DIR log_dir.mkdir(parents=True, exist_ok=True) sid = datetime.now().strftime("%Y%m%d-%H%M%S") path = log_dir / f"session-{sid}.jsonl" return cls(path=path, session_id=sid, echo_to_console=echo_to_console) # --- yasam dongusu -------------------------------------------------- def __enter__(self) -> "SessionLogger": return self def __exit__(self, exc_type, exc, tb) -> None: if exc_type is not None: self.error(f"oturum istisna ile kapaniyor: {exc_type.__name__}: {exc}") self.close() def close(self) -> None: if self._fp.closed: return try: self.event("session_end", session_id=self.session_id) finally: self._fp.close() # --- ana API -------------------------------------------------------- def event(self, event_type: str, **fields: Any) -> None: record = { "ts": _utc_now_iso(), "session": self.session_id, "type": event_type, **{k: _safe_json(v) for k, v in fields.items()}, } line = json.dumps(record, ensure_ascii=False) with self._lock: if not self._fp.closed: self._fp.write(line + "\n") if self.echo: self._echo(event_type, fields) def info(self, msg: str, **extra: Any) -> None: self.event("info", message=msg, **extra) def warn(self, msg: str, **extra: Any) -> None: self.event("warn", message=msg, **extra) def error(self, msg: str, **extra: Any) -> None: self.event("error", message=msg, **extra) def model(self, content: str, *, role: str = "assistant", **extra: Any) -> None: """Modelden donen icerigi logla (thought + response).""" self.event("model", role=role, content=content, **extra) def tool_call(self, name: str, args: dict[str, Any], **extra: Any) -> None: self.event("tool_call", name=name, args=args, **extra) def command(self, result: Any) -> None: """CommandResult'i (ssh_client) dogrudan logla.""" # Geri dongusel import olmasin diye duck-typing yapiyoruz. self.event( "command", command=getattr(result, "command", None), exit_code=getattr(result, "exit_code", None), duration_s=round(float(getattr(result, "duration_s", 0.0)), 3), stdout=getattr(result, "stdout", ""), stderr=getattr(result, "stderr", ""), ) # --- konsol formatlama --------------------------------------------- def _echo(self, event_type: str, fields: dict[str, Any]) -> None: color = _LEVEL_COLORS.get(event_type, "") ts = datetime.now().strftime("%H:%M:%S") prefix = f"{_C_DIM}{ts}{_C_RESET} {color}{event_type:>13}{_C_RESET}" if event_type == "command": cmd = _short(str(fields.get("command", "")), 120) ec = fields.get("exit_code") dur = fields.get("duration_s") tag = _C_GREEN if ec == 0 else _C_RED print(f"{prefix} {tag}exit={ec}{_C_RESET} {dur}s $ {cmd}") stdout = _short(str(fields.get("stdout", "")), 200) stderr = _short(str(fields.get("stderr", "")), 200) if stdout: print(f" {_C_DIM}stdout:{_C_RESET} {stdout}") if stderr: print(f" {_C_RED}stderr:{_C_RESET} {stderr}") return if event_type == "tool_call": name = fields.get("name") args = _short(json.dumps(_safe_json(fields.get("args", {})), ensure_ascii=False), 200) print(f"{prefix} {name}({args})") return if event_type == "model": role = fields.get("role", "assistant") content = _short(str(fields.get("content", "")), 280) print(f"{prefix} [{role}] {content}") return # info / warn / error / session_* msg = fields.get("message") or fields.get("session_id") or "" extras = {k: v for k, v in fields.items() if k not in {"message", "session_id"}} extras_str = f" {_C_DIM}{extras}{_C_RESET}" if extras else "" print(f"{prefix} {_short(str(msg), 280)}{extras_str}", file=sys.stderr if event_type == "error" else sys.stdout)