feat(TASK-023): deep-research hardening — fail-safe scripts, fault fallbacks, pipeline notes

R3 guards: parse_request (crash → ask_topic), bootstrap_research,
combine_findings (crash → fault-text findings; also lifts branch-local
PIPELINE-FAULT findings into pipeline_faults, deduped), reflexion_gate
(crash fails forward to synthesize), incorporate_feedback (crash still
loops with a generic feedback note), verify_sources (top-level guard +
pipeline_notes fold).

Fallbacks per the plan's naming pins: plan → plan_fault → end_fault
(fail-closed; gains state_updates plan_failure per the TASK-016 rule);
research_one_question → branch-local question_fault (map collects the
fault finding); vet_sources/critique → degrade-visibly markers that
continue to the stage each node's own next pointed to; synthesize →
synth_fault → end_fault. end_fault renders
'DEEP_RESEARCH FAILED — PIPELINE-FAULT: <fault text>'.

Accepted-path visibility: verify_sources folds pipeline_faults into
pipeline_notes ('' when fault-free), appended to end_accepted's output
and the approval question — happy-path output stays byte-identical.
initial_state seeds questions: [] so a dying plan inside the static
fan-out leaves the sibling knowledge_lookup lane mapping over zero
items while end_fault renders.

R5: 19 python3-guarded regression tests + GraphParser wiring test;
bundled warning baseline unchanged.
This commit is contained in:
Dark-Alex-17 committed 2026-09-14 10:50:33 -06:00
1 parent c98b6bc9cd
commit 0ce2654173
13 files changed
+1111 -10

No files matched your search

+68 -2
View File
@@ -38,6 +38,14 @@ initial_state:
research_attempts: 0 research_attempts: 0
local_context: '' local_context: ''
local_sources: '' local_sources: ''
# Fault-path defaults (TASK-023 hardening). `questions: []` also keeps the
# knowledge_lookup lane alive for one super-step when `plan` dies inside the
# static fan-out: research_each_question maps over zero items while the
# plan_fault → end_fault route renders the failure sentinel.
pipeline_faults: []
pipeline_notes: ''
fault_text: ''
questions: []
start: parse_request start: parse_request
@@ -91,8 +99,22 @@ nodes:
maxItems: 6 maxItems: 6
description: 3-5 specific, self-contained sub-questions. description: 3-5 specific, self-contained sub-questions.
required: [research_plan, questions] required: [research_plan, questions]
# On success plan_failure holds the node's structured JSON output; on
# failure the engine writes "LLM node …failed: <chain>" into it BEFORE
# routing to the fallback (schema properties are only merged on success).
state_updates:
plan_failure: '{{output}}'
fallback: plan_fault
next: research_each_question next: research_each_question
plan_fault:
id: plan_fault
type: script
description: Fail-closed marker for a dead planning stage (reached only via plan's fallback) — normalizes the failure text captured in plan_failure into a PIPELINE-FAULT and continues to end_fault; no plan means nothing to research, so the DEEP_RESEARCH FAILED sentinel always renders. The sibling knowledge_lookup lane survives one more super-step by mapping over the empty initial_state `questions` list.
script: scripts/plan_fault.py
fallback: end_fault
next: end_fault
knowledge_lookup: knowledge_lookup:
id: knowledge_lookup id: knowledge_lookup
type: rag type: rag
@@ -153,6 +175,16 @@ nodes:
max_attempts: 2 max_attempts: 2
state_updates: state_updates:
finding: '{{output}}' finding: '{{output}}'
# On failure the engine writes "LLM node …failed: <chain>" into `finding`
# via the state_updates above BEFORE routing to the fallback. The marker
# is branch-local: a map branch's fallback must stay inside the branch.
fallback: question_fault
question_fault:
id: question_fault
type: script
description: Fail-closed marker for a dead research lane (reached only via research_one_question's fallback route inside the map branch). Normalizes the engine's failure text in `finding` into a PIPELINE-FAULT finding the map collects — one dead lane never sinks the whole map. combine_findings later lifts PIPELINE-FAULT findings into pipeline_faults for the report's Pipeline notes.
script: scripts/question_fault.py
combine_findings: combine_findings:
id: combine_findings id: combine_findings
@@ -179,6 +211,15 @@ nodes:
max_iterations: 15 max_iterations: 15
state_updates: state_updates:
source_assessment: '{{output}}' source_assessment: '{{output}}'
fallback: vet_fault
next: critique
vet_fault:
id: vet_fault
type: script
description: Degrade-visibly marker for a dead vetting stage (reached only via vet_sources' fallback) — vetting is a quality lane, not the product. Records a PIPELINE-FAULT, rewrites source_assessment to a neutral "unvetted" note, and continues to critique, the stage vet_sources' own next pointed to.
script: scripts/vet_fault.py
fallback: critique
next: critique next: critique
critique: critique:
@@ -219,6 +260,15 @@ nodes:
tools: [] tools: []
state_updates: state_updates:
critique: '{{output}}' critique: '{{output}}'
fallback: critique_fault
next: reflexion_gate
critique_fault:
id: critique_fault
type: script
description: Degrade-visibly marker for a dead critique stage (reached only via critique's fallback) — critique is a quality lane, not the product. Records a PIPELINE-FAULT and rewrites `critique` to a neutral note with deliberately NO "VERDICT:" line, so reflexion_gate's malformed-critique PASS default proceeds to synthesis; then continues to reflexion_gate, the stage critique's own next pointed to.
script: scripts/critique_fault.py
fallback: reflexion_gate
next: reflexion_gate next: reflexion_gate
reflexion_gate: reflexion_gate:
@@ -245,8 +295,19 @@ nodes:
max_attempts: 2 max_attempts: 2
state_updates: state_updates:
report: '{{output}}' report: '{{output}}'
# On failure the engine writes "Agent node failed: <chain>" into `report`
# via the state_updates above BEFORE routing to the fallback.
fallback: synth_fault
next: verify_sources next: verify_sources
synth_fault:
id: synth_fault
type: script
description: Fail-closed marker for a dead synthesis stage (reached only via synthesize's fallback) — no report means no product. Normalizes the "Agent node failed:" text captured in `report` into a PIPELINE-FAULT and continues to end_fault, which renders the DEEP_RESEARCH FAILED sentinel.
script: scripts/synth_fault.py
fallback: end_fault
next: end_fault
verify_sources: verify_sources:
id: verify_sources id: verify_sources
type: script type: script
@@ -262,7 +323,7 @@ nodes:
{{report}} {{report}}
---- ----
{{source_check}} {{source_check}}{{pipeline_notes}}
---- ----
Accept this report? Pick "accept" or "reject", or type specific Accept this report? Pick "accept" or "reject", or type specific
@@ -285,9 +346,14 @@ nodes:
end_accepted: end_accepted:
id: end_accepted id: end_accepted
type: end type: end
output: '{{report}}' output: '{{report}}{{pipeline_notes}}'
end_rejected: end_rejected:
id: end_rejected id: end_rejected
type: end type: end
output: "Research on '{{topic}}' was rejected and discarded." output: "Research on '{{topic}}' was rejected and discarded."
end_fault:
id: end_fault
type: end
output: 'DEEP_RESEARCH FAILED — PIPELINE-FAULT: {{fault_text}}'
@@ -6,6 +6,12 @@ list on this node fans out into two parallel branches (the LLM planner and
the RAG knowledge lookup) as a single super-step. The validator requires the RAG knowledge lookup) as a single super-step. The validator requires
declared parallel-branch script outputs, so we emit an empty JSON object declared parallel-branch script outputs, so we emit an empty JSON object
explicitly here. explicitly here.
Fail-safe (R3): the node reads no state and computes nothing, so its
happy-path output IS the sane degraded output — the guard emits the same
empty object on any crash. No PIPELINE-FAULT is recorded: appending one
would require reading state (the very thing that crashed) or clobbering
faults recorded upstream by parse_request, and a crash here loses nothing.
""" """
import json import json
@@ -14,5 +20,7 @@ def main():
print(json.dumps({})) print(json.dumps({}))
if __name__ == "__main__": try:
main() main()
except Exception: # noqa: BLE001 — the fan-out source must never crash the graph
print(json.dumps({}))
@@ -6,6 +6,18 @@ one entry per sub-question, in input order). Downstream nodes
(`vet_sources`, `critique`, `synthesize`) read `{{findings}}` as a (`vet_sources`, `critique`, `synthesize`) read `{{findings}}` as a
single block, so this script renders the array as a Markdown document single block, so this script renders the array as a Markdown document
with one section per question. with one section per question.
Fault visibility: a research lane that died lands here as a finding
starting with "PIPELINE-FAULT:" (written by question_fault inside the map
branch, where `pipeline_faults` itself is out of reach — map branches only
surface their collected output_key). This script lifts such findings into
`pipeline_faults` (deduplicated, since a reflexion or feedback loop can
re-run the map) so dead lanes surface in the report's Pipeline notes.
Fail-safe (R3): on a crash (e.g. malformed GRAPH_STATE), `findings`
becomes the fault text itself and a PIPELINE-FAULT entry is recorded —
the pipeline proceeds visibly degraded, and the human approval gate is
the backstop. Existing faults are preserved best-effort.
""" """
import json import json
import os import os
@@ -19,21 +31,40 @@ def load_state():
return json.loads(os.environ.get("GRAPH_STATE", "{}")) return json.loads(os.environ.get("GRAPH_STATE", "{}"))
def safe_faults():
try:
state = load_state()
return [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)]
except Exception: # noqa: BLE001 — best-effort preservation only
return []
def main(): def main():
state = load_state() state = load_state()
questions = state.get("questions") or [] questions = state.get("questions") or []
per_question = state.get("question_findings") or [] per_question = state.get("question_findings") or []
faults = [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)]
sections = [] sections = []
for idx, q in enumerate(questions): for idx, q in enumerate(questions):
body = per_question[idx] if idx < len(per_question) else "" body = per_question[idx] if idx < len(per_question) else ""
if isinstance(body, dict) or isinstance(body, list): if isinstance(body, dict) or isinstance(body, list):
body = json.dumps(body, indent=2) body = json.dumps(body, indent=2)
if isinstance(body, str) and body.startswith("PIPELINE-FAULT:") and body not in faults:
faults.append(body)
sections.append(f"## {q}\n\n{body}") sections.append(f"## {q}\n\n{body}")
findings = "\n\n".join(sections) if sections else "No findings gathered." findings = "\n\n".join(sections) if sections else "No findings gathered."
print(json.dumps({"findings": findings})) print(json.dumps({"findings": findings, "pipeline_faults": faults}))
if __name__ == "__main__": try:
main() main()
except Exception as e: # noqa: BLE001 — a crashed join must degrade, not die
faults = safe_faults()
fault = (
f"PIPELINE-FAULT: combining findings crashed — {e}; per-question "
"findings may exist in state but could not be joined"
)
faults.append(fault)
print(json.dumps({"findings": fault, "pipeline_faults": faults}))
@@ -0,0 +1,61 @@
#!/usr/bin/env python3
"""Degrade-visibly marker for a dead critique stage of deep-research.
Reached ONLY via critique's `fallback:` route. Grounded in engine behavior:
on llm-node failure the engine writes "LLM node …failed: <chain>" into
`critique` via the node's state_updates BEFORE routing to the fallback.
Critique is a quality lane, not the product, so this path degrades instead
of failing closed: it records a PIPELINE-FAULT (surfaced later in the
report's Pipeline notes) and rewrites `critique` to a neutral note, then
continues to `reflexion_gate` — the stage critique's own `next` pointed to.
The rewritten note deliberately contains NO "VERDICT:" line:
reflexion_gate's parse_verdict defaults to PASS when no verdict line is
found (its documented malformed-critique behavior), so a dead critique
proceeds to synthesis instead of burning the revision budget on a fault.
"""
import json
import os
MAX_DETAIL_CHARS = 500
CRITIQUE_NOTE = (
"Critique unavailable — the critique stage failed (pipeline fault); "
"proceeding to synthesis without an automated review pass."
)
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 main():
state = load_state()
faults = [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)]
raw = state.get("critique")
detail = raw.strip().replace("\n", " ") if isinstance(raw, str) else ""
if not detail.startswith("LLM node"):
detail = "the critique stage died without recording a failure detail"
if len(detail) > MAX_DETAIL_CHARS:
detail = detail[: MAX_DETAIL_CHARS - 1] + "…"
faults.append(f"PIPELINE-FAULT: critique failed — {detail}")
print(json.dumps({"pipeline_faults": faults, "critique": CRITIQUE_NOTE}))
try:
main()
except Exception as e: # noqa: BLE001 — the fault marker itself must never crash
print(
json.dumps(
{
"pipeline_faults": [
f"PIPELINE-FAULT: critique failed — fault-marker script error: {e}"
],
"critique": CRITIQUE_NOTE,
}
)
)
@@ -10,6 +10,13 @@ reflexion counter is reset so the user-driven pass gets a fresh revision
budget. budget.
Routing (`_next`): always research_each_question. Routing (`_next`): always research_each_question.
Fail-safe (R3): a crash here means the user's feedback text cannot be
recovered, but their INTENT — another research pass — is unambiguous
(they chose neither accept nor reject). The degraded route therefore
still loops to `research_each_question`, with a generic feedback note
naming the loss and a PIPELINE-FAULT entry (best-effort preserving
existing faults) that surfaces in the next approval's Pipeline notes.
""" """
import json import json
import os import os
@@ -23,6 +30,14 @@ def load_state():
return json.loads(os.environ.get("GRAPH_STATE", "{}")) return json.loads(os.environ.get("GRAPH_STATE", "{}"))
def safe_faults():
try:
state = load_state()
return [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)]
except Exception: # noqa: BLE001 — best-effort preservation only
return []
def main(): def main():
state = load_state() state = load_state()
feedback = (state.get("decision") or "").strip() feedback = (state.get("decision") or "").strip()
@@ -37,5 +52,26 @@ def main():
print(json.dumps(output)) print(json.dumps(output))
if __name__ == "__main__": try:
main() main()
except Exception as e: # noqa: BLE001 — a crashed feedback fold must degrade, not die
faults = safe_faults()
faults.append(
f"PIPELINE-FAULT: feedback incorporation crashed — {e}; the reviewer's "
"feedback text could not be recovered"
)
print(
json.dumps(
{
"_next": "research_each_question",
"research_attempts": 0,
"research_feedback": (
"The user reviewed the report and asked for changes, but "
"the feedback text could not be recovered (pipeline "
"fault). Re-run the research pass with extra care and "
"note this loss."
),
"pipeline_faults": faults,
}
)
)
@@ -9,6 +9,13 @@ supply one interactively.
Routing (`_next`): Routing (`_next`):
- prompt present -> (no _next; static next: plan) - prompt present -> (no _next; static next: plan)
- prompt empty -> ask_topic - prompt empty -> ask_topic
Fail-safe (R3): a crash here (e.g. malformed GRAPH_STATE) means the
caller's prompt cannot be read at all, so proceeding silently is
meaningless — the degraded route is `ask_topic`, which asks the user for
the topic directly, plus a PIPELINE-FAULT entry so the loss is visible in
the final report's Pipeline notes. Clobbering `pipeline_faults` is safe
here: this is the start node, so the list can only hold its initial [].
""" """
import json import json
import os import os
@@ -31,5 +38,17 @@ def main():
print(json.dumps({"_next": "ask_topic"})) print(json.dumps({"_next": "ask_topic"}))
if __name__ == "__main__": try:
main() main()
except Exception as e: # noqa: BLE001 — a crashed entry router must degrade, not die
print(
json.dumps(
{
"_next": "ask_topic",
"pipeline_faults": [
f"PIPELINE-FAULT: request parsing crashed — {e}; asked the "
"user for the research topic instead"
],
}
)
)
@@ -0,0 +1,58 @@
#!/usr/bin/env python3
"""Fail-closed marker for a dead planning stage of the deep-research graph.
Reached ONLY via plan's `fallback:` route. Grounded in engine behavior: on
llm-node failure the engine writes "LLM node …failed: <chain>" into
`plan_failure` via the node's state_updates BEFORE routing to the fallback
(on success the same key holds the node's structured JSON output, which
never starts with the "LLM node" prefix). No plan means no sub-questions —
there is nothing to research — so this path is fail-closed: record the
PIPELINE-FAULT and continue to `end_fault`, which renders the
DEEP_RESEARCH FAILED sentinel.
The planner runs inside the static fan-out [plan, knowledge_lookup]. While
this marker routes to end_fault, the sibling branch proceeds one more
super-step into `research_each_question`, whose `{{questions}}` resolves to
the initial_state [] — the doomed lane maps over zero items and the graph
returns end_fault's output at the next join.
"""
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 main():
state = load_state()
faults = [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)]
raw = state.get("plan_failure")
detail = raw.strip().replace("\n", " ") if isinstance(raw, str) else ""
if not detail.startswith("LLM node"):
detail = "the planning stage died without recording a failure detail"
if len(detail) > MAX_DETAIL_CHARS:
detail = detail[: MAX_DETAIL_CHARS - 1] + "…"
fault_text = f"planning failed — nothing to research: {detail}"
faults.append(f"PIPELINE-FAULT: {fault_text}")
print(json.dumps({"fault_text": fault_text, "pipeline_faults": faults}))
try:
main()
except Exception as e: # noqa: BLE001 — the fault marker itself must never crash
fault_text = f"planning failed — fault-marker script error: {e}"
print(
json.dumps(
{
"fault_text": fault_text,
"pipeline_faults": [f"PIPELINE-FAULT: {fault_text}"],
}
)
)
@@ -0,0 +1,52 @@
#!/usr/bin/env python3
"""Fail-closed marker for a dead research lane of the deep-research graph.
Reached ONLY via research_one_question's `fallback:` route inside the
`research_each_question` map branch (branch-local — a map branch's fallback
must stay inside the branch subgraph). Grounded in engine behavior: on
llm-node failure the engine writes "LLM node …failed: <chain>" into
`finding` via the node's state_updates BEFORE routing to the fallback (on
success the same key holds the researched findings, which never start with
the "LLM node" prefix). This node normalizes the failure text into a
PIPELINE-FAULT finding and the chain ends here — the map collects the fault
as the lane's `finding`, so one dead research lane never sinks the whole
map. combine_findings later lifts PIPELINE-FAULT findings into
`pipeline_faults` so the fault also surfaces in the report's Pipeline notes.
"""
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 fault_finding(detail):
return {
"finding": (
f"PIPELINE-FAULT: question research failed — finding unavailable — {detail}"
)
}
def main():
state = load_state()
raw = state.get("finding")
detail = raw.strip().replace("\n", " ") if isinstance(raw, str) else ""
if not detail.startswith("LLM node"):
detail = "the research lane died without recording a failure detail"
if len(detail) > MAX_DETAIL_CHARS:
detail = detail[: MAX_DETAIL_CHARS - 1] + "…"
print(json.dumps(fault_finding(detail)))
try:
main()
except Exception as e: # noqa: BLE001 — the fault marker itself must never crash
print(json.dumps(fault_finding(f"fault-marker script error: {e}")))
@@ -15,6 +15,12 @@ Routing (`_next`):
Reflexion is a best-effort quality booster, not a hard gate: once the Reflexion is a best-effort quality booster, not a hard gate: once the
budget is spent the workflow proceeds anyway, and the human approval budget is spent the workflow proceeds anyway, and the human approval
step is the final backstop. step is the final backstop.
Fail-safe (R3): a crash here fails FORWARD to `synthesize` — the same
direction as the existing malformed-critique PASS default — with a
PIPELINE-FAULT entry recorded (best-effort preserving existing faults),
so a broken gate costs the run its automated critique routing, never the
run itself.
""" """
import json import json
import os import os
@@ -33,6 +39,14 @@ def load_state():
return json.loads(os.environ.get("GRAPH_STATE", "{}")) return json.loads(os.environ.get("GRAPH_STATE", "{}"))
def safe_faults():
try:
state = load_state()
return [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)]
except Exception: # noqa: BLE001 — best-effort preservation only
return []
def as_int(value, default=0): def as_int(value, default=0):
try: try:
return int(value) return int(value)
@@ -72,5 +86,12 @@ def main():
print(json.dumps(output)) print(json.dumps(output))
if __name__ == "__main__": try:
main() main()
except Exception as e: # noqa: BLE001 — a crashed gate fails forward, never dies
faults = safe_faults()
faults.append(
f"PIPELINE-FAULT: reflexion gate crashed — {e}; proceeded to synthesis "
"without automated critique routing"
)
print(json.dumps({"_next": "synthesize", "pipeline_faults": faults}))
@@ -0,0 +1,51 @@
#!/usr/bin/env python3
"""Fail-closed marker for a dead synthesis stage of the deep-research graph.
Reached ONLY via synthesize's `fallback:` route. Grounded in engine
behavior: on agent-node failure the engine writes "Agent node failed:
{e:#}" into `report` via the node's state_updates BEFORE routing to the
fallback (on success the same key holds the report-writer's report, which
never starts with that prefix). No report means no product, so this path is
fail-closed: normalize the failure text, record the PIPELINE-FAULT, and
continue to `end_fault`, which renders the DEEP_RESEARCH FAILED sentinel.
"""
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 main():
state = load_state()
faults = [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)]
raw = state.get("report")
detail = raw.strip().replace("\n", " ") if isinstance(raw, str) else ""
if not detail.startswith("Agent node failed:"):
detail = "the synthesis stage died without recording a failure detail"
if len(detail) > MAX_DETAIL_CHARS:
detail = detail[: MAX_DETAIL_CHARS - 1] + "…"
fault_text = f"report synthesis failed — no report was produced: {detail}"
faults.append(f"PIPELINE-FAULT: {fault_text}")
print(json.dumps({"fault_text": fault_text, "pipeline_faults": faults}))
try:
main()
except Exception as e: # noqa: BLE001 — the fault marker itself must never crash
fault_text = f"report synthesis failed — fault-marker script error: {e}"
print(
json.dumps(
{
"fault_text": fault_text,
"pipeline_faults": [f"PIPELINE-FAULT: {fault_text}"],
}
)
)
@@ -6,6 +6,19 @@ request, and writes a `source_check` summary into state so the human
reviewer sees broken citations at the approval step. reviewer sees broken citations at the approval step.
Times out per request so a slow source cannot stall the graph. Times out per request so a slow source cannot stall the graph.
Pipeline-notes fold: end-node templates render unconditionally, so the
conditional "## Pipeline notes" section of the accepted report is built
HERE — the last stop before the approval gate on every accepted path,
including reflexion and feedback re-loops. `pipeline_notes` is the
rendered section when `pipeline_faults` is non-empty and the empty
string otherwise, keeping the happy-path `{{report}}{{pipeline_notes}}`
output byte-identical to the pre-hardening `{{report}}`.
Fail-safe (R3): per-URL probes were already guarded; the top level now
is too. On a crash the summary says sources were NOT checked, a
PIPELINE-FAULT is recorded (best-effort preserving existing faults), and
pipeline_notes still renders — the approval gate remains the backstop.
""" """
import json import json
import os import os
@@ -25,6 +38,21 @@ def load_state():
return json.loads(os.environ.get("GRAPH_STATE", "{}")) return json.loads(os.environ.get("GRAPH_STATE", "{}"))
def safe_faults():
try:
state = load_state()
return [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)]
except Exception: # noqa: BLE001 — best-effort preservation only
return []
def render_notes(faults):
entries = [f for f in (faults or []) if isinstance(f, str) and f.strip()]
if not entries:
return ""
return "\n\n## Pipeline notes\n\n" + "\n".join(f"- {e}" for e in entries)
def reachable(url, timeout=5.0): def reachable(url, timeout=5.0):
req = urllib.request.Request(url, method="HEAD") req = urllib.request.Request(url, method="HEAD")
try: try:
@@ -62,8 +90,27 @@ def main():
+ "\n".join(results) + "\n".join(results)
) )
print(json.dumps({"source_check": summary})) notes = render_notes(state.get("pipeline_faults"))
print(json.dumps({"source_check": summary, "pipeline_notes": notes}))
if __name__ == "__main__": try:
main() main()
except Exception as e: # noqa: BLE001 — a crashed source check must degrade, not die
faults = safe_faults()
faults.append(
f"PIPELINE-FAULT: source verification crashed — {e}; cited sources "
"were not checked for reachability"
)
print(
json.dumps(
{
"source_check": (
"Source verification crashed (pipeline fault) — cited "
"sources were NOT checked for reachability."
),
"pipeline_notes": render_notes(faults),
"pipeline_faults": faults,
}
)
)
@@ -0,0 +1,60 @@
#!/usr/bin/env python3
"""Degrade-visibly marker for a dead source-vetting stage of deep-research.
Reached ONLY via vet_sources' `fallback:` route. Grounded in engine
behavior: on llm-node failure the engine writes "LLM node …failed: <chain>"
into `source_assessment` via the node's state_updates BEFORE routing to the
fallback. Vetting is a quality lane, not the product, so this path degrades
instead of failing closed: it records a PIPELINE-FAULT (surfaced later in
the report's Pipeline notes) and rewrites `source_assessment` to a neutral
"unvetted" note so downstream prompts (critique, synthesize) never carry
raw engine error text, then continues to `critique` — the stage
vet_sources' own `next` pointed to.
"""
import json
import os
MAX_DETAIL_CHARS = 500
UNVETTED_NOTE = (
"Source credibility assessment unavailable — the vetting stage failed "
"(pipeline fault); treat every cited source as unvetted and needing "
"corroboration."
)
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 main():
state = load_state()
faults = [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)]
raw = state.get("source_assessment")
detail = raw.strip().replace("\n", " ") if isinstance(raw, str) else ""
if not detail.startswith("LLM node"):
detail = "the vetting stage died without recording a failure detail"
if len(detail) > MAX_DETAIL_CHARS:
detail = detail[: MAX_DETAIL_CHARS - 1] + "…"
faults.append(f"PIPELINE-FAULT: source vetting failed — {detail}")
print(json.dumps({"pipeline_faults": faults, "source_assessment": UNVETTED_NOTE}))
try:
main()
except Exception as e: # noqa: BLE001 — the fault marker itself must never crash
print(
json.dumps(
{
"pipeline_faults": [
f"PIPELINE-FAULT: source vetting failed — "
f"fault-marker script error: {e}"
],
"source_assessment": UNVETTED_NOTE,
}
)
)
+591
View File
@@ -10378,6 +10378,597 @@ mod tests {
} }
} }
// ---- deep-research suite-script regression tests (R5) ----
//
// The deep-research graph's R3 guards and fault markers, exercised by
// invoking `python3 <script>` with a synthetic GRAPH_STATE env. The raw
// runner feeds deliberately malformed JSON to prove every guarded script
// emits a sane degraded output instead of crashing the node.
fn run_deep_research_script(script: &str, state: &serde_json::Value) -> serde_json::Value {
run_deep_research_script_raw(script, &state.to_string())
}
fn run_deep_research_script_raw(script: &str, raw_state: &str) -> serde_json::Value {
let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("assets/agents/deep-research/scripts")
.join(script);
let out = std::process::Command::new("python3")
.arg(&path)
.env("GRAPH_STATE", raw_state)
.output()
.unwrap_or_else(|e| panic!("failed to invoke python3 {script}: {e}"));
assert!(
out.status.success(),
"{script} exited nonzero: {}",
String::from_utf8_lossy(&out.stderr)
);
serde_json::from_slice(&out.stdout).unwrap_or_else(|e| {
panic!(
"{script} stdout is not JSON ({e}): {}",
String::from_utf8_lossy(&out.stdout)
)
})
}
#[test]
fn deep_research_scripts_survive_malformed_state() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
// R3: every guarded script and fault marker must exit 0 with JSON
// stdout even when GRAPH_STATE is not JSON at all (the runner
// asserts both); per-script degraded shapes are pinned below.
for script in [
"parse_request.py",
"bootstrap_research.py",
"combine_findings.py",
"reflexion_gate.py",
"incorporate_feedback.py",
"verify_sources.py",
"plan_fault.py",
"question_fault.py",
"vet_fault.py",
"critique_fault.py",
"synth_fault.py",
] {
let _ = run_deep_research_script_raw(script, "not json {");
}
}
#[test]
fn deep_research_parse_request_crash_asks_user() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
// An unreadable state means the caller's prompt is lost — the sane
// degraded route is asking the user for the topic directly.
let out = run_deep_research_script_raw("parse_request.py", "not json {");
assert_eq!(out["_next"], "ask_topic", "{out}");
assert!(
out["pipeline_faults"][0]
.as_str()
.unwrap()
.starts_with("PIPELINE-FAULT: request parsing crashed"),
"{out}"
);
}
#[test]
fn deep_research_parse_request_happy_path_unchanged() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let out =
run_deep_research_script("parse_request.py", &json!({"initial_prompt": " quantum "}));
assert_eq!(out, json!({"topic": "quantum"}));
let out = run_deep_research_script("parse_request.py", &json!({"initial_prompt": ""}));
assert_eq!(out, json!({"_next": "ask_topic"}));
}
#[test]
fn deep_research_bootstrap_survives_malformed_state() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
// The fan-out source's happy-path output IS its degraded output.
assert_eq!(
run_deep_research_script_raw("bootstrap_research.py", "not json {"),
json!({})
);
}
#[test]
fn deep_research_combine_findings_crash_degrades() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let out = run_deep_research_script_raw("combine_findings.py", "not json {");
let findings = out["findings"].as_str().unwrap();
assert!(
findings.starts_with("PIPELINE-FAULT: combining findings crashed"),
"{out}"
);
assert_eq!(out["pipeline_faults"], json!([findings]), "{out}");
}
#[test]
fn deep_research_combine_findings_happy_path_unchanged() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let state = json!({
"questions": ["Q1", "Q2"],
"question_findings": ["finding one", "finding two"],
"pipeline_faults": []
});
let out = run_deep_research_script("combine_findings.py", &state);
assert_eq!(
out,
json!({
"findings": "## Q1\n\nfinding one\n\n## Q2\n\nfinding two",
"pipeline_faults": []
})
);
}
#[test]
fn deep_research_combine_findings_lifts_question_faults() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
// A dead research lane's PIPELINE-FAULT finding (written branch-local
// by question_fault, where pipeline_faults is out of reach) must be
// lifted into pipeline_faults so it reaches the Pipeline notes.
let fault = "PIPELINE-FAULT: question research failed — finding unavailable — LLM node failed: boom";
let out = run_deep_research_script(
"combine_findings.py",
&json!({
"questions": ["Q1", "Q2"],
"question_findings": ["finding one", fault],
"pipeline_faults": []
}),
);
assert_eq!(out["pipeline_faults"], json!([fault]), "{out}");
assert!(out["findings"].as_str().unwrap().contains(fault), "{out}");
// Reflexion/feedback loops re-run the map: the lift must deduplicate.
let out = run_deep_research_script(
"combine_findings.py",
&json!({
"questions": ["Q1"],
"question_findings": [fault],
"pipeline_faults": [fault]
}),
);
assert_eq!(out["pipeline_faults"], json!([fault]), "{out}");
}
#[test]
fn deep_research_reflexion_gate_crash_fails_forward() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
// Same direction as the documented malformed-critique PASS default:
// a broken gate costs the automated critique routing, never the run.
let out = run_deep_research_script_raw("reflexion_gate.py", "not json {");
assert_eq!(out["_next"], "synthesize", "{out}");
assert!(
out["pipeline_faults"][0]
.as_str()
.unwrap()
.starts_with("PIPELINE-FAULT: reflexion gate crashed"),
"{out}"
);
}
#[test]
fn deep_research_reflexion_gate_happy_paths_unchanged() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let out = run_deep_research_script(
"reflexion_gate.py",
&json!({"critique": "VERDICT: REVISE\nFEEDBACK: missing X", "research_attempts": 0}),
);
assert_eq!(out["_next"], "research_each_question", "{out}");
assert_eq!(out["research_attempts"], 1, "{out}");
let out = run_deep_research_script(
"reflexion_gate.py",
&json!({"critique": "VERDICT: PASS\nFEEDBACK: none", "research_attempts": 0}),
);
assert_eq!(out, json!({"_next": "synthesize"}));
}
#[test]
fn deep_research_incorporate_feedback_crash_still_loops() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
// The user's intent (another pass) is unambiguous even when their
// feedback text is lost — the degraded route still loops.
let out = run_deep_research_script_raw("incorporate_feedback.py", "not json {");
assert_eq!(out["_next"], "research_each_question", "{out}");
assert_eq!(out["research_attempts"], 0, "{out}");
assert!(
out["research_feedback"]
.as_str()
.unwrap()
.contains("could not be recovered"),
"{out}"
);
assert!(
out["pipeline_faults"][0]
.as_str()
.unwrap()
.starts_with("PIPELINE-FAULT: feedback incorporation crashed"),
"{out}"
);
}
#[test]
fn deep_research_incorporate_feedback_happy_path_unchanged() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let out =
run_deep_research_script("incorporate_feedback.py", &json!({"decision": "add X"}));
assert_eq!(out["_next"], "research_each_question", "{out}");
assert_eq!(out["research_attempts"], 0, "{out}");
assert!(
out["research_feedback"].as_str().unwrap().contains("add X"),
"{out}"
);
assert!(out.get("pipeline_faults").is_none(), "{out}");
}
#[test]
fn deep_research_verify_sources_crash_degrades() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let out = run_deep_research_script_raw("verify_sources.py", "not json {");
assert!(
out["source_check"]
.as_str()
.unwrap()
.contains("NOT checked"),
"{out}"
);
assert!(
out["pipeline_notes"]
.as_str()
.unwrap()
.contains("## Pipeline notes"),
"the crash fault must still render in the notes: {out}"
);
assert!(
out["pipeline_faults"][0]
.as_str()
.unwrap()
.starts_with("PIPELINE-FAULT: source verification crashed"),
"{out}"
);
}
#[test]
fn deep_research_verify_sources_folds_pipeline_notes() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
// No URLs in the report → no network probes; the notes fold is what
// this pins. Non-empty faults render the "## Pipeline notes" section…
let out = run_deep_research_script(
"verify_sources.py",
&json!({
"report": "plain text, no links",
"pipeline_faults": ["PIPELINE-FAULT: critique failed — LLM node failed: x"]
}),
);
assert_eq!(
out["source_check"],
"No web sources were cited in the report."
);
assert_eq!(
out["pipeline_notes"],
"\n\n## Pipeline notes\n\n- PIPELINE-FAULT: critique failed — LLM node failed: x"
);
// …and a fault-free run folds to "" so the accepted-path output
// ({{report}}{{pipeline_notes}}) stays byte-identical.
let out = run_deep_research_script(
"verify_sources.py",
&json!({"report": "plain text, no links", "pipeline_faults": []}),
);
assert_eq!(out["pipeline_notes"], "");
}
#[test]
fn deep_research_plan_fault_normalizes_capture() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let out = run_deep_research_script(
"plan_fault.py",
&json!({"plan_failure": "LLM node failed: boom", "pipeline_faults": []}),
);
assert_eq!(
out["fault_text"],
"planning failed — nothing to research: LLM node failed: boom"
);
assert_eq!(
out["pipeline_faults"],
json!(["PIPELINE-FAULT: planning failed — nothing to research: LLM node failed: boom"])
);
// On success plan_failure holds the node's structured JSON output —
// the prefix anchor must never mistake it for a failure detail.
let out = run_deep_research_script(
"plan_fault.py",
&json!({"plan_failure": "{\"research_plan\":\"ok\"}"}),
);
assert!(
out["fault_text"]
.as_str()
.unwrap()
.contains("died without recording"),
"{out}"
);
}
#[test]
fn deep_research_question_fault_normalizes_finding() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let out = run_deep_research_script(
"question_fault.py",
&json!({"finding": "LLM node failed: boom"}),
);
assert_eq!(
out,
json!({
"finding":
"PIPELINE-FAULT: question research failed — finding unavailable — LLM node failed: boom"
})
);
// state_updates run on success too — a real finding (no "LLM node"
// prefix) must not leak into the fault detail.
let out = run_deep_research_script(
"question_fault.py",
&json!({"finding": "real research findings"}),
);
let finding = out["finding"].as_str().unwrap();
assert!(
finding.starts_with("PIPELINE-FAULT: question research failed — finding unavailable —"),
"{out}"
);
assert!(finding.contains("died without recording"), "{out}");
}
#[test]
fn deep_research_vet_fault_degrades_and_preserves() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let out = run_deep_research_script(
"vet_fault.py",
&json!({
"source_assessment": "LLM node failed: kaboom",
"pipeline_faults": ["PIPELINE-FAULT: earlier"]
}),
);
assert_eq!(
out["pipeline_faults"],
json!([
"PIPELINE-FAULT: earlier",
"PIPELINE-FAULT: source vetting failed — LLM node failed: kaboom"
]),
"existing faults must be preserved"
);
let assessment = out["source_assessment"].as_str().unwrap();
assert!(
!assessment.starts_with("LLM node"),
"raw engine error text must not flow into downstream prompts: {out}"
);
assert!(assessment.contains("unvetted"), "{out}");
}
#[test]
fn deep_research_critique_fault_rides_the_pass_default() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let out = run_deep_research_script(
"critique_fault.py",
&json!({"critique": "LLM node failed: dead", "pipeline_faults": []}),
);
assert_eq!(
out["pipeline_faults"],
json!(["PIPELINE-FAULT: critique failed — LLM node failed: dead"])
);
let note = out["critique"].as_str().unwrap();
assert!(
!note.contains("VERDICT:"),
"the note must not synthesize a verdict line: {note}"
);
// Integration: the rewritten critique rides reflexion_gate's
// malformed-critique PASS default straight to synthesis.
let gate = run_deep_research_script(
"reflexion_gate.py",
&json!({"critique": note, "research_attempts": 0}),
);
assert_eq!(gate, json!({"_next": "synthesize"}));
}
#[test]
fn deep_research_synth_fault_normalizes_report() {
if !cmd_available("python3") {
eprintln!("skipping: python3 not available");
return;
}
let out = run_deep_research_script(
"synth_fault.py",
&json!({"report": "Agent node failed: writer died", "pipeline_faults": []}),
);
assert_eq!(
out["fault_text"],
"report synthesis failed — no report was produced: Agent node failed: writer died"
);
assert_eq!(
out["pipeline_faults"],
json!([
"PIPELINE-FAULT: report synthesis failed — no report was produced: Agent node failed: writer died"
])
);
// A real report never starts with the engine's failure prefix.
let out = run_deep_research_script("synth_fault.py", &json!({"report": "# A real report"}));
assert!(
out["fault_text"]
.as_str()
.unwrap()
.contains("died without recording"),
"{out}"
);
}
#[test]
fn deep_research_fault_wiring_preserves_sentinel() {
use crate::graph::{GraphParser, NodeType};
let dir =
std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("assets/agents/deep-research");
let graph = GraphParser::new(&dir)
.load_from_file(dir.join("graph.yaml"))
.unwrap();
// plan: fail-closed. Its output_schema + fallback combination requires
// a state_updates capture (the TASK-016 warning) so plan_fault can see
// why it was reached.
let NodeType::Llm(plan) = &graph.get_node("plan").unwrap().node_type else {
panic!("plan must be an llm node")
};
assert!(plan.output_schema.is_some());
assert_eq!(plan.fallback.as_deref(), Some("plan_fault"));
assert!(
plan.state_updates
.as_ref()
.is_some_and(|u| u.contains_key("plan_failure")),
"plan must capture its failure text for the fault marker"
);
let plan_fault = graph.get_node("plan_fault").unwrap();
assert_eq!(plan_fault.next_target(), Some("end_fault"));
let NodeType::Script(pf) = &plan_fault.node_type else {
panic!("plan_fault must be a script node")
};
assert_eq!(pf.fallback.as_deref(), Some("end_fault"));
// research_one_question: branch-local marker; no `next` so the map
// collects the fault as the lane's finding.
let NodeType::Llm(question) = &graph.get_node("research_one_question").unwrap().node_type
else {
panic!("research_one_question must be an llm node")
};
assert_eq!(question.fallback.as_deref(), Some("question_fault"));
assert!(
question
.state_updates
.as_ref()
.is_some_and(|u| u.contains_key("finding")),
"the failure text must land in the key question_fault normalizes"
);
assert!(
graph.get_node("question_fault").unwrap().next.is_none(),
"question_fault must end the branch chain so the map collects the finding"
);
let NodeType::Map(map) = &graph.get_node("research_each_question").unwrap().node_type
else {
panic!("research_each_question must be a map node")
};
assert_eq!(map.over, "{{questions}}");
assert_eq!(map.output_key, "finding");
// vet_sources / critique: degrade-visibly markers that continue to the
// stage each node's own `next` pointed to.
let NodeType::Llm(vet) = &graph.get_node("vet_sources").unwrap().node_type else {
panic!("vet_sources must be an llm node")
};
assert_eq!(vet.fallback.as_deref(), Some("vet_fault"));
assert_eq!(
graph.get_node("vet_fault").unwrap().next_target(),
Some("critique")
);
let NodeType::Llm(critique) = &graph.get_node("critique").unwrap().node_type else {
panic!("critique must be an llm node")
};
assert_eq!(critique.fallback.as_deref(), Some("critique_fault"));
assert_eq!(
graph.get_node("critique_fault").unwrap().next_target(),
Some("reflexion_gate")
);
// synthesize: fail-closed via synth_fault → end_fault; the failure
// text lands in `report` via state_updates.
let NodeType::Agent(synthesize) = &graph.get_node("synthesize").unwrap().node_type else {
panic!("synthesize must be an agent node")
};
assert_eq!(synthesize.fallback.as_deref(), Some("synth_fault"));
assert_eq!(synthesize.max_attempts, 2);
assert!(
synthesize
.state_updates
.as_ref()
.is_some_and(|u| u.contains_key("report"))
);
assert_eq!(
graph.get_node("synth_fault").unwrap().next_target(),
Some("end_fault")
);
// end_fault renders the naming-pinned sentinel.
let NodeType::End(end) = &graph.get_node("end_fault").unwrap().node_type else {
panic!("end_fault must be an end node")
};
assert_eq!(
end.output,
"DEEP_RESEARCH FAILED — PIPELINE-FAULT: {{fault_text}}"
);
// Accepted-path visibility: pipeline_notes ("" on a fault-free run,
// keeping the happy-path output byte-identical) is appended to the
// report and shown at the approval gate.
let NodeType::End(accepted) = &graph.get_node("end_accepted").unwrap().node_type else {
panic!("end_accepted must be an end node")
};
assert_eq!(accepted.output, "{{report}}{{pipeline_notes}}");
let NodeType::Approval(approve) = &graph.get_node("approve").unwrap().node_type else {
panic!("approve must be an approval node")
};
assert!(approve.question.contains("{{pipeline_notes}}"));
// Every fault-path state key has an initial_state default; questions'
// [] also keeps the sibling knowledge_lookup lane alive for the one
// super-step where a dead plan races end_fault (the map resolves
// {{questions}} to zero items instead of erroring).
assert_eq!(graph.initial_state.get("pipeline_faults"), Some(&json!([])));
assert_eq!(graph.initial_state.get("pipeline_notes"), Some(&json!("")));
assert_eq!(graph.initial_state.get("fault_text"), Some(&json!("")));
assert_eq!(graph.initial_state.get("questions"), Some(&json!([])));
}
#[test] #[test]
#[serial] #[serial]
fn install_functions_force_preserves_user_mcp_json() { fn install_functions_force_preserves_user_mcp_json() {