Enrich issue technical details with local Git blame

strix/core/repository_history.py

@@ -0,0 +1,95 @@
+"""Best-effort history lookups against an existing local Git checkout."""
+
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class BlameInfo:
+    author_name: str
+    author_email: str
+    commit_sha: str
+    commit_timestamp: str
+    commit_summary: str
+
+
+def _parse_blame(output: str) -> BlameInfo | None:
+    lines = output.splitlines()
+    if not lines or "\x00" in output:
+        return None
+    header = lines[0].split()
+    if not header or not re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", header[0]):
+        return None
+    commit_sha = header[0]
+    if not commit_sha.strip("0"):
+        # Git uses an all-zero object ID for a line with uncommitted changes.
+        return None
+    metadata: dict[str, str] = {}
+    for raw_line in lines[1:]:
+        if raw_line.startswith("\t"):
+            break
+        key, _, value = raw_line.partition(" ")
+        metadata[key] = value
+    try:
+        timestamp = datetime.fromtimestamp(int(metadata["committer-time"]), tz=UTC).isoformat()
+        return BlameInfo(
+            author_name=metadata["author"],
+            author_email=metadata["author-mail"].removeprefix("<").removesuffix(">"),
+            commit_sha=commit_sha,
+            commit_timestamp=timestamp,
+            commit_summary=metadata["summary"],
+        )
+    except (KeyError, ValueError, OverflowError, OSError):
+        return None
+
+
+def blame_line(repository: Path, file_path: object, line: int) -> 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.
+    """
+    if type(line) is not int or line < 1 or not isinstance(file_path, str) or not file_path:
+        return None
+    try:
+        relative = Path(file_path)
+        if relative.is_absolute() or ".." in relative.parts:
+            return None
+        root = repository.resolve(strict=True)
+        path = (root / relative).resolve(strict=True)
+        if not path.is_relative_to(root) or not path.is_file():
+            return None
+        with path.open("rb") as source:
+            if b"\x00" in source.read(8192):
+                return None
+        result = subprocess.run(  # noqa: S603
+            [  # noqa: S607
+                "git",
+                "-C",
+                str(root),
+                "blame",
+                "--line-porcelain",
+                "--no-textconv",
+                "-L",
+                f"{line},{line}",
+                "--",
+                file_path,
+            ],
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            check=False,
+            timeout=3,
+            env={**os.environ, "GIT_NO_LAZY_FETCH": "1"},
+        )
+    except (OSError, ValueError, RuntimeError, subprocess.SubprocessError):
+        return None
+    return _parse_blame(result.stdout) if result.returncode == 0 else None

strix/report/history.py

@@ -0,0 +1,112 @@
+"""Optional repository history inside the existing technical analysis text."""
+
+from __future__ import annotations
+
+import html
+import logging
+import re
+from pathlib import Path
+from typing import Any
+
+from strix.core.repository_history import blame_line
+
+
+logger = logging.getLogger(__name__)
+_START = "<!-- strix:repository-history -->"
+_END = "<!-- /strix:repository-history -->"
+
+
+def _text(value: str) -> str:
+    """Keep repository-authored metadata inert in Markdown and HTML renderers."""
+    return re.sub(r"([\\`*_\[\]])", r"\\\1", html.escape(" ".join(value.split())))
+
+
+def _repositories(run_record: dict[str, Any], target: str | None) -> list[Path]:
+    sources: list[dict[str, Any]] = run_record.get("local_sources") or []
+    roots = {Path(s["source_path"]).resolve() for s in sources if s.get("source_path")}
+    matched: set[Path] = set()
+    for source in sources:
+        if target and target in (
+            source.get("source_path"),
+            source.get("workspace_subdir"),
+            f"/workspace/{source.get('workspace_subdir')}",
+        ):
+            matched.add(Path(source["source_path"]).resolve())
+    targets: list[dict[str, Any]] = run_record.get("targets_info") or []
+    for entry in targets:
+        details: dict[str, Any] = entry.get("details") or {}
+        path = details.get("cloned_repo_path") or details.get("target_path")
+        if entry.get("type") not in {"repository", "local_code"} or not path:
+            continue
+        root = Path(path).resolve()
+        roots.add(root)
+        if target and target in (
+            entry.get("original"),
+            details.get("target_repo"),
+            details.get("target_path"),
+            details.get("workspace_subdir"),
+            f"/workspace/{details.get('workspace_subdir')}",
+        ):
+            matched.add(root)
+    return sorted(matched or roots)
+
+
+def _location_line(location: dict[str, Any]) -> int | None:
+    start = location.get("start_line")
+    if type(start) is not int or start < 1:
+        return None
+    primary = location.get("primary_line")
+    end = location.get("end_line", start)
+    if type(primary) is int and type(end) is int and start <= primary <= end:
+        return primary
+    return start
+
+
+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
+    revisions and resumed scans replace only the generated part of the analysis.
+    """
+    try:
+        analysis = re.sub(
+            re.escape(_START) + r".*?" + re.escape(_END),
+            "",
+            report.get("technical_analysis") or "",
+            flags=re.DOTALL,
+        ).rstrip()
+        if "technical_analysis" in report:
+            report["technical_analysis"] = analysis
+        locations: list[dict[str, Any]] = report.get("code_locations") or []
+        if not locations:
+            return
+        roots = _repositories(run_record, report.get("target"))
+        entries: list[str] = []
+        seen: set[tuple[str, int]] = set()
+        for location in locations:
+            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:
+                continue
+            seen.add((file_path, line))
+            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)
+            if info is None:
+                continue
+            entries.append(
+                f"**{_text(file_path)}:{line}**\n\n"
+                f"- Author: {_text(info.author_name)} ({_text(info.author_email)})\n"
+                f"- Commit: {_text(info.commit_sha)}\n"
+                f"- Commit timestamp (UTC): {_text(info.commit_timestamp)}\n"
+                f"- Commit summary: {_text(info.commit_summary)}"
+            )
+        if entries:
+            block = "\n\n".join(entries)
+            report["technical_analysis"] = (
+                f"{analysis}\n\n{_START}\n### Last modified by\n\n{block}\n{_END}"
+            ).lstrip()
+    except Exception:  # noqa: BLE001 - history must never prevent issue creation or reporting.
+        logger.debug("Repository history enrichment failed (non-fatal)", exc_info=True)

strix/report/state.py

@@ -14,6 +14,7 @@ from strix.config import codex
 from strix.config.loader import load_settings
 from strix.core.paths import run_dir_for, runtime_state_dir
 from strix.report.coverage import write_coverage
+from strix.report.history import enrich_report
 from strix.report.pricing import resolve_litellm_model
 from strix.report.sarif import write_sarif
 from strix.report.writer import (
@@ -407,6 +408,7 @@ class ReportState:
         if agent_name:
             report["agent_name"] = agent_name
 
+        enrich_report(report, self.run_record)
         if self.vulnerability_found_callback:
             self.vulnerability_found_callback(report)
 
@@ -496,6 +498,9 @@ class ReportState:
         revised["update_history"] = history
         revised["updated_at"] = entry["timestamp"]
 
+        if {"technical_analysis", "code_locations", "target"} & changed.keys():
+            enrich_report(revised, self.run_record)
+
         # Persistence must accept the revision before local state changes. A
         # failed callback leaves the old evidence intact and the update retryable.
         if self.vulnerability_updated_callback:

strix/tools/reporting/tool.py

@@ -43,6 +43,7 @@ _CODE_LOCATION_FIELDS = (
     "file",
     "start_line",
     "end_line",
+    "primary_line",
     "snippet",
     "label",
     "fix_before",
@@ -73,7 +74,9 @@ def _normalize_code_locations(
             if field not in loc or loc[field] is None:
                 continue
             value = loc[field]
-            if field in ("start_line", "end_line"):
+            if field in ("start_line", "end_line", "primary_line"):
+                if isinstance(value, (bool, float)):
+                    continue
                 try:
                     normalized[field] = int(value)
                 except (TypeError, ValueError):
@@ -1123,7 +1126,9 @@ async def create_vulnerability_report(
             but unverified follow-on risks separate; do not use them to
             set CVSS metrics.
         target: Affected URL / domain / repository.
-        technical_analysis: The mechanism and root cause.
+        technical_analysis: The mechanism and root cause. When local repository
+            history is available, the report automatically appends "Last modified
+            by" details for the code locations. Do not invent attribution.
         poc_description: Step-by-step reproduction (steps only, no code).
         poc_script_code: Working PoC (Python preferred).
         remediation_steps: Specific, actionable fix (prose, no code).
@@ -1221,6 +1226,9 @@ async def create_vulnerability_report(
             - ``end_line`` (REQUIRED): 1-based; ``>= start_line``.
               Only equal to ``start_line`` when the block truly is one
               line.
+            - ``primary_line`` (optional): the primary vulnerable line within
+              this range. Local Git blame uses this line, or ``start_line``
+              when omitted or outside the range. History enrichment is best-effort.
             - ``snippet`` (optional): verbatim source at this range.
             - ``label`` (optional): short role description; especially
               important for multi-part fixes.

tests/test_report_history.py

@@ -0,0 +1,284 @@
+"""Repository history through the reporting tool, callbacks, and persisted artifacts."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+from typing import TYPE_CHECKING, Any
+
+import pytest
+from agents.tool_context import ToolContext
+
+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
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from pathlib import Path
+
+
+_ANALYSIS = "The query interpolates attacker-controlled input."
+_TARGET = "https://example.test/team/application.git"
+_FIRST_AUTHOR = "Alice Original"
+_LATEST_AUTHOR = "Bea Reviewer"
+_CVSS = {
+    "attack_vector": "N",
+    "attack_complexity": "L",
+    "privileges_required": "N",
+    "user_interaction": "N",
+    "scope": "U",
+    "confidentiality": "H",
+    "integrity": "H",
+    "availability": "H",
+}
+
+
+def _git(path: Path, *args: str, author: str = _FIRST_AUTHOR) -> str:
+    result = subprocess.run(  # noqa: S603
+        ["git", "-c", "commit.gpgsign=false", "-C", str(path), *args],  # noqa: S607
+        env={
+            **os.environ,
+            "GIT_AUTHOR_NAME": author,
+            "GIT_AUTHOR_EMAIL": "author@example.test",
+            "GIT_COMMITTER_NAME": author,
+            "GIT_COMMITTER_EMAIL": "committer@example.test",
+            "GIT_AUTHOR_DATE": "2024-01-02T03:04:05+00:00",
+            "GIT_COMMITTER_DATE": "2024-01-02T03:04:05+00:00",
+        },
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return result.stdout.strip()
+
+
+@pytest.fixture
+def history_run(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> Iterator[tuple[ReportState, Path, dict[str, str]]]:
+    """Use an existing full clone, including an older author for the range start."""
+    monkeypatch.chdir(tmp_path)
+    origin = tmp_path / "origin"
+    origin.mkdir()
+    _git(origin, "init", "--quiet")
+    (origin / "app.py").write_text("query = 'initial'\nresult = query\n", encoding="utf-8")
+    _git(origin, "add", "app.py")
+    _git(origin, "commit", "--quiet", "-m", "Add query handler")
+    first_sha = _git(origin, "rev-parse", "HEAD")
+    (origin / "app.py").write_text("query = 'initial'\nresult = execute(query)\n", encoding="utf-8")
+    _git(origin, "commit", "--quiet", "-am", "Execute the query", author=_LATEST_AUTHOR)
+    latest_sha = _git(origin, "rev-parse", "HEAD")
+    clone = tmp_path / "application"
+    _git(tmp_path, "clone", "--quiet", "--no-hardlinks", str(origin), str(clone))
+    assert _git(clone, "rev-parse", "--is-shallow-repository") == "false"
+
+    state = ReportState(run_name="history-run")
+    state.set_scan_config(
+        {
+            "targets": [
+                {
+                    "type": "repository",
+                    "original": _TARGET,
+                    "details": {
+                        "target_repo": _TARGET,
+                        "cloned_repo_path": str(clone),
+                        "workspace_subdir": "application",
+                    },
+                }
+            ],
+            "local_sources": [{"source_path": str(clone), "workspace_subdir": "application"}],
+        }
+    )
+    set_global_report_state(state)
+    try:
+        yield state, clone, {"first": first_sha, "latest": latest_sha}
+    finally:
+        set_global_report_state(None)
+
+
+async def _create(location: dict[str, Any]) -> dict[str, Any]:
+    arguments = {
+        "title": "SQL injection in the query handler",
+        "description": "The query handler executes unsanitized input.",
+        "impact": "An anonymous caller can access other users' records.",
+        "target": _TARGET,
+        "technical_analysis": _ANALYSIS,
+        "poc_description": "Submit a quote in the query parameter.",
+        "poc_script_code": "GET /query?q='",
+        "remediation_steps": "Use parameterized queries.",
+        "evidence": "The response includes another user's record.",
+        "assumptions": "The observed database role is available in production.",
+        "counterevidence": "No input validation runs before the query.",
+        "confidence": "high",
+        "severity_change_conditions": "A read-only database role would limit the impact.",
+        "fix_effort": "low",
+        "cvss_breakdown": _CVSS,
+        "code_locations": [location],
+    }
+    context = ToolContext(
+        context={"agent_id": "root"},
+        tool_name="create_vulnerability_report",
+        tool_call_id="create-1",
+        tool_arguments=json.dumps(arguments),
+    )
+    result: dict[str, Any] = json.loads(
+        await create_vulnerability_report.on_invoke_tool(context, json.dumps(arguments))
+    )
+    return result
+
+
+async def _update(report_id: str, **fields: Any) -> dict[str, Any]:
+    arguments = {
+        "report_id": report_id,
+        "update_reason": "Refined the vulnerable location.",
+        **fields,
+    }
+    context = ToolContext(
+        context={"agent_id": "root"},
+        tool_name="update_vulnerability_report",
+        tool_call_id="update-1",
+        tool_arguments=json.dumps(arguments),
+    )
+    result: dict[str, Any] = json.loads(
+        await update_vulnerability_report.on_invoke_tool(context, json.dumps(arguments))
+    )
+    return result
+
+
+@pytest.mark.parametrize("primary_line", [2, None, -1, 3, "invalid", True])
+async def test_tool_enriches_callbacks_and_artifacts_from_full_clone(
+    history_run: tuple[ReportState, Path, dict[str, str]], primary_line: Any
+) -> None:
+    state, _clone, commits = history_run
+    callbacks: list[dict[str, Any]] = []
+    state.vulnerability_found_callback = lambda report: callbacks.append(dict(report))
+    location = {"file": "app.py", "start_line": 1, "end_line": 2, "primary_line": primary_line}
+
+    result = await _create(location)
+
+    assert result["success"] is True
+    report = state.vulnerability_reports[0]
+    analysis = report["technical_analysis"]
+    expected_author = _LATEST_AUTHOR if primary_line == 2 else _FIRST_AUTHOR
+    expected_sha = commits["latest"] if primary_line == 2 else commits["first"]
+    expected_summary = "Execute the query" if primary_line == 2 else "Add query handler"
+    assert analysis.startswith(_ANALYSIS)
+    assert analysis.count("Last modified by") == 1
+    for value in (expected_author, "author@example.test", expected_sha, expected_summary):
+        assert value in analysis
+    assert "2024-01-02T03:04:05+00:00" in analysis
+    assert callbacks == [report]
+    assert "blame" not in report
+    assert "author_name" not in report
+    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()
+    technical_section = markdown.split("## Technical Analysis", 1)[1].split(
+        "## Proof of Concept", 1
+    )[0]
+    assert analysis in technical_section
+
+
+async def test_resume_and_updates_refresh_history_without_stale_or_duplicate_details(
+    history_run: tuple[ReportState, Path, dict[str, str]],
+) -> None:
+    state, _clone, commits = history_run
+    assert (await _create({"file": "app.py", "start_line": 1, "end_line": 2}))["success"]
+    resumed = ReportState(run_name=state.run_name)
+    resumed.hydrate_from_run_dir()
+    set_global_report_state(resumed)
+    callbacks: list[dict[str, Any]] = []
+    resumed.vulnerability_updated_callback = lambda report: callbacks.append(dict(report))
+    report_id = resumed.vulnerability_reports[0]["id"]
+    assert commits["first"] in resumed.vulnerability_reports[0]["technical_analysis"]
+
+    result = await _update(
+        report_id,
+        code_locations=[{"file": "app.py", "start_line": 1, "end_line": 2, "primary_line": 2}],
+    )
+    assert result["success"] is True
+    analysis = resumed.vulnerability_reports[0]["technical_analysis"]
+    assert commits["latest"] in analysis
+    assert commits["first"] not in analysis
+    result = await _update(report_id, technical_analysis="Revised root cause.\n\n" + analysis)
+    assert result["success"] is True
+    analysis = resumed.vulnerability_reports[0]["technical_analysis"]
+    assert analysis.count("Last modified by") == 1
+    assert analysis.startswith("Revised root cause.")
+
+    result = await _update(
+        report_id, code_locations=[{"file": "deleted.py", "start_line": 1, "end_line": 2}]
+    )
+    assert result["success"] is True
+    final_report = resumed.vulnerability_reports[0]
+    assert "Last modified by" not in final_report["technical_analysis"]
+    assert callbacks[-1] == final_report
+    saved = json.loads((resumed.get_run_dir() / "vulnerabilities.json").read_text())
+    assert saved == [final_report]
+    markdown = (resumed.get_run_dir() / "vulnerabilities" / f"{report_id}.md").read_text()
+    assert "Last modified by" not in markdown
+
+
+@pytest.mark.parametrize(
+    "location",
+    [
+        {"file": "deleted.py", "start_line": 1, "end_line": 1},
+        {"file": "app.py", "start_line": 999, "end_line": 999},
+        {"file": "app.py", "start_line": True, "end_line": 1},
+        {"file": "app.py", "start_line": 1.5, "end_line": 2},
+        {"file": "app.py", "end_line": 1},
+    ],
+)
+async def test_unavailable_history_does_not_block_reporting(
+    history_run: tuple[ReportState, Path, dict[str, str]], location: dict[str, Any]
+) -> None:
+    state, _clone, _commits = history_run
+    assert (await _create(location))["success"] is True
+    assert state.vulnerability_reports[0]["technical_analysis"] == _ANALYSIS
+    assert (state.get_run_dir() / "vulnerabilities.json").exists()
+
+
+async def test_unexpected_history_failure_does_not_block_reporting(
+    history_run: tuple[ReportState, Path, dict[str, str]], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    state, _clone, _commits = history_run
+
+    def broken_blame(*_args: Any) -> None:
+        raise RuntimeError("history lookup failed")
+
+    monkeypatch.setattr("strix.report.history.blame_line", broken_blame)
+    assert (await _create({"file": "app.py", "start_line": 1, "end_line": 1}))["success"] is True
+    assert state.vulnerability_reports[0]["technical_analysis"] == _ANALYSIS
+    assert (state.get_run_dir() / "vulnerabilities" / "vuln-0001.md").exists()
+
+
+@pytest.mark.parametrize("target_alias", [_TARGET, "application", "/workspace/application"])
+def test_ambiguous_repository_requires_matching_target_and_target_updates_refresh_history(
+    history_run: tuple[ReportState, Path, dict[str, str]], tmp_path: Path, target_alias: str
+) -> None:
+    state, clone, commits = history_run
+    other = tmp_path / "other"
+    _git(tmp_path, "clone", "--quiet", str(clone), str(other))
+    (other / "app.py").write_text("query = 'other'\nresult = execute(query)\n", encoding="utf-8")
+    _git(other, "commit", "--quiet", "-am", "Other repository change", author="Other Author")
+    state.run_record["local_sources"].append(
+        {"source_path": str(other), "workspace_subdir": "other"}
+    )
+    report = {
+        "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
+
+    report["target"] = target_alias
+    enrich_report(report, state.run_record)
+    assert commits["first"] in report["technical_analysis"]
+    report["target"] = "other"
+    enrich_report(report, state.run_record)
+    assert "Other Author" in report["technical_analysis"]
+    assert commits["first"] not in report["technical_analysis"]
+    assert report["technical_analysis"].count("Last modified by") == 1

tests/test_repository_history.py

@@ -0,0 +1,198 @@
+"""Local history lookup coverage using real Git history and failure injection."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from typing import TYPE_CHECKING, Any
+from unittest.mock import patch
+
+import pytest
+
+from strix.core.repository_history import blame_line
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _git(repository: Path, *args: str, env: dict[str, str] | None = None) -> str:
+    return subprocess.run(  # noqa: S603
+        ["git", "-C", str(repository), *args],  # noqa: S607
+        env={**os.environ, **(env or {})},
+        check=True,
+        capture_output=True,
+        text=True,
+    ).stdout.strip()
+
+
+@pytest.fixture
+def repository(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    monkeypatch.setenv("GIT_CONFIG_GLOBAL", os.devnull)
+    monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1")
+    root = tmp_path / "repository"
+    root.mkdir()
+    _git(root, "init")
+    _git(root, "config", "user.name", "First Author")
+    _git(root, "config", "user.email", "first@example.test")
+    (root / "vulnerable.py").write_text("first line\nvulnerable line\nlast line\n")
+    _git(root, "add", "vulnerable.py")
+    _git(
+        root,
+        "commit",
+        "-m",
+        "Original source",
+        env={
+            "GIT_AUTHOR_DATE": "2020-01-01T00:00:00Z",
+            "GIT_COMMITTER_DATE": "2020-01-02T00:00:00Z",
+        },
+    )
+    (root / "vulnerable.py").write_text("first line\nchanged vulnerable line\nlast line\n")
+    _git(root, "add", "vulnerable.py")
+    _git(
+        root,
+        "commit",
+        "-m",
+        "Change vulnerable line\n\nCommit body.",
+        env={
+            "GIT_AUTHOR_NAME": "Second Author",
+            "GIT_AUTHOR_EMAIL": "second@example.test",
+            "GIT_AUTHOR_DATE": "2021-02-03T04:05:06Z",
+            "GIT_COMMITTER_DATE": "2021-02-04T05:06:07Z",
+        },
+    )
+    return root
+
+
+def test_blame_attributes_requested_line_and_commit_timestamp(repository: Path) -> None:
+    result = blame_line(repository, "vulnerable.py", 2)
+
+    assert result is not None
+    assert result.author_name == "Second Author"
+    assert result.author_email == "second@example.test"
+    assert result.commit_sha == _git(repository, "rev-parse", "HEAD")
+    assert result.commit_timestamp == "2021-02-04T05:06:07+00:00"
+    assert result.commit_summary == "Change vulnerable line"
+    unchanged = blame_line(repository, "vulnerable.py", 1)
+    assert unchanged is not None
+    assert unchanged.author_name == "First Author"
+    assert unchanged.commit_sha == _git(repository, "rev-parse", "HEAD~1")
+
+
+def test_committed_rename_follows_history_and_missing_old_path_is_skipped(repository: Path) -> None:
+    original_sha = _git(repository, "rev-parse", "HEAD")
+    _git(repository, "mv", "vulnerable.py", "renamed.py")
+    _git(repository, "commit", "-m", "Rename source")
+
+    result = blame_line(repository, "renamed.py", 2)
+    assert result is not None
+    assert result.commit_sha == original_sha
+    assert blame_line(repository, "vulnerable.py", 2) is None
+
+
+def test_uncommitted_line_is_skipped_but_unchanged_line_retains_history(repository: Path) -> None:
+    (repository / "vulnerable.py").write_text("first line\nuncommitted change\nlast line\n")
+
+    assert blame_line(repository, "vulnerable.py", 2) is None
+    assert blame_line(repository, "vulnerable.py", 1) is not None
+    _git(repository, "add", "vulnerable.py")
+    assert blame_line(repository, "vulnerable.py", 2) is None
+
+
+@pytest.mark.parametrize("line", [None, 0, -1, True, "2", 1.5, 500])
+def test_invalid_missing_or_out_of_bounds_lines_are_skipped(repository: Path, line: Any) -> None:
+    assert blame_line(repository, "vulnerable.py", line) is None
+
+
+@pytest.mark.parametrize(
+    "file_path", ["", "missing.py", "../vulnerable.py", "/etc/passwd", "bad\x00"]
+)
+def test_invalid_or_missing_paths_are_skipped(repository: Path, file_path: str) -> None:
+    assert blame_line(repository, file_path, 1) is None
+
+
+def test_binary_untracked_and_non_repository_files_are_skipped(
+    repository: Path, tmp_path: Path
+) -> None:
+    (repository / "binary.dat").write_bytes(b"binary\x00data\n")
+    _git(repository, "add", "binary.dat")
+    _git(repository, "commit", "-m", "Add binary")
+    (repository / "generated.py").write_text("generated source\n")
+    (tmp_path / "plain.py").write_text("not a repository\n")
+
+    assert blame_line(repository, "binary.dat", 1) is None
+    assert blame_line(repository, "generated.py", 1) is None
+    assert blame_line(tmp_path, "plain.py", 1) is None
+    assert blame_line(tmp_path / "absent", "plain.py", 1) is None
+
+
+def test_tracked_generated_text_can_be_attributed(repository: Path) -> None:
+    (repository / "generated.py").write_text("# Generated file\nvalue = 1\n")
+    _git(repository, "add", "generated.py")
+    _git(repository, "commit", "-m", "Generate source")
+
+    result = blame_line(repository, "generated.py", 2)
+    assert result is not None
+    assert result.commit_summary == "Generate source"
+
+
+def test_existing_linked_worktree_uses_its_local_history(repository: Path, tmp_path: Path) -> None:
+    worktree = tmp_path / "worktree"
+    _git(repository, "worktree", "add", "--detach", str(worktree), "HEAD")
+
+    result = blame_line(worktree, "vulnerable.py", 2)
+    assert result is not None
+    assert result.commit_sha == _git(repository, "rev-parse", "HEAD")
+
+
+def test_symlink_cannot_escape_repository(repository: Path, tmp_path: Path) -> None:
+    outside = tmp_path / "outside.py"
+    outside.write_text("outside source\n")
+    (repository / "escape.py").symlink_to(outside)
+
+    with patch("strix.core.repository_history.subprocess.run") as run:
+        assert blame_line(repository, "escape.py", 1) is None
+    run.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "failure",
+    [
+        FileNotFoundError("git missing"),
+        PermissionError("denied"),
+        subprocess.TimeoutExpired("git", 3),
+    ],
+)
+def test_git_failures_do_not_escape(repository: Path, failure: Exception) -> None:
+    with patch("strix.core.repository_history.subprocess.run", side_effect=failure):
+        assert blame_line(repository, "vulnerable.py", 2) is None
+
+
+@pytest.mark.parametrize("output", ["", "malformed", "a" * 40 + " 1 1 1\nauthor Someone\n"])
+def test_malformed_output_is_skipped(repository: Path, output: str) -> None:
+    result = subprocess.CompletedProcess(["git"], 0, stdout=output)
+    with patch("strix.core.repository_history.subprocess.run", return_value=result):
+        assert blame_line(repository, "vulnerable.py", 2) is None
+
+
+@pytest.mark.parametrize("timestamp", ["not-a-time", "9999999999999999999999999999999"])
+def test_invalid_commit_timestamp_is_skipped(repository: Path, timestamp: str) -> None:
+    output = (
+        f"{'a' * 40} 2 2 1\nauthor Name\nauthor-mail <name@example.test>\n"
+        f"committer-time {timestamp}\nsummary Message\n\tcode\n"
+    )
+    result = subprocess.CompletedProcess(["git"], 0, stdout=output)
+    with patch("strix.core.repository_history.subprocess.run", return_value=result):
+        assert blame_line(repository, "vulnerable.py", 2) is None
+
+
+def test_nonzero_exit_is_skipped_and_git_is_local_bounded(repository: Path) -> 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) is None
+
+    args, kwargs = run.call_args
+    assert args[0][-4:] == ["-L", "2,2", "--", "vulnerable.py"]
+    assert kwargs["timeout"] <= 3
+    assert kwargs["env"]["GIT_NO_LAZY_FETCH"] == "1"
+    assert "--no-textconv" in args[0]