mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 19:59:26 +00:00
Marius CiepluchandClaude Opus 5
3224884d8f
Add group 9 to the approved task plan as the bounded owner amendment release milestone M1a requires, implement the task-local recorded-veto evaluation flag, and run the three pinned NodeGoat revisions under it. The flag re-runs the production comparison and admission over a lifted copy of the same evidence and records each veto instead of applying it. Recognized completeness vetoes are lifted and logged; failures never are, because silence from a failure is not a quiet fixed twin. It is off by default, accepted only on explicit base/head replay, refused in Git stdin mode, labels every payload it produces, and admits no defect even when a lifted candidate otherwise would. M1a outcome, inside its one-week box and with expectations frozen on disk first: the three entailed <lambda>2 witnesses are present at contributions.js 32-34; the fixed twin was genuinely analyzed and quiet, answering its target question with zero rows and no deadline reason while the same findings appear as base-only; the unchanged revision was analyzed and recorded unchanged. The kill criterion did not trigger. Five vetoes fired and were recorded. M1a also corrected the diagnosis: with all five lifted the detection still stops at operation_correspondence_unresolved, because a newly introduced operation sits on a changed line and has no base anchor. The unchanged revision resolves its base location on the same lines and classifies as unchanged, which isolates correspondence as the mechanism. That fifth veto was not in the earlier four-veto diagnosis and is now M1b's central repair. Diagnosis only. No defect admitted, no capability enabled, no gate changed, no latency claim; the 900-second development budget cannot measure the hook gate. Rollout and task 8.3 remain NO-GO. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
267 lines
12 KiB
Python
267 lines
12 KiB
Python
"""M1a recorded-veto evaluation: vetoes are recorded beside the production result, never applied."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
from test_push_policy import context, scan
|
|
|
|
from openultrasast.cpg.backend import NullBackend
|
|
from openultrasast.model.contracts import QuestionOutcome
|
|
from openultrasast.push.contracts import PushComparison
|
|
from openultrasast.push.policy import _compare_evidence, admit_candidates
|
|
from openultrasast.push.report import PushReport, render_report
|
|
from openultrasast.push.vetoes import FLAG, LABEL, RecordedVetoReport, boundary_kind, record_vetoes
|
|
|
|
COMPARISON = PushComparison("head", "base", "explicit", ("refs/heads/main",))
|
|
|
|
|
|
def vetoed(result, reason, *, boundaries=(), vendor=True):
|
|
outcome = result.question_outcomes[0]
|
|
demoted = replace(outcome, status="unresolved", reason=reason)
|
|
return replace(
|
|
result,
|
|
question_outcomes=(demoted,),
|
|
degradations=({"stage": "cpg", "reason": "vendor_semantics_unresolved"},) if vendor else (),
|
|
scope=replace(result.scope, unresolved_boundaries=(*boundaries, *(("vendor_semantics_unresolved",) if vendor else ()))),
|
|
)
|
|
|
|
|
|
def production(head, base, ctx):
|
|
candidates = _compare_evidence(head, base, context=ctx, head_semantics="same", base_semantics="same")
|
|
return candidates, {c.defect_id: ("capability_unavailable",) for c in candidates}
|
|
|
|
|
|
def test_recognized_vetoes_are_lifted_recorded_and_never_applied():
|
|
head = scan()
|
|
question = head.scope.selected[0].identity
|
|
owned = "context_projection_unavailable:contributor-scan:" + question.question_id
|
|
head = vetoed(head, "change_context_incomplete", boundaries=(owned,))
|
|
base = vetoed(scan(rows=[], findings=False), "graph_incomplete")
|
|
ctx = replace(context(), unresolved_boundaries=("612e6a73:ambiguous_line_correspondence",))
|
|
applied, admission = production(head, base, ctx)
|
|
assert applied[0].novelty == "unknown" and applied[0].reason == "witness_identity_unresolved"
|
|
|
|
report = record_vetoes(
|
|
head=head,
|
|
base=base,
|
|
context=ctx,
|
|
comparison=COMPARISON,
|
|
semantics="same",
|
|
production=applied,
|
|
production_admission=admission,
|
|
coverage_reasons=("vendor_semantics_unresolved", "612e6a73:ambiguous_line_correspondence", "deadline_exhausted"),
|
|
)
|
|
assert report.label == LABEL and report.flag == FLAG and report.label.startswith("EXPERIMENTAL")
|
|
finding = report.findings[0]
|
|
assert (finding.production_novelty, finding.production_reason) == ("unknown", "witness_identity_unresolved")
|
|
assert (finding.recorded_novelty, finding.recorded_reason) == ("new", "source_connection_absent_from_comparable_base")
|
|
assert finding.question == question and finding.operation is not None and finding.base_location == "a.js:5"
|
|
lifted = {(v.stage, v.veto) for v in finding.vetoes if v.lifted}
|
|
assert lifted == {
|
|
("head_question", "change_context_incomplete"),
|
|
("head_scan", "context_projection_unavailable"),
|
|
("head_scan", "vendor_semantics_unresolved"),
|
|
("head_scan", "degradation:vendor_semantics_unresolved"),
|
|
("base_question", "graph_incomplete"),
|
|
("base_scan", "vendor_semantics_unresolved"),
|
|
("base_scan", "degradation:vendor_semantics_unresolved"),
|
|
("change_context", "ambiguous_line_correspondence"),
|
|
}
|
|
remaining = {(v.stage, v.veto) for v in finding.vetoes if not v.lifted}
|
|
# The unlifted deadline reason still reaches admission as a dependency gap.
|
|
assert remaining == {("admission", "capability_unavailable"), ("admission", "dependency_unresolved")}
|
|
assert finding.recorded_admission == ("capability_unavailable", "dependency_unresolved")
|
|
assert report.lifted_coverage_reasons == ("vendor_semantics_unresolved", "612e6a73:ambiguous_line_correspondence")
|
|
assert report.remaining_coverage_reasons == ("deadline_exhausted",)
|
|
assert {(c.status, c.reason) for c in report.head.lifted} == {
|
|
("question", "change_context_incomplete"),
|
|
("degradation", "vendor_semantics_unresolved"),
|
|
}
|
|
assert ("question", "graph_incomplete") in {(c.status, c.reason) for c in report.base.lifted}
|
|
assert report.head.changed_path_questions[0].lifted and report.head.changed_path_questions[0].rows == 1
|
|
# Round trip and the production evidence are untouched.
|
|
assert RecordedVetoReport.from_payload(json.loads(json.dumps(report.to_payload()))) == report
|
|
assert head.question_outcomes[0].status == "unresolved" and base.question_outcomes[0].status == "unresolved"
|
|
|
|
|
|
def test_failures_and_unrecognized_boundaries_are_never_lifted():
|
|
head = replace(scan(), degradations=({"stage": "cpg", "reason": "cpg_empty"},))
|
|
base = scan(rows=[], status="unanswered", findings=False)
|
|
applied, admission = production(head, base, context())
|
|
report = record_vetoes(
|
|
head=head,
|
|
base=base,
|
|
context=context(),
|
|
comparison=COMPARISON,
|
|
semantics="same",
|
|
production=applied,
|
|
production_admission=admission,
|
|
coverage_reasons=("cpg_empty",),
|
|
)
|
|
finding = report.findings[0]
|
|
assert finding.recorded_novelty == "unknown" and finding.recorded_reason == "head_context_incomplete"
|
|
assert not any(v.lifted for v in finding.vetoes)
|
|
assert ("head_scan", "degradation:cpg_empty") in {(v.stage, v.veto) for v in finding.vetoes}
|
|
assert ("base_question", "test_answer") in {(v.stage, v.veto) for v in finding.vetoes}
|
|
assert report.remaining_coverage_reasons == ("cpg_empty",) and report.lifted_coverage_reasons == ()
|
|
assert boundary_kind("deadline_exhausted") is None and boundary_kind("snapshot:unreadable") is None
|
|
|
|
|
|
def test_unanswered_lifted_reason_without_rows_stays_applied():
|
|
head = scan()
|
|
outcome = QuestionOutcome(head.question_outcomes[0].identity, "unanswered", "graph_incomplete", None)
|
|
head = replace(head, question_outcomes=(outcome,))
|
|
applied, admission = production(head, scan(rows=[], findings=False), context())
|
|
report = record_vetoes(
|
|
head=head,
|
|
base=scan(rows=[], findings=False),
|
|
context=context(),
|
|
comparison=COMPARISON,
|
|
semantics="same",
|
|
production=applied,
|
|
production_admission=admission,
|
|
coverage_reasons=(),
|
|
)
|
|
assert report.head.lifted == () and report.findings[0].recorded_novelty == "unknown"
|
|
assert report.findings[0].operation is None
|
|
|
|
|
|
def test_quiet_fixed_twin_reports_base_only_finding_with_answered_head_counterpart():
|
|
head = scan(rows=[], findings=False)
|
|
base = scan()
|
|
applied, admission = production(head, base, context())
|
|
report = record_vetoes(
|
|
head=head,
|
|
base=base,
|
|
context=context(),
|
|
comparison=COMPARISON,
|
|
semantics="same",
|
|
production=applied,
|
|
production_admission=admission,
|
|
coverage_reasons=(),
|
|
)
|
|
assert report.findings == ()
|
|
only = report.base_only_findings[0]
|
|
assert only.site == "a.js:4:handler" and only.head_counterparts[0].status == "completed" and only.head_counterparts[0].rows == 0
|
|
|
|
|
|
def test_query_identity_difference_is_recorded_not_applied():
|
|
head = scan()
|
|
question = replace(head.scope.selected[0].identity, function=None)
|
|
rows = json.loads(head.question_outcomes[0].raw_rows_json)
|
|
rows[0]["sinkMethod"] = "<lambda>2"
|
|
head = replace(
|
|
head,
|
|
scope=replace(head.scope, selected=(replace(head.scope.selected[0], identity=question),)),
|
|
question_outcomes=(QuestionOutcome(question, "completed", "test_answer", json.dumps(rows)),),
|
|
findings=(replace(head.findings[0], site="a.js:4:<lambda>2"),),
|
|
)
|
|
base = scan(rows=[], findings=False)
|
|
base = replace(
|
|
base,
|
|
scope=replace(base.scope, selected=(replace(base.scope.selected[0], identity=question),)),
|
|
question_outcomes=(QuestionOutcome(question, "completed", "test_answer", "[]"),),
|
|
)
|
|
applied, admission = production(head, base, context())
|
|
report = record_vetoes(
|
|
head=head,
|
|
base=base,
|
|
context=context(),
|
|
comparison=COMPARISON,
|
|
semantics="same",
|
|
production=applied,
|
|
production_admission=admission,
|
|
coverage_reasons=(),
|
|
)
|
|
finding = report.findings[0]
|
|
identity = [v for v in finding.vetoes if v.stage == "identity"]
|
|
assert identity and identity[0].veto == "query_identity_differs_from_witness" and "<lambda>2" in identity[0].detail
|
|
assert finding.operation_method == "<lambda>2" and finding.recorded_novelty == "new"
|
|
|
|
|
|
def test_recorded_report_never_admits_even_when_lifted_candidate_would_pass():
|
|
from test_push_admission import control
|
|
|
|
candidate, capability, _ = control()
|
|
head = scan()
|
|
base = scan(rows=[], findings=False)
|
|
applied, admission = production(head, base, context())
|
|
report = record_vetoes(
|
|
head=head,
|
|
base=base,
|
|
context=context(),
|
|
comparison=COMPARISON,
|
|
semantics="same",
|
|
production=applied,
|
|
production_admission=admission,
|
|
coverage_reasons=(),
|
|
capabilities=(capability,),
|
|
)
|
|
assert admit_candidates((candidate,), capabilities=(capability,)).defects # the control itself admits
|
|
payload = report.to_payload()
|
|
assert "defects" not in payload and all("admitted" not in f for f in payload["findings"])
|
|
|
|
|
|
def test_report_requires_label_and_renders_experimental_line(tmp_path):
|
|
from test_push_report import report_with
|
|
|
|
base = report_with()
|
|
with pytest.raises(ValueError, match="experimental label"):
|
|
replace(base, experimental={"status": "evaluated"})
|
|
labeled = replace(base, experimental={"label": LABEL, "flag": FLAG, "status": "evaluated", "findings": [{"vetoes": [{}, {}]}]})
|
|
text = render_report(labeled, artifact=None)
|
|
assert "Experimental: recorded-veto evaluation retained 1 raw finding(s) with 2 recorded veto(es)" in text
|
|
assert "not an alert" in text
|
|
pending = replace(base, experimental={"label": LABEL, "flag": FLAG, "status": "not_evaluated", "reason": "head_failed:TimeoutError"})
|
|
assert "was not produced (head_failed:TimeoutError)" in render_report(pending, artifact=None)
|
|
assert "Experimental:" not in render_report(base, artifact=None)
|
|
assert isinstance(base, PushReport)
|
|
|
|
|
|
def test_replay_flag_is_off_by_default_and_labels_its_artifact(tmp_path):
|
|
from test_push_runner import history
|
|
|
|
from openultrasast.push.runner import replay
|
|
|
|
root, base, head, _ = history(tmp_path)
|
|
plain = tmp_path / "plain.json"
|
|
replay(root, base=base, head=head, artifact=plain, backend=NullBackend())
|
|
assert json.loads(plain.read_text())["experimental"] == {}
|
|
recorded = tmp_path / "recorded.json"
|
|
delivery = replay(root, base=base, head=head, artifact=recorded, backend=NullBackend(), record_vetoes=True)
|
|
data = json.loads(recorded.read_text())
|
|
assert data["experimental"]["label"] == LABEL and data["experimental"]["flag"] == FLAG
|
|
assert data["experimental"]["status"] in ("evaluated", "not_evaluated")
|
|
assert data["result"]["finding_status"] == "none" and data["admission"]["defects"] == []
|
|
assert "Experimental:" in delivery.text and delivery.exit_code == 0
|
|
|
|
|
|
def test_cli_refuses_the_flag_outside_explicit_replay(tmp_path, capsys):
|
|
from test_push_runner import history
|
|
|
|
from openultrasast.cli import main
|
|
|
|
root, base, head, _ = history(tmp_path)
|
|
with pytest.raises(SystemExit):
|
|
main(["pre-push", str(root), "--remote", "origin", "/x", "--artifact", str(tmp_path / "x.json"), "--experimental-record-vetoes"])
|
|
assert "requires explicit --base/--head replay" in capsys.readouterr().err
|
|
code = main(
|
|
[
|
|
"pre-push",
|
|
str(root),
|
|
"--base",
|
|
base,
|
|
"--head",
|
|
head,
|
|
"--artifact",
|
|
str(tmp_path / "cli.json"),
|
|
"--deadline",
|
|
"20",
|
|
"--experimental-record-vetoes",
|
|
]
|
|
)
|
|
assert code == 0 and "Experimental:" in capsys.readouterr().out
|