Private
Public Access
0
0

refactor(src): startup_profiler.py Phase 11.3.2 - extract _log_phase_output

Phase 11.3.2. CONTEXT-MANAGER EXCEPTION.

The plan claimed 'StartupProfiler.phase() is NOT a context manager;
tier-2's claim is factually wrong.' This is incorrect. phase() IS a
context manager:
- Decorated with @contextmanager (src/startup_profiler.py:26)
- Used in 13 'with startup_profiler.phase(...)' call sites in
  src/gui_2.py (lines 308, 311, 327, 338, 343, 627, 629, 631, 669,
  672, 711, 729, 739)

It cannot return Result[None] because:
- @contextmanager requires the function to yield (not return)
- The except body is inside a finally block (which cannot return)

Best partial migration: extract _log_phase_output helper that returns
Result[None]; phase() calls it and ignores the Result (we're in a
finally block).

Audit post-migration:
- _log_phase_output L28 = INTERNAL_COMPLIANT (Heuristic A) ✓
- phase() L54 try/finally = INTERNAL_COMPLIANT (canonical cleanup) ✓

Tests: 12/12 pass (test_audit_allowlist_2d, test_gui_startup_smoke,
test_headless_service, test_startup_profiler, test_warmup_canaries).

This site is documented in the per-site report as a CONTEXT-MANAGER
EXCEPTION. The Heuristic #19 (catch+log) classification remains valid;
the partial migration adds explicit Result-returning helpers where
possible without breaking the context manager pattern.
This commit is contained in:
2026-06-18 00:10:16 -04:00
parent 4c42bd0545
commit 2ed449ee5f
2 changed files with 109 additions and 5 deletions
@@ -0,0 +1,83 @@
"""Phase 11.3.2 partial migration for startup_profiler.py.
CONTEXT-MANAGER EXCEPTION: StartupProfiler.phase() IS a context manager
(decorated with @contextmanager; used in 13 'with profiler.phase(...)'
call sites in src/gui_2.py). It CANNOT return Result[None] from the
except body because @contextmanager requires the function to yield
(not return), and the except body is inside a finally block.
The plan claimed "phase() is NOT a context manager" - this is factually
wrong. We do the best partial migration: extract a Result-returning
helper for the stderr.write, and document the constraint.
The audit classifies the existing site as INTERNAL_COMPLIANT via
Heuristic #19 (catch+log). The plan's rejection was based on the
incorrect assumption that phase() is a regular method.
"""
from __future__ import annotations
from pathlib import Path
p = Path("src/startup_profiler.py")
content = p.read_text(encoding="utf-8")
# 1. Add Result import
old_imports = "import time\nimport sys\nfrom contextlib import contextmanager\nfrom dataclasses import dataclass, field\nfrom typing import Any, Iterator"
new_imports = (
"import time\n"
"import sys\n"
"from contextlib import contextmanager\n"
"from dataclasses import dataclass, field\n"
"from typing import Any, Iterator\n"
"\n"
"from src.result_types import ErrorInfo, ErrorKind, Result"
)
assert old_imports in content, "imports marker not found"
content = content.replace(old_imports, new_imports)
# 2. Add _log_phase_output helper BEFORE @dataclass StartupProfiler
helper = '''
def _log_phase_output(line: str, phase_name: str) -> Result[None]:
"""Best-effort stderr write for phase timing output. Returns Result[None].
Used by phase() (which is a @contextmanager; cannot return Result from
its except body because @contextmanager requires yield, not return, and
the except is in a finally block).
"""
try:
sys.stderr.write(line)
sys.stderr.flush()
return Result(data=None)
except OSError as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"phase output failed for {phase_name}: {e}",
source="startup_profiler._log_phase_output",
original=e,
)])
'''
old_class_marker = "\n\n@dataclass\nclass StartupProfiler:"
new_class_marker = helper + "\n\n@dataclass\nclass StartupProfiler:"
assert old_class_marker in content, "class marker not found"
content = content.replace(old_class_marker, new_class_marker)
# 3. Replace the except body in phase() to use _log_phase_output
old_except = (
" try:\n"
" sys.stderr.write(f\"[startup] {name}: {(p.end_ts - p.start_ts) * 1000.0:.1f}ms\\n\")\n"
" sys.stderr.flush()\n"
" except OSError as e:\n"
" sys.stderr.write(f\"[startup] phase output failed for {name}: {e}\\n\")"
)
new_except = (
" log_line = f\"[startup] {name}: {(p.end_ts - p.start_ts) * 1000.0:.1f}ms\\n\"\n"
" log_result = _log_phase_output(log_line, name)\n"
" if not log_result.ok:\n"
" _log_phase_output(f\"[startup] phase output failed for {name}: {log_result.errors[0].message}\\n\", name)"
)
assert old_except in content, "except marker not found"
content = content.replace(old_except, new_except)
p.write_text(content, encoding="utf-8", newline="")
print("ok")
+26 -5
View File
@@ -4,6 +4,8 @@ from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Any, Iterator
from src.result_types import ErrorInfo, ErrorKind, Result
@dataclass
class _Phase:
@@ -12,6 +14,26 @@ class _Phase:
end_ts: float = 0.0
def _log_phase_output(line: str, phase_name: str) -> Result[None]:
"""Best-effort stderr write for phase timing output. Returns Result[None].
Used by phase() (which is a @contextmanager; cannot return Result from
its except body because @contextmanager requires yield, not return, and
the except is in a finally block).
"""
try:
sys.stderr.write(line)
sys.stderr.flush()
return Result(data=None)
except OSError as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"phase output failed for {phase_name}: {e}",
source="startup_profiler._log_phase_output",
original=e,
)])
@dataclass
class StartupProfiler:
_phases: list[_Phase] = field(default_factory=list)
@@ -34,11 +56,10 @@ class StartupProfiler:
finally:
p.end_ts = time.perf_counter()
self._phases.append(p)
try:
sys.stderr.write(f"[startup] {name}: {(p.end_ts - p.start_ts) * 1000.0:.1f}ms\n")
sys.stderr.flush()
except OSError as e:
sys.stderr.write(f"[startup] phase output failed for {name}: {e}\n")
log_line = f"[startup] {name}: {(p.end_ts - p.start_ts) * 1000.0:.1f}ms\n"
log_result = _log_phase_output(log_line, name)
if not log_result.ok:
_log_phase_output(f"[startup] phase output failed for {name}: {log_result.errors[0].message}\n", name)
def snapshot(self) -> dict[str, Any]:
phases: dict[str, dict[str, float]] = {}