fix: Scope the parse verdict to the DAG-integrity session

`astro dev parse` captures the image build and the integrity run in one stream,
and a Dockerfile `RUN pytest` contributes its own pytest output. Buildkit
prefixes those lines so they missed the anchored patterns, but the legacy builder
emits RUN stdout verbatim — and a stray "=== 4 passed in 0.21s ===" satisfied
the completion check for a run that never tested a DAG. Three false passes were
reachable: a CLI refusal (which also discarded its own reason), an integrity run
killed mid-flight, and a session that collected nothing.

Scope every pytest-shaped signal to the last session in the log (the image must
exist before its container runs, so the integrity session is always last), let
the CLI's error wrapper outrank any summary, and treat counted-but-unnameable
failures as no verdict rather than a pass. In verify.sh, decide a timeout on the
kill we own instead of on what the output resembles.

Real failures now outrank both, so a run cut short after finding them reports
them instead of discarding them.

Found by adversarial review.

Author: kaxil

scripts/parse_check.py

@@ -33,6 +33,15 @@
 from report_fmt import failure
 
 BUILD_MARKER = "an error was encountered while building the image"
+# `astro dev parse` captures the image build and the DAG-integrity run in ONE
+# stream, and the build can contribute pytest output of its own (a Dockerfile
+# `RUN pytest`). Buildkit prefixes those lines ("#12 1.204 ") so they miss the
+# anchored patterns below, but the legacy builder emits RUN stdout verbatim — and
+# a stray "=== 4 passed in 0.21s ===" was enough to read a run that never tested
+# a DAG as a pass. Every pytest-shaped signal is therefore scoped to the LAST
+# session in the log: the image must exist before its container runs, so the
+# integrity session is always the final one.
+_SESSION_START = "test session starts"
 _COLLECTED = re.compile(r"^collected (\d+) items?")
 # pytest's closing summary line — its presence is the evidence the run FINISHED.
 # A timeout kill mid-run leaves "collected N items" with no summary; that must
@@ -44,6 +53,11 @@
     r"^=+ .*\b(passed|failed|error|no tests ran)\b.* in [0-9.]+s(?: \([0-9:]+\))?",
     re.MULTILINE,
 )
+# The same summary, but reporting something broken. Used to catch the one shape
+# where a completed run still must not read as a pass: pytest counted failures
+# that no test_file_imports[...] entry accounted for, so we cannot name what
+# broke and have nothing honest to report.
+_SUMMARY_BROKEN = re.compile(r"^=+ .*\b\d+ (?:failed|error)", re.MULTILINE)
 _FAILED_LINE = re.compile(r"^FAILED .*::test_file_imports\[(?P<path>[^\]]+)\]")
 _EXC_HEAD = re.compile(r"^E\s+Exception: (?P<path>\S+) failed to import with message")
 _EXC_CLASS = re.compile(r"^([A-Za-z_][A-Za-z0-9_.]*(?:Error|Exception|Warning))\b")
@@ -61,17 +75,28 @@
 def parse_output(text: str) -> tuple[int, dict]:
     if BUILD_MARKER in text:
         return 4, {"checked": 0, "failures": []}
-    if "error during collection" in text or "errors during collection" in text:
+
+    idx = text.rfind(_SESSION_START)
+    body = text[idx:] if idx != -1 else text
+
+    if "error during collection" in body or "errors during collection" in body:
         # Surface the terse cause (e.g. "TypeError: DagBag.__init__() got...")
         # from pytest's short summary so the caller can show why.
         cause = ""
-        for line in text.splitlines():
+        for line in body.splitlines():
             if line.startswith("ERROR ") and " - " in line:
                 cause = line.split(" - ", 1)[1].strip()
                 break
         return 5, {"checked": 0, "failures": [], "collection_error": cause}
 
-    lines = text.splitlines()
+    # The CLI wraps every Pytest outcome it cannot read as "tests failed" (exit 1
+    # gets its own "See above for errors detected in your DAGs") in this line: a
+    # missing venv, a venv without pytest, or pytest exiting 5 for "no tests ran".
+    # Its presence means this run produced no verdict, so it outranks any summary
+    # the stream happens to contain.
+    cli_err = _CLI_ERROR.search(text)
+
+    lines = body.splitlines()
     checked = 0
     for line in lines:
         m = _COLLECTED.match(line.strip())
@@ -107,18 +132,24 @@ def parse_output(text: str) -> tuple[int, dict]:
             failures[m.group("path")] = failure(
                 m.group("path"), "", "failed to import (see the CI log for the traceback)")
 
-    # A verdict requires evidence the run COMPLETED, not merely started:
-    # the astro clean marker or pytest's closing summary line. "collected N
-    # items" alone is what a timeout kill leaves behind.
-    completed = _CLEAN in text or _SUMMARY.search(text) is not None
-    if not completed:
+    sorted_failures = sorted(failures.values(), key=lambda f: f["path"])
+    # Real failures are reported even when the run was cut short or the CLI
+    # errored: they are evidence of genuine breakage, and reporting them is
+    # fail-closed where degrading to the fallback would discard them.
+    if failures:
+        return 3, {"checked": checked, "failures": sorted_failures}
+
+    # A pass requires evidence the run COMPLETED, not merely started: the astro
+    # clean marker (which only the CLI prints, and only on success) or the closing
+    # summary of the integrity session. "collected N items" alone is what a
+    # timeout kill leaves behind.
+    completed = _CLEAN in text or _SUMMARY.search(body) is not None
+    if cli_err is not None or not completed or _SUMMARY_BROKEN.search(body):
         result: dict = {"checked": 0, "failures": []}
-        m = _CLI_ERROR.search(text)
-        if m:
-            result["cli_error"] = m.group("cause").strip()
+        if cli_err is not None:
+            result["cli_error"] = cli_err.group("cause").strip()
         return 2, result
-    return (3 if failures else 0), {"checked": checked, "failures": sorted(
-        failures.values(), key=lambda f: f["path"])}
+    return 0, {"checked": checked, "failures": []}
 
 
 def main() -> int:

scripts/verify.sh

@@ -227,6 +227,16 @@ verify_parse_level() {
   tail -n 40 "$WORKDIR/parse-target.log"
   echo "::endgroup::"
 
+  # The kill outranks the scrape. A `timeout` can land on a log that still looks
+  # finished — the image build's own pytest output, a session cut off after its
+  # summary — and nothing was fully verified either way, so decide on the signal
+  # we own rather than on what the output resembles.
+  if [[ "$target_run_rc" -eq 124 ]]; then
+    fallback_note="ℹ️ Image-level verification could not run (\`astro dev parse\` timed out before completing). Results below come from the import-level fallback."
+    echo "::warning::verify-level parse: astro dev parse timed out before completing; falling back to import-level verification."
+    return 1
+  fi
+
   # Both "harness produced no verdict" classes degrade to the import-level
   # check rather than reporting nothing: rc 5 = the project's integrity test is
   # incompatible with the target Airflow; rc 2 = the run never completed
@@ -237,19 +247,14 @@ verify_parse_level() {
     echo "::warning::verify-level parse: the project's DAG integrity test is incompatible with the target Airflow; falling back to import-level verification."
     return 1
   elif [[ "$target_rc" -eq 2 ]]; then
+    # The CLI usually states exactly why it refused (a project configured for
+    # standalone dev mode, a venv without pytest, …). Its words beat the
+    # catch-all: "produced no recognizable result" is what made a field case take
+    # a full log dig to explain. Timeouts already returned above.
     local reason="produced no recognizable result"
-    local cli_error
-    if [[ "$target_run_rc" -eq 124 ]]; then
-      # A timeout kill leaves no explanation of its own, so keep ours.
-      reason="timed out before completing"
-    else
-      # The CLI usually states exactly why it refused (a project configured for
-      # standalone dev mode, a venv without pytest, …). Its words beat the
-      # catch-all: "produced no recognizable result" is what made a field case
-      # take a full log dig to explain.
-      cli_error=$(jq -r '.cli_error // empty' "$WORKDIR/import-failures.json" 2>/dev/null || true)
-      [[ -n "$cli_error" ]] && reason="failed: ${cli_error}"
-    fi
+    local cli_error=""
+    cli_error=$(jq -r '.cli_error // empty' "$WORKDIR/import-failures.json" 2>/dev/null || true)
+    if [[ -n "$cli_error" ]]; then reason="failed: ${cli_error}"; fi
     fallback_note="ℹ️ Image-level verification could not run (\`astro dev parse\` ${reason}). Results below come from the import-level fallback."
     echo "::warning::verify-level parse: astro dev parse ${reason}; falling back to import-level verification."
     return 1

tests/test_parse_check.py

@@ -158,15 +158,105 @@ def test_build_failure_wrapper_is_not_demoted_to_cli_error(tmp_path):
     assert "cli_error" not in result
 
 
-def test_completed_run_never_gains_a_cli_error(tmp_path):
-    # Defensive: a finished run is a verdict. Even if the wrapper somehow appears
-    # alongside a closing summary, the summary wins.
+def test_failing_run_with_wrapper_still_reports_the_failures(tmp_path):
+    # A wrapper alongside real failures means the run was cut short after finding
+    # them (pytest exit 1 gets its own message, not the wrapper). Reporting the
+    # failures is fail-closed; degrading to the fallback would discard them.
     text = FAILING_RUN + "Error: something went wrong while parsing your DAGs: noise\n"
     rc, result, *_ = _run(tmp_path, text)
     assert rc == 3
+    assert result["failures"][0]["path"] == "dags/format_probe.py"
     assert "cli_error" not in result
 
 
+# A Dockerfile `RUN pytest` lands in the same captured stream as the integrity
+# run. Buildkit prefixes those lines ("#12 1.204 ") so they miss the anchored
+# patterns, but the legacy builder emits RUN stdout verbatim — and an unprefixed
+# summary was enough to manufacture a pass. Found by adversarial review.
+BUILD_PYTEST_NOISE = """\
+Checking your DAGs for errors…
+Step 6/7 : RUN pytest tests/unit -q
+ ---> Running in a1b2c3
+collected 4 items
+=========================== 4 passed in 0.21s ===========================
+ ---> d4e5f6
+Successfully built d4e5f6
+"""
+
+
+def test_build_pytest_summary_cannot_pass_a_refused_parse(tmp_path):
+    # The worst shape: the build's summary satisfies "completed" while the CLI is
+    # saying it never ran the parse at all.
+    text = BUILD_PYTEST_NOISE + (
+        "Error: something went wrong while parsing your DAGs: no virtual "
+        "environment found — run 'astro dev start' first\n"
+    )
+    rc, result, *_ = _run(tmp_path, text)
+    assert rc == 2, "a build's pytest summary must not stand in for the integrity run"
+    assert result["checked"] == 0, "the build's collected count must not be reported"
+    assert "virtual environment" in result["cli_error"]
+
+
+def test_build_pytest_summary_cannot_pass_a_truncated_integrity_run(tmp_path):
+    truncated = (
+        "============================= test session starts ==============================\n"
+        "collected 37 items\n\n"
+        ".astro/test_dag_integrity_default.py ......\n"
+    )
+    rc, result, *_ = _run(tmp_path, BUILD_PYTEST_NOISE + truncated)
+    assert rc == 2
+    assert result["checked"] == 0
+
+
+def test_no_tests_ran_is_not_a_pass(tmp_path):
+    # pytest exits 5 when it collects nothing, which the CLI wraps. Reporting
+    # "all 0 DAG file(s) import cleanly" over an untested project is the exact
+    # false green this level exists to prevent.
+    text = (
+        "Checking your DAGs for errors…\n"
+        "============================= test session starts ==============================\n"
+        "collected 0 items\n\n"
+        "==================== no tests ran in 0.12s ====================\n"
+        "Error: something went wrong while parsing your DAGs: "
+        "something went wrong while Pytesting your DAGs\n"
+    )
+    rc, result, *_ = _run(tmp_path, text)
+    assert rc == 2
+    assert "Pytesting" in result["cli_error"]
+
+
+def test_build_noise_does_not_hide_a_real_collection_error(tmp_path):
+    # The integrity session's own collection error must still win rc 5 even with a
+    # clean build-stage pytest run ahead of it in the stream.
+    rc, result, *_ = _run(tmp_path, BUILD_PYTEST_NOISE + COLLECTION_ERROR_RUN)
+    assert rc == 5
+    assert "DagBag.__init__()" in result["collection_error"]
+
+
+def test_counted_failures_we_cannot_name_are_not_a_pass(tmp_path):
+    # pytest counted a failure that no test_file_imports[...] entry accounts for
+    # (an unexpected harness shape). We can't name what broke, so the only honest
+    # answers are "no verdict" or a fabricated one — take the former.
+    text = (
+        "Checking your DAGs for errors…\n"
+        "============================= test session starts ==============================\n"
+        "collected 1 item\n\n"
+        "tests/test_something_else.py F                                           [100%]\n"
+        "=========================== 1 failed in 1.93s ===========================\n"
+    )
+    rc, result, *_ = _run(tmp_path, text)
+    assert rc == 2
+    assert result["checked"] == 0
+
+
+def test_build_noise_does_not_inflate_a_real_pass(tmp_path):
+    # The complement: with a genuine integrity run after the build noise, the
+    # count must come from the integrity session (21), not the build's (4).
+    rc, result, *_ = _run(tmp_path, BUILD_PYTEST_NOISE + CLEAN_RUN)
+    assert rc == 0
+    assert result["checked"] == 21
+
+
 def test_long_run_summary_with_hms_suffix_is_recognized(tmp_path):
     # pytest switches the summary duration format at 60s: "in 62.50s (0:01:02)".
     # Missing that form silently disabled parse-level verdicts for any project