feat(agents): evidence discipline, shared coverage ledger, and target-scoped threat models

Adds the negative-path half of the analysis process: a mutable coverage ledger recording what was reviewed and cleared, a target-scoped threat model shared across agents and runs, a required counterevidence pass before filing, and a fix-verification gate on inline patches.
This commit is contained in:
Ahmed Allam
2026-08-03 01:41:28 +00:00
parent c071db79ee
commit 5cf49f8527
3 changed files with 23 additions and 21 deletions
+12 -12
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
import pytest
@@ -26,7 +26,7 @@ def coverage_store(tmp_path: Path) -> Path:
return tmp_path
def _record(**overrides: str) -> dict[str, object]:
def _record(**overrides: str) -> dict[str, Any]:
kwargs = {
"surface": "POST /api/orders/{id}",
"risk_area": "object-level authorization",
@@ -36,7 +36,7 @@ def _record(**overrides: str) -> dict[str, object]:
"agent_name": "authz-tester",
}
kwargs.update(overrides)
return _record_impl(**kwargs) # type: ignore[arg-type]
return _record_impl(**kwargs)
def test_record_persists_entry(coverage_store: Path) -> None:
@@ -59,14 +59,14 @@ def test_record_normalizes_outcome() -> None:
def test_record_rejects_unknown_outcome() -> None:
result = _record(outcome="looks fine")
assert result["success"] is False
assert any("Invalid outcome" in e for e in result["errors"]) # type: ignore[union-attr]
assert any("Invalid outcome" in e for e in result["errors"])
assert not get_coverage_entries()
def test_record_requires_surface_and_risk_area() -> None:
result = _record(surface=" ", risk_area="")
assert result["success"] is False
joined = " ".join(result["errors"]) # type: ignore[arg-type]
joined = " ".join(result["errors"])
assert "surface" in joined
assert "risk_area" in joined
@@ -75,7 +75,7 @@ def test_record_requires_surface_and_risk_area() -> None:
def test_evidence_required_for_asserted_outcomes(outcome: str) -> None:
result = _record(outcome=outcome, evidence=" ")
assert result["success"] is False
assert any("evidence is required" in e for e in result["errors"]) # type: ignore[union-attr]
assert any("evidence is required" in e for e in result["errors"])
def test_evidence_optional_for_reported() -> None:
@@ -91,12 +91,12 @@ def test_outcome_counts_and_filtering() -> None:
listed = _list_impl(outcome="needs_follow_up", surface=None, caller_agent_id="agent-1")
assert listed["filtered_count"] == 1
assert listed["entries"][0]["surface"] == "/upload" # type: ignore[index]
assert listed["entries"][0]["by_you"] is True # type: ignore[index]
assert listed["entries"][0]["surface"] == "/upload"
assert listed["entries"][0]["by_you"] is True
by_surface = _list_impl(outcome=None, surface="sea", caller_agent_id=None)
assert by_surface["filtered_count"] == 1
assert by_surface["entries"][0]["surface"] == "/search" # type: ignore[index]
assert by_surface["entries"][0]["surface"] == "/search"
def test_list_rejects_unknown_outcome_filter() -> None:
@@ -112,7 +112,7 @@ def test_hydrate_reloads_from_disk(coverage_store: Path) -> None:
assert entries[0]["risk_area"] == "object-level authorization"
def _update(entry_id: str, **overrides: str) -> dict[str, object]:
def _update(entry_id: str, **overrides: str) -> dict[str, Any]:
kwargs = {
"entry_id": entry_id,
"outcome": "reported",
@@ -121,7 +121,7 @@ def _update(entry_id: str, **overrides: str) -> dict[str, object]:
"agent_name": "followup-tester",
}
kwargs.update(overrides)
return _update_impl(**kwargs) # type: ignore[arg-type]
return _update_impl(**kwargs)
def test_update_moves_outcome_and_keeps_history() -> None:
@@ -162,7 +162,7 @@ def test_update_can_reopen_a_closed_entry() -> None:
assert outcome_counts() == {"needs_follow_up": 1}
listed = _list_impl(outcome=None, surface=None, caller_agent_id=None)
assert listed["entries"][0]["previous_outcomes"] == ["ruled_out"] # type: ignore[index]
assert listed["entries"][0]["previous_outcomes"] == ["ruled_out"]
def test_update_enforces_evidence_for_closing_outcomes() -> None:
+7 -7
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
import pytest
@@ -141,7 +141,7 @@ async def test_create_report_rejects_invalid_fix_effort(report_state: ReportStat
assert not report_state.vulnerability_reports
async def _create_with(report_state: ReportState, **overrides: object) -> dict[str, object]:
async def _create_with(report_state: ReportState, **overrides: object) -> dict[str, Any]:
kwargs: dict[str, object] = {
"title": "X",
"description": "d",
@@ -172,7 +172,7 @@ async def _create_with(report_state: ReportState, **overrides: object) -> dict[s
async def test_create_report_requires_counterevidence(report_state: ReportState) -> None:
result = await _create_with(report_state, counterevidence=" ")
assert result["success"] is False
assert any("Counterevidence" in e for e in result["errors"]) # type: ignore[union-attr]
assert any("Counterevidence" in e for e in result["errors"])
assert not report_state.vulnerability_reports
@@ -181,14 +181,14 @@ async def test_create_report_requires_severity_change_conditions(
) -> None:
result = await _create_with(report_state, severity_change_conditions="")
assert result["success"] is False
assert any("severity_change_conditions" in e for e in result["errors"]) # type: ignore[union-attr]
assert any("severity_change_conditions" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_create_report_rejects_invalid_confidence(report_state: ReportState) -> None:
result = await _create_with(report_state, confidence="pretty sure")
assert result["success"] is False
assert any("confidence" in e for e in result["errors"]) # type: ignore[union-attr]
assert any("confidence" in e for e in result["errors"])
assert not report_state.vulnerability_reports
@@ -197,7 +197,7 @@ async def test_create_report_requires_rationale_when_confidence_not_high(
) -> None:
result = await _create_with(report_state, confidence="medium")
assert result["success"] is False
assert any("confidence_rationale" in e for e in result["errors"]) # type: ignore[union-attr]
assert any("confidence_rationale" in e for e in result["errors"])
assert not report_state.vulnerability_reports
@@ -669,7 +669,7 @@ _INFO_LOCATION = {
async def test_fix_after_requires_verification(report_state: ReportState) -> None:
result = await _create_with(report_state, code_locations=[_FIX_LOCATION])
assert result["success"] is False
assert any("fix_verification" in e for e in result["errors"]) # type: ignore[union-attr]
assert any("fix_verification" in e for e in result["errors"])
assert not report_state.vulnerability_reports
+4 -2
View File
@@ -1,3 +1,4 @@
from collections.abc import Iterator
from pathlib import Path
import pytest
@@ -13,10 +14,11 @@ from strix.skills import (
skill_search_dirs,
validate_requested_skills,
)
from strix.utils.resource_paths import get_strix_resource_path
@pytest.fixture(autouse=True)
def _clear_extra_dirs() -> None:
def _clear_extra_dirs() -> Iterator[None]:
original = list(skills_mod._EXTRA_SKILL_DIRS)
skills_mod._EXTRA_SKILL_DIRS.clear()
try:
@@ -38,7 +40,7 @@ def _write_root_skill(root: Path, name: str, body: str) -> None:
def test_no_registration_leaves_builtin_only() -> None:
assert registered_skill_dirs() == ()
builtin = skills_mod.get_strix_resource_path("skills")
builtin = get_strix_resource_path("skills")
assert skill_search_dirs() == (builtin,)
assert {"nmap", "subfinder"}.issubset(get_available_skills()["tooling"])