| """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
|
|
|
|
|
| _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()
|
|
|
| self._fp = path.open("a", encoding="utf-8", buffering=1)
|
| self.event(
|
| "session_start",
|
| session_id=session_id,
|
| log_path=str(path),
|
| )
|
|
|
|
|
| @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)
|
|
|
|
|
| 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()
|
|
|
|
|
| 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."""
|
|
|
| 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", ""),
|
| )
|
|
|
|
|
| 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
|
|
|
|
|
| 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)
|
|
|
