orchestrata_ultimate.agents.volks.pm.goal_report

Reference › orchestrata_ultimate › agents › volks › pm › goal_report

goal_report generates goal completion reports for the Project Manager (volk; specification 012, reporting stories). GoalReportGenerator.generate takes a goal id, description, success criteria, the latest iteration result, the iteration number and an optional history, and fills a GoalReport: progress percentage, criteria met and not met, gaps, recommendations and a list of ReportSection entries. _evaluate_criteria checks each criterion with the module-level helpers _evaluate_tests, _evaluate_coverage and _evaluate_spec_coverage. _analyze_gaps finds causes for unmet criteria, including a coverage deficit (_check_coverage_gap). _generate_recommendations adds advice from oscillation in fail counts (_check_oscillation, which recommends escalation) and from nearness to the iteration limit (_check_iteration_limit). _build_sections adds summary, test-results and trend sections, and to_markdown renders the report as markdown.

At a glance

Class diagram

Dependency diagram

API

Classes

ReportSection (dataclass)

Decorators: dataclass, traces(spec='012', story='US-012-Reporting', reqs=['REQ-012-041']) · Line 22

A section of the goal completion report.

GoalReport (dataclass)

Decorators: dataclass, traces(spec='012', story='US-012-Reporting', reqs=['REQ-012-041']) · Line 40

Structured goal completion report.

GoalReportGenerator

Decorators: traces(spec='012', story='US-012-Reporting', reqs=['REQ-012-041', 'REQ-012-050']) · Line 76

Generates goal completion reports from iteration results.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/volks/pm/goal_report.py

"""Goal Completion Report generator.

Produces structured markdown reports comparing iteration results
against goal success criteria. Identifies gaps, calculates progress,
and recommends next actions.

SPEC-012: Goal progress reporting and gap analysis (REQ-012-041 through REQ-012-050).
"""

import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


@dataclass
@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-041"])
class ReportSection:
    """A section of the goal completion report.

    Attributes:
        title: Section heading.
        content: Section body text.
        status: pass, fail, partial, or info.
    """

    title: str = ""
    content: str = ""
    status: str = "info"


@dataclass
@traces(
    spec="012", story="US-012-Reporting", reqs=["REQ-012-041"]
)  # GoalReport — the structured object generate() returns
class GoalReport:
    """Structured goal completion report.

    Attributes:
        goal_id: Goal identifier.
        goal_description: Goal text.
        generated_at: ISO timestamp of report generation.
        iteration_number: Current iteration number.
        progress_pct: Overall progress percentage.
        criteria_met: List of criteria that are satisfied.
        criteria_not_met: List of criteria not yet satisfied.
        gaps: Root cause analysis of gaps.
        recommendations: Suggested next actions.
        sections: Detailed report sections.
    """

    goal_id: str = ""
    goal_description: str = ""
    generated_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
    iteration_number: int = 0
    progress_pct: float = 0.0
    criteria_met: list[str] = field(default_factory=list)
    criteria_not_met: list[str] = field(default_factory=list)
    gaps: list[str] = field(default_factory=list)
    recommendations: list[str] = field(default_factory=list)
    sections: list[ReportSection] = field(default_factory=list)


@traces(
    spec="012",
    story="US-012-Reporting",
    reqs=[
        "REQ-012-041",  # generate() accepts Goal and iteration results
        "REQ-012-050",  # Pure function - no mutation
    ],
)
class GoalReportGenerator:
    """Generates goal completion reports from iteration results.

    Compares actual results against success criteria, identifies
    gaps, and produces structured reports with recommendations.
    """

    @traces(
        spec="012",
        story="US-012-Reporting",
        reqs=[
            "REQ-012-041",  # accepts Goal and iteration results, returns GoalReport
            "REQ-012-050",  # pure function - no mutation
        ],
    )
    def generate(
        self,
        goal_id: str,
        description: str,
        criteria: dict[str, Any],
        result: dict[str, Any],
        iteration_number: int,
        history: list[dict[str, Any]] | None = None,
    ) -> GoalReport:
        """Generate a completion report for a goal iteration.

        Args:
            goal_id: Goal identifier.
            description: Goal description text.
            criteria: Success criteria dict.
            result: Current iteration result dict.
            iteration_number: Current iteration number.
            history: Previous iteration results for trend analysis.

        Returns:
            GoalReport with all sections populated.
        """
        report = GoalReport(
            goal_id=goal_id,
            goal_description=description,
            iteration_number=iteration_number,
        )
        self._evaluate_criteria(report, criteria, result)
        self._analyze_gaps(report, criteria, result)
        self._generate_recommendations(report, criteria, result, history or [])
        self._build_sections(report, result, history or [])
        report.progress_pct = _calculate_progress(
            report.criteria_met, report.criteria_not_met
        )
        return report

    @traces(
        spec="012",
        story="US-012-Reporting",
        reqs=[
            "REQ-012-043",  # Test results section
            "REQ-012-044",  # Coverage section
            "REQ-012-045",  # Spec coverage section
        ],
    )
    def _evaluate_criteria(
        self,
        report: GoalReport,
        criteria: dict[str, Any],
        result: dict[str, Any],
    ) -> None:
        """Evaluate each criterion against the result.

        Args:
            report: Report to populate.
            criteria: Success criteria dict.
            result: Iteration result dict.
        """
        if criteria.get("tests_pass", False):
            _evaluate_tests(report, result)
        min_cov = criteria.get("min_coverage_pct", 0)
        if min_cov > 0:
            _evaluate_coverage(report, result, min_cov)
        min_spec = criteria.get("min_spec_coverage_pct", 0)
        if min_spec > 0:
            _evaluate_spec_coverage(report, result, min_spec)

    @traces(
        spec="012",
        story="US-012-Reporting",
        reqs=["REQ-012-047"],  # Gap analysis
    )
    def _analyze_gaps(
        self,
        report: GoalReport,
        criteria: dict[str, Any],
        result: dict[str, Any],
    ) -> None:
        """Identify root causes for unmet criteria.

        Args:
            report: Report to populate.
            criteria: Success criteria dict.
            result: Iteration result dict.
        """
        fail_count = result.get("fail_count", 0)
        if fail_count > 0:
            report.gaps.append(
                f"{fail_count} test(s) failing — "
                "review test output for assertion errors"
            )
        _check_coverage_gap(report, criteria, result)
        if not result.get("completed", True):
            report.gaps.append(
                "Iteration did not complete — "
                "check for timeouts or infrastructure issues"
            )

    @traces(
        spec="012",
        story="US-012-Reporting",
        reqs=["REQ-012-048"],  # Recommendations
    )
    def _generate_recommendations(
        self,
        report: GoalReport,
        criteria: dict[str, Any],
        result: dict[str, Any],
        history: list[dict[str, Any]],
    ) -> None:
        """Generate actionable recommendations.

        Args:
            report: Report to populate.
            criteria: Success criteria dict.
            result: Iteration result dict.
            history: Previous iteration results.
        """
        if not report.criteria_not_met:
            report.recommendations.append("All criteria met — mark goal as completed")
            return
        fail_count = result.get("fail_count", 0)
        if fail_count > 0:
            report.recommendations.append(
                f"Fix {fail_count} failing test(s) before next iteration"
            )
        _check_oscillation(report, history)
        _check_iteration_limit(report, criteria, history)

    @traces(
        spec="012",
        story="US-012-Reporting",
        reqs=[
            "REQ-012-042",  # Summary section
            "REQ-012-043",  # Test results section
            "REQ-012-046",  # Trend analysis
        ],
    )
    def _build_sections(
        self,
        report: GoalReport,
        result: dict[str, Any],
        history: list[dict[str, Any]],
    ) -> None:
        """Build detailed report sections.

        Args:
            report: Report to populate.
            result: Iteration result dict.
            history: Previous iteration results.
        """
        _add_summary_section(report)
        _add_test_results_section(report, result)
        if history:
            _add_trend_section(report, history)

    def to_markdown(self, report: GoalReport) -> str:
        """Render a GoalReport as markdown text.

        Args:
            report: The report to render.

        Returns:
            Markdown string.
        """
        lines = [
            f"# Goal Completion Report — {report.goal_id}",
            "",
            f"**Goal:** {report.goal_description}",
            f"**Iteration:** {report.iteration_number}",
            f"**Progress:** {report.progress_pct:.1f}%",
            f"**Generated:** {report.generated_at}",
            "",
        ]
        _render_criteria_sections(lines, report)
        _render_gaps_and_recs(lines, report)
        _render_detail_sections(lines, report)
        return "\n".join(lines)


@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-043"])  # Test results
def _evaluate_tests(report: GoalReport, result: dict[str, Any]) -> None:
    """Evaluate test pass/fail criterion.

    Args:
        report: Report to populate.
        result: Iteration result dict.
    """
    pass_count = result.get("pass_count", 0)
    fail_count = result.get("fail_count", 0)
    if result.get("tests_passed", False):
        report.criteria_met.append(
            f"Tests pass: {pass_count} passed, {fail_count} failed"
        )
    else:
        report.criteria_not_met.append(
            f"Tests failing: {pass_count} passed, {fail_count} failed"
        )


@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-044"])  # Coverage
def _evaluate_coverage(
    report: GoalReport,
    result: dict[str, Any],
    min_cov: float,
) -> None:
    """Evaluate code coverage criterion.

    Args:
        report: Report to populate.
        result: Iteration result dict.
        min_cov: Minimum required coverage percentage.
    """
    actual_cov = result.get("coverage_pct", 0.0)
    if actual_cov >= min_cov:
        report.criteria_met.append(f"Coverage {actual_cov:.1f}% >= {min_cov}% target")
    else:
        gap = min_cov - actual_cov
        report.criteria_not_met.append(
            f"Coverage {actual_cov:.1f}% < {min_cov}% target (gap: {gap:.1f}%)"
        )


@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-045"])  # Spec coverage
def _evaluate_spec_coverage(
    report: GoalReport,
    result: dict[str, Any],
    min_spec: float,
) -> None:
    """Evaluate spec coverage criterion.

    Args:
        report: Report to populate.
        result: Iteration result dict.
        min_spec: Minimum required spec coverage percentage.
    """
    actual_spec = result.get("spec_coverage_pct", 0.0)
    if actual_spec >= min_spec:
        report.criteria_met.append(f"Spec coverage {actual_spec:.1f}% >= {min_spec}%")
    else:
        gap = min_spec - actual_spec
        report.criteria_not_met.append(
            f"Spec coverage {actual_spec:.1f}% < {min_spec}% (gap: {gap:.1f}%)"
        )


@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-047"])  # Coverage gap
def _check_coverage_gap(
    report: GoalReport,
    criteria: dict[str, Any],
    result: dict[str, Any],
) -> None:
    """Add coverage deficit gap if applicable.

    Args:
        report: Report to populate.
        criteria: Success criteria dict.
        result: Iteration result dict.
    """
    min_cov = criteria.get("min_coverage_pct", 0)
    actual_cov = result.get("coverage_pct", 0.0)
    if min_cov > 0 and actual_cov < min_cov:
        deficit = min_cov - actual_cov
        report.gaps.append(
            f"Coverage deficit of {deficit:.1f}% — add tests for uncovered branches"
        )


@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-048"])
def _check_oscillation(
    report: GoalReport,
    history: list[dict[str, Any]],
) -> None:
    """Detect oscillation in fail counts and recommend escalation.

    Left as a recommendation-only string, not a CXP dispatch to U:SD
    (2026-08-19 wiring audit): no spec.md acceptance criterion mandates
    auto-escalation — REQ-012-048 (the only requirement citing this
    function) is untraced to spec text, and RTM's other pointer here,
    REQ-012-046, governs the unrelated trend-analysis section. This
    function is also a bare pure function with no CXP/postman/self access,
    inside a class documented as pure ("REQ-012-050: Pure function - no
    mutation" on GoalReportGenerator.generate()) — sending a packet would
    require a real design decision (who owns the send: PM volk itself?
    on what trigger besides this string match?) that belongs to the
    orchestrator, not a wiring change made here on its own authority.

    Args:
        report: Report to populate.
        history: Previous iteration results.
    """
    if len(history) < 4:
        return
    recent_fails = [h.get("fail_count", 0) for h in history[-4:]]
    if all(f > 0 for f in recent_fails):
        if recent_fails[-1] >= recent_fails[0]:
            report.recommendations.append(
                "OSCILLATION DETECTED — fail count not "
                "decreasing. Consider escalating to "
                "Senior Dev or changing approach"
            )


@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-048"])
def _check_iteration_limit(
    report: GoalReport,
    criteria: dict[str, Any],
    history: list[dict[str, Any]],
) -> None:
    """Warn when near max iteration limit.

    Args:
        report: Report to populate.
        criteria: Success criteria dict.
        history: Previous iteration results.
    """
    max_iter = criteria.get("max_iterations", 20)
    iterations_used = len(history) + 1
    remaining = max_iter - iterations_used
    if remaining <= 3:
        report.recommendations.append(
            f"WARNING: Only {remaining} iteration(s) remaining before max ({max_iter})"
        )


@traces(
    spec="012", story="US-012-Reporting", reqs=["REQ-012-049"]
)  # Progress calculation
def _calculate_progress(met: list[str], not_met: list[str]) -> float:
    """Calculate overall progress percentage.

    Args:
        met: List of met criteria descriptions.
        not_met: List of unmet criteria descriptions.

    Returns:
        Progress percentage (0-100).
    """
    total = len(met) + len(not_met)
    if total == 0:
        return 0.0
    return round((len(met) / total) * 100, 1)


@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-042"])  # Summary section
def _add_summary_section(report: GoalReport) -> None:
    """Add summary section to report.

    Args:
        report: Report to populate.
    """
    total = len(report.criteria_met) + len(report.criteria_not_met)
    met_count = len(report.criteria_met)
    status = "pass" if not report.criteria_not_met else "fail"
    report.sections.append(
        ReportSection(
            title="Summary",
            content=(
                f"Iteration {report.iteration_number}: {met_count}/{total} criteria met"
            ),
            status=status,
        )
    )


@traces(
    spec="012", story="US-012-Reporting", reqs=["REQ-012-043"]
)  # Test results section
def _add_test_results_section(report: GoalReport, result: dict[str, Any]) -> None:
    """Add test results section to report.

    Args:
        report: Report to populate.
        result: Iteration result dict.
    """
    status = "pass" if result.get("tests_passed") else "fail"
    report.sections.append(
        ReportSection(
            title="Test Results",
            content=(
                f"Passed: {result.get('pass_count', 0)}, "
                f"Failed: {result.get('fail_count', 0)}, "
                f"Coverage: {result.get('coverage_pct', 0.0):.1f}%"
            ),
            status=status,
        )
    )


@traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-046"])  # Trend section
def _add_trend_section(
    report: GoalReport,
    history: list[dict[str, Any]],
) -> None:
    """Add trend section showing recent iteration history.

    Args:
        report: Report to populate.
        history: Previous iteration results.
    """
    trend_lines = [
        f"  iter {idx + 1}: "
        f"pass={h.get('pass_count', 0)}, "
        f"fail={h.get('fail_count', 0)}, "
        f"cov={h.get('coverage_pct', 0.0):.1f}%"
        for idx, h in enumerate(history[-5:])
    ]
    report.sections.append(
        ReportSection(
            title="Trend (last 5)",
            content="\n".join(trend_lines),
            status="info",
        )
    )


def _render_criteria_sections(lines: list[str], report: GoalReport) -> None:
    """Render criteria met/not-met markdown sections.

    Args:
        lines: Output lines list to append to.
        report: The report to render.
    """
    if report.criteria_met:
        lines.append("## Criteria Met")
        for c in report.criteria_met:
            lines.append(f"- [x] {c}")
        lines.append("")
    if report.criteria_not_met:
        lines.append("## Criteria Not Met")
        for c in report.criteria_not_met:
            lines.append(f"- [ ] {c}")
        lines.append("")


def _render_gaps_and_recs(lines: list[str], report: GoalReport) -> None:
    """Render gap analysis and recommendations sections.

    Args:
        lines: Output lines list to append to.
        report: The report to render.
    """
    if report.gaps:
        lines.append("## Gap Analysis")
        for g in report.gaps:
            lines.append(f"- {g}")
        lines.append("")
    if report.recommendations:
        lines.append("## Recommendations")
        for r in report.recommendations:
            lines.append(f"- {r}")
        lines.append("")


def _render_detail_sections(lines: list[str], report: GoalReport) -> None:
    """Render detailed report sections as markdown.

    Args:
        lines: Output lines list to append to.
        report: The report to render.
    """
    status_icons = {
        "pass": "[PASS]",
        "fail": "[FAIL]",
        "partial": "[PARTIAL]",
    }
    for section in report.sections:
        icon = status_icons.get(section.status, "")
        lines.append(f"## {section.title} {icon}")
        lines.append(section.content)
        lines.append("")