[codex] harden dependency pin guard

scripts/check_qpk_pin_consistency.py

@@ -1,44 +1,100 @@
 #!/usr/bin/env python3
-"""Check that all QPK git references match the canonical QPK_PIN.
+"""Check QuantStrategyLab direct git dependency pins.
+
+Guards two failure modes:
+- QuantPlatformKit refs must match the canonical QPK_PIN on main.
+- Any QuantStrategyLab git dependency appearing in multiple dependency files
+  must use one consistent ref across requirements/constraints/pyproject.
+
 Usage: python scripts/check_qpk_pin_consistency.py [--fix]
 """
-import re, sys
+from __future__ import annotations
+
+import re
+import subprocess
+import sys
+import urllib.request
 from pathlib import Path
 
 QPK_PIN_URL = "https://raw.githubusercontent.com/QuantStrategyLab/QuantPlatformKit/main/QPK_PIN"
 QPK_REF_RE = re.compile(r"QuantPlatformKit\.git@([a-f0-9]{40})")
+QSL_REF_RE = re.compile(r"github\.com/QuantStrategyLab/([A-Za-z0-9_.-]+)\.git@([A-Za-z0-9._/-]+)")
+PINNED_FILE_GLOBS = ("**/requirements*.txt", "**/constraints*.txt", "**/pyproject.toml")
 
-def fetch_pin() -> str:
-    import urllib.request
-    with urllib.request.urlopen(QPK_PIN_URL, timeout=10) as r:
-        sha = r.read().decode().strip().split()[0]
-    if len(sha) == 40: return sha
+
+def _extract_pin(raw: str) -> str:
+    sha = raw.strip().split()[0]
+    if len(sha) == 40:
+        return sha
     raise RuntimeError(f"Invalid QPK_PIN: {sha}")
 
-def main():
+
+def fetch_pin() -> str:
+    errors: list[str] = []
+    try:
+        with urllib.request.urlopen(QPK_PIN_URL, timeout=10) as response:
+            return _extract_pin(response.read().decode())
+    except Exception as exc:  # pragma: no cover - fallback path depends on host certs
+        errors.append(f"urllib: {exc}")
+
+    try:
+        output = subprocess.check_output(["curl", "-fsSL", QPK_PIN_URL], text=True, timeout=10)
+        return _extract_pin(output)
+    except Exception as exc:
+        errors.append(f"curl: {exc}")
+
+    raise RuntimeError("Unable to fetch QPK_PIN: " + "; ".join(errors))
+
+
+def iter_pinned_files() -> list[Path]:
+    paths: dict[str, Path] = {}
+    for pattern in PINNED_FILE_GLOBS:
+        for path in Path.cwd().glob(pattern):
+            if "external" in path.parts:
+                continue
+            paths[str(path)] = path
+    return [paths[name] for name in sorted(paths)]
+
+
+def main() -> int:
     fix = "--fix" in sys.argv
     target = fetch_pin()
     print(f"QPK_PIN: {target[:12]}...")
     errors = 0
-    for path in sorted(Path.cwd().glob("**/requirements*.txt")) + sorted(Path.cwd().glob("**/constraints*.txt")) + sorted(Path.cwd().glob("**/pyproject.toml")):
-        if "external" in str(path): continue
-        content = path.read_text()
+    qsl_refs: dict[str, list[tuple[Path, int, str]]] = {}
+
+    for path in iter_pinned_files():
+        content = path.read_text(encoding="utf-8")
         updated = content
-        for m in QPK_REF_RE.finditer(content):
-            sha = m.group(1)
+        for line_no, line in enumerate(content.splitlines(), start=1):
+            for repo, ref in QSL_REF_RE.findall(line):
+                qsl_refs.setdefault(repo, []).append((path, line_no, ref))
+        for match in QPK_REF_RE.finditer(content):
+            sha = match.group(1)
             if sha != target:
                 errors += 1
                 print(f"  ❌ {path}: QPK@{sha[:12]} (expected {target[:12]})")
                 if fix:
                     updated = updated.replace(f"QuantPlatformKit.git@{sha}", f"QuantPlatformKit.git@{target}")
                     print("     → fixed")
         if fix and updated != content:
-            path.write_text(updated)
+            path.write_text(updated, encoding="utf-8")
+
+    for repo, matches in sorted(qsl_refs.items()):
+        refs = sorted({ref for _, _, ref in matches})
+        if len(refs) <= 1:
+            continue
+        errors += 1
+        print(f"  ❌ inconsistent QuantStrategyLab dependency pin for {repo}:")
+        for path, line_no, ref in matches:
+            print(f"     {path}:{line_no}: {ref}")
+
     if errors:
-        print(f"\n{errors} mismatch(es). Run with --fix to auto-fix.")
+        print(f"\n{errors} dependency pin mismatch(es). Run with --fix to auto-fix QPK refs only.")
         return 1
-    print("✅ All QPK pins match.")
+    print("✅ QuantStrategyLab dependency pins are consistent.")
     return 0
 
+
 if __name__ == "__main__":
     raise SystemExit(main())

tests/test_dependency_pin_guard.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+from types import ModuleType
+
+
+SCRIPT = Path("scripts/check_qpk_pin_consistency.py")
+CI_WORKFLOW = Path(".github/workflows/ci.yml")
+
+
+def _load_guard_module() -> ModuleType:
+    spec = importlib.util.spec_from_file_location("check_qpk_pin_consistency_for_test", SCRIPT)
+    assert spec is not None
+    assert spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_dependency_pin_guard_checks_constraints_and_all_qsl_git_refs() -> None:
+    script = SCRIPT.read_text(encoding="utf-8")
+
+    assert '"**/constraints*.txt"' in script
+    assert "QSL_REF_RE" in script
+    assert "inconsistent QuantStrategyLab dependency pin" in script
+
+
+def test_dependency_pin_guard_rejects_internal_qsl_git_ref_drift(tmp_path, monkeypatch, capsys) -> None:
+    module = _load_guard_module()
+    qpk_ref = "0" * 40
+    old_strategy_ref = "1" * 40
+    new_strategy_ref = "2" * 40
+    (tmp_path / "requirements.txt").write_text(
+        "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@" + qpk_ref + "\n"
+        "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@"
+        + old_strategy_ref
+        + "\n",
+        encoding="utf-8",
+    )
+    (tmp_path / "constraints.txt").write_text(
+        "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@"
+        + new_strategy_ref
+        + "\n",
+        encoding="utf-8",
+    )
+
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr(module, "fetch_pin", lambda: qpk_ref)
+    monkeypatch.setattr(sys, "argv", ["check_qpk_pin_consistency.py"])
+
+    assert module.main() == 1
+    output = capsys.readouterr().out
+    assert "inconsistent QuantStrategyLab dependency pin for UsEquityStrategies" in output
+
+
+def test_dependency_pin_guard_is_blocking_in_ci() -> None:
+    workflow = CI_WORKFLOW.read_text(encoding="utf-8")
+    step_start = workflow.index("name: Check QPK pin consistency")
+    next_step = workflow.find("\n      - name:", step_start + 1)
+    step = workflow[step_start : next_step if next_step != -1 else len(workflow)]
+
+    assert "python scripts/check_qpk_pin_consistency.py" in step
+    assert "continue-on-error" not in step