manual_slop/src/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/sessions/<session_id>/
    comms.log          - every comms entry (direction/kind/payload) as JSON-L
    toolcalls.log      - sequential record of every tool invocation
    apihooks.log       - sequential record of every API hook call
    clicalls.log       - sequential record of every CLI subprocess call
    scripts/           - subdir containing the AI-generated PowerShell scripts
    outputs/           - subdir containing tool outputs saved via log_tool_output()

scripts/generated/
    <ts>_<seq:04d>.ps1  - top-level copy of every PowerShell script the AI
                          generated, in order. The session_dir/scripts/ subdir
                          also gets a copy per session.

Where <ts> = YYYYMMDD_HHMMSS of when this session was started, and
<session_id> = <ts>[_<Label>] (the optional label is sanitized to
alnum + dash + underscore via open_session(label=...)).

Note: the FILENAMES are plain (comms.log, toolcalls.log, apihooks.log,
clicalls.log) — the <ts>/session_id is the PARENT DIRECTORY, not a filename
prefix. The docstring above previously said comms_<ts>.log which is
incorrect since commit 73e1a36d restructured to per-session subdirs.
"""

import atexit
import datetime
import json
import threading

from pathlib import Path
from typing  import Any, Optional, TextIO

from src import paths
from src.result_types import Result, ErrorInfo, ErrorKind


_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
_output_seq:      int = 0               # monotonic counter for output files this session
_seq_lock:        threading.Lock = threading.Lock()
_output_seq_lock: threading.Lock = threading.Lock()

_comms_fh: Optional[TextIO] = None # file handle: logs/sessions/<session_id>/comms.log
_tool_fh:  Optional[TextIO] = None # file handle: logs/sessions/<session_id>/toolcalls.log
_api_fh:   Optional[TextIO] = None # file handle: logs/sessions/<session_id>/apihooks.log
_cli_fh:   Optional[TextIO] = None # file handle: logs/sessions/<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.
 [C: tests/test_app_controller_offloading.py:tmp_session_dir, tests/test_logging_e2e.py:test_logging_e2e, tests/test_session_logger_optimization.py:test_log_tool_call_saves_in_session_scripts, tests/test_session_logger_optimization.py:test_log_tool_output_saves_in_session_outputs, tests/test_session_logger_optimization.py:test_session_directory_and_subdirectories_creation, tests/test_session_logger_reset.py:test_reset_session, tests/test_session_logging.py:test_open_session_creates_subdir_and_registry]
 """
 global _ts, _session_id, _session_dir, _comms_fh, _tool_fh, _api_fh, _cli_fh, _seq, _output_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 = paths.get_logs_dir() / _session_id
 _session_dir.mkdir(parents=True, exist_ok=True)
 (_session_dir / "scripts").mkdir(exist_ok=True)
 (_session_dir / "outputs").mkdir(exist_ok=True)
 
 paths.get_scripts_dir().mkdir(parents=True, exist_ok=True)

 _seq        = 0
 _output_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 src.log_registry import LogRegistry

  registry = LogRegistry(str(paths.get_logs_dir() / "log_registry.toml"))
  registry.register_session(_session_id, str(_session_dir), datetime.datetime.now())
 except (OSError, KeyError, AttributeError, TypeError) 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.
 [C: tests/test_app_controller_offloading.py:tmp_session_dir, tests/test_logging_e2e.py:e2e_setup, tests/test_logging_e2e.py:test_logging_e2e, tests/test_session_logger_optimization.py:temp_session_setup, tests/test_session_logger_reset.py:temp_logs, tests/test_session_logging.py:temp_logs]
 """
 global _comms_fh, _tool_fh, _api_fh, _cli_fh, _session_id
 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 src.log_registry import LogRegistry

  registry = LogRegistry(str(paths.get_logs_dir() / "log_registry.toml"))
  registry.update_auto_whitelist_status(_session_id)
 except (OSError, KeyError, AttributeError, TypeError) as e:
  print(f"Warning: Could not update auto-whitelist on close: {e}")

def reset_session(label: Optional[str] = None) -> None:
 """Closes the current session and opens a new one with the given label."""
 close_session()
 open_session(label)

def log_api_hook(method: str, path: str, payload: str) -> Result[bool]:
 """Log an API hook invocation."""
 if _api_fh is None:
  return Result(data=False)
 ts_entry = datetime.datetime.now().strftime("%H:%M:%S")
 try:
  _api_fh.write(f"[{ts_entry}] {method} {path} - Payload: {payload}\n")
  _api_fh.flush()
  return Result(data=True)
 except (OSError, UnicodeEncodeError, ValueError) as e:
  return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="session_logger.log_api_hook", original=e)])

def log_comms(entry: dict[str, Any]) -> Result[bool]:
 """
 Append one comms entry to the comms log file as a JSON-L line.
 Thread-safe (GIL + line-buffered file).
 [C: tests/test_logging_e2e.py:test_logging_e2e]
 """
 if _comms_fh is None:
  return Result(data=False)
 try:
  _comms_fh.write(json.dumps(entry, ensure_ascii=False, default=str) + "\n")
  return Result(data=True)
 except (OSError, TypeError, ValueError) as e:
  return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="session_logger.log_comms", original=e)])

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
 the session's scripts directory. Returns the path of the written script file.
 [C: tests/test_session_logger_optimization.py:test_log_tool_call_saves_in_session_scripts]
 """
 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"script_{seq:04d}.ps1"
 
 if _session_dir:
  ps1_path: Optional[Path] = _session_dir / "scripts" / ps1_name
 else:
  ps1_path = paths.get_scripts_dir() / f"{_ts}_{seq:04d}.ps1"

 try:
  if ps1_path:
   ps1_path.write_text(script, encoding="utf-8")
 except (OSError, UnicodeEncodeError) as exc:
  _write_err = Result(data=None, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"write error: {exc}", source="session_logger.log_tool_call.write_ps1", original=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()
  return Result(data=str(ps1_path) if ps1_path else None)
 except (OSError, UnicodeEncodeError, ValueError) as e:
  return Result(data=None, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="session_logger.log_tool_call", original=e)])

 return str(ps1_path) if ps1_path else None

def log_tool_output(content: str) -> Optional[str]:
 """
 Save tool output content to a unique file in the session's outputs directory.
 Returns the path of the written file.
 [C: tests/test_session_logger_optimization.py:test_log_tool_output_returns_none_if_no_session, tests/test_session_logger_optimization.py:test_log_tool_output_saves_in_session_outputs]
 """
 global _output_seq
 if _session_dir is None:
  return None

 with _output_seq_lock:
  _output_seq += 1
  seq = _output_seq

 out_name = f"output_{seq:04d}.txt"
 out_path = _session_dir / "outputs" / out_name

 try:
  out_path.write_text(content, encoding="utf-8")
  return str(out_path)
 except (OSError, UnicodeEncodeError):
  return None

def log_cli_call(command: str, stdin_content: Optional[str], stdout_content: Optional[str], stderr_content: Optional[str], latency: float) -> Result[bool]:
 """Log details of a CLI subprocess execution."""
 if _cli_fh is None:
  return Result(data=False)
 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()
  return Result(data=True)
 except (OSError, TypeError, ValueError) as e:
  return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="session_logger.log_cli_call", original=e)])