diff --git a/strix/report/coverage.py b/strix/report/coverage.py index 0f7a5e8a..d3e70e04 100644 --- a/strix/report/coverage.py +++ b/strix/report/coverage.py @@ -315,6 +315,26 @@ def _entry_is_about(entry: dict[str, Any], phrasings: list[list[str]]) -> bool: return any(all(term in haystack for term in terms) for terms in phrasings) +def _entry_authored_by(entry: dict[str, Any], owner_ids: set[str]) -> bool: + """True when this row — now, or in a superseded revision — was written by + one of *owner_ids*. + + ``update_coverage`` overwrites ``agent_id`` with the updater's, so a surface + a carrier genuinely assessed would look unexamined after an unrelated agent + edits the shared row. History preserves the earlier author, so an earlier + assessment by a carrier still counts. + """ + if str(entry.get("agent_id") or "") in owner_ids: + return True + history = entry.get("history") + if isinstance(history, list): + return any( + isinstance(item, dict) and str(item.get("agent_id") or "") in owner_ids + for item in history + ) + return False + + def _topics_for(skill: str) -> list[tuple[str | None, list[list[str]]]]: """The independently-accountable topics a carried skill must cover. @@ -377,7 +397,7 @@ def skill_coverage_gaps( vclass = VULN_CLASSES.get(skill) if vclass is not None and vclass.sub_topics: owner_ids = carrier_ids.get(skill, set()) - relevant = [e for e in entries if str(e.get("agent_id") or "") in owner_ids] + relevant = [e for e in entries if _entry_authored_by(e, owner_ids)] else: relevant = entries uncovered = [ diff --git a/strix/tools/coverage/tools.py b/strix/tools/coverage/tools.py index 94c31fb9..8c019738 100644 --- a/strix/tools/coverage/tools.py +++ b/strix/tools/coverage/tools.py @@ -292,6 +292,12 @@ def _update_impl( superseded["evidence"] = existing["evidence"] if existing.get("agent_name"): superseded["agent_name"] = existing["agent_name"] + # Preserve the superseded author's id, not just their name: a later + # update by a different agent overwrites ``agent_id``, and coverage + # accounting attributes a bundled skill's surface to the agent that + # actually assessed it, so the earlier authorship has to survive. + if existing.get("agent_id"): + superseded["agent_id"] = existing["agent_id"] history = existing.get("history") existing["history"] = [*history, superseded] if isinstance(history, list) else [superseded] diff --git a/tests/test_report_coverage.py b/tests/test_report_coverage.py index 29da948e..d53d298d 100644 --- a/tests/test_report_coverage.py +++ b/tests/test_report_coverage.py @@ -203,6 +203,39 @@ def test_unrelated_agents_row_does_not_cover_a_browser_subtopic() -> None: assert "postMessage" not in sub # the browser agent's own row still counts +def test_updated_browser_row_keeps_its_carrier_authorship() -> None: + """update_coverage overwrites agent_id with the updater's; a browser surface a + carrier assessed must still count via the author preserved in history.""" + doc = _document( + entries=[ + _entry( + risk_area="client-side path traversal", + surface="fetch from route param", + agent_id="agent-2", # current author after an unrelated update + agent_name="redirect-tester", + history=[ + { + "outcome": "no_issue_found", + "agent_id": "agent-1", # the carrier who first assessed it + "agent_name": "browser-tester", + } + ], + ), + ], + agent_graph=_graph( + statuses={"agent-1": "completed", "agent-2": "completed"}, + names={"agent-1": "browser-tester", "agent-2": "redirect-tester"}, + metadata={ + "agent-1": {"skills": ["browser_security"]}, + "agent-2": {"skills": ["open_redirect"]}, + }, + ), + ) + + sub = {g["risk_area"] for g in doc["gaps"] if g["kind"] == "unrecorded_sub_topic"} + assert "client-side path traversal" not in sub # preserved carrier authorship counts + + def test_registry_backs_skill_phrasings_and_cwe() -> None: """_SKILL_PHRASINGS is derived from the registry, so they cannot drift.""" assert _SKILL_PHRASINGS == {name: v.aliases for name, v in VULN_CLASSES.items()}