Bound report history enrichment and require unambiguous repository identity

strix/core/repository_history.py

@@ -49,14 +49,23 @@ def _parse_blame(output: str) -> BlameInfo | None:
         return None
 
 
-def blame_line(repository: Path, file_path: object, line: int) -> BlameInfo | None:
+def blame_line(
+    repository: Path, file_path: object, line: int, *, timeout: float = 3.0
+) -> BlameInfo | None:
     """Attribute one working-tree line, or return None when history is unavailable.
 
     Paths are relative to ``repository``. Git follows committed renames itself;
     missing paths, binary files and uncommitted lines are intentionally omitted.
     Lookups never fetch history and are bounded so enrichment stays optional.
+    Callers may supply a shorter timeout to share a budget across lookups.
     """
-    if type(line) is not int or line < 1 or not isinstance(file_path, str) or not file_path:
+    if (
+        type(line) is not int
+        or line < 1
+        or not isinstance(file_path, str)
+        or not file_path
+        or timeout <= 0
+    ):
         return None
     try:
         relative = Path(file_path)
@@ -87,7 +96,7 @@ def blame_line(repository: Path, file_path: object, line: int) -> BlameInfo | No
             encoding="utf-8",
             errors="replace",
             check=False,
-            timeout=3,
+            timeout=min(timeout, 3.0),
             env={**os.environ, "GIT_NO_LAZY_FETCH": "1"},
         )
     except (OSError, ValueError, RuntimeError, subprocess.SubprocessError):

strix/report/history.py

@@ -6,6 +6,7 @@ import html
 import logging
 import re
 from pathlib import Path
+from time import monotonic
 from typing import Any
 
 from strix.core.repository_history import blame_line
@@ -14,6 +15,7 @@ from strix.core.repository_history import blame_line
 logger = logging.getLogger(__name__)
 _START = "<!-- strix:repository-history -->"
 _END = "<!-- /strix:repository-history -->"
+_ENRICHMENT_TIMEOUT = 3.0
 
 
 def _text(value: str) -> str:
@@ -48,7 +50,10 @@ def _repositories(run_record: dict[str, Any], target: str | None) -> list[Path]:
             f"/workspace/{details.get('workspace_subdir')}",
         ):
             matched.add(root)
-    return sorted(matched or roots)
+    # File existence cannot identify the repository for an unmatched target.
+    # Fall back only when the scan itself has exactly one possible checkout.
+    selected = matched or roots
+    return sorted(selected) if len(selected) == 1 else []
 
 
 def _location_line(location: dict[str, Any]) -> int | None:
@@ -65,11 +70,12 @@ def _location_line(location: dict[str, Any]) -> int | None:
 def enrich_report(report: dict[str, Any], run_record: dict[str, Any]) -> None:
     """Refresh blame for the current locations, never failing a report operation.
 
-    No new checkout is made. Ambiguous paths across multiple repositories are
-    omitted unless the finding's target identifies its repository. Markers let
+    No new checkout is made. Multiple repositories require a uniquely matching
+    target. A shared time budget bounds all lookups in a report. Markers let
     revisions and resumed scans replace only the generated part of the analysis.
     """
     try:
+        deadline = monotonic() + _ENRICHMENT_TIMEOUT
         analysis = re.sub(
             re.escape(_START) + r".*?" + re.escape(_END),
             "",
@@ -82,9 +88,13 @@ def enrich_report(report: dict[str, Any], run_record: dict[str, Any]) -> None:
         if not locations:
             return
         roots = _repositories(run_record, report.get("target"))
+        if not roots:
+            return
         entries: list[str] = []
         seen: set[tuple[str, int]] = set()
         for location in locations:
+            if monotonic() >= deadline:
+                break
             file_path = location.get("file")
             line = _location_line(location)
             if not isinstance(file_path, str) or line is None or (file_path, line) in seen:
@@ -93,7 +103,10 @@ def enrich_report(report: dict[str, Any], run_record: dict[str, Any]) -> None:
             candidates = [root for root in roots if (root / file_path).is_file()]
             if len(candidates) != 1:
                 continue
-            info = blame_line(candidates[0], file_path, line)
+            remaining = deadline - monotonic()
+            if remaining <= 0:
+                break
+            info = blame_line(candidates[0], file_path, line, timeout=remaining)
             if info is None:
                 continue
             entries.append(

tests/test_report_history.py

@@ -10,6 +10,8 @@ from typing import TYPE_CHECKING, Any
 import pytest
 from agents.tool_context import ToolContext
 
+from strix.core.repository_history import BlameInfo
+from strix.report import history
 from strix.report.history import enrich_report
 from strix.report.state import ReportState, set_global_report_state
 from strix.tools.reporting.tool import create_vulnerability_report, update_vulnerability_report
@@ -99,7 +101,9 @@ def history_run(
         set_global_report_state(None)
 
 
-async def _create(location: dict[str, Any]) -> dict[str, Any]:
+async def _create(
+    location: dict[str, Any], *additional_locations: dict[str, Any]
+) -> dict[str, Any]:
     arguments = {
         "title": "SQL injection in the query handler",
         "description": "The query handler executes unsanitized input.",
@@ -116,7 +120,7 @@ async def _create(location: dict[str, Any]) -> dict[str, Any]:
         "severity_change_conditions": "A read-only database role would limit the impact.",
         "fix_effort": "low",
         "cvss_breakdown": _CVSS,
-        "code_locations": [location],
+        "code_locations": [location, *additional_locations],
     }
     context = ToolContext(
         context={"agent_id": "root"},
@@ -246,7 +250,7 @@ async def test_unexpected_history_failure_does_not_block_reporting(
 ) -> None:
     state, _clone, _commits = history_run
 
-    def broken_blame(*_args: Any) -> None:
+    def broken_blame(*_args: Any, **_kwargs: Any) -> None:
         raise RuntimeError("history lookup failed")
 
     monkeypatch.setattr("strix.report.history.blame_line", broken_blame)
@@ -282,3 +286,133 @@ def test_ambiguous_repository_requires_matching_target_and_target_updates_refres
     assert "Other Author" in report["technical_analysis"]
     assert commits["first"] not in report["technical_analysis"]
     assert report["technical_analysis"].count("Last modified by") == 1
+
+
+@pytest.mark.parametrize("target", [None, "https://example.test/query", "query handler"])
+def test_unmatched_target_cannot_select_repository_by_unique_file(
+    history_run: tuple[ReportState, Path, dict[str, str]], tmp_path: Path, target: str | None
+) -> None:
+    state, _clone, _commits = history_run
+    other = tmp_path / "other"
+    other.mkdir()
+    state.run_record["local_sources"].append(
+        {"source_path": str(other), "workspace_subdir": "other"}
+    )
+    report = {
+        "target": target,
+        "technical_analysis": _ANALYSIS,
+        "code_locations": [{"file": "app.py", "start_line": 1, "end_line": 1}],
+    }
+
+    enrich_report(report, state.run_record)
+
+    assert report["technical_analysis"] == _ANALYSIS
+
+
+@pytest.mark.parametrize("target", ["application", "/workspace/application"])
+def test_colliding_workspace_alias_cannot_select_repository_by_unique_file(
+    history_run: tuple[ReportState, Path, dict[str, str]], tmp_path: Path, target: str
+) -> None:
+    state, _clone, _commits = history_run
+    other = tmp_path / "other"
+    other.mkdir()
+    state.run_record["local_sources"].append(
+        {"source_path": str(other), "workspace_subdir": "application"}
+    )
+    report = {
+        "target": target,
+        "technical_analysis": _ANALYSIS,
+        "code_locations": [{"file": "app.py", "start_line": 1, "end_line": 1}],
+    }
+
+    enrich_report(report, state.run_record)
+
+    assert report["technical_analysis"] == _ANALYSIS
+
+
+def test_unmatched_target_can_use_only_configured_repository(
+    history_run: tuple[ReportState, Path, dict[str, str]],
+) -> None:
+    state, _clone, commits = history_run
+    report = {
+        "target": "https://example.test/query",
+        "technical_analysis": _ANALYSIS,
+        "code_locations": [{"file": "app.py", "start_line": 1, "end_line": 1}],
+    }
+
+    enrich_report(report, state.run_record)
+
+    assert commits["first"] in report["technical_analysis"]
+
+
+@pytest.mark.parametrize("first_lookup_succeeds", [True, False])
+async def test_history_budget_is_shared_and_partial_results_are_persisted(
+    history_run: tuple[ReportState, Path, dict[str, str]],
+    monkeypatch: pytest.MonkeyPatch,
+    first_lookup_succeeds: bool,
+) -> None:
+    state, _clone, commits = history_run
+    clock = [0.0]
+    timeouts: list[float] = []
+    info = BlameInfo(
+        author_name=_FIRST_AUTHOR,
+        author_email="author@example.test",
+        commit_sha=commits["first"],
+        commit_timestamp="2024-01-02T03:04:05+00:00",
+        commit_summary="Add query handler",
+    )
+
+    def slow_blame(*_args: Any, timeout: float) -> BlameInfo | None:
+        timeouts.append(timeout)
+        clock[0] += min(1.0 if len(timeouts) == 1 else 3.0, timeout)
+        return info if first_lookup_succeeds and len(timeouts) == 1 else None
+
+    monkeypatch.setattr(history, "monotonic", lambda: clock[0])
+    monkeypatch.setattr(history, "blame_line", slow_blame)
+
+    result = await _create(
+        {"file": "app.py", "start_line": 1, "end_line": 1},
+        {"file": "app.py", "start_line": 2, "end_line": 2},
+        {"file": "app.py", "start_line": 999, "end_line": 999},
+    )
+
+    assert result["success"] is True
+    assert timeouts == pytest.approx([3.0, 2.0])
+    assert clock[0] == pytest.approx(3.0)
+    report = state.vulnerability_reports[0]
+    analysis = report["technical_analysis"]
+    if first_lookup_succeeds:
+        assert commits["first"] in analysis
+        assert "**app.py:1**" in analysis
+        assert "**app.py:2**" not in analysis
+    else:
+        assert analysis == _ANALYSIS
+    saved = json.loads((state.get_run_dir() / "vulnerabilities.json").read_text())
+    assert saved == [report]
+    markdown = (state.get_run_dir() / "vulnerabilities" / f"{report['id']}.md").read_text()
+    assert analysis in markdown
+
+
+def test_invalid_locations_still_consume_shared_history_budget(
+    history_run: tuple[ReportState, Path, dict[str, str]], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    state, _clone, _commits = history_run
+    clock = [0.0]
+    location_line = history._location_line
+
+    def slow_location_line(location: dict[str, Any]) -> int | None:
+        clock[0] += 0.5
+        return location_line(location)
+
+    monkeypatch.setattr(history, "monotonic", lambda: clock[0])
+    monkeypatch.setattr(history, "_location_line", slow_location_line)
+    report = {
+        "target": _TARGET,
+        "technical_analysis": _ANALYSIS,
+        "code_locations": [{"file": "app.py"} for _ in range(20)],
+    }
+
+    enrich_report(report, state.run_record)
+
+    assert clock[0] <= 3.0
+    assert report["technical_analysis"] == _ANALYSIS

tests/test_repository_history.py

@@ -196,3 +196,22 @@ def test_nonzero_exit_is_skipped_and_git_is_local_bounded(repository: Path) -> N
     assert kwargs["timeout"] <= 3
     assert kwargs["env"]["GIT_NO_LAZY_FETCH"] == "1"
     assert "--no-textconv" in args[0]
+
+
+@pytest.mark.parametrize(("timeout", "expected"), [(0.25, 0.25), (1.5, 1.5), (30, 3)])
+def test_blame_timeout_respects_remaining_budget_and_per_lookup_limit(
+    repository: Path, timeout: float, expected: float
+) -> None:
+    result = subprocess.CompletedProcess(["git"], 128, stdout="", stderr="fatal")
+    with patch("strix.core.repository_history.subprocess.run", return_value=result) as run:
+        assert blame_line(repository, "vulnerable.py", 2, timeout=timeout) is None
+
+    assert run.call_args.kwargs["timeout"] == pytest.approx(expected)
+
+
+@pytest.mark.parametrize("timeout", [0, -1])
+def test_expired_blame_budget_does_not_start_git(repository: Path, timeout: float) -> None:
+    with patch("strix.core.repository_history.subprocess.run") as run:
+        assert blame_line(repository, "vulnerable.py", 2, timeout=timeout) is None
+
+    run.assert_not_called()