You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by ak...@apache.org on 2026/05/29 18:01:57 UTC
(tooling-agents) branch main updated: guardrails: distinguish no_relevant_code from all_batches_failed; surface pipeline failures loud
This is an automated email from the ASF dual-hosted git repository.
andrewmusselman pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tooling-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 7ece820 guardrails: distinguish no_relevant_code from all_batches_failed; surface pipeline failures loud
7ece820 is described below
commit 7ece820887205142df72e28bf89e555dc8eb3256
Author: Andrew Musselman <ak...@apache.org>
AuthorDate: Fri May 29 11:01:47 2026 -0700
guardrails: distinguish no_relevant_code from all_batches_failed; surface pipeline failures loud
Closes the silent-failure path observed on superset-core L3 (and
previously on mina V10, airflow task-sdk V3, airflow providers/google
V17) where a bundle with zero relevant files emitted the same error
envelope as a bundle whose batches all crashed, the orchestrator
silently mis-attributed the envelope as per-section output, and the
consolidator extracted those stubs as fake N/A.
asvs_bundle.py: separate code path for the empty-batches case emits
a proper bundled envelope with per-section N/A reports and
bundle_status: no_relevant_code. The existing all-batches-failed
envelope now carries bundle_status: all_batches_failed and
attempted_batches for diagnosis.
asvs_orchestrate.py _parse_audit_output rewritten to return tuples
(report_text, signal) where signal is ok / no_relevant_code /
bundle_error / missing_section / malformed_multi. The parser checks
for error envelopes explicitly before the bundled-mode check so
legacy mis-attribution paths can no longer reach the per-section
stub fallback. The caller routes failure signals to local_failures,
so the audit-phase summary reflects reality instead of counting
ERROR stubs as successes.
asvs_consolidate.py adds a pre-extraction scan for known
failure-shape markers in stored report content and short-circuits
the Sonnet extraction call when found, recording the section as
ERROR rather than the LLM default-N/A interpretation. The Quality
Checks block now reports pipeline-failure section counts and prints
RUN SHOULD NOT BE PUBLISHED WITHOUT REVIEW when any are present.
Re-consolidating older runs will surface sections that previously
slipped through silently. This is the correct outcome.
---
ASVS/agents/asvs_bundle.py | 70 +++++++++++++++++
ASVS/agents/asvs_consolidate.py | 86 +++++++++++++++++++++
ASVS/agents/asvs_orchestrate.py | 162 +++++++++++++++++++++++++++++++++-------
3 files changed, 293 insertions(+), 25 deletions(-)
diff --git a/ASVS/agents/asvs_bundle.py b/ASVS/agents/asvs_bundle.py
index 773b74b..4b221eb 100644
--- a/ASVS/agents/asvs_bundle.py
+++ b/ASVS/agents/asvs_bundle.py
@@ -1331,17 +1331,87 @@ Be thorough but precise. If something is done correctly, acknowledge it as a pos
print(f"[bundle {bundle_label}] Opus batch {i+1} FAILED: {e}", flush=True)
return f"[Analysis failed for batch {i+1}: {str(e)[:200]}]"
+ # =============================================================
+ # GUARDRAIL: distinguish "no work needed" from "all work failed"
+ # =============================================================
+ # Before this guardrail, a bundle with zero opus_batches (code
+ # inventory determined nothing relevant to audit) fell through
+ # to the same `analysis_results == []` path as a bundle whose
+ # batches all crashed, and both returned the same `error:
+ # "All analysis batches failed"` envelope. The orchestrator's
+ # parser couldn't recognize the envelope as bundled output,
+ # silently attributed the error JSON to the first section,
+ # and emitted "did not return per-section output" stubs for
+ # the rest. The stored stubs were then read by the consolidator
+ # as legitimate N/A.
+ #
+ # The two cases need different handling:
+ # - 0 batches => no relevant code; legitimate N/A
+ # - all crashed => real failure; surface loudly
+ if not opus_batches:
+ print(
+ f"[bundle {bundle_label}] no Opus batches needed (code inventory "
+ f"determined no relevant implementation); emitting per-section N/A",
+ flush=True,
+ )
+ per_section_na = {}
+ for sid in asvs_sections:
+ desc = asvs_descriptions.get(sid, f"ASVS Requirement {sid}")
+ na_body = (
+ f"## ASVS-{sid}\n\n"
+ f"**Status:** N/A\n\n"
+ f"**Reason:** Code inventory of the audited scope determined "
+ f"that no implementation relevant to this requirement exists "
+ f"in the repository. The framework does not appear to provide "
+ f"functionality covered by ASVS {sid}.\n"
+ )
+ report = _format_section_report(
+ sid, desc,
+ repo_name, audit_date,
+ len(relevant_files), len(filtered_files), skipped_count,
+ {"Critical": 0, "High": 0, "Medium": 0, "Low": 0},
+ na_body, "",
+ )
+ per_section_na[sid] = {
+ "report": report,
+ "findings": {"Critical": 0, "High": 0, "Medium": 0, "Low": 0},
+ "files_analyzed": len(relevant_files),
+ "files_total": len(filtered_files),
+ "files_skipped": skipped_count,
+ "status": "N/A",
+ "reason": "no_relevant_code",
+ }
+ return {"outputText": json.dumps({
+ "mode": "bundled",
+ "asvs_sections": asvs_sections,
+ "per_section": per_section_na,
+ "bundle_status": "no_relevant_code",
+ })}
+
analysis_results = await asyncio.gather(*[
analyze_batch(i, batch)
for i, batch in enumerate(opus_batches)
])
+ attempted = len(opus_batches)
analysis_results = [r for r in analysis_results if r and not r.startswith("[Analysis failed")]
if not analysis_results:
+ # GUARDRAIL: opus_batches existed (work was attempted) but
+ # every batch crashed. Tag the envelope so the orchestrator's
+ # parser recognizes this as a real failure rather than a
+ # bundled mode-output. See asvs_orchestrate.py
+ # `_parse_audit_output` error-envelope branch.
+ print(
+ f"[bundle {bundle_label}] ALL OPUS BATCHES FAILED "
+ f"(attempted={attempted}); returning error envelope",
+ flush=True,
+ )
return {"outputText": json.dumps({
"error": "All analysis batches failed",
"asvs_sections": asvs_sections,
+ "attempted_batches": attempted,
+ "bundle_status": "all_batches_failed",
})}
# =============================================================
diff --git a/ASVS/agents/asvs_consolidate.py b/ASVS/agents/asvs_consolidate.py
index 9ea4b75..bd32fcd 100644
--- a/ASVS/agents/asvs_consolidate.py
+++ b/ASVS/agents/asvs_consolidate.py
@@ -452,11 +452,72 @@ JSON schema:
"positive_controls": [{"control": "description", "evidence": "where observed", "files": ["file:line"]}]
}"""
+ # GUARDRAIL: failure-shape patterns that may appear in stored
+ # per-section reports. These indicate the audit pipeline failed
+ # for the section rather than producing a real audit. The LLM
+ # extraction step, given such content, tends to silently return
+ # `{"findings": [], "asvs_status": "N/A"}` because the failure
+ # text doesn't look like findings. Detecting these patterns up
+ # front lets us tag the section as ERROR and surface in the
+ # quality check rather than burying the failure as fake-N/A.
+ #
+ # The patterns must match content produced by:
+ # asvs_bundle.py all-batches-failed envelope, no-files envelope
+ # asvs_orchestrate.py _parse_audit_output ERROR stubs
+ _FAILURE_PATTERNS = (
+ "**Status:** ERROR", # orchestrator stubs
+ '"error": "All analysis batches failed"', # raw bundle envelope
+ '"error": "No files found in namespaces', # raw bundle envelope
+ "did not return per-section output", # legacy stub (pre-guardrails)
+ "Bundled audit produced no output", # legacy stub
+ )
+
+ def _detect_pipeline_failure(content):
+ """Return (is_failure, marker_text) if the content looks like a
+ pipeline-failure stub rather than a real audit report."""
+ if not content:
+ return False, None
+ head = content[:1500]
+ for pat in _FAILURE_PATTERNS:
+ if pat in head:
+ return True, pat
+ return False, None
+
extraction_semaphore = asyncio.Semaphore(8) # raised from 5
all_extracted = {}
extraction_errors = []
+ pipeline_failure_sections = [] # surfaced in Quality Checks
async def extract_report(report_key, content):
+ # GUARDRAIL: short-circuit pipeline-failure content before
+ # the LLM extraction call. This both saves a Sonnet call per
+ # failed section and ensures the failure is recorded as ERROR
+ # rather than the LLM's default-N/A interpretation.
+ is_failure, marker = _detect_pipeline_failure(content)
+ if is_failure:
+ # Derive a best-effort ASVS section ID from the report key
+ # (e.g., "oauth_openid_integration/10.1.1.md" -> "10.1.1").
+ import re as _re
+ m = _re.search(r'(\d+\.\d+(?:\.\d+)?)\.md$', report_key)
+ section_id = m.group(1) if m else report_key
+ print(
+ f" {report_key}: PIPELINE FAILURE detected "
+ f"(marker={marker!r}); short-circuit extraction",
+ flush=True,
+ )
+ synthetic = {
+ "source_report": report_key,
+ "asvs_section": section_id,
+ "asvs_section_title": "",
+ "asvs_status": "ERROR",
+ "findings": [],
+ "positive_controls": [],
+ "_pipeline_failure": True,
+ "_pipeline_failure_marker": marker,
+ }
+ pipeline_failure_sections.append((report_key, section_id, marker))
+ return report_key, synthetic, None
+
# Cache key includes content hash so re-running against an updated
# report file produces fresh extraction. Without this, the same
# filename in the same directory always hits the cache regardless
@@ -1478,6 +1539,31 @@ For multi-value table cells (Files, Source Reports, etc.): use ", " (comma-space
issue_headers = set(re.findall(r'## Issue: (FINDING-\d{3})', issues_md))
print(f"Issues: {len(issue_headers)} (expected: {len(actionable)})")
+ # GUARDRAIL: pipeline-failure surfacing. Sections whose stored
+ # report content matched a known failure pattern were extracted
+ # as ERROR (not silently as N/A). If any are present, the run is
+ # not safe to publish without manual review — the consolidated
+ # report will be missing real audit content for those sections,
+ # and the silent failure mode this guardrail is closing has in
+ # the past produced reports where V10 (OAuth) was entirely empty
+ # because every section came through as a stub.
+ print(f"Pipeline-failure sections: {len(pipeline_failure_sections)}")
+ if pipeline_failure_sections:
+ print(" *** RUN SHOULD NOT BE PUBLISHED WITHOUT REVIEW ***")
+ # Group by marker so the operator sees the failure shape at
+ # a glance rather than scrolling through N individual lines.
+ by_marker = {}
+ for rk, sid, marker in pipeline_failure_sections:
+ by_marker.setdefault(marker, []).append((rk, sid))
+ for marker, items in sorted(by_marker.items()):
+ print(f" marker={marker!r}: {len(items)} section(s)")
+ # Cap displayed list at 20 to avoid log spam on a fully
+ # broken run; the count above gives the full picture.
+ for rk, sid in items[:20]:
+ print(f" - {sid} (from {rk})")
+ if len(items) > 20:
+ print(f" ... and {len(items) - 20} more")
+
# Push to GitHub
print("\n=== Pushing files ===")
diff --git a/ASVS/agents/asvs_orchestrate.py b/ASVS/agents/asvs_orchestrate.py
index 0913b45..c0a90c3 100644
--- a/ASVS/agents/asvs_orchestrate.py
+++ b/ASVS/agents/asvs_orchestrate.py
@@ -60,41 +60,128 @@ async def run(input_dict, tools):
def _parse_audit_output(audit_output_text, section_chunk):
"""Decode either a bundled JSON envelope or a single-section markdown report.
- Returns dict: {section_id: report_markdown}.
+ Returns dict: {section_id: (report_markdown, status_signal)}.
+
+ status_signal is one of:
+ - "ok" => section has real content (findings or
+ the auditor's own Pass/Fail/N/A judgment)
+ - "no_relevant_code" => bundle determined nothing applies (the
+ asvs_bundle.py "no Opus batches needed"
+ path); legitimate N/A, NOT a failure
+ - "bundle_error" => bundle emitted an error envelope. The
+ section's result is missing and the run
+ must not be published without manual review.
+ - "missing_section" => bundle returned `per_section` with this
+ section's entry absent or None. Partial
+ bundle failure; flag for review.
+ - "malformed_multi" => bundle was expected to return a JSON
+ envelope for a multi-section chunk but
+ returned raw markdown. Treat as a partial
+ failure: first section gets the markdown,
+ the rest get error stubs.
"""
import json
- # Try bundled envelope first
- if audit_output_text.strip().startswith("{"):
+
+ # ---- Bundled JSON envelope path ----
+ stripped = audit_output_text.strip()
+ if stripped.startswith("{"):
try:
envelope = json.loads(audit_output_text)
+ except json.JSONDecodeError:
+ envelope = None
+
+ if envelope is not None:
+ # GUARDRAIL: explicit error envelope check BEFORE the
+ # bundled-mode check. The legacy code path treated any
+ # non-bundled JSON as "unparseable" and fell through to
+ # the multi-section fallback, which silently attributed
+ # the error JSON to the first section. Detect the error
+ # shape explicitly and emit per-section ERROR stubs so
+ # downstream can fail the quality check.
+ err_msg = envelope.get("error")
+ if err_msg:
+ bundle_status = envelope.get("bundle_status", "bundle_error")
+ print(
+ f" [WARN] bundle returned error envelope "
+ f"(status={bundle_status}): {err_msg}",
+ flush=True,
+ )
+ out = {}
+ for sid in section_chunk:
+ out[sid] = (
+ (
+ f"# ASVS {sid}\n\n"
+ f"**Status:** ERROR\n\n"
+ f"**Reason:** Bundle audit agent returned an "
+ f"error envelope. This section's per-section "
+ f"report is missing.\n\n"
+ f"**Bundle status:** `{bundle_status}`\n\n"
+ f"**Error:** {err_msg}\n\n"
+ f"_The consolidated report should not be "
+ f"published until this section is re-audited "
+ f"or this status is reviewed._\n"
+ ),
+ "bundle_error",
+ )
+ return out
+
if envelope.get("mode") == "bundled":
per_section = envelope.get("per_section", {})
+ bundle_status = envelope.get("bundle_status", "ok")
out = {}
for sid in section_chunk:
entry = per_section.get(sid)
if entry is None:
out[sid] = (
- f"# ASVS {sid}\n\n"
- f"_Bundled audit produced no output for this section. "
- f"This may indicate the section is not applicable to "
- f"the audited file scope._\n"
+ (
+ f"# ASVS {sid}\n\n"
+ f"**Status:** ERROR\n\n"
+ f"**Reason:** Bundle audit produced no "
+ f"output for this section. The bundle "
+ f"completed but this section's slot in "
+ f"the per_section map was empty. This "
+ f"may indicate the analysis didn't "
+ f"return a recognizable section header "
+ f"for this requirement.\n"
+ ),
+ "missing_section",
)
else:
- out[sid] = entry.get("report", "")
+ section_signal = (
+ "no_relevant_code"
+ if bundle_status == "no_relevant_code"
+ else "ok"
+ )
+ out[sid] = (entry.get("report", ""), section_signal)
return out
- except json.JSONDecodeError:
- pass
- # Fallback: single-section markdown report
+
+ # ---- Fallback: single-section markdown report ----
if len(section_chunk) == 1:
- return {section_chunk[0]: audit_output_text}
- # Multiple sections expected but didn't get JSON envelope — attribute
- # the markdown to the first and emit stubs for the rest
- out = {section_chunk[0]: audit_output_text}
+ return {section_chunk[0]: (audit_output_text, "ok")}
+
+ # ---- Fallback: multiple sections expected but no JSON envelope ----
+ # This is a malformed response. Attribute markdown to first section
+ # and emit ERROR stubs for the rest so the loss is surfaced rather
+ # than silently treated as "did not return per-section output" and
+ # then read as N/A by the consolidator.
+ print(
+ f" [WARN] bundle returned non-JSON output but multiple sections "
+ f"({len(section_chunk)}) were expected; first section keeps the "
+ f"output, others marked ERROR",
+ flush=True,
+ )
+ out = {section_chunk[0]: (audit_output_text, "malformed_multi")}
for sid in section_chunk[1:]:
out[sid] = (
- f"# ASVS {sid}\n\n"
- f"_Audit agent did not return per-section output for this requirement. "
- f"See ASVS {section_chunk[0]} report for the bundle's full output._\n"
+ (
+ f"# ASVS {sid}\n\n"
+ f"**Status:** ERROR\n\n"
+ f"**Reason:** Audit agent did not return a parseable "
+ f"per-section envelope for this multi-section bundle. "
+ f"See ASVS {section_chunk[0]} for the bundle's raw "
+ f"output.\n"
+ ),
+ "malformed_multi",
)
return out
@@ -2030,7 +2117,11 @@ async def run(input_dict, tools):
return local_successes, local_failures, local_outputs, pass_output_dir
# ----- Parse output: bundled JSON envelope or single-section markdown -----
- per_section_reports = _parse_audit_output(audit_output_text, section_chunk)
+ # GUARDRAIL: _parse_audit_output now returns dict of
+ # {sid: (report_text, status_signal)}. Signals: "ok" /
+ # "no_relevant_code" (both success), "bundle_error" /
+ # "missing_section" / "malformed_multi" (all failure shapes).
+ per_section_parsed = _parse_audit_output(audit_output_text, section_chunk)
# ----- Push per-section reports in parallel (throttled) -----
# Uses the SHARED github_push_sem from outer scope so all
@@ -2059,16 +2150,37 @@ async def run(input_dict, tools):
return section_id, err_str
push_results = await asyncio.gather(*[
- store_one(sid, txt) for sid, txt in per_section_reports.items()
+ store_one(sid, parsed[0]) for sid, parsed in per_section_parsed.items()
])
+ # GUARDRAIL: a successful CouchDB write of an ERROR-status
+ # stub is NOT an audit success. Sections whose parser signal
+ # is anything other than "ok" or "no_relevant_code" go into
+ # local_failures even though the store itself succeeded, so
+ # the run summary reflects the real audit-phase outcome.
+ section_signals = {sid: parsed[1] for sid, parsed in per_section_parsed.items()}
+ _failure_signals = {"bundle_error", "missing_section", "malformed_multi"}
for sid, err in push_results:
- if err is None:
- local_successes.append(sid)
- print(f" [{pass_name}] {sid}: stored", flush=True)
- else:
+ if err is not None:
local_failures.append(f"{sid} (store): {err}")
print(f" [{pass_name}] {sid}: store failed: {err}", flush=True)
- local_outputs.extend(per_section_reports.values())
+ continue
+ signal = section_signals.get(sid, "ok")
+ if signal in _failure_signals:
+ local_failures.append(f"{sid} (bundle): {signal}")
+ print(
+ f" [{pass_name}] {sid}: stored as ERROR stub "
+ f"(signal={signal}) — counted as failure",
+ flush=True,
+ )
+ else:
+ local_successes.append(sid)
+ # Annotate the success log line when the success is
+ # a deliberate N/A so an operator skimming the log
+ # can see at a glance which sections were genuinely
+ # not applicable vs. which were affirmatively audited.
+ suffix = " [N/A: no relevant code]" if signal == "no_relevant_code" else ""
+ print(f" [{pass_name}] {sid}: stored{suffix}", flush=True)
+ local_outputs.extend(parsed[0] for parsed in per_section_parsed.values())
return local_successes, local_failures, local_outputs, pass_output_dir
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org