orchestrata_ultimate.harness.reporting

Reference › orchestrata_ultimate › harness › reporting

Reports and trend queries for harness results, used by the report and trends subcommands of the cli. generate_report(history_dir, output_path) loads every JSON file written by the JSON history tracker, computes overall statistics (total runs, pass rate, average score), per-volk run and pass counts, per-scenario score trends and the flattened list of snapshot regressions, and renders HTML with Jinja2 when it is available, or a built-in rendering otherwise. generate_junit_xml(results, output_path) writes a JUnit XML file with one test case per trial (_trial_to_junit_case) for CI consumers. TrendQueries is a set of static methods that return Cypher query strings for the graph history: the pass rate of a volk over the last 30 days, the regression count of a scenario and its average score trend. The class builds strings and does not execute them. Result types are in models.

At a glance

Class diagram

Dependency diagram

API

Classes

TrendQueries

Decorators: traces(spec='102', story='US-002', reqs=['REQ-102-028']) · Line 385

Static Cypher queries for historical trend analysis in the graph.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/harness/reporting.py

"""HTML report generation and JUnit XML export for Volk Test Harness."""

from __future__ import annotations

import json
import logging
import xml.etree.ElementTree as ET
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.harness.models import ScenarioResult, TrialResult

_logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# HTML Report
# ---------------------------------------------------------------------------

_HTML_TEMPLATE = """\
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Volk Harness Report</title>
<style>
body { font-family: system-ui, -apple-system, sans-serif; margin: 2rem; background: #f8f9fa; color: #212529; }
h1, h2 { color: #1a1a2e; border-bottom: 2px solid #dee2e6; padding-bottom: .3rem; }
table { border-collapse: collapse; width: 100%; margin: 1rem 0; background: #fff; }
th, td { border: 1px solid #dee2e6; padding: .5rem .75rem; text-align: left; }
th { background: #e9ecef; font-weight: 600; }
tr:nth-child(even) { background: #f8f9fa; }
.bar-container { background: #e9ecef; height: 1.2rem; border-radius: .25rem; overflow: hidden; width: 100%; }
.bar { height: 100%; background: #198754; border-radius: .25rem; }
.bar.warn { background: #ffc107; }
.bar.danger { background: #dc3545; }
.pass { color: #198754; font-weight: 600; }
.fail { color: #dc3545; font-weight: 600; }
.regression { background: #fff3cd; }
.summary { display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 1rem; margin: 1rem 0; }
.card { background: #fff; padding: 1rem; border-radius: .5rem; box-shadow: 0 1px 3px rgba(0,0,0,.1); }
.card h3 { margin: 0 0 .5rem; font-size: 1rem; color: #495057; }
.card .value { font-size: 1.5rem; font-weight: 700; }
</style>
</head>
<body>
<h1>Volk Harness Report</h1>
<p>Generated: {{ generated_at }}</p>

<div class="summary">
  <div class="card">
    <h3>Total Runs</h3>
    <div class="value">{{ total_runs }}</div>
  </div>
  <div class="card">
    <h3>Pass Rate</h3>
    <div class="value">{{ "%.1f"|format(pass_rate) }}%</div>
  </div>
  <div class="card">
    <h3>Avg Score</h3>
    <div class="value">{{ "%.3f"|format(avg_score) }}</div>
  </div>
  <div class="card">
    <h3>Regressions</h3>
    <div class="value">{{ regression_count }}</div>
  </div>
</div>

<h2>Per-Volk Pass Rates</h2>
<table>
  <thead>
    <tr><th>Volk</th><th>Runs</th><th>Passed</th><th>Pass Rate</th><th>Bar</th></tr>
  </thead>
  <tbody>
    {% for v in volk_stats|sort(attribute='pass_rate', reverse=True) %}
    <tr>
      <td>{{ v.name }}</td>
      <td>{{ v.runs }}</td>
      <td>{{ v.passes }}</td>
      <td class="{{ 'pass' if v.pass_rate >= 90 else 'fail' }}">{{ "%.1f"|format(v.pass_rate) }}%</td>
      <td style="width: 30%;">
        <div class="bar-container">
          <div class="bar {{ 'danger' if v.pass_rate < 70 else 'warn' if v.pass_rate < 90 else '' }}"
               style="width: {{ v.pass_rate }}%;"></div>
        </div>
      </td>
    </tr>
    {% endfor %}
  </tbody>
</table>

<h2>Per-Scenario Score Trends</h2>
<table>
  <thead>
    <tr><th>Scenario</th><th>Volk</th><th>Latest Score</th><th>Avg Score</th><th>Runs</th></tr>
  </thead>
  <tbody>
    {% for s in scenario_stats|sort(attribute='scenario_id') %}
    <tr>
      <td>{{ s.scenario_id }}</td>
      <td>{{ s.volk }}</td>
      <td>{{ "%.3f"|format(s.latest_score) }}</td>
      <td>{{ "%.3f"|format(s.avg_score) }}</td>
      <td>{{ s.runs }}</td>
    </tr>
    {% endfor %}
  </tbody>
</table>

<h2>Regressions</h2>
{% if regressions %}
<table>
  <thead>
    <tr><th>Timestamp</th><th>Scenario</th><th>Volk</th><th>Trial</th><th>Diff Preview</th></tr>
  </thead>
  <tbody>
    {% for r in regressions %}
    <tr class="regression">
      <td>{{ r.timestamp }}</td>
      <td>{{ r.scenario_id }}</td>
      <td>{{ r.volk }}</td>
      <td>{{ r.trial_number }}</td>
      <td><pre style="max-width: 400px; overflow: auto; font-size: .8rem;">{{ r.diff_preview }}</pre></td>
    </tr>
    {% endfor %}
  </tbody>
</table>
{% else %}
<p>No regressions detected. 🎉</p>
{% endif %}
</body>
</html>
"""


@traces(spec="102", story="US-002", reqs=["REQ-102-022"])
def _load_history_entries(history_dir: Path) -> list[dict[str, Any]]:
    """Load all JSON history files and return a flat list of entries."""
    entries: list[dict[str, Any]] = []
    if not history_dir.exists():
        return entries
    for path in sorted(history_dir.glob("*.json")):
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
            if isinstance(data, list):
                entries.extend(data)
        except (json.JSONDecodeError, OSError) as exc:
            _logger.warning("Skipping corrupt history file %s: %s", path, exc)
    return entries


@traces(spec="102", story="US-002", reqs=["REQ-102-030"])
def _compute_overall_stats(entries: list[dict[str, Any]]) -> tuple[int, float, float]:
    """Return (total_runs, pass_rate, avg_score) across all history entries."""
    total_runs = len(entries)
    passes = sum(1 for e in entries if e.get("passed"))
    pass_rate = (passes / total_runs * 100) if total_runs else 0.0
    scores = [e.get("aggregate_score", 0.0) for e in entries]
    avg_score = sum(scores) / len(scores) if scores else 0.0
    return total_runs, pass_rate, avg_score


@traces(spec="102", story="US-002", reqs=["REQ-102-031"])
def _compute_volk_stats(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Per-volk run/pass counts and pass rate."""
    volk_data: dict[str, dict[str, Any]] = {}
    for e in entries:
        name = e.get("volk", "unknown")
        volk_data.setdefault(name, {"runs": 0, "passes": 0})
        volk_data[name]["runs"] += 1
        if e.get("passed"):
            volk_data[name]["passes"] += 1

    return [
        {
            "name": name,
            "runs": d["runs"],
            "passes": d["passes"],
            "pass_rate": (d["passes"] / d["runs"] * 100) if d["runs"] else 0.0,
        }
        for name, d in volk_data.items()
    ]


@traces(spec="102", story="US-002", reqs=["REQ-102-032"])
def _compute_scenario_stats(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Per-(volk, scenario) score trend: latest score, average score, run count."""
    scenario_data: dict[str, dict[str, Any]] = {}
    for e in entries:
        sid = e.get("scenario_id", "unknown")
        volk = e.get("volk", "unknown")
        key = f"{volk}/{sid}"
        if key not in scenario_data:
            scenario_data[key] = {
                "scenario_id": sid,
                "volk": volk,
                "scores": [],
                "timestamps": [],
            }
        scenario_data[key]["scores"].append(e.get("aggregate_score", 0.0))
        scenario_data[key]["timestamps"].append(e.get("timestamp", ""))

    return [
        {
            "scenario_id": d["scenario_id"],
            "volk": d["volk"],
            "latest_score": d["scores"][-1] if d["scores"] else 0.0,
            "avg_score": sum(d["scores"]) / len(d["scores"]) if d["scores"] else 0.0,
            "runs": len(d["scores"]),
        }
        for d in scenario_data.values()
    ]


@traces(spec="102", story="US-002", reqs=["REQ-102-033"])
def _compute_regressions(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Flatten per-trial snapshot-diff regressions across all history entries."""
    regressions: list[dict[str, Any]] = []
    for e in entries:
        for trial in e.get("trials", []):
            if trial.get("snapshot_diff"):
                diff = trial["snapshot_diff"]
                regressions.append(
                    {
                        "timestamp": e.get("timestamp", "unknown"),
                        "scenario_id": e.get("scenario_id", "unknown"),
                        "volk": e.get("volk", "unknown"),
                        "trial_number": trial.get("trial_number", 0),
                        "diff_preview": diff[:300] + ("..." if len(diff) > 300 else ""),
                    }
                )
    return regressions


@traces(spec="102", story="US-002", reqs=["REQ-102-029"])
def _render_report_html(context: dict[str, Any]) -> str:
    """Render the report context into HTML — Jinja2 if available, else a
    partially-rendered fallback using the stdlib ``string.Template``.
    """
    try:
        from jinja2 import Template

        template = Template(_HTML_TEMPLATE)
        return template.render(**context)
    except ImportError:
        from string import Template as StringTemplate

        st = StringTemplate(_HTML_TEMPLATE)
        html = st.safe_substitute(
            {
                k: str(v)
                for k, v in context.items()
                if k not in ("volk_stats", "scenario_stats", "regressions")
            }
        )
        return html + "\n<!-- Jinja2 not available; report partially rendered -->\n"


@traces(
    spec="102",
    story="US-002",
    reqs=[
        "REQ-102-029",
        "REQ-102-030",
        "REQ-102-031",
        "REQ-102-032",
        "REQ-102-033",
    ],
)
def generate_report(history_dir: Path, output_path: Path) -> Path:
    """Read all JSON history files and generate an HTML report.

    Returns:
        Path to the written HTML file.
    """
    entries = _load_history_entries(history_dir)

    total_runs, pass_rate, avg_score = _compute_overall_stats(entries)
    volk_stats = _compute_volk_stats(entries)
    scenario_stats = _compute_scenario_stats(entries)
    regressions = _compute_regressions(entries)

    context = {
        "generated_at": datetime.now(UTC).isoformat(),
        "total_runs": total_runs,
        "pass_rate": pass_rate,
        "avg_score": avg_score,
        "regression_count": len(regressions),
        "volk_stats": volk_stats,
        "scenario_stats": scenario_stats,
        "regressions": regressions,
    }

    html = _render_report_html(context)

    output_path.write_text(html, encoding="utf-8")
    _logger.info("Report written to %s", output_path)
    return output_path


# ---------------------------------------------------------------------------
# JUnit XML
# ---------------------------------------------------------------------------


def _trial_to_junit_case(
    trial: TrialResult,
    scenario_id: str,
    volk_name: str,
) -> ET.Element:
    """Build a <testcase> element for a single trial."""
    case = ET.Element("testcase")
    case.set("name", f"{scenario_id} trial {trial.trial_number}")
    case.set("classname", volk_name)
    case.set("time", str(trial.execution_time_ms / 1000.0))

    if not trial.passed:
        failure = ET.SubElement(case, "failure")
        failure.set("message", f"Trial {trial.trial_number} failed")
        details: list[str] = []
        for er in trial.evaluator_results:
            status = "PASS" if er.passed else "FAIL"
            details.append(f"  {er.name}: {status} (score={er.score}) — {er.details}")
        if trial.error:
            details.append(f"Error: {trial.error}")
        if trial.snapshot_diff:
            details.append(f"Snapshot diff:\n{trial.snapshot_diff[:500]}")
        failure.text = "\n".join(details)

    if trial.error and trial.passed:
        # Error that didn't cause failure (unexpected)
        err = ET.SubElement(case, "error")
        err.set("message", str(trial.error))
        err.text = str(trial.error)

    return case


def generate_junit_xml(results: list[ScenarioResult], output_path: Path) -> Path:
    """Generate a JUnit XML file from scenario results.

    Returns:
        Path to the written XML file.
    """
    testsuites = ET.Element("testsuites")

    # Group by volk
    by_volk: dict[str, list[ScenarioResult]] = {}
    for result in results:
        by_volk.setdefault(result.volk_name, []).append(result)

    for volk_name, volk_results in by_volk.items():
        suite = ET.SubElement(testsuites, "testsuite")
        suite.set("name", f"volk_harness.{volk_name}")
        total_tests = sum(len(r.trials) for r in volk_results)
        total_failures = sum(
            sum(1 for t in r.trials if not t.passed) for r in volk_results
        )
        total_time = sum(r.execution_time_ms for r in volk_results) / 1000.0
        suite.set("tests", str(total_tests))
        suite.set("failures", str(total_failures))
        suite.set("errors", "0")
        suite.set("time", str(total_time))
        suite.set("timestamp", datetime.now(UTC).isoformat())

        for result in volk_results:
            for trial in result.trials:
                case = _trial_to_junit_case(trial, result.scenario_id, volk_name)
                suite.append(case)

    tree = ET.ElementTree(testsuites)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    tree.write(output_path, encoding="utf-8", xml_declaration=True)
    _logger.info("JUnit XML written to %s", output_path)
    return output_path


# ---------------------------------------------------------------------------
# Neo4j Trend Queries
# ---------------------------------------------------------------------------


@traces(spec="102", story="US-002", reqs=["REQ-102-028"])
class TrendQueries:
    """Static Cypher queries for historical trend analysis in the graph."""

    @staticmethod
    @traces(spec="102", story="US-002", reqs=["REQ-102-028"])
    def pass_rate_last_30_days(volk_name: str) -> str:
        """Return Cypher query string for pass rate of *volk_name* over last 30 days."""
        return (
            "MATCH (r:Run)-[:RESULT_OF]->(s:Scenario)-[:EVALUATED_BY]->(v:Volk {alias: $volk_name}) "
            "WHERE datetime(r.timestamp) > datetime() - duration('P30D') "
            "RETURN count(r) AS total_runs, "
            "       sum(CASE WHEN r.passed THEN 1 ELSE 0 END) AS passes, "
            "       (sum(CASE WHEN r.passed THEN 1 ELSE 0 END) * 100.0 / count(r)) AS pass_rate"
        )

    @staticmethod
    @traces(spec="102", story="US-002", reqs=["REQ-102-028"])
    def regression_count(scenario_id: str) -> str:
        """Return Cypher query string for number of regressions for *scenario_id*."""
        return (
            "MATCH (reg:Regression)-[:DETECTED_IN]->(s:Scenario {id: $scenario_id}) "
            "RETURN count(reg) AS regression_count"
        )

    @staticmethod
    @traces(spec="102", story="US-002", reqs=["REQ-102-028"])
    def avg_score_trend(scenario_id: str) -> str:
        """Return Cypher query string for average score trend of *scenario_id*."""
        return (
            "MATCH (r:Run)-[:RESULT_OF]->(s:Scenario {id: $scenario_id}) "
            "RETURN date(datetime(r.timestamp)) AS run_date, "
            "       avg(r.aggregate_score) AS avg_score "
            "ORDER BY run_date DESC "
            "LIMIT 30"
        )