fix(coverage): keep a bundled surface attributed after an unrelated update

update_coverage overwrites a row's agent_id with the updater's, so a
browser_security surface a carrier genuinely assessed looked unexamined
once another agent edited the shared row. Preserve the superseded
author's id in history and let the carrier filter count a row whose
current OR earlier author carried the skill.
This commit is contained in:
Abdelrahman Mostafa 2026-10-06 11:34:10 +03:00
parent e975cf7171
commit 0d3f2595b8
3 changed files with 60 additions and 1 deletions

View file

@ -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 = [

View file

@ -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]

View file

@ -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()}