script-build-and-framework-tmp/visualization/backend/app/evaluation_event_projection.py

from __future__ import annotations

import re
from collections import defaultdict
from typing import Any

from .evaluation_report_parser import EvaluationReportParser
from .runtime_event_index import (
    RuntimeEventIndex,
    event_input,
    event_output,
    event_summary,
    integer,
)


class EvaluationEventProjector:
    """Project evaluator calls using event metadata for identity and one parser.

    The report body never decides build/round association.  It only supplies
    business-facing evaluation content after the event has been placed by its
    structured metadata.
    """

    def __init__(self, parser: EvaluationReportParser | None = None):
        self._parser = parser or EvaluationReportParser()

    def project(
        self,
        index: RuntimeEventIndex,
        *,
        valid_rounds: set[int],
        valid_branches: set[int],
    ) -> dict[str, Any]:
        multipath: dict[int, list[dict[str, Any]]] = defaultdict(list)
        overall: dict[int, list[dict[str, Any]]] = defaultdict(list)
        unassigned: list[dict[str, Any]] = []

        for event in index.ordered:
            if str(event.get("event_type") or "") != "agent_invoke":
                continue
            name = str(event.get("event_name") or "")
            if name not in {"script_multipath_evaluator", "script_evaluator"}:
                continue
            round_index = integer(event.get("round_index"))
            if round_index not in valid_rounds:
                unassigned.append(event_summary(event, "missing-business-context"))
                continue

            report = _agent_summary(event) or str(event.get("output_preview") or "")
            parsed = self._parser.parse(
                report,
                kind="multipath" if name == "script_multipath_evaluator" else "overall",
            )
            if name == "script_evaluator":
                overall[round_index].append(
                    {
                        **event_summary(event, "structured-round"),
                        **parsed,
                        "id": f"overall-evaluation:{integer(event.get('id')) or 0}",
                        "roundIndex": round_index,
                        "detailRef": f"event:{integer(event.get('id')) or 0}",
                        "promptRef": f"prompt:event:{integer(event.get('id')) or 0}",
                    }
                )
                continue

            task = _agent_task(event) or str(event.get("input_preview") or "")
            branch_ids = [value for value in _branch_ids(task) if value in valid_branches]
            review = {
                **event_summary(event, "explicit-task-branch-ids"),
                **parsed,
                "id": f"multipath-review:{integer(event.get('id')) or 0}",
                "roundIndex": round_index,
                "branchIds": branch_ids,
                "branchConclusions": _branch_conclusions(parsed, branch_ids),
                "comparison": parsed.get("recommendation"),
                "detailRef": f"event:{integer(event.get('id')) or 0}",
                "promptRef": f"prompt:event:{integer(event.get('id')) or 0}",
            }
            if branch_ids:
                multipath[round_index].append(review)
            else:
                unassigned.append({**review, "association": "missing-branch-scope"})

        return {
            "multipathReviewsByRound": dict(multipath),
            "overallEvaluationsByRound": dict(overall),
            "unassigned": unassigned,
        }


def _branch_ids(text: str) -> list[int]:
    values: list[int] = []
    source = text or ""

    # Agent tasks commonly express one batch as ``branch_id = 4, 5, 6, 7``.
    # Capture the complete labelled list instead of keeping only its first id.
    list_pattern = (
        r"\bbranch[_\s-]*ids?\s*[:=]\s*\[?\s*"
        r"(\d+(?:\s*[,，、/]\s*\d+)*)\s*\]?"
    )
    for match in re.finditer(list_pattern, source, flags=re.IGNORECASE):
        for raw_value in re.findall(r"\d+", match.group(1)):
            value = int(raw_value)
            if value not in values:
                values.append(value)

    # Keep supporting prose that names branches individually.
    for match in re.finditer(r"(?:方案|分支)\s*(\d+)", source, flags=re.IGNORECASE):
        value = int(match.group(1))
        if value not in values:
            values.append(value)
    return values


def _branch_conclusions(
    parsed: dict[str, Any], branch_ids: list[int]
) -> list[dict[str, Any]]:
    by_subject = {
        str(item.get("subject") or ""): item
        for item in parsed.get("itemConclusions") or []
        if isinstance(item, dict)
    }
    return [
        {
            "branchId": branch_id,
            "conclusion": (by_subject.get(f"方案 {branch_id}") or {}).get("conclusion"),
            "summary": _item_summary(by_subject.get(f"方案 {branch_id}")),
        }
        for branch_id in branch_ids
    ]


def _item_summary(item: Any) -> str | None:
    if not isinstance(item, dict):
        return None
    values = [
        *(str(value) for value in item.get("achievements") or [] if value),
        *(str(value) for value in item.get("problems") or [] if value),
    ]
    return "；".join(values[:2]) or None


def _agent_task(event: dict[str, Any]) -> str | None:
    value = event_input(event)
    if isinstance(value, dict) and isinstance(value.get("task"), str):
        return value["task"]
    return None


def _agent_summary(event: dict[str, Any]) -> str | None:
    value = event_output(event)
    if isinstance(value, dict) and isinstance(value.get("summary"), str):
        return value["summary"]
    return None