From 73ab2778cad2955c3642c0e7a8d51547f819722f Mon Sep 17 00:00:00 2001 From: Ed_ Date: Tue, 16 Jun 2026 19:13:30 -0400 Subject: [PATCH] feat(report): implement write_failure_report + 8 tests, 100% coverage --- scripts/tier2/write_report.py | 157 ++++++++++++++++++++++++++++++ tests/test_tier2_report_writer.py | 33 +++++++ 2 files changed, 190 insertions(+) create mode 100644 scripts/tier2/write_report.py diff --git a/scripts/tier2/write_report.py b/scripts/tier2/write_report.py new file mode 100644 index 00000000..f6d41d16 --- /dev/null +++ b/scripts/tier2/write_report.py @@ -0,0 +1,157 @@ +"""Markdown failure report writer for Tier 2 give-up events. + +Writes a 7-section markdown report to the failures dir on give-up, plus +a .STOPPED flag file. Pure logic, no external deps beyond the stdlib. +""" +from __future__ import annotations + +import os +import subprocess +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Literal + +from scripts.tier2.failcount import FailcountState + + +def _failures_dir() -> Path: + return Path(os.environ.get( + "TIER2_FAILURES_DIR", + r"C:\Users\Ed\AppData\Local\manual_slop\tier2_failures", + )) + + +def compute_report_path(track_name: str, now: datetime) -> Path: + utc_ts = now.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + return _failures_dir() / f"{track_name}_{utc_ts}.md" + + +def compute_stopped_flag_path(track_name: str) -> Path: + return _failures_dir() / f"{track_name}.STOPPED" + + +@dataclass +class TaskResult: + task_id: str + phase: Literal["Red", "Green", "Refactor", "Commit"] + commit_sha: str + summary: str + error: str | None = None + + +def _git_log_for_branch(branch_name: str, repo_path: Path) -> str: + try: + result = subprocess.run( + ["git", "log", "--oneline", f"{branch_name}", "^origin/main"], + cwd=repo_path, + capture_output=True, + text=True, + timeout=10, + ) + return result.stdout.strip() if result.returncode == 0 else "(git log failed)" + except (subprocess.TimeoutExpired, FileNotFoundError): + return "(git not available)" + + +def _recommend(state: FailcountState, current_task: TaskResult | None) -> str: + if state.red_phase_failures >= state.green_phase_failures: + return "The red-phase (test-writing) is stuck. Consider whether the spec needs a clearer test plan or whether external dependencies are missing." + return "The green-phase (implementation) is stuck. Consider whether the spec describes behavior the available APIs can produce." + + +def _format_duration(started: datetime, stopped: datetime) -> str: + delta = stopped - started + total_seconds = int(delta.total_seconds()) + hours, remainder = divmod(total_seconds, 3600) + minutes, seconds = divmod(remainder, 60) + return f"{hours}h {minutes}m {seconds}s" + + +def _truncate(s: str, max_lines: int = 50) -> str: + lines = s.splitlines() + if len(lines) <= max_lines: + return s + return "\n".join(lines[:max_lines]) + f"\n... (truncated, {len(lines) - max_lines} more lines)" + + +def write_failure_report( + track_name: str, + branch_name: str, + started_at: datetime, + stopped_at: datetime, + give_up_signal: str, + completed_tasks: list[TaskResult], + current_task: TaskResult | None, + last_failures: list[str], + state: FailcountState, + repo_path: Path, +) -> Path: + failures_dir = _failures_dir() + failures_dir.mkdir(parents=True, exist_ok=True) + report_path = compute_report_path(track_name, stopped_at) + flag_path = compute_stopped_flag_path(track_name) + + duration = _format_duration(started_at, stopped_at) + completed_lines = "\n".join( + f"- **{t.task_id}** ({t.phase}) `{t.commit_sha[:7]}`: {t.summary}" + for t in completed_tasks + ) or "- (none)" + failures_text = "\n\n".join(f"```\n{_truncate(f)}\n```" for f in last_failures[:3]) or "_(none)_" + state_text = ( + f"```\n" + f"red_phase_failures: {state.red_phase_failures}\n" + f"green_phase_failures: {state.green_phase_failures}\n" + f"no_progress_started_at: {state.no_progress_started_at.isoformat() if state.no_progress_started_at else 'None'}\n" + f"```" + ) + git_log = _git_log_for_branch(branch_name, repo_path) + recommendation = _recommend(state, current_task) + current_text = ( + f"- **Task:** {current_task.task_id}\n" + f"- **Phase:** {current_task.phase}\n" + f"- **Summary:** {current_task.summary}\n" + f"- **Error:**\n```\n{_truncate(current_task.error or '(none)')}\n```" + if current_task else "_(no current task)_" + ) + + content = f"""# Tier 2 Failure Report: {track_name} + +## 1. Header + +- **Track:** {track_name} +- **Branch:** {branch_name} +- **Started:** {started_at.isoformat()} +- **Stopped:** {stopped_at.isoformat()} +- **Duration:** {duration} +- **Give-up signal:** {give_up_signal} + +## 2. Tasks Completed + +{completed_lines} + +## 3. Current Task + +{current_text} + +## 4. Last 3 Failures + +{failures_text} + +## 5. Failcount State + +{state_text} + +## 6. Git State + +``` +{git_log} +``` + +## 7. Recommendation + +{recommendation} +""" + report_path.write_text(content, encoding="utf-8") + flag_path.write_text(stopped_at.isoformat(), encoding="utf-8") + return report_path diff --git a/tests/test_tier2_report_writer.py b/tests/test_tier2_report_writer.py index f44ed836..8f5501e5 100644 --- a/tests/test_tier2_report_writer.py +++ b/tests/test_tier2_report_writer.py @@ -85,3 +85,36 @@ def test_stopped_flag_created(monkeypatch, tmp_path) -> None: ) flag = compute_stopped_flag_path("my_track") assert flag.exists() + + +def test_recommend_green_phase_stuck() -> None: + from scripts.tier2.write_report import _recommend + state = FailcountState(red_phase_failures=0, green_phase_failures=3) + rec = _recommend(state, None) + assert "green-phase" in rec.lower() + + +def test_truncate_long_text() -> None: + from scripts.tier2.write_report import _truncate + long_text = "\n".join(f"line {i}" for i in range(100)) + truncated = _truncate(long_text, max_lines=10) + assert "line 0" in truncated + assert "line 9" in truncated + assert "truncated" in truncated + assert "90 more lines" in truncated + + +def test_truncate_short_text_unchanged() -> None: + from scripts.tier2.write_report import _truncate + text = "short\ntext" + assert _truncate(text) == text + + +def test_git_log_fallback_on_subprocess_error(monkeypatch, tmp_path) -> None: + from scripts.tier2.write_report import _git_log_for_branch + import scripts.tier2.write_report as wr + def fake_run(*args, **kwargs): + raise FileNotFoundError("no git") + monkeypatch.setattr(wr.subprocess, "run", fake_run) + result = _git_log_for_branch("any", tmp_path) + assert "not available" in result.lower()