scripts/test_review.py

"""End-to-end smoke test through the Review Agent.

Pipeline: Image Analyzer -> Planner -> Vision Pass -> Aggregator -> Review.

Usage:
    uv run python scripts/test_review.py
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

from dotenv import load_dotenv

from ergo_agentic.datasources import DatasourceRegistry
from ergo_agentic.models import DEFAULT_MODEL_CONFIG
from ergo_agentic.nodes.aggregator import aggregate
from ergo_agentic.nodes.image_analyzer import analyze_image
from ergo_agentic.nodes.parameter_planner import plan_parameters
from ergo_agentic.nodes.review_agent import _make_node as _make_review
from ergo_agentic.nodes.vision_pass import _make_node as _make_vision
from ergo_agentic.state import ImageInput

REPORT_SAMPLE_FILE = (
    Path(__file__).resolve().parents[1] / "docs" / "datasources" / "report-sample.json"
)


def main() -> int:
    load_dotenv()
    registry = DatasourceRegistry.from_knowledge_base()
    run_vision_pass = _make_vision(registry)
    run_review = _make_review(registry)

    with REPORT_SAMPLE_FILE.open() as f:
        report = json.load(f)
    urls = report.get("uploadedImages", [])
    images: list[ImageInput] = [
        {"image_id": f"img_{i}", "url": url, "label": None}
        for i, url in enumerate(urls, start=1)
    ]

    print(f"Step 1: analyzing {len(images)} images...")
    manifests = []
    for img in images:
        result = analyze_image(
            {"image": img, "model_id": DEFAULT_MODEL_CONFIG.image_analyzer}
        )
        manifests.append(result["image_manifests"][0])

    print("Step 2: planning parameters...")
    plan_result = plan_parameters({"image_manifests": manifests}, registry=registry)

    print("Step 3: running vision passes...")
    images_by_id = {img["image_id"]: img for img in images}
    model_id = DEFAULT_MODEL_CONFIG.vision_passes[0]
    observations = []
    for fg_name, fg_plan in plan_result["execution_plan"]["focus_groups"].items():
        if not fg_plan["parameter_ids"] or not fg_plan["image_ids"]:
            continue
        for image_id in fg_plan["image_ids"]:
            result = run_vision_pass({
                "image": images_by_id[image_id],
                "model_id": model_id,
                "focus_group": fg_name,
                "parameter_ids": fg_plan["parameter_ids"],
            })
            observations.extend(result["observations"])
    print(f"  collected {len(observations)} observations")

    print("Step 4: aggregating...")
    agg_result = aggregate({"observations": observations}, registry=registry)
    findings = agg_result["aggregated_findings"]
    conflicts = agg_result["conflicts"]
    print(f"  {len(findings)} findings, {len(conflicts)} conflicts")

    print(f"\nStep 5: review agent (model={DEFAULT_MODEL_CONFIG.review_agent})...\n")
    review_state = {
        "aggregated_findings": findings,
        "conflicts": conflicts,
        "images": images,
    }
    review_result = run_review(review_state)

    decisions = review_result["review_decisions"]
    final_outcomes = review_result["final_outcomes"]
    evidence_trail = review_result["evidence_trail"]

    for d in decisions:
        param = registry.get_parameter(d.parameter_id)
        label = param.parameter_text if param else d.parameter_id
        print(f"[{label}] decision={d.decision.value}")
        if d.decision.value == "insufficient_evidence":
            if d.override_reason:
                print(f"  reason: {d.override_reason}")
            continue
        original = d.original_candidates
        print(f"  candidates seen: {original}")
        print(f"  final: {d.final_outcomes}")
        if d.decision.value == "overridden" and d.override_reason:
            print(f"  override reason: {d.override_reason}")
        for o in d.final_outcomes:
            opt = registry.get_option(o)
            risk = opt.risk_level.value if opt and opt.risk_level else "good"
            print(f"    -> {o} [risk={risk}, score={opt.posture_score if opt else '?'}]")
        print()

    print(f"Total final outcomes: {len(final_outcomes)}")
    print(f"Evidence trail entries: {len(evidence_trail)}")
    return 0


if __name__ == "__main__":
    sys.exit(main())