Blame for logger.py

"""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)