feat(report): implement write_failure_report + 8 tests, 100% coverage

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

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()