research: reviewed D1–D4 chain, reproducible CI and pilot gates

.github/workflows/convergence.yml

@@ -0,0 +1,99 @@
+name: Convergence review checks
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, ready_for_review]
+    paths:
+      - 'docs/research/source-trust/**'
+      - '.github/workflows/convergence.yml'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+env:
+  PYTHONUTF8: '1'
+  PYTHONIOENCODING: utf-8
+
+jobs:
+  convergence:
+    name: Convergence (${{ matrix.os }}, Python 3.12)
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, windows-latest]
+    defaults:
+      run:
+        shell: bash
+    steps:
+      - name: Preserve exact committed bytes on Windows too
+        run: git config --global core.autocrlf false
+      - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
+        with:
+          fetch-depth: 0
+      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97
+        with:
+          python-version: '3.12'
+      - name: Install pinned offline test dependency
+        run: python -m pip install -r docs/research/source-trust/convergence/d5-calibration-v2/requirements.txt
+      - name: CI log gate regression tests (outside the frozen 212-test suite)
+        run: |
+          mkdir -p logs/convergence
+          python -I -m unittest discover -s docs/research/source-trust/ci -p 'test_*.py' -v 2>&1 | tee logs/convergence/log-gate-tests.log
+      - name: D5 v2 engineering tests (not a calibration run or pilot)
+        run: |
+          python -I -X utf8 -m unittest discover -s docs/research/source-trust/convergence/d5-calibration-v2/tests -p 'test_*.py' -v 2>&1 | tee logs/convergence/d5-v2-tests.log
+      - name: Prospective design metadata tests (no admission or pilot)
+        run: |
+          python -I -X utf8 -c "import unittest; s=unittest.defaultTestLoader.discover('docs/research/source-trust/convergence/prospective-design-v1/tests', pattern='test_*.py'); r=unittest.TextTestRunner(verbosity=2).run(s); raise SystemExit(not (r.wasSuccessful() and r.testsRun == 15 and len(r.skipped) == 0))" 2>&1 | tee logs/convergence/prospective-design-tests.log
+          for packet in prospective-design-v1/proposal.json ledger-input-validation-v1/run-001/sampling-contract.json; do
+            name=$(basename "$packet" .json)
+            if python -I -X utf8 docs/research/source-trust/convergence/prospective-design-v1/design_binding.py --commit HEAD --packet "docs/research/source-trust/convergence/$packet" > "logs/convergence/design-$name.json"; then
+              exit 1
+            else
+              result=$?
+              test "$result" -eq 2
+            fi
+          done
+          python -I -c "import json; from pathlib import Path; p=json.loads(Path('logs/convergence/design-proposal.json').read_text(encoding='utf-8')); h=json.loads(Path('logs/convergence/design-sampling-contract.json').read_text(encoding='utf-8')); assert p['state']=='PENDING_REAL_DESIGN_INPUTS' and h['state']=='REJECTED_POST_OUTCOME_DESIGN'; assert all(x['experiment_authorized'] is False and x['interval_selection_ready'] is False and x['formal_materials_admitted']==0 for x in (p,h))"
+      - name: Full convergence suite (platform skips remain visible)
+        run: |
+          mkdir -p logs/convergence
+          if python -u -m unittest discover -s docs/research/source-trust/convergence -p 'test_*.py' -v > logs/convergence/tests.log 2>&1; then
+            result=0
+          else
+            result=$?
+          fi
+          tail -n 20 logs/convergence/tests.log
+          exit "$result"
+      - name: Enforce frozen test count and platform skip limits
+        run: |
+          python -I docs/research/source-trust/ci/check_convergence_log.py logs/convergence/tests.log --runner-os "${{ runner.os }}" | tee logs/convergence/test-counts.json
+      - name: Original design and frozen-file checks
+        run: python docs/research/source-trust/convergence/verify_convergence_design.py > logs/convergence/design.json
+      - name: Commit-bound snapshot verification
+        run: python docs/research/source-trust/convergence/evidence_snapshots.py > logs/convergence/snapshots.json
+      - name: Provider gate remains blocked
+        run: |
+          if python docs/research/source-trust/convergence/pilot_readiness.py docs/research/source-trust/convergence/pilot-startup-v1/startup-check.json > logs/convergence/pilot.json; then
+            echo 'Unexpected dispatchable pilot'
+            exit 1
+          else
+            result=$?
+            test "$result" -eq 2
+          fi
+      - name: Replay immutable historical archives at their measured commits
+        if: runner.os == 'Linux'
+        run: |
+          git worktree add --detach "$RUNNER_TEMP/agentshield-v2" 3ebce7de93781fb5e3c742529c8a6f5be882f2ed
+          python "$RUNNER_TEMP/agentshield-v2/docs/research/source-trust/convergence/run_development_v2.py" verify --root "$RUNNER_TEMP/agentshield-v2/docs/research/source-trust/convergence/development-v2/run-001" > logs/convergence/historical-v2.json
+          git worktree add --detach "$RUNNER_TEMP/agentshield-v3" c57563d7a34c3bf6cdd2f5b48b85a5031143596c
+          python "$RUNNER_TEMP/agentshield-v3/docs/research/source-trust/convergence/model_workflow_v3.py" verify --root "$RUNNER_TEMP/agentshield-v3/docs/research/source-trust/convergence/model-capture-v3/run-001" > logs/convergence/historical-v3.json
+      - name: Upload logs including skips and blocked pilot state
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02
+        with:
+          name: convergence-${{ matrix.os }}
+          path: logs/convergence/
+          retention-days: 14

docs/research/source-trust/ci/check_convergence_log.py

@@ -0,0 +1,62 @@
+"""Enforce the frozen convergence suite size and OS skip budget offline."""
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+import re
+
+
+EXPECTED_TESTS = 212
+MAX_SKIPPED = {"Linux": 0, "Windows": 50}
+SUMMARY = re.compile(
+    r"^Ran (?P<tests>\d+) tests? in [0-9]+(?:\.[0-9]+)?s\n\n"
+    r"OK(?: \(skipped=(?P<skipped>\d+)\))?$",
+    re.MULTILINE,
+)
+
+
+def validate_test_log(text: str, runner_os: str) -> dict:
+    if runner_os not in MAX_SKIPPED:
+        raise ValueError(f"Unsupported runner OS: {runner_os!r}")
+    normalized = text.replace("\r\n", "\n").rstrip()
+    run_headers = re.findall(r"^Ran \d+ tests? in .*?$", normalized, re.MULTILINE)
+    if len(run_headers) != 1:
+        raise ValueError("Expected exactly one unittest run summary")
+    match = SUMMARY.search(normalized)
+    if match is None or match.end() != len(normalized):
+        raise ValueError("Missing final successful unittest summary")
+    tests = int(match["tests"])
+    skipped = int(match["skipped"] or 0)
+    if tests != EXPECTED_TESTS:
+        raise ValueError(f"Expected Ran {EXPECTED_TESTS}, found Ran {tests}")
+    if skipped > MAX_SKIPPED[runner_os]:
+        raise ValueError(
+            f"{runner_os} skipped={skipped} exceeds limit={MAX_SKIPPED[runner_os]}"
+        )
+    return {
+        "status": "PASS",
+        "runner_os": runner_os,
+        "tests": tests,
+        "skipped": skipped,
+        "expected_tests": EXPECTED_TESTS,
+        "max_skipped": MAX_SKIPPED[runner_os],
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("log", type=Path)
+    parser.add_argument("--runner-os", required=True)
+    args = parser.parse_args()
+    try:
+        report = validate_test_log(args.log.read_text(encoding="utf-8"), args.runner_os)
+    except (OSError, UnicodeError, ValueError) as exc:
+        print(json.dumps({"status": "FAIL", "reason": str(exc)}, ensure_ascii=True))
+        return 1
+    print(json.dumps(report, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

docs/research/source-trust/ci/test_check_convergence_log.py

@@ -0,0 +1,123 @@
+"""Offline regression tests for the frozen 212-test CI contract.
+
+This sibling directory is intentionally outside convergence/ discovery, so
+adding gate regression tests cannot silently change the measured suite size.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+CHECKER = Path(__file__).with_name("check_convergence_log.py")
+
+
+def suite_log(tests: int = 212, skipped: int = 0) -> str:
+    status = f"OK (skipped={skipped})" if skipped else "OK"
+    return f"test_example ... ok\n\n{'-' * 70}\nRan {tests} tests in 1.234s\n\n{status}\n"
+
+
+class ConvergenceLogGateTests(unittest.TestCase):
+    def run_gate(self, text: str, runner_os: str) -> subprocess.CompletedProcess[str]:
+        with tempfile.TemporaryDirectory() as temp:
+            log = Path(temp) / "tests.log"
+            log.write_text(text, encoding="utf-8", newline="")
+            return subprocess.run(
+                [sys.executable, "-I", str(CHECKER), str(log), "--runner-os", runner_os],
+                capture_output=True,
+                text=True,
+                encoding="utf-8",
+            )
+
+    def test_frozen_count_and_platform_skip_boundaries(self):
+        cases = [
+            ("Linux", 212, 0, True),
+            ("Windows", 212, 0, True),
+            ("Windows", 212, 49, True),
+            ("Windows", 212, 50, True),
+            ("Linux", 212, 1, False),
+            ("Windows", 212, 51, False),
+            ("Linux", 211, 0, False),
+            ("Linux", 213, 0, False),
+            ("Windows", 211, 50, False),
+            ("Windows", 213, 50, False),
+            ("macOS", 212, 0, False),
+        ]
+        for runner_os, tests, skipped, accepted in cases:
+            with self.subTest(runner_os=runner_os, tests=tests, skipped=skipped):
+                result = self.run_gate(suite_log(tests, skipped), runner_os)
+                self.assertEqual(
+                    result.returncode, 0 if accepted else 1,
+                    f"Gate must enforce the frozen contract: {result.stderr}",
+                )
+                report = json.loads(result.stdout)
+                self.assertEqual(report["status"], "PASS" if accepted else "FAIL")
+                if accepted:
+                    self.assertEqual(report["tests"], tests)
+                    self.assertEqual(report["skipped"], skipped)
+                    self.assertEqual(report["runner_os"], runner_os)
+                    self.assertEqual(report["expected_tests"], 212)
+                    self.assertEqual(report["max_skipped"], 0 if runner_os == "Linux" else 50)
+
+    def test_rejects_missing_ambiguous_or_nonfinal_success_summary(self):
+        valid = suite_log()
+        cases = {
+            "empty": "",
+            "truncated": valid.rsplit("OK", 1)[0],
+            "duplicate_summary": valid + valid,
+            "earlier_failed_run": valid.replace("OK", "FAILED (failures=1)") + valid,
+            "trailing_failure": valid + "FAILED (failures=1)\n",
+            "trailing_traceback": valid + "Traceback (most recent call last):\nRuntimeError: broken\n",
+            "unexpected_success": valid.replace("OK", "OK (unexpected successes=1)"),
+            "expected_failure": valid.replace("OK", "OK (expected failures=1)"),
+            "negative_skips": valid.replace("OK", "OK (skipped=-1)"),
+        }
+        for name, text in cases.items():
+            with self.subTest(case=name):
+                result = self.run_gate(text, "Linux")
+                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
+                self.assertEqual(json.loads(result.stdout)["status"], "FAIL")
+
+    def test_cli_rejects_missing_and_invalid_utf8_logs(self):
+        with tempfile.TemporaryDirectory() as temp:
+            missing = Path(temp) / "missing.log"
+            invalid = Path(temp) / "invalid.log"
+            invalid.write_bytes(b"\xff\xfe\x00")
+            for log in (missing, invalid):
+                with self.subTest(log=log.name):
+                    result = subprocess.run(
+                        [sys.executable, "-I", str(CHECKER), str(log), "--runner-os", "Linux"],
+                        capture_output=True, text=True, encoding="utf-8",
+                    )
+                    self.assertEqual(result.returncode, 1, result.stderr)
+                    self.assertEqual(json.loads(result.stdout)["status"], "FAIL")
+
+    def test_accepts_crlf_summary(self):
+        result = self.run_gate(suite_log(212, 50).replace("\n", "\r\n"), "Windows")
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertEqual(json.loads(result.stdout)["skipped"], 50)
+
+    def test_workflow_checks_actual_log_and_runs_gate_regressions(self):
+        repo = CHECKER.parents[4]
+        workflow = (repo / ".github/workflows/convergence.yml").read_text(encoding="utf-8")
+        command = (
+            'python -I docs/research/source-trust/ci/check_convergence_log.py '
+            'logs/convergence/tests.log --runner-os "${{ runner.os }}"'
+        )
+        self.assertIn(command, workflow)
+        self.assertIn('tee logs/convergence/test-counts.json', workflow)
+        self.assertIn(
+            "python -I -m unittest discover -s docs/research/source-trust/ci -p 'test_*.py' -v",
+            workflow,
+        )
+        suite_position = workflow.index("- name: Full convergence suite")
+        self.assertGreater(workflow.index(command), suite_position)
+        self.assertLess(workflow.index(command), workflow.index("- name: Original design"))
+
+
+if __name__ == "__main__":
+    unittest.main()