feat(logs): Implement file-based offloading for scripts and tool outputs

This commit is contained in:
2026-03-08 20:51:27 -04:00
parent 07b0f83794
commit 7063bead12
4 changed files with 263 additions and 5 deletions

View File

@@ -0,0 +1,95 @@
import pytest
from pathlib import Path
from typing import Generator
from src import session_logger
from src import paths
@pytest.fixture
def temp_session_setup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[tuple[Path, Path], None, None]:
# Ensure session is closed and state is reset
session_logger.close_session()
monkeypatch.setattr(session_logger, "_comms_fh", None)
monkeypatch.setattr(session_logger, "_session_dir", None)
monkeypatch.setattr(session_logger, "_seq", 0)
monkeypatch.setattr(session_logger, "_output_seq", 0)
log_dir = tmp_path / "logs"
scripts_dir = tmp_path / "scripts" / "generated"
log_dir.mkdir(parents=True, exist_ok=True)
scripts_dir.mkdir(parents=True, exist_ok=True)
monkeypatch.setattr(paths, "get_logs_dir", lambda: log_dir)
monkeypatch.setattr(paths, "get_scripts_dir", lambda: scripts_dir)
yield log_dir, scripts_dir
# Cleanup
session_logger.close_session()
def test_session_directory_and_subdirectories_creation(temp_session_setup: tuple[Path, Path]) -> None:
log_dir, _ = temp_session_setup
session_logger.open_session(label="opt-test")
# Find the session directory
session_dirs = [d for d in log_dir.iterdir() if d.is_dir()]
assert len(session_dirs) == 1
session_dir = session_dirs[0]
assert (session_dir / "scripts").exists()
assert (session_dir / "outputs").exists()
assert (session_dir / "comms.log").exists()
assert (session_dir / "toolcalls.log").exists()
def test_log_tool_call_saves_in_session_scripts(temp_session_setup: tuple[Path, Path]) -> None:
log_dir, _ = temp_session_setup
session_logger.open_session(label="tool-call-test")
# Find the session directory
session_dir = next(d for d in log_dir.iterdir() if d.is_dir())
scripts_subdir = session_dir / "scripts"
script_content = "Write-Host 'Hello from test'"
result_content = "Success"
# Call log_tool_call with script_path=None
ps1_path_str = session_logger.log_tool_call(script_content, result_content, None)
assert ps1_path_str is not None
ps1_path = Path(ps1_path_str)
assert ps1_path.parent == scripts_subdir
assert ps1_path.name == "script_0001.ps1"
assert ps1_path.read_text(encoding="utf-8") == script_content
# Verify second call increments sequence
ps1_path_str_2 = session_logger.log_tool_call("Get-Date", "2026-03-08", None)
assert ps1_path_str_2 is not None
assert Path(ps1_path_str_2).name == "script_0002.ps1"
def test_log_tool_output_saves_in_session_outputs(temp_session_setup: tuple[Path, Path]) -> None:
log_dir, _ = temp_session_setup
session_logger.open_session(label="output-test")
# Find the session directory
session_dir = next(d for d in log_dir.iterdir() if d.is_dir())
outputs_subdir = session_dir / "outputs"
output_content = "This is some tool output content."
# Call log_tool_output
output_path_str = session_logger.log_tool_output(output_content)
assert output_path_str is not None
output_path = Path(output_path_str)
assert output_path.parent == outputs_subdir
assert output_path.name == "output_0001.txt"
assert output_path.read_text(encoding="utf-8") == output_content
# Verify second call increments sequence
output_path_str_2 = session_logger.log_tool_output("More content")
assert output_path_str_2 is not None
assert Path(output_path_str_2).name == "output_0002.txt"
def test_log_tool_output_returns_none_if_no_session(temp_session_setup: tuple[Path, Path]) -> None:
# We don't call open_session here
output_path_str = session_logger.log_tool_output("Should not save")
assert output_path_str is None