fix(acceptance): reject nonpassing unittest summaries (#6916)

* fix(acceptance): require passing unittest counts

* test(acceptance): retain unittest output attribution guard

* test(acceptance): cover unittest errors and recorded failures

* test(acceptance): honor unittest empty-selection exit status

* docs(subagents): keep unittest guidance within inherited budget
This commit is contained in:
Guo-Yixin 2026-10-15 02:26:47 -05:00 committed by GitHub
parent 6bc1020d34
commit 8d994cc1e7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 207 additions and 6 deletions

View File

@ -2179,6 +2179,12 @@ summaries do not veto that sibling evidence. All-empty or all-ignored runs and
bare `test result: ok` banners remain `UNVERIFIED`; incomplete zero-test summaries
are kept as vetoes, and failures still take precedence.
For `tests_passed:python -m unittest`, a complete `Ran N tests in ...s` / `OK`
summary must include at least one ordinary passing test. Skips and expected
failures do not count as passes. Successful runs with all-skipped,
all-expected-failure, empty or incomplete summaries remain `UNVERIFIED`;
nonzero exits, failures and unexpected successes still take precedence.
For Maven and Gradle test acceptance, a `BUILD SUCCESS` / `BUILD SUCCESSFUL`
banner alone remains `UNVERIFIED`: builds can succeed without running tests.
Complete `Tests run: N, Failures: F, Errors: E, Skipped: S` or `N tests completed`

View File

@ -90,6 +90,7 @@ empty/all-skipped modules contribute no proof but do not veto tested modules.
These summaries also participate in the command/output attribution guard.
Cargo: complete positive siblings permit empty targets. All-empty/ignored or bare banners stay UNVERIFIED; failures win.
Unittest needs complete counts with ordinary passes; uppercase OK is not Go output. Failures win.
**Events**: `task_started`, `task_running`, `task_completed`/`task_failed`/`task_timed_out`
**Handled LLM failures**: `LLMErrorHandlingMiddleware` deliberately converts provider/model exceptions into an `AIMessage` so the graph can end cleanly, stamping `additional_kwargs.deerflow_error_fallback=true` plus error metadata. Clean graph termination does not imply subagent success: `SubagentExecutor` inspects the last assistant message at terminalization and maps a marked fallback to `SubagentStatus.FAILED`, which then emits `task_failed` and the existing structured `subagent_error`. Only the marker is authoritative — error-looking assistant prose without it remains a normal completed result, so neither the executor nor frontend parses display text as a status protocol.

View File

@ -116,12 +116,36 @@ _FILE_CONTENT_READ_CAP_BYTES = 50_000
#: way (UNVERIFIED), and fail shapes win over pass shapes when both appear.
_TEST_PASS_SHAPE_RE = re.compile(
r"\b[1-9]\d*\s+passed\b" # pytest / jest: "5 passed" (zero is not a pass)
r"|^OK$" # unittest: bare OK line
r"|^ok\s+\S" # go test: "ok \tpkg/path"
r"|\ball tests passed\b",
re.IGNORECASE | re.MULTILINE,
)
# Go's lowercase, same-line prefix must not consume unittest's uppercase OK
# annotations (or a package name on a later line).
_GO_TEST_PASS_SHAPE_RE = re.compile(r"^ok[ \t]+\S", re.MULTILINE)
_UNITTEST_RESULT_LINE_RE = re.compile(r"^OK(?:[ \t]+\([^\r\n]*\))?[ \t]*\r?$", re.MULTILINE)
_UNITTEST_SUCCESS_SUMMARY_RE = re.compile(
r"^Ran (?P<total>0|[1-9][0-9]{0,8}) tests? in [0-9]+(?:\.[0-9]+)?s\r?\n\r?\n"
r"OK(?: \((?P<counts>[^\r\n()]*)\))?[ \t]*\r?$",
re.MULTILINE,
)
def _unittest_test_summary_holds(text: str) -> bool:
"""Require at least one ordinary pass in a complete unittest summary."""
for match in _UNITTEST_SUCCESS_SUMMARY_RE.finditer(text):
counts: dict[str, int] = {}
for field in match["counts"].split(", ") if match["counts"] is not None else []:
counter = re.fullmatch(r"(skipped|expected failures)=(0|[1-9][0-9]{0,8})", field)
if counter is None or counter[1] in counts:
break
counts[counter[1]] = int(counter[2])
else:
if int(match["total"]) > sum(counts.values()):
return True
return False
#: Zero-test evidence vetoes the pass shapes the count-bearing alternatives
#: cannot see: "0 passed", go's no-test markers, unittest "Ran 0 tests".
_TEST_ZERO_SHAPE_RE = re.compile(r"\b0\s+passed\b|\[no test files\]|\[no tests to run\]|\bRan 0 tests\b", re.IGNORECASE)
@ -846,7 +870,16 @@ def _carries_summary_shape(text: str) -> bool:
or zero-test veto). The pass shapes match as substrings, so
subagent-chosen strings — a ``cd`` argument, a ``CDPATH`` value — must
not be allowed to lend the recorded tail a summary shape."""
return bool(_TEST_PASS_SHAPE_RE.search(text) or _TEST_FAIL_SHAPE_RE.search(text) or _TEST_ZERO_SHAPE_RE.search(text) or _MAVEN_TEST_SUMMARY_RE.search(text) or _GRADLE_TEST_SUMMARY_RE.search(text) or _BUILD_SUCCESS_RE.search(text))
return bool(
_TEST_PASS_SHAPE_RE.search(text)
or _GO_TEST_PASS_SHAPE_RE.search(text)
or _UNITTEST_RESULT_LINE_RE.search(text)
or _TEST_FAIL_SHAPE_RE.search(text)
or _TEST_ZERO_SHAPE_RE.search(text)
or _MAVEN_TEST_SUMMARY_RE.search(text)
or _GRADLE_TEST_SUMMARY_RE.search(text)
or _BUILD_SUCCESS_RE.search(text)
)
def _normalize_cd_scope_path(path: str) -> tuple[str, bool] | None:
@ -2010,7 +2043,8 @@ def _check_tests_passed_leaf(command: str, bash_executions: list[dict[str, Any]]
return base
# Empty packages/targets cannot veto complete positive sibling evidence.
summary_tail = _without_empty_cargo_targets(_GO_ZERO_TEST_LINE_RE.sub("", output_tail))
if (_TEST_PASS_SHAPE_RE.search(summary_tail) or build_summary is True) and not _TEST_ZERO_SHAPE_RE.search(summary_tail):
passing = _TEST_PASS_SHAPE_RE.search(summary_tail) or _GO_TEST_PASS_SHAPE_RE.search(summary_tail) or _unittest_test_summary_holds(summary_tail) or build_summary is True
if passing and not _TEST_ZERO_SHAPE_RE.search(summary_tail):
base["checked"] = True
base["holds"] = True
base["detail"] = "recorded output carries a passing test summary"

View File

@ -1848,7 +1848,7 @@ class TestTestsPassedLeaf:
assert verdict["leaves"][0]["checked"] is False
def test_bare_criterion_exact_run_still_holds(self):
executions = [_bash_execution("python -m unittest", output_tail="Ran 12 tests\n\nOK")]
executions = [_bash_execution("python -m unittest", output_tail="Ran 12 tests in 0.001s\n\nOK")]
verdict = check_acceptance_criteria(["tests_passed:python -m unittest"], bash_executions=executions)
assert verdict["leaves"][0]["holds"] is True
@ -2500,7 +2500,7 @@ class TestTestsPassedLeaf:
@pytest.mark.parametrize(
"output",
[
".....\nOK\n",
".....\nRan 5 tests in 0.001s\n\nOK\n",
"test result: ok. 5 passed; 0 failed",
"ok \tgithub.com/example/pkg\t0.5s",
"3 tests completed, 0 failed\nBUILD SUCCESSFUL",

View File

@ -0,0 +1,160 @@
"""Unittest acceptance requires actual passing tests, including annotated OK summaries."""
import asyncio
import importlib.util
import os
import shlex
import sys
from pathlib import Path
from types import SimpleNamespace
import pytest
from langchain_core.messages import AIMessage
from langchain_core.tools import StructuredTool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode
from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware
from deerflow.sandbox.local.local_sandbox import LocalSandbox, PathMapping
from deerflow.subagents.acceptance_checks import check_acceptance_criteria, render_acceptance_section, render_acceptance_segment
def _summary(total=2, options="", newline="\n"):
return newline.join([f"Ran {total} tests in 0.001s", "", f"OK{options}"])
def _verdict(output, **overrides):
execution = {"command": "python -m unittest", "status": "success", "shell_persistent": False, "output_tail": output, **overrides}
return check_acceptance_criteria(["tests_passed:python -m unittest"], bash_executions=[execution])
@pytest.mark.parametrize("options", ["", " (skipped=1)", " (expected failures=1)", " (skipped=1, expected failures=1)"])
@pytest.mark.parametrize("newline", ["\n", "\r\n"])
def test_passing_tests_survive_unittest_annotations(options, newline):
verdict = _verdict(_summary(total=3, options=options, newline=newline))
assert verdict["all_hold"] is True
assert "[holds]" in render_acceptance_section(verdict)
assert "acceptance:" in render_acceptance_segment(verdict)
assert "execution evidence only" in render_acceptance_section(verdict)
@pytest.mark.parametrize(
"output",
[
"OK",
"OK (skipped=1)",
"ok\nexample/package",
_summary(0),
_summary(2, " (skipped=2)"),
_summary(2, " (expected failures=2)"),
_summary(2, " (skipped=1, expected failures=1)"),
_summary(2, " (skipped=3)"),
_summary(2, " (skipped=1, skipped=0)"),
_summary(2, " (unknown=1)"),
_summary("9" * 100),
_summary(2, " (skipped=" + "9" * 100 + ")"),
"Ran 2 tests in\n\nOK",
"Ran 2 tests in 0.001s\n\nOK (skipped=",
"warning: " + _summary(),
],
)
def test_incomplete_empty_or_nonpassing_unittest_evidence_remains_unverified(output):
verdict = _verdict(output)
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == (False, False)
assert "[UNVERIFIED]" in render_acceptance_section(verdict)
@pytest.mark.parametrize("failure", ["FAILED (failures=1)", "FAILED (errors=1)", "FAILED (unexpected successes=1)"])
@pytest.mark.parametrize("reverse", [False, True])
def test_failure_records_still_win_over_a_passing_summary(failure, reverse):
records = [_summary(2, " (skipped=1)"), failure]
verdict = _verdict("\n".join(reversed(records) if reverse else records))
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == (True, False)
@pytest.mark.parametrize(
"overrides",
[
{"shell_persistent": True},
{"shell_persistent": None},
{"command_truncated": True},
{"command": "python -m unittest sample.OneTest"},
{"command": "echo 'other'; python -m unittest"},
{"command": 'cd "OK (skipped=1)" && python -m unittest'},
],
)
def test_annotations_do_not_bypass_evidence_admission(overrides):
assert _verdict(_summary(2, " (skipped=1)"), **overrides)["leaves"][0]["checked"] is False
def test_recorded_error_status_still_overrides_positive_counts():
verdict = _verdict(_summary(2, " (skipped=1)"), status="error")
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == (True, False)
@pytest.fixture
def executor_module(monkeypatch):
# conftest substitutes the executor; use its actual harvest implementation.
path = Path(__file__).parents[1] / "packages/harness/deerflow/subagents/executor.py"
spec = importlib.util.spec_from_file_location("_unittest_evidence_executor", path)
module = importlib.util.module_from_spec(spec)
monkeypatch.setitem(sys.modules, spec.name, module)
spec.loader.exec_module(module)
return module
class SandboxMessages(MessagesState):
sandbox: dict
@pytest.mark.skipif(os.name == "nt", reason="Real LocalSandbox command integration uses a POSIX Python executable")
@pytest.mark.asyncio
@pytest.mark.parametrize(
"methods,expected",
[
(" def test_pass(self): pass\n", (True, True)),
(' def test_pass(self): pass\n @unittest.skip("optional")\n def test_skip(self): pass\n', (True, True)),
(' @unittest.skip("optional")\n def test_skip(self): pass\n', (False, False)),
(" pass\n", (True, False)),
(" @unittest.expectedFailure\n def test_xfail(self): self.fail()\n", (False, False)),
(" def test_pass(self): pass\n @unittest.expectedFailure\n def test_xfail(self): self.fail()\n", (True, True)),
(" def test_fail(self): self.fail()\n", (True, False)),
(' def test_error(self): raise RuntimeError("fixture")\n', (True, False)),
(" @unittest.expectedFailure\n def test_xpass(self): pass\n", (True, False)),
],
ids=["pass", "pass-and-skip", "all-skipped", "empty", "all-expected-failure", "pass-and-expected-failure", "failure", "error", "unexpected-success"],
)
async def test_native_unittest_results_reach_parent_acceptance(tmp_path, monkeypatch, executor_module, methods, expected):
(tmp_path / "test_sample.py").write_text("import unittest\nclass Sample(unittest.TestCase):\n" + methods, encoding="utf-8")
sandbox = LocalSandbox("unittest-evidence", path_mappings=[PathMapping("/mnt/user-data/workspace", str(tmp_path))])
monkeypatch.setattr("deerflow.sandbox.sandbox_provider.get_sandbox_provider", lambda: SimpleNamespace(get=lambda sandbox_id: sandbox if sandbox_id == sandbox.id else None))
outputs = []
async def bash(command: str) -> str:
output = await asyncio.to_thread(sandbox.execute_command, command, timeout=30)
outputs.append(output)
return output
tool = StructuredTool.from_function(coroutine=bash, name="bash", description="Execute the offline Python unittest fixture.")
graph = StateGraph(SandboxMessages)
graph.add_node("tools", ToolNode([tool], awrap_tool_call=ToolErrorHandlingMiddleware().awrap_tool_call))
graph.add_edge(START, "tools")
graph.add_edge("tools", END)
command = shlex.quote(sys.executable) + " -m unittest discover -s /mnt/user-data/workspace"
state = await graph.compile().ainvoke(
{
"sandbox": {"sandbox_id": sandbox.id},
"messages": [AIMessage(content="", tool_calls=[{"name": "bash", "id": "unittest-call", "type": "tool_call", "args": {"command": command}}])],
}
)
assert len(outputs) == 1 and "Ran " in outputs[0], outputs
executions = executor_module._harvest_bash_executions(state)
assert len(executions) == 1 and executions[0]["tool_call_id"] == "unittest-call"
assert executions[0]["shell_persistent"] is False
verdict = check_acceptance_criteria([f"tests_passed:{command}"], bash_executions=executions)
assert (verdict["leaves"][0]["checked"], verdict["all_hold"]) == expected, (outputs, verdict)
if expected == (True, False):
# Python 3.12+ exits 5 for an empty selection, 1 for test errors/failures.
exit_code = 5 if methods == " pass\n" else 1
assert f"Exit Code: {exit_code}" in outputs[0]
assert executions[0]["status"] == "error"