manual_slop/session_logger.py

# session_logger.py
"""
Opens timestamped log/script files at startup and keeps them open for the
lifetime of the process.  The next run of the GUI creates new files; the
previous run's files are simply closed when the process exits.
File layout
-----------
logs/
    comms_<ts>.log      - every comms entry (direction/kind/payload) as JSON-L
    toolcalls_<ts>.log  - sequential record of every tool invocation
    clicalls_<ts>.log   - sequential record of every CLI subprocess call
scripts/generated/
    <ts>_<seq:04d>.ps1  - each PowerShell script the AI generated, in order
Where <ts> = YYYYMMDD_HHMMSS of when this session was started.
"""
import atexit
import datetime
import json
import threading
from typing import Any, Optional, TextIO
from pathlib import Path
_LOG_DIR: Path = Path("./logs")
_SCRIPTS_DIR: Path = Path("./scripts/generated")
_ts: str = ""            # session timestamp string  e.g. "20260301_142233"
_session_id: str = ""    # YYYYMMDD_HHMMSS[_Label]
_session_dir: Optional[Path] = None # Path to the sub-directory for this session
_seq: int = 0            # monotonic counter for script files this session
_seq_lock: threading.Lock = threading.Lock()
_comms_fh: Optional[TextIO] = None # file handle: logs/<session_id>/comms.log
_tool_fh: Optional[TextIO] = None  # file handle: logs/<session_id>/toolcalls.log
_api_fh: Optional[TextIO] = None   # file handle: logs/<session_id>/apihooks.log
_cli_fh: Optional[TextIO] = None   # file handle: logs/<session_id>/clicalls.log

def _now_ts() -> str:
 return datetime.datetime.now().strftime("%Y%m%d_%H%M%S")

def open_session(label: Optional[str] = None) -> None:
 """
    Called once at GUI startup.  Creates the log directories if needed and
    opens the log files for this session within a sub-directory.
    """
 global _ts, _session_id, _session_dir, _comms_fh, _tool_fh, _api_fh, _cli_fh, _seq
 if _comms_fh is not None:
  return
 _ts = _now_ts()
 _session_id = _ts
 if label:
  safe_label = "".join(c if c.isalnum() or c in ("-", "_") else "_" for c in label)
  _session_id += f"_{safe_label}"
 _session_dir = _LOG_DIR / _session_id
 _session_dir.mkdir(parents=True, exist_ok=True)
 _SCRIPTS_DIR.mkdir(parents=True, exist_ok=True)
 _seq = 0
 _comms_fh = open(_session_dir / "comms.log", "w", encoding="utf-8", buffering=1)
 _tool_fh = open(_session_dir / "toolcalls.log", "w", encoding="utf-8", buffering=1)
 _api_fh = open(_session_dir / "apihooks.log", "w", encoding="utf-8", buffering=1)
 _cli_fh = open(_session_dir / "clicalls.log", "w", encoding="utf-8", buffering=1)
 _tool_fh.write(f"# Tool-call log — session {_session_id}\n\n")
 _tool_fh.flush()
 _cli_fh.write(f"# CLI Subprocess Call Log — session {_session_id}\n\n")
 _cli_fh.flush()
 try:
  from log_registry import LogRegistry
  registry = LogRegistry(str(_LOG_DIR / "log_registry.toml"))
  registry.register_session(_session_id, str(_session_dir), datetime.datetime.now())
 except Exception as e:
  print(f"Warning: Could not register session in LogRegistry: {e}")
 atexit.register(close_session)

def close_session() -> None:
 """Flush and close all log files. Called on clean exit."""
 global _comms_fh, _tool_fh, _api_fh, _cli_fh, _session_id, _LOG_DIR
 if _comms_fh is None:
  return
 if _comms_fh:
  _comms_fh.close()
  _comms_fh = None
 if _tool_fh:
  _tool_fh.close()
  _tool_fh = None
 if _api_fh:
  _api_fh.close()
  _api_fh = None
 if _cli_fh:
  _cli_fh.close()
  _cli_fh = None
 try:
  from log_registry import LogRegistry
  registry = LogRegistry(str(_LOG_DIR / "log_registry.toml"))
  registry.update_auto_whitelist_status(_session_id)
 except Exception as e:
  print(f"Warning: Could not update auto-whitelist on close: {e}")

def log_api_hook(method: str, path: str, payload: str) -> None:
 """Log an API hook invocation."""
 if _api_fh is None:
  return
 ts_entry = datetime.datetime.now().strftime("%H:%M:%S")
 try:
  _api_fh.write(f"[{ts_entry}] {method} {path} - Payload: {payload}\n")
  _api_fh.flush()
 except Exception:
  pass

def log_comms(entry: dict[str, Any]) -> None:
 """
    Append one comms entry to the comms log file as a JSON-L line.
    Thread-safe (GIL + line-buffered file).
    """
 if _comms_fh is None:
  return
 try:
  _comms_fh.write(json.dumps(entry, ensure_ascii=False, default=str) + "\n")
 except Exception:
  pass

def log_tool_call(script: str, result: str, script_path: Optional[str]) -> Optional[str]:
 """
    Append a tool-call record to the toolcalls log and write the PS1 script to
    scripts/generated/.  Returns the path of the written script file.
    """
 global _seq
 if _tool_fh is None:
  return script_path
 with _seq_lock:
  _seq += 1
  seq = _seq
 ts_entry = datetime.datetime.now().strftime("%H:%M:%S")
 ps1_name = f"{_ts}_{seq:04d}.ps1"
 ps1_path: Optional[Path] = _SCRIPTS_DIR / ps1_name
 try:
  ps1_path.write_text(script, encoding="utf-8")
 except Exception as exc:
  ps1_path = None
  ps1_name = f"(write error: {exc})"
 try:
  _tool_fh.write(
   f"## Call #{seq}  [{ts_entry}]\n"
   f"Script file: {ps1_path}\n\n"
   f"### Result\n\n"
   f"```\n{result}\n```\n\n"
   f"---\n\n"
  )
  _tool_fh.flush()
 except Exception:
  pass
 return str(ps1_path) if ps1_path else None

def log_cli_call(command: str, stdin_content: Optional[str], stdout_content: Optional[str], stderr_content: Optional[str], latency: float) -> None:
 """Log details of a CLI subprocess execution."""
 if _cli_fh is None:
  return
 ts_entry = datetime.datetime.now().strftime("%H:%M:%S")
 try:
  log_data = {
   "timestamp": ts_entry,
   "command": command,
   "stdin": stdin_content,
   "stdout": stdout_content,
   "stderr": stderr_content,
   "latency_sec": latency
  }
  _cli_fh.write(json.dumps(log_data, ensure_ascii=False, default=str) + "\n")
  _cli_fh.flush()
 except Exception:
  pass