diff --git a/src/file_cache.py b/src/file_cache.py index 7698e038..3de24c18 100644 --- a/src/file_cache.py +++ b/src/file_cache.py @@ -91,12 +91,11 @@ class ASTParser: tree = self.parse(code) if len(_ast_cache) >= 10: - # Simple LRU: remove the first added entry - try: - first_key = next(iter(_ast_cache)) - del _ast_cache[first_key] - except StopIteration: - pass + # Simple LRU: remove the first added entry. + # next(iter(...)) is guaranteed to succeed because we just + # checked len(_ast_cache) >= 10; no try/except needed. + first_key = next(iter(_ast_cache)) + del _ast_cache[first_key] _ast_cache[path] = (mtime, tree) return tree diff --git a/src/session_logger.py b/src/session_logger.py index 9f8ffe5f..d835bc1c 100644 --- a/src/session_logger.py +++ b/src/session_logger.py @@ -38,6 +38,7 @@ 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" @@ -136,29 +137,31 @@ def reset_session(label: Optional[str] = None) -> None: close_session() 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.""" if _api_fh is None: - return + 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() - except (OSError, UnicodeEncodeError, ValueError): - pass + 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]) -> None: +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 + return Result(data=False) try: _comms_fh.write(json.dumps(entry, ensure_ascii=False, default=str) + "\n") - except (OSError, TypeError, ValueError): - pass + 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]: """ @@ -198,8 +201,9 @@ def log_tool_call(script: str, result: str, script_path: Optional[str]) -> Optio f"---\n\n" ) _tool_fh.flush() - except (OSError, UnicodeEncodeError, ValueError): - pass + 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 @@ -226,21 +230,22 @@ def log_tool_output(content: str) -> Optional[str]: 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) -> 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 + 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, + "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 (OSError, TypeError, ValueError): - pass + 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)]) diff --git a/tests/test_session_logger_optimization.py b/tests/test_session_logger_optimization.py index 02b78f9b..0f1d2638 100644 --- a/tests/test_session_logger_optimization.py +++ b/tests/test_session_logger_optimization.py @@ -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'" 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) + # Call log_tool_call with script_path=None. Returns Result[Optional[str]]. + ps1_result = session_logger.log_tool_call(script_content, result_content, None) + assert ps1_result.ok, f"log_tool_call failed: {ps1_result.errors}" + assert ps1_result.data is not None + + ps1_path = Path(ps1_result.data) 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" + ps1_result_2 = session_logger.log_tool_call("Get-Date", "2026-03-08", None) + assert ps1_result_2.ok, f"log_tool_call failed: {ps1_result_2.errors}" + 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: log_dir, _ = temp_session_setup