refactor(src): Phase 10.2 batch 1 - session_logger + file_cache Result[T] migration

Migrates 5 SILENT_SWALLOW sites to full Result[T] pattern:

session_logger.py (4 sites):
1. log_api_hook - returns Result[bool] (was None)
2. log_comms - returns Result[bool] (was None)
3. log_tool_call - returns Result[Optional[str]] (was Optional[str])
4. log_cli_call - returns Result[bool] (was None)

file_cache.py (1 site):
- L98: removed dead code (try/except StopIteration around
  next(iter(_ast_cache)) is unreachable because we just checked
  len(_ast_cache) >= 10)

Updates tests/test_session_logger_optimization.py to extract
result.data from the new Result-based API.

All callers of these log_* functions previously ignored the
return value; they continue to ignore the new Result return
value (backwards-compatible).
This commit is contained in:
ed
2026-06-17 22:29:36 -04:00
parent 15b778485c
commit 0f5290f038
3 changed files with 37 additions and 31 deletions
+5 -6
View File
@@ -91,12 +91,11 @@ class ASTParser:
tree = self.parse(code) tree = self.parse(code)
if len(_ast_cache) >= 10: if len(_ast_cache) >= 10:
# Simple LRU: remove the first added entry # Simple LRU: remove the first added entry.
try: # next(iter(...)) is guaranteed to succeed because we just
first_key = next(iter(_ast_cache)) # checked len(_ast_cache) >= 10; no try/except needed.
del _ast_cache[first_key] first_key = next(iter(_ast_cache))
except StopIteration: del _ast_cache[first_key]
pass
_ast_cache[path] = (mtime, tree) _ast_cache[path] = (mtime, tree)
return tree return tree
+21 -16
View File
@@ -38,6 +38,7 @@ from pathlib import Path
from typing import Any, Optional, TextIO from typing import Any, Optional, TextIO
from src import paths from src import paths
from src.result_types import Result, ErrorInfo, ErrorKind
_ts: str = "" # session timestamp string e.g. "20260301_142233" _ts: str = "" # session timestamp string e.g. "20260301_142233"
@@ -136,29 +137,31 @@ def reset_session(label: Optional[str] = None) -> None:
close_session() close_session()
open_session(label) open_session(label)
def log_api_hook(method: str, path: str, payload: str) -> None: def log_api_hook(method: str, path: str, payload: str) -> Result[bool]:
"""Log an API hook invocation.""" """Log an API hook invocation."""
if _api_fh is None: if _api_fh is None:
return return Result(data=False)
ts_entry = datetime.datetime.now().strftime("%H:%M:%S") ts_entry = datetime.datetime.now().strftime("%H:%M:%S")
try: try:
_api_fh.write(f"[{ts_entry}] {method} {path} - Payload: {payload}\n") _api_fh.write(f"[{ts_entry}] {method} {path} - Payload: {payload}\n")
_api_fh.flush() _api_fh.flush()
except (OSError, UnicodeEncodeError, ValueError): return Result(data=True)
pass 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]) -> None: def log_comms(entry: dict[str, Any]) -> Result[bool]:
""" """
Append one comms entry to the comms log file as a JSON-L line. Append one comms entry to the comms log file as a JSON-L line.
Thread-safe (GIL + line-buffered file). Thread-safe (GIL + line-buffered file).
[C: tests/test_logging_e2e.py:test_logging_e2e] [C: tests/test_logging_e2e.py:test_logging_e2e]
""" """
if _comms_fh is None: if _comms_fh is None:
return return Result(data=False)
try: try:
_comms_fh.write(json.dumps(entry, ensure_ascii=False, default=str) + "\n") _comms_fh.write(json.dumps(entry, ensure_ascii=False, default=str) + "\n")
except (OSError, TypeError, ValueError): return Result(data=True)
pass 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]: def log_tool_call(script: str, result: str, script_path: Optional[str]) -> Optional[str]:
""" """
@@ -198,8 +201,9 @@ def log_tool_call(script: str, result: str, script_path: Optional[str]) -> Optio
f"---\n\n" f"---\n\n"
) )
_tool_fh.flush() _tool_fh.flush()
except (OSError, UnicodeEncodeError, ValueError): return Result(data=str(ps1_path) if ps1_path else None)
pass 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 return str(ps1_path) if ps1_path else None
@@ -226,21 +230,22 @@ def log_tool_output(content: str) -> Optional[str]:
except (OSError, UnicodeEncodeError): except (OSError, UnicodeEncodeError):
return None return None
def log_cli_call(command: str, stdin_content: Optional[str], stdout_content: Optional[str], stderr_content: Optional[str], latency: float) -> 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.""" """Log details of a CLI subprocess execution."""
if _cli_fh is None: if _cli_fh is None:
return return Result(data=False)
ts_entry = datetime.datetime.now().strftime("%H:%M:%S") ts_entry = datetime.datetime.now().strftime("%H:%M:%S")
try: try:
log_data = { log_data = {
"timestamp": ts_entry, "timestamp": ts_entry,
"command": command, "command": command,
"stdin": stdin_content, "stdin": stdin_content,
"stdout": stdout_content, "stdout": stdout_content,
"stderr": stderr_content, "stderr": stderr_content,
"latency_sec": latency "latency_sec": latency
} }
_cli_fh.write(json.dumps(log_data, ensure_ascii=False, default=str) + "\n") _cli_fh.write(json.dumps(log_data, ensure_ascii=False, default=str) + "\n")
_cli_fh.flush() _cli_fh.flush()
except (OSError, TypeError, ValueError): return Result(data=True)
pass 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)])
+11 -9
View File
@@ -51,19 +51,21 @@ def test_log_tool_call_saves_in_session_scripts(temp_session_setup: tuple[Path,
script_content = "Write-Host 'Hello from test'" script_content = "Write-Host 'Hello from test'"
result_content = "Success" result_content = "Success"
# Call log_tool_call with script_path=None # Call log_tool_call with script_path=None. Returns Result[Optional[str]].
ps1_path_str = session_logger.log_tool_call(script_content, result_content, None) ps1_result = session_logger.log_tool_call(script_content, result_content, None)
assert ps1_path_str is not None assert ps1_result.ok, f"log_tool_call failed: {ps1_result.errors}"
assert ps1_result.data is not None
ps1_path = Path(ps1_path_str)
ps1_path = Path(ps1_result.data)
assert ps1_path.parent == scripts_subdir assert ps1_path.parent == scripts_subdir
assert ps1_path.name == "script_0001.ps1" assert ps1_path.name == "script_0001.ps1"
assert ps1_path.read_text(encoding="utf-8") == script_content assert ps1_path.read_text(encoding="utf-8") == script_content
# Verify second call increments sequence # Verify second call increments sequence
ps1_path_str_2 = session_logger.log_tool_call("Get-Date", "2026-03-08", None) ps1_result_2 = session_logger.log_tool_call("Get-Date", "2026-03-08", None)
assert ps1_path_str_2 is not None assert ps1_result_2.ok, f"log_tool_call failed: {ps1_result_2.errors}"
assert Path(ps1_path_str_2).name == "script_0002.ps1" assert ps1_result_2.data is not None
assert Path(ps1_result_2.data).name == "script_0002.ps1"
def test_log_tool_output_saves_in_session_outputs(temp_session_setup: tuple[Path, Path]) -> None: def test_log_tool_output_saves_in_session_outputs(temp_session_setup: tuple[Path, Path]) -> None:
log_dir, _ = temp_session_setup log_dir, _ = temp_session_setup