158 lines
4.3 KiB
Python
158 lines
4.3 KiB
Python
"""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
|