src/agents/slow_path/checkpoint.py

"""Quality checkpoint (S1a) — deterministic run inspection (SPINE_V2_PLAN §3, W1).

Sits between `TaskRunner.run` and `Assembler.assemble` in the coordinator. 0 LLM,
never-throw: `assess` inspects the completed `RunState` against the plan and turns
today's *silent* degrade-and-continue into an *explained* one — the Assembler gets
told what specifically failed/degraded, and every flag is structlog'd as a
`repair_candidate` so the S1b (targeted repair, gated) decision has an evidence
base. INV-6 untouched: nothing here re-plans or calls a model.

v1 checks:
  CK1  all tasks failed            -> overall "failed" (coordinator answers honestly, no LLM)
  CK2  empty retrieve consumed     -> flag the retrieve + its transitive dependents
  CK3  table truncated at the cap  -> flag (answer must scope itself to the first 10k rows)
  CK4  trend collapsed to 1 bucket -> flag (the pr/13 1970-bucket class)
  CK5  all-null column consumed    -> flag the consuming analyze_* task
  CK6  chart-spec sanity (§4.6)    -> flag empty/mismatched/overcrowded/non-numeric charts
"""

from __future__ import annotations

from collections import defaultdict, deque
from typing import Any

from src.middlewares.logging import get_logger

from ..planner.schemas import PLACEHOLDER_RE, TaskList
from .schemas import RepairCandidate, RunAssessment, RunState, TaskAssessment

logger = get_logger("slow_path_checkpoint")

_TABLE_ROW_CAP = 10_000  # the query pipeline's LIMIT defense cap (CK3)
_BAR_CATEGORY_CAP = 30   # §4.6 — a bar chart beyond this is unreadable
_PIE_CATEGORY_CAP = 8    # §4.6 — a pie chart beyond this is unreadable


def assess(run_state: RunState, task_list: TaskList) -> RunAssessment:
    """Inspect a completed run. NEVER raises — on an internal slip the checkpoint
    degrades to today's behavior (an all-ok pass-through), never blocks the answer."""
    try:
        return _assess(run_state, task_list)
    except Exception as exc:  # noqa: BLE001 — never-throw seam (house pattern)
        logger.error("checkpoint failed", error=repr(exc))
        return RunAssessment(overall="ok")


def _assess(run_state: RunState, task_list: TaskList) -> RunAssessment:
    results = run_state.results
    notes: dict[str, list[str]] = {tid: [] for tid in results}
    candidates: list[RepairCandidate] = []

    def flag(task_id: str, check: str, reason: str, *, repairable: bool) -> None:
        if task_id not in notes:
            return
        notes[task_id].append(reason)
        # Telemetry for the S1b decision (SPINE_V2_PLAN §3): every flag is logged,
        # repairable or not — the hit-rate decides whether a repair pass pays.
        logger.info("repair_candidate", task_id=task_id, check=check, reason=reason)
        if repairable:
            candidates.append(RepairCandidate(task_id=task_id, reason=reason))

    # CK1 — everything failed: no partial result worth prose. The coordinator
    # returns a deterministic honest-failure answer (0 LLM), mirroring the
    # infeasible path; the record stays non-substantive.
    if results and all(r.status == "failure" for r in results.values()):
        logger.info(
            "repair_candidate",
            task_id="*",
            check="CK1",
            reason="all tasks failed",
        )
        return RunAssessment(
            overall="failed",
            tasks=[
                TaskAssessment(
                    task_id=tid, verdict="failed", notes=[r.error] if r.error else []
                )
                for tid, r in results.items()
            ],
        )

    dependents = _transitive_dependents(task_list)

    for tid, result in results.items():
        for out in result.outputs:
            # CK2 — an empty retrieval, with downstream consumers flagged too.
            if out.tool == "retrieve_data" and out.kind == "table" and not (out.rows or []):
                flag(tid, "CK2", "retrieve_data returned 0 rows (the filters "
                     "matched nothing)", repairable=True)
                for dep in sorted(dependents.get(tid, ())):
                    flag(dep, "CK2", f"consumed the empty result of {tid}",
                         repairable=False)
            # CK3 — the retrieval cap was hit; findings only cover the cap window.
            elif out.kind == "table" and out.rows is not None and len(out.rows) >= _TABLE_ROW_CAP:
                flag(tid, "CK3", f"table hit the {_TABLE_ROW_CAP:,}-row retrieval "
                     "cap — findings describe only the first "
                     f"{_TABLE_ROW_CAP:,} rows", repairable=False)

            # CK4 — a single trend bucket (e.g. every date parsed into one period).
            if out.tool == "analyze_trend" and out.kind == "series" and isinstance(out.value, dict):
                points = out.value.get("points")
                if isinstance(points, list) and len(points) == 1:
                    freq = out.value.get("freq", "period")
                    flag(tid, "CK4", f"trend collapsed into a single {freq} bucket "
                         "(1 point) — no movement can be described",
                         repairable=True)

            # CK6 — chart-spec sanity (§4.6; lands with W2's render_chart).
            if out.kind == "chart" and isinstance(out.value, dict):
                for issue in _chart_issues(out.value):
                    flag(tid, "CK6", issue, repairable=True)

    # CK5 — an analyze_* consumed a table whose column(s) are entirely null.
    # Consumption is read from the PLAN (the `data`/`data_right` placeholders);
    # the runner resolves the same references at execution time.
    for task in task_list.tasks:
        for call in task.tool_calls:
            if not call.tool.startswith("analyze_"):
                continue
            for arg_name in ("data", "data_right"):
                ref = _placeholder_ref(call.args.get(arg_name))
                table = _last_table_output(results.get(ref)) if ref else None
                if table is None:
                    continue
                null_cols = _all_null_columns(table)
                if null_cols:
                    flag(task.id, "CK5",
                         f"input column(s) {null_cols} from {ref} are entirely "
                         "null — results based on them are meaningless",
                         repairable=True)

    assessments: list[TaskAssessment] = []
    for tid, result in results.items():
        if result.status == "failure":
            verdict = "failed"
        elif notes[tid] or result.status == "partial":
            verdict = "degraded"
        else:
            verdict = "ok"
        assessments.append(TaskAssessment(task_id=tid, verdict=verdict, notes=notes[tid]))

    overall = "degraded" if any(a.verdict != "ok" for a in assessments) else "ok"
    return RunAssessment(overall=overall, tasks=assessments, repair_candidates=candidates)


def _transitive_dependents(task_list: TaskList) -> dict[str, set[str]]:
    """dependents[id] = every task downstream of id via depends_on edges."""
    direct: dict[str, set[str]] = defaultdict(set)
    for task in task_list.tasks:
        for dep in task.depends_on:
            direct[dep].add(task.id)
    out: dict[str, set[str]] = {}
    for tid in list(direct):
        seen: set[str] = set()
        queue = deque(direct[tid])
        while queue:
            node = queue.popleft()
            if node in seen:
                continue
            seen.add(node)
            queue.extend(direct.get(node, ()))
        out[tid] = seen
    return out


def _placeholder_ref(value: Any) -> str | None:
    if not isinstance(value, str):
        return None
    match = PLACEHOLDER_RE.fullmatch(value.strip())
    return match.group(1) if match else None


def _last_table_output(result: Any) -> Any | None:
    """The referenced task's representative output (matches TaskRunner's
    `outputs[-1]` resolution), if it is a non-empty table."""
    if result is None or not result.outputs:
        return None
    out = result.outputs[-1]
    if out.kind != "table" or not out.columns or not out.rows:
        return None
    return out


def _all_null_columns(table: Any) -> list[str]:
    cols: list[str] = []
    for i, name in enumerate(table.columns):
        if all(row[i] is None for row in table.rows if i < len(row)):
            cols.append(name)
    return cols


def _chart_issues(envelope: dict[str, Any]) -> list[str]:
    """§4.6 spec checks on a dataeyond.chart.v1 envelope. Purely structural —
    the spec builder is deterministic, so an issue here means the *plan args*
    picked a bad shape (wrong column, oversized category set), not a tool bug."""
    issues: list[str] = []
    chart_type = envelope.get("chart_type", "chart")
    plotly = envelope.get("plotly")
    traces = plotly.get("data") if isinstance(plotly, dict) else None
    if not isinstance(traces, list) or not traces:
        return [f"{chart_type} chart has no data traces"]

    for trace in traces:
        if not isinstance(trace, dict):
            continue
        name = trace.get("name")
        label = f"series {name!r}" if name else "a series"
        xs = trace.get("x") if trace.get("x") is not None else trace.get("labels")
        ys = trace.get("y") if trace.get("y") is not None else trace.get("values")
        xs = xs if isinstance(xs, list) else []
        ys = ys if isinstance(ys, list) else []
        if not xs or not ys or all(v is None for v in ys):
            issues.append(f"{chart_type} chart: {label} is empty")
            continue
        if len(xs) != len(ys):
            issues.append(
                f"{chart_type} chart: {label} has mismatched lengths "
                f"(x={len(xs)}, y={len(ys)})"
            )
        if chart_type == "bar" and len(xs) > _BAR_CATEGORY_CAP:
            issues.append(
                f"bar chart: {label} has {len(xs)} categories "
                f"(> {_BAR_CATEGORY_CAP}) — unreadable; aggregate or top-N first"
            )
        if chart_type == "pie" and len(xs) > _PIE_CATEGORY_CAP:
            issues.append(
                f"pie chart has {len(xs)} slices (> {_PIE_CATEGORY_CAP}) — "
                "unreadable; group the tail into 'other' or use a bar chart"
            )
        if chart_type in ("bar", "line", "scatter") and any(
            v is not None and not isinstance(v, int | float) for v in ys
        ):
            issues.append(f"{chart_type} chart: {label} has non-numeric y values")
    return issues