fix(adversary): fail closed per criterion when check_criterion dies

A live run died at map_criteria: one heavy criterion exhausted
max_iterations (10) with no fallback and sank the whole 17-criterion
review. check_criterion now has a 30-iteration budget, a tool-budget
instruction, and falls back to criterion_fault, which records a
schema-shaped UNMET verdict with a PIPELINE-FAULT evidence marker for
that one criterion; verdict.py renders it as DIED rather than judged.
This commit is contained in:
Dark-Alex-17 committed 2026-09-14 10:50:34 -06:00
1 parent 7dac4620a8
commit 3f49db86b2
5 files changed
+391 -8

No files matched your search

@@ -0,0 +1,91 @@
#!/usr/bin/env python3
"""Fail-closed per-criterion fault marker for the adversary graph.
Reached ONLY via check_criterion's `fallback:` route, after the engine has
written its "LLM node …failed: <chain>" text into `crit_verdict` through
state_updates. Converts that text into a schema-shaped UNMET verdict whose
evidence carries a "PIPELINE-FAULT:" marker, so one criterion check that dies
(max_iterations exhausted, API failure) is recorded as DIED rather than
judged, the other criteria still get verdicts, and the run is DIVERGES. The
map collects it like any other verdict — the branch never sinks the map.
"""
import json
import os
MAX_DETAIL_CHARS = 500
def load_state():
if path := os.environ.get("GRAPH_STATE_FILE"):
with open(path) as f:
return json.load(f)
return json.loads(os.environ.get("GRAPH_STATE", "{}"))
def criterion_fields(state):
c = state.get("criterion")
if isinstance(c, dict):
cid = c.get("id")
text = c.get("text")
return (
cid.strip() if isinstance(cid, str) and cid.strip() else "unknown",
text.strip() if isinstance(text, str) else "",
)
return "unknown", ""
def failure_detail(state):
captured = state.get("crit_verdict")
detail = captured.strip() if isinstance(captured, str) else ""
if not detail.startswith("LLM node"):
return "no failure text captured"
detail = detail.replace("\n", " ")
if len(detail) > MAX_DETAIL_CHARS:
detail = detail[: MAX_DETAIL_CHARS - 1] + "…"
return detail
def main():
state = load_state()
cid, ctext = criterion_fields(state)
detail = failure_detail(state)
print(
json.dumps(
{
"crit_verdict": {
"id": cid,
"text": ctext,
"status": "UNMET",
"evidence": f"PIPELINE-FAULT: criterion check failed — {detail}",
"complaint": (
f"criterion check DIED (pipeline fault) — {detail}; the "
"criterion was NOT verified and is treated as unmet "
"(fail-closed)"
),
}
}
)
)
try:
main()
except Exception as e: # noqa: BLE001 — the fault marker itself must never crash
print(
json.dumps(
{
"crit_verdict": {
"id": "unknown",
"text": "",
"status": "UNMET",
"evidence": f"PIPELINE-FAULT: criterion fault-marker script error: {e}",
"complaint": (
f"criterion check DIED (pipeline fault) — fault-marker script "
f"error: {e}; the criterion was NOT verified and is treated as "
"unmet (fail-closed)"
),
}
}
)
)
+16 -6
View File
@@ -6,14 +6,17 @@ holistic complaints, and zero pipeline faults. Everything else — including
a missing plan or a degraded (faulted) pipeline — is DIVERGES (fail-closed).
Any "PIPELINE-FAULT:" marker recorded in `pipeline_faults` (or a holistic
llm-node failure captured in `holistic_failure`) becomes complaint #1;
per-criterion results are still reported. Assembles the exact sentinel
format the callers route on. Never crashes into a silent verdict.
per-criterion results are still reported. A criterion whose check DIED
(criterion_fault's UNMET verdict, evidence prefixed "PIPELINE-FAULT:") is
rendered as died rather than judged. Assembles the exact sentinel format the
callers route on. Never crashes into a silent verdict.
"""
import json
import os
MAX_FAULT_DETAIL_CHARS = 300
DIED_MARKER = "criterion check DIED (pipeline fault)"
def load_state():
@@ -57,10 +60,17 @@ def append_criterion_complaints(lines, verdicts, extra, start):
text = (v.get("text") or "").strip().replace("\n", " ")
if len(text) > 140:
text = text[:137] + "…"
lines.append(
f'{i}. Acceptance criterion "{text}" — {v.get("status", "?").title()} — '
f'{(v.get("complaint") or "").strip()}'
)
complaint = (v.get("complaint") or "").strip()
evidence = v.get("evidence")
if isinstance(evidence, str) and evidence.startswith("PIPELINE-FAULT:"):
if not complaint.startswith(DIED_MARKER):
complaint = f"{DIED_MARKER} — {complaint}"
lines.append(f'{i}. Acceptance criterion "{text}" — {complaint}')
else:
lines.append(
f'{i}. Acceptance criterion "{text}" — {v.get("status", "?").title()} — '
f"{complaint}"
)
for c in extra:
i += 1
lines.append(