Hugging Face's logo

Spaces:
DataEyond
/
Agentic-Service-Data-Eyond-Catalog
Running

App Files Community
27
Agentic-Service-Data-Eyond-Catalog / src /agents /slow_path /checkpoint.py
ishaq101's picture
ishaq101
/fix planner and unstructured pipeline (#23)
7124acf
Raw History Blame
15.1 kB
"""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 a 10k-row subset)
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
CK7 ML model below its floor -> flag the tool whose own `quality` block says weak
"""
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.
#
# `repairable` is now ON the line (A23, 2026-09-21). It was the field that
# decided the whole trigger design and it was the one the log left out, so
# the hit rate could not be counted from logs without re-deriving which
# check produced which flag.
logger.info(
"repair_candidate",
task_id=task_id,
check=check,
reason=reason,
repairable=repairable,
)
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:
# "a subset", not "the first N rows" (D-2, 2026-09-23): the rows kept are
# stable now (SQL gets a deterministic ORDER BY, Parquet keeps file order)
# but they are NOT the earliest or a random sample — only a subset.
flag(tid, "CK3", f"table hit the {_TABLE_ROW_CAP:,}-row retrieval "
f"cap — findings describe a {_TABLE_ROW_CAP:,}-row subset of the "
"data, not all of it", 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)
# CK7 — an ML tool scored below its OWN floor, so its finding is weak
# evidence. The tool owns the judgement (it knows what "good" means for
# its method) and publishes it in the reserved `quality` block
# ({metric, value, floor, weak} — see the tool conventions); the
# checkpoint only surfaces it, so a NEW ML tool is covered the moment it
# emits the block, with no change here. Absent block -> nothing to check.
quality = out.value.get("quality") if isinstance(out.value, dict) else None
if isinstance(quality, dict) and quality.get("weak"):
# Direction-agnostic wording on purpose: "below the floor" is bad for
# r2 and silhouette but GOOD for MASE and p-value, so the message
# states the pair and lets the tool's own `note` explain it. The tool
# owns the verdict AND its explanation; the checkpoint only surfaces
# both. Without a note, fall back to the neutral phrasing.
detail = quality.get("note") or "the result is weak evidence"
flag(tid, "CK7",
f"{out.tool} reported {quality.get('metric')}="
f"{quality.get('value')} against a {quality.get('floor')} "
f"floor — {detail}",
repairable=False)
# 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.
consumed_refs: set[str] = set()
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))
if ref:
consumed_refs.add(ref)
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)
# CK5b — the same defect on a retrieval NO analyze_* consumed, i.e. one that goes
# straight to the assembler. Planner recipes R2/R2b answer a question with exactly
# one aggregate `retrieve_data` and no analyze_* step, so the consumer-side sweep
# above never runs for them and an all-null aggregate column (e.g. avg(PA) over a
# window where PA was never recorded) reached the answer unflagged. Found
# 2026-07-23, same root cause as the report-body bug. Restricted to
# `retrieve_data`: `check_*` outputs legitimately carry all-null columns (an
# uncounted table surfaces `table_row_count = None`).
for tid, result in results.items():
if tid in consumed_refs or result.status != "success":
continue
table = _last_table_output(result)
if table is None or table.tool != "retrieve_data":
continue
null_cols = _all_null_columns(table)
if null_cols:
flag(tid, "CK5",
f"column(s) {null_cols} in this result are entirely null — "
"findings 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"
# Per-run rollup — the measurement the repair edge is gated on (A23, 2026-09-21).
#
# Trigger B fires on a repairable flag or an outright task failure, budget 1. This
# line says what WOULD have happened on this turn, without anything happening: how
# many flags, how many repairable, whether a repair pass would have fired, and on
# which task. Counting `would_fire=True` across runs gives the real hit rate, which
# is what sizes the cost before the edge is built — checkpoint.py's own note above
# asked for exactly this ("the hit-rate decides whether a repair pass pays").
#
# Deliberately one line per run, not per flag: a rate is a property of the run.
failed = [tid for tid, r in results.items() if r.status == "failure"]
target = candidates[0].task_id if candidates else (failed[0] if failed else None)
logger.info(
"repair_hit_rate",
n_tasks=len(results),
n_flags=sum(len(v) for v in notes.values()),
n_repairable=len(candidates),
n_failed=len(failed),
would_fire=bool(candidates or failed),
would_repair=target,
overall=overall,
)
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