"""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