feat(taxonomy): Redirect logs and artifacts to dedicated sub-folders

This commit is contained in:
2026-03-01 09:03:02 -05:00
parent 09bedbf4f0
commit 6326546005
10 changed files with 36 additions and 16 deletions

View File

@@ -3,33 +3,39 @@
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/
logs/sessions/
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")
_LOG_DIR: Path = Path("./logs/sessions")
_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
_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")
@@ -42,29 +48,35 @@ def open_session(label: Optional[str] = None) -> None:
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:
@@ -72,6 +84,7 @@ def close_session() -> None:
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
@@ -84,6 +97,7 @@ def close_session() -> None:
if _cli_fh:
_cli_fh.close()
_cli_fh = None
try:
from log_registry import LogRegistry
registry = LogRegistry(str(_LOG_DIR / "log_registry.toml"))
@@ -122,17 +136,21 @@ def log_tool_call(script: str, result: str, script_path: Optional[str]) -> Optio
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"
@@ -144,6 +162,7 @@ def log_tool_call(script: str, result: str, script_path: Optional[str]) -> Optio
_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: