feat: report safety-citation coverage deltas

.github/workflows/coverage-delta.yml

@@ -0,0 +1,35 @@
+---
+name: Coverage delta
+
+on:
+  pull_request:
+
+concurrency:
+  group: coverage-delta-${{ github.event.pull_request.number }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+  issues: write
+  pull-requests: write
+
+jobs:
+  coverage-delta:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          fetch-depth: 0
+      - name: Compare safety-linter citations
+        env:
+          GH_TOKEN: ${{ github.token }}
+          CAN_COMMENT: ${{ github.event.pull_request.head.repo.full_name == github.repository }}
+        run: |-
+          args=()
+          [ "$CAN_COMMENT" = 'true' ] || args+=(--no-comment)
+          python3 -m tools.change_control.coverage_delta \
+            --base '${{ github.event.pull_request.base.sha }}' \
+            --head '${{ github.event.pull_request.head.sha }}' \
+            --repository '${{ github.repository }}' \
+            --pr '${{ github.event.pull_request.number }}' \
+            "${args[@]}"

tools/change_control/coverage_delta.py

@@ -0,0 +1,161 @@
+# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""Compare deterministic safety-linter output between two Git revisions."""
+
+import argparse
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+from .__main__ import GhApi
+
+MARKER = '<!-- coverage-delta -->'
+
+
+def compare_reports(base, head):
+    """Render coverage and citation changes without assigning safety significance."""
+    if head.get('unavailable'):
+        raise RuntimeError(f'head coverage unavailable: {head["unavailable"]}')
+    if base.get('unavailable'):
+        return (
+            'Coverage before: unavailable. The base predates the stacked dependency on '
+            f'change-0001 (`tools/safety_lint`): {base["unavailable"]}\n'
+            f'Coverage after: {head.get("coverage", {}).get("cited_tests", "?")}/{head.get("coverage", {}).get("total", "?")} cited.'
+        )
+    before = base.get('coverage', {})
+    after = head.get('coverage', {})
+    lines = [
+        f'Coverage before: {before.get("cited_tests", "?")}/{before.get("total", "?")} cited.',
+        f'Coverage after: {after.get("cited_tests", "?")}/{after.get("total", "?")} cited.',
+    ]
+    base_citations = base.get('citations', {})
+    head_citations = head.get('citations', {})
+    for sr_id in sorted(set(base_citations) | set(head_citations)):
+        old = set(base_citations.get(sr_id, []))
+        new = set(head_citations.get(sr_id, []))
+        if old - new:
+            lines.append(f'- {sr_id} lost citation(s): {", ".join(sorted(old - new))}')
+        if new - old:
+            lines.append(f'- {sr_id} gained citation(s): {", ".join(sorted(new - old))}')
+    old_unresolved = {
+        (item.get('check_id'), item.get('subject'), item.get('message'))
+        for item in base.get('findings', [])
+        if item.get('check_id') in ('C3', 'C4')
+    }
+    new_unresolved = {
+        (item.get('check_id'), item.get('subject'), item.get('message'))
+        for item in head.get('findings', [])
+        if item.get('check_id') in ('C3', 'C4')
+    }
+    for _, subject, message in sorted(new_unresolved - old_unresolved):
+        lines.append(f'- Newly unresolvable citation for {subject}: {message}')
+    if len(lines) == 2:
+        lines.append('- No citation gains, losses, or newly unresolvable citations.')
+    lines.append(
+        'Limitation: this is deterministic citation resolution, not evidence that a cited test executed or passed.'
+    )
+    return '\n'.join(lines)
+
+
+def run_linter_at_tree(worktree):
+    """Run the revision's own unchanged linter and add its parsed citation map."""
+    if not (worktree / 'tools/safety_lint/__main__.py').is_file():
+        return {'unavailable': 'tools/safety_lint is absent at this revision'}
+    child_environment = os.environ.copy()
+    for name in ('GH_TOKEN', 'GITHUB_TOKEN'):
+        child_environment.pop(name, None)
+    result = subprocess.run(
+        [sys.executable, '-m', 'tools.safety_lint', '--json'],
+        cwd=worktree,
+        env=child_environment,
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode == 2:
+        raise RuntimeError(result.stderr.strip() or 'safety linter could not run')
+    try:
+        report = json.loads(result.stdout)
+    except json.JSONDecodeError as error:
+        raise RuntimeError('safety linter emitted invalid JSON') from error
+    citation_code = (
+        'import json; from tools.safety_lint.runner import analyze; '
+        'print(json.dumps({r.sr_id: sorted(set(r.test_refs)) for r in analyze(".").trace}, sort_keys=True))'
+    )
+    citations = subprocess.run(
+        [sys.executable, '-c', citation_code],
+        cwd=worktree,
+        env=child_environment,
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if citations.returncode:
+        raise RuntimeError(citations.stderr.strip() or 'cannot extract linter citations')
+    report['citations'] = json.loads(citations.stdout)
+    return report
+
+
+def report_at_revision(root, revision):
+    """Run the unchanged checked-in linter at one detached revision."""
+    temporary = Path(tempfile.mkdtemp(prefix='pstop-coverage-delta-'))
+    try:
+        result = subprocess.run(
+            ['git', 'worktree', 'add', '--detach', str(temporary), revision],
+            cwd=root,
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        if result.returncode:
+            raise RuntimeError(result.stderr.strip() or f'cannot materialize revision {revision}')
+        return run_linter_at_tree(temporary)
+    finally:
+        subprocess.run(
+            ['git', 'worktree', 'remove', '--force', str(temporary)], cwd=root, check=False, capture_output=True
+        )
+        shutil.rmtree(temporary, ignore_errors=True)
+
+
+def upsert_coverage_comment(api, repository, pr, report):
+    """Create or update the single marker-owned deterministic coverage comment."""
+    comments = api('GET', f'repos/{repository}/issues/{pr}/comments', paginate=True)
+    existing = next((comment for comment in comments if MARKER in comment.get('body', '')), None)
+    body = f'{MARKER}\n{report}'
+    if existing:
+        api('PATCH', f'repos/{repository}/issues/comments/{existing["id"]}', {'body': body})
+    else:
+        api('POST', f'repos/{repository}/issues/{pr}/comments', {'body': body})
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--root', default='.')
+    parser.add_argument('--base', required=True)
+    parser.add_argument('--head', default='HEAD')
+    parser.add_argument('--repository', required=True)
+    parser.add_argument('--pr', required=True, type=int)
+    parser.add_argument('--gh', default='gh')
+    parser.add_argument('--no-comment', action='store_true')
+    args = parser.parse_args(argv)
+    try:
+        root = Path(args.root).resolve()
+        report = compare_reports(report_at_revision(root, args.base), report_at_revision(root, args.head))
+        print(report)
+        if not args.no_comment:
+            try:
+                upsert_coverage_comment(GhApi(args.gh, args.repository), args.repository, args.pr, report)
+            except RuntimeError as error:
+                print(f'coverage-delta: comment publication warning: {error}', file=sys.stderr)
+        return 0
+    except (OSError, RuntimeError, ValueError) as error:
+        print(f'coverage-delta: cannot run: {error}', file=sys.stderr)
+        return 2
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tools/change_control/test_coverage_delta.py

@@ -0,0 +1,148 @@
+# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""Spec-driven tests for deterministic safety-citation coverage delta reporting."""
+
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from tools.change_control.coverage_delta import (  # noqa: E402
+    compare_reports,
+    run_linter_at_tree,
+    upsert_coverage_comment,
+)
+
+
+class CoverageDeltaTests(unittest.TestCase):
+    def test_coverage_delta_detects_lost_citation(self):
+        """Deleting real cited evidence in a scratch repository must name the affected requirement."""
+        with tempfile.TemporaryDirectory() as directory:
+            scratch = Path(directory) / 'repository'
+            shutil.copytree(ROOT / 'tools/safety_lint/fixtures/repository', scratch)
+            (scratch / 'tools').mkdir(exist_ok=True)
+            shutil.copytree(
+                ROOT / 'tools/safety_lint',
+                scratch / 'tools/safety_lint',
+                ignore=shutil.ignore_patterns('__pycache__'),
+            )
+            base = run_linter_at_tree(scratch)
+            (scratch / 'tests/test_unique_probe.py').unlink()
+            head = run_linter_at_tree(scratch)
+            delta = compare_reports(base, head)
+        self.assertIn('SR-R-01', delta)
+        self.assertIn('unresolvable', delta.lower())
+
+    def test_revision_linter_subprocesses_cannot_observe_actions_credentials(self):
+        """Code from an untrusted revision must receive no Actions token while producing a normal report."""
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            package = root / 'tools/safety_lint'
+            package.mkdir(parents=True)
+            (root / 'tools/__init__.py').write_text('', encoding='utf-8')
+            (package / '__init__.py').write_text('', encoding='utf-8')
+            guard = (
+                "if os.environ.get('GH_TOKEN') or os.environ.get('GITHUB_TOKEN'): raise RuntimeError('token leaked')"
+            )
+            (package / '__main__.py').write_text(
+                'import json, os\n'
+                f'{guard}\n'
+                "print(json.dumps({'coverage': {'cited_tests': 1, 'total': 1}, 'findings': []}))\n",
+                encoding='utf-8',
+            )
+            (package / 'runner.py').write_text(
+                'import os\nfrom types import SimpleNamespace\n'
+                f'{guard}\n'
+                "def analyze(root): return SimpleNamespace(trace=[SimpleNamespace(sr_id='SR-X-01', test_refs=['probe'])])\n",
+                encoding='utf-8',
+            )
+            with mock.patch.dict(os.environ, {'GH_TOKEN': 'gh-sentinel', 'GITHUB_TOKEN': 'github-sentinel'}):
+                report = run_linter_at_tree(root)
+                self.assertEqual(os.environ['GH_TOKEN'], 'gh-sentinel')
+                self.assertEqual(os.environ['GITHUB_TOKEN'], 'github-sentinel')
+        self.assertEqual(report['coverage'], {'cited_tests': 1, 'total': 1})
+        self.assertEqual(report['citations'], {'SR-X-01': ['probe']})
+
+    def test_base_without_linter_exposes_stacked_dependency(self):
+        """A base predating change-0001 must be reported as unavailable, never treated as zero coverage."""
+        delta = compare_reports(
+            {'unavailable': 'tools/safety_lint absent'},
+            {'coverage': {'total': 1, 'cited_tests': 1}, 'findings': [], 'citations': {}},
+        )
+        self.assertIn('stacked dependency', delta.lower())
+
+    def test_head_without_linter_is_an_execution_error(self):
+        """Missing head coverage must fail explicitly rather than render unknown values as a delta."""
+        with self.assertRaisesRegex(RuntimeError, 'head coverage unavailable'):
+            compare_reports(
+                {'coverage': {'total': 1, 'cited_tests': 1}, 'findings': [], 'citations': {}},
+                {'unavailable': 'tools/safety_lint absent'},
+            )
+
+    def test_coverage_comment_updates_instead_of_appending(self):
+        """Coverage delta must update its marker-owned comment rather than append on every run."""
+        writes = []
+
+        def api(method, path, body=None, paginate=False):
+            if method == 'GET':
+                self.assertTrue(paginate)
+                return [{'id': 12, 'body': '<!-- coverage-delta -->\nold'}]
+            writes.append((method, path, body))
+            return {}
+
+        upsert_coverage_comment(api, 'acme/project', 7, 'new')
+        self.assertEqual(writes[0][0:2], ('PATCH', 'repos/acme/project/issues/comments/12'))
+
+    def test_coverage_comment_marker_on_second_page_is_updated(self):
+        """Comment lookup must paginate so a marker beyond page one is updated rather than duplicated."""
+        calls = []
+
+        def api(method, path, body=None, paginate=False):
+            calls.append((method, path, body, paginate))
+            if method == 'GET':
+                self.assertTrue(paginate)
+                return [{'id': 1, 'body': 'first page'}, {'id': 12, 'body': '<!-- coverage-delta -->\nold'}]
+            return {}
+
+        upsert_coverage_comment(api, 'acme/project', 7, 'new')
+        self.assertIn(
+            ('PATCH', 'repos/acme/project/issues/comments/12', {'body': '<!-- coverage-delta -->\nnew'}, False), calls
+        )
+        self.assertFalse(any(call[0] == 'POST' for call in calls))
+
+    def test_coverage_cli_no_comment_avoids_write_api(self):
+        """Fork-safe coverage reporting must support stdout-only operation when write tokens are unavailable."""
+        result = subprocess.run(
+            [sys.executable, '-m', 'tools.change_control.coverage_delta', '--help'],
+            cwd=ROOT,
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        self.assertEqual(result.returncode, 0)
+        self.assertIn('--no-comment', result.stdout)
+
+
+class WorkflowTests(unittest.TestCase):
+    def test_workflows_disable_writes_for_fork_pull_requests(self):
+        """Fork pull requests must still report coverage without attempting unavailable comment writes."""
+        coverage = (ROOT / '.github/workflows/coverage-delta.yml').read_text(encoding='utf-8')
+        self.assertIn('CAN_COMMENT', coverage)
+        self.assertIn('--no-comment', coverage)
+
+    def test_marker_comment_workflows_cancel_superseded_pr_runs(self):
+        """The marker-comment writer must cancel stale runs in its per-PR concurrency group."""
+        coverage = (ROOT / '.github/workflows/coverage-delta.yml').read_text(encoding='utf-8')
+        self.assertIn('group: coverage-delta-${{ github.event.pull_request.number }}', coverage)
+        self.assertEqual(coverage.count('cancel-in-progress: true'), 1)
+
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)