""" core/epoch/epoch_report_generator.py Story 9.07: EpochReportGenerator — Epoch Summary Report Produces a structured Markdown summary of a Nightly Epoch run. All logic is pure Python — NO LLM calls are made here. Reports are written to: data/observability/epoch_reports/epoch_{epoch_id}.md # VERIFICATION_STAMP # Story: 9.07 # Verified By: parallel-builder # Verified At: 2026-02-25T00:00:00Z # Tests: 12/12 # Coverage: 100% """ from __future__ import annotations import os from dataclasses import dataclass, field from datetime import date, datetime, timezone from typing import Optional # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- REPORT_DIR = "/mnt/e/genesis-system/data/observability/epoch_reports" # Total number of phases the Nightly Epoch runs. All 7 completed = COMPLETED. _TOTAL_PHASES = 7 # --------------------------------------------------------------------------- # Data classes # --------------------------------------------------------------------------- @dataclass class EpochResult: """Input data collected during a Nightly Epoch run. Attributes: epoch_id: Unique identifier for this epoch run (e.g. "2026_02_25"). phases_completed: Names of the phases that finished successfully. axioms: List of axiom objects (or dicts) distilled this epoch. pr_url: Optional GitHub PR URL opened by the epoch. tier1_updates: Count of Tier-1 memory updates applied. week_summary: 1–2 sentence prose summary of the week. shadow_pass_rate: Optional float (0.0–1.0) from the Shadow Arena gate. """ epoch_id: str phases_completed: list[str] = field(default_factory=list) axioms: list = field(default_factory=list) pr_url: Optional[str] = None tier1_updates: int = 0 week_summary: str = "" shadow_pass_rate: Optional[float] = None @dataclass class EpochReport: """Output produced by EpochReportGenerator. Attributes: markdown_content: The full Markdown text of the report. file_path: Absolute path where the report was written on disk. epoch_id: The epoch identifier this report covers. """ markdown_content: str file_path: str epoch_id: str # --------------------------------------------------------------------------- # EpochReportGenerator # --------------------------------------------------------------------------- class EpochReportGenerator: """Generates and persists a Markdown epoch summary report. Args: report_dir: Directory to write reports into. Defaults to ``REPORT_DIR``. Callers (especially tests) may supply a ``tempfile.mkdtemp()`` path to avoid touching production storage. """ def __init__(self, report_dir: str = REPORT_DIR) -> None: self.report_dir = report_dir # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def generate(self, result: EpochResult) -> EpochReport: """Build a Markdown report from *result*, write it to disk and return it. Args: result: The ``EpochResult`` from a completed (or partial/failed) Nightly Epoch run. Returns: An ``EpochReport`` containing the Markdown content, file path, and epoch identifier. """ status = self._determine_status(result) report_date = self._epoch_date(result.epoch_id) content = self._build_markdown(result, status, report_date) file_path = self._write_report(result.epoch_id, content) return EpochReport( markdown_content=content, file_path=file_path, epoch_id=result.epoch_id, ) # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ @staticmethod def _determine_status(result: EpochResult) -> str: """Return "COMPLETED", "PARTIAL", or "FAILED" based on phases_completed.""" n = len(result.phases_completed) if n == 0: return "FAILED" if n >= _TOTAL_PHASES: return "COMPLETED" return "PARTIAL" @staticmethod def _epoch_date(epoch_id: str) -> str: """Derive a human-readable date string from the epoch_id or fall back to today.""" # epoch_id format is typically YYYY_MM_DD; try to parse it try: parts = epoch_id.split("_") if len(parts) >= 3: year, month, day = int(parts[0]), int(parts[1]), int(parts[2]) return date(year, month, day).strftime("%Y-%m-%d") except (ValueError, IndexError): pass # Fallback: use today's UTC date return datetime.now(timezone.utc).strftime("%Y-%m-%d") @staticmethod def _format_axiom(axiom) -> str: """Format a single axiom as a Markdown list item.""" if isinstance(axiom, dict): axiom_id = axiom.get("id", "unknown") content = axiom.get("content", str(axiom)) confidence = axiom.get("confidence") if confidence is not None: return f"- `{axiom_id}` (confidence: {confidence:.2f}) — {content}" return f"- `{axiom_id}` — {content}" # Dataclass / object with attributes axiom_id = getattr(axiom, "id", "unknown") content = getattr(axiom, "content", str(axiom)) confidence = getattr(axiom, "confidence", None) if confidence is not None: return f"- `{axiom_id}` (confidence: {confidence:.2f}) — {content}" return f"- `{axiom_id}` — {content}" def _build_markdown( self, result: EpochResult, status: str, report_date: str ) -> str: """Assemble the full Markdown report string.""" lines: list[str] = [] # --- Header --- lines.append(f"# Genesis Nightly Epoch — {report_date}") lines.append("") lines.append(f"## Status: {status}") lines.append("") # --- Overview table --- lines.append("## Overview") lines.append("") lines.append("| Field | Value |") lines.append("|-------|-------|") lines.append(f"| Epoch ID | `{result.epoch_id}` |") lines.append(f"| Date | {report_date} |") lines.append(f"| Status | **{status}** |") lines.append( f"| Phases Completed | {len(result.phases_completed)} / {_TOTAL_PHASES} |" ) lines.append(f"| Axioms Distilled | {len(result.axioms)} |") lines.append(f"| Tier-1 Updates | {result.tier1_updates} |") lines.append("") # --- Phases --- lines.append("## Phases Completed") lines.append("") if result.phases_completed: for phase in result.phases_completed: lines.append(f"- {phase}") else: lines.append("_No phases completed._") lines.append("") # --- Axioms --- lines.append("## Axioms Distilled") lines.append("") if result.axioms: for axiom in result.axioms: lines.append(self._format_axiom(axiom)) else: lines.append("_No axioms distilled this epoch._") lines.append("") # --- Week Summary --- lines.append("## Week Summary") lines.append("") if result.week_summary: lines.append(result.week_summary) else: lines.append("_No week summary available._") lines.append("") # --- PR / Shadow Arena (only when PR opened) --- if result.pr_url: lines.append("## Pull Request") lines.append("") lines.append(f"PR opened: [{result.pr_url}]({result.pr_url})") lines.append("") if result.shadow_pass_rate is not None: pct = result.shadow_pass_rate * 100 lines.append( f"Shadow Arena pass rate: **{pct:.1f}%**" ) lines.append("") return "\n".join(lines) def _write_report(self, epoch_id: str, content: str) -> str: """Ensure the report directory exists, write the file and return its path.""" os.makedirs(self.report_dir, exist_ok=True) file_path = os.path.join(self.report_dir, f"epoch_{epoch_id}.md") with open(file_path, "w", encoding="utf-8") as fh: fh.write(content) return file_path