ed
dc397db7ed
TIER-3 READ AGENTS.md + conductor/workflow.md + conductor/code_styleguides/error_handling.md + the 4 source files + 3 test files before this commit. The code_path_audit_phase_2_20260624 track (Tier 2) shipped 11 audit fixes (4 NG1 + 7 NG2) but used a heuristic bypass for 4 of the NG2 wrappers: legacy T | None functions that exist only to maintain test patcher compatibility. Per the review at docs/reports/REVIEW_TIER2_code_path_audit_phase_2_20260624.md Finding 8, this track eliminates the legacy wrappers properly. 11 wrappers eliminated (8 main + 3 _legacy_compat inner): - src/ai_client.py: get_current_tier (1 src + 1 test consumer) - src/ai_client.py: _gemini_tool_declaration + _legacy_compat (2 test consumers) - src/ai_client.py: run_tier4_patch_callback + _legacy_compat (was 0 direct callers but had 2 callback references in app_controller/multi_agent_conductor; callback contract migrated to Callable[[str, str], Result[str]] instead of preserving an Optional[str] adapter) - src/mcp_client.py: _get_symbol_node + _legacy_compat (8 in-file consumers) - src/mcp_client.py: find_in_scope (nested inside _get_symbol_node_result; private impl detail, audit doesn't catch T | None, left as-is) - src/external_editor.py: launch_diff (1 src + 3 test + 1 live_gui test consumer) - src/external_editor.py: launch_editor (no consumers; deleted) - src/session_logger.py: log_tool_output (2 src + 3 test consumers) - src/project_manager.py: parse_ts (no consumers; deleted) For each consumer: replace legacy_fn(args) with legacy_fn_result(args).data. For T | None checks: replace if x is None: with if not result.ok: or if not result.ok or not isinstance(result.data, ...) (depending on pattern). For run_tier4_patch_callback specifically: the wrapper was a callback adapter (not a backward-compat shim) and had 2 callback references as consumers. Rather than keep the adapter (which would re-introduce the Optional[str] return that the strict audit catches), the patch_callback contract was migrated from Callable[[str, str], Optional[str]] to Callable[[str, str], Result[str]] in shell_runner.py + app_controller.py + 9 _send_<vendor>_result signatures in ai_client.py. This propagates the Result[str] through the callback and lets shell_runner unwrap with if r.ok and r.data instead of if patch_text. Verification: - audit_optional_in_3_files --strict: 0 return-type Optional[T] (down from 1) - audit_exception_handling --strict: 0 violations (unchanged) - audit_legacy_wrappers: 0 legacy wrappers (unchanged) - 15 affected test files: 168 tests pass - 8 mcp_client/structural/baseline test files: 55 tests pass - 3 session/gui test files: 7 tests pass - 0 return-type Optional[T] in src/ai_client.py (was 1: run_tier4_patch_callback)
101 lines
3.9 KiB
Python
101 lines
3.9 KiB
Python
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. 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_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
|
|
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_result = session_logger.log_tool_output_result(output_content)
|
|
assert output_result.ok, f"log_tool_output failed: {output_result.errors}"
|
|
assert output_result.data is not None
|
|
|
|
output_path = Path(output_result.data)
|
|
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_result_2 = session_logger.log_tool_output_result("More content")
|
|
assert output_result_2.ok, f"log_tool_output failed: {output_result_2.errors}"
|
|
assert output_result_2.data is not None
|
|
assert Path(output_result_2.data).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_result = session_logger.log_tool_output_result("Should not save")
|
|
assert not output_result.ok
|
|
assert output_result.data is None
|