Fail closed on untrusted AI signal provenance

src/quant_advisor_research/advisory_report.py

@@ -2,8 +2,11 @@
 
 import argparse
 import datetime as dt
+import hashlib
 import json
 import os
+import re
+import subprocess
 from collections import defaultdict
 from collections.abc import Mapping
 from dataclasses import dataclass
@@ -96,6 +99,13 @@
 AI_POLICY_ALLOWED_KEYS = frozenset({"execution_allowed", "portfolio_allocation_allowed", "downstream_use"})
 AI_POLICY_FORBIDDEN_TERMS = frozenset({"live", "allocation", "broker", "execution", "order", "position", "account"})
 AI_POLICY_BLOCKING_TERMS = frozenset({"blocked", "not allowed", "do not", "never", "no "})
+AI_SIGNAL_MANIFEST_NAME = "latest_signal.manifest.json"
+AI_SIGNAL_MANIFEST_PATH = "data/output/latest_signal.manifest.json"
+AI_SIGNAL_PATH = "data/output/latest_signal.json"
+AI_SIGNAL_PRODUCER_REPOSITORY = "QuantStrategyLab/ResearchSignalContextPipelines"
+AI_SIGNAL_PROVENANCE_WARNING = "ai_signal_provenance_untrusted"
+SHA256_PATTERN = re.compile(r"[0-9a-f]{64}")
+GIT_SHA_PATTERN = re.compile(r"[0-9a-f]{40}")
 _UNAVAILABLE_INPUT = object()
 
 HORIZON_WINDOWS = {
@@ -501,6 +511,114 @@ def load_ai_signal(path: str | Path | None, *, source_bytes: bytes | None = None
     return payload
 
 
+def _ai_provenance_untrusted() -> None:
+    raise AISignalValidationError(AI_SIGNAL_PROVENANCE_WARNING)
+
+
+def _git_output(repo: Path, *args: str) -> bytes:
+    result = subprocess.run(
+        ["git", "-C", str(repo), *args],
+        check=False,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.DEVNULL,
+    )
+    if result.returncode != 0:
+        _ai_provenance_untrusted()
+    return result.stdout
+
+
+def _repository_from_remote(remote: str) -> str:
+    value = remote.strip().removesuffix("/").removesuffix(".git")
+    for prefix in ("https://github.com/", "git@github.com:", "ssh://git@github.com/"):
+        if value.startswith(prefix):
+            return value.removeprefix(prefix)
+    return ""
+
+
+def load_trusted_ai_signal(
+    path: str | Path | None,
+    *,
+    source_bytes: bytes | None = None,
+) -> dict[str, Any] | None:
+    if path is None:
+        return None
+    signal_path = Path(path).resolve()
+    signal_bytes = source_bytes
+    if signal_bytes is None:
+        try:
+            signal_bytes = signal_path.read_bytes()
+        except OSError:
+            raise AISignalValidationError("ai_signal_unavailable") from None
+    payload = load_ai_signal(signal_path, source_bytes=signal_bytes)
+    if payload is None:
+        return None
+    manifest_path = signal_path.with_name(AI_SIGNAL_MANIFEST_NAME)
+    try:
+        manifest_bytes = manifest_path.read_bytes()
+        manifest = json.loads(manifest_bytes.decode("utf-8"))
+    except (OSError, UnicodeError, json.JSONDecodeError):
+        _ai_provenance_untrusted()
+    if not isinstance(manifest, Mapping):
+        _ai_provenance_untrusted()
+    required_keys = {
+        "manifest_type",
+        "schema_version",
+        "artifact",
+        "as_of",
+        "generated_at",
+        "expires_at",
+        "mode",
+        "producer",
+        "input_digest",
+        "policy",
+    }
+    artifact = manifest.get("artifact")
+    producer = manifest.get("producer")
+    policy = manifest.get("policy")
+    if (
+        set(manifest) != required_keys
+        or manifest.get("manifest_type") != "research_signal_context"
+        or manifest.get("schema_version") != 2
+        or not isinstance(artifact, Mapping)
+        or set(artifact) != {"path", "sha256"}
+        or artifact.get("path") != AI_SIGNAL_PATH
+        or not isinstance(artifact.get("sha256"), str)
+        or SHA256_PATTERN.fullmatch(artifact["sha256"]) is None
+        or artifact["sha256"] != hashlib.sha256(signal_bytes).hexdigest()
+        or not isinstance(producer, Mapping)
+        or set(producer) != {"repository", "commit_sha"}
+        or producer.get("repository") != AI_SIGNAL_PRODUCER_REPOSITORY
+        or not isinstance(producer.get("commit_sha"), str)
+        or GIT_SHA_PATTERN.fullmatch(producer["commit_sha"]) is None
+        or not isinstance(manifest.get("input_digest"), str)
+        or re.fullmatch(r"sha256:[0-9a-f]{64}", manifest["input_digest"]) is None
+        or not isinstance(policy, Mapping)
+        or set(policy) != {"execution_allowed"}
+        or policy.get("execution_allowed") is not False
+        or any(manifest.get(key) != payload.get(key) for key in ("as_of", "generated_at", "expires_at", "mode"))
+    ):
+        _ai_provenance_untrusted()
+
+    try:
+        repo = Path(_git_output(signal_path.parent, "rev-parse", "--show-toplevel").decode("utf-8").strip()).resolve()
+        signal_relative = signal_path.relative_to(repo).as_posix()
+        manifest_relative = manifest_path.resolve().relative_to(repo).as_posix()
+        head = _git_output(repo, "rev-parse", "HEAD").decode("ascii").strip()
+        remote = _git_output(repo, "remote", "get-url", "origin").decode("utf-8").strip()
+    except (UnicodeError, ValueError):
+        _ai_provenance_untrusted()
+    if (
+        signal_relative != AI_SIGNAL_PATH
+        or manifest_relative != AI_SIGNAL_MANIFEST_PATH
+        or GIT_SHA_PATTERN.fullmatch(head) is None
+        or _repository_from_remote(remote) != AI_SIGNAL_PRODUCER_REPOSITORY
+        or _git_output(repo, "show", f"{head}:{signal_relative}") != signal_bytes
+        or _git_output(repo, "show", f"{head}:{manifest_relative}") != manifest_bytes
+    ):
+        _ai_provenance_untrusted()
+    return payload
+
+
 
 def load_theme_momentum(path: str | Path | None, *, source_bytes: bytes | None = None) -> dict[str, Any] | None:
     if path is None:
@@ -1721,7 +1839,7 @@ def build_advisory_report(
             ai_quality_warnings.append("ai_signal_unavailable")
         else:
             try:
-                candidate_ai_signal = load_ai_signal(
+                candidate_ai_signal = load_trusted_ai_signal(
                     ai_signal_path,
                     source_bytes=ai_bytes if isinstance(ai_bytes, bytes) else None,
                 )

src/quant_advisor_research/build_pipeline.py

@@ -11,9 +11,17 @@
 from urllib.parse import quote
 from urllib.request import urlopen
 
-from .advisory_report import build_advisory_report, render_markdown, write_json, write_text
+from .advisory_report import (
+    AISignalValidationError,
+    build_advisory_report,
+    load_trusted_ai_signal,
+    render_markdown,
+    write_json,
+    write_text,
+)
 from .artifacts import write_report_manifest
 from .market_confirmation import (
+    EXCLUDED_SYMBOLS,
     build_market_confirmation_rows,
     collect_symbols,
     load_proxy_urls,
@@ -119,12 +127,29 @@ def build_market_confirmation_artifact(
     cache_max_age_days: int,
 ) -> Path:
     theme_momentum = load_theme_momentum(theme_momentum_path) if theme_momentum_path else None
-    symbols = collect_symbols(
-        political_watchlist_path=political_watchlist_path,
-        ai_signal_path=ai_signal_path,
-        theme_momentum=theme_momentum,
-        max_symbols=max_symbols,
+    ai_signal = None
+    if ai_signal_path:
+        try:
+            ai_signal = load_trusted_ai_signal(ai_signal_path)
+        except AISignalValidationError:
+            pass
+    symbols = set(
+        collect_symbols(
+            political_watchlist_path=political_watchlist_path,
+            ai_signal_path=None,
+            theme_momentum=theme_momentum,
+            max_symbols=2**31 - 1,
+        )
     )
+    if ai_signal:
+        for symbol in ai_signal.get("universe", []):
+            if isinstance(symbol, str) and symbol.strip():
+                symbols.add(symbol.upper())
+        for key in ("candidate_bias", "research_bias", "symbol_bias", "symbol_theme_exposure"):
+            value = ai_signal.get(key)
+            if isinstance(value, dict):
+                symbols.update(str(symbol).upper() for symbol in value if str(symbol).strip())
+    symbols = sorted(symbols - EXCLUDED_SYMBOLS)[:max_symbols]
     proxy_urls = load_proxy_urls(
         proxy_list_path=proxy_list,
         proxy_urls_text=proxy_urls_text,

tests/test_advisory_report.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import datetime as dt
+import hashlib
 import json
+import subprocess
 from pathlib import Path
 
 import pytest
@@ -25,6 +27,55 @@
 ROOT = Path(__file__).resolve().parents[1]
 
 
+def write_trusted_example_ai_signal(tmp_path: Path) -> Path:
+    repo = tmp_path / "ResearchSignalContextPipelines"
+    signal = repo / "data/output/latest_signal.json"
+    signal.parent.mkdir(parents=True)
+    signal.write_bytes((ROOT / "examples/research_signal_context.example.json").read_bytes())
+    payload = json.loads(signal.read_text(encoding="utf-8"))
+    manifest = {
+        "manifest_type": "research_signal_context",
+        "schema_version": 2,
+        "artifact": {
+            "path": "data/output/latest_signal.json",
+            "sha256": hashlib.sha256(signal.read_bytes()).hexdigest(),
+        },
+        "as_of": payload["as_of"],
+        "generated_at": payload["generated_at"],
+        "expires_at": payload["expires_at"],
+        "mode": payload["mode"],
+        "producer": {
+            "repository": "QuantStrategyLab/ResearchSignalContextPipelines",
+            "commit_sha": "a" * 40,
+        },
+        "input_digest": f"sha256:{'b' * 64}",
+        "policy": {"execution_allowed": False},
+    }
+    signal.with_name("latest_signal.manifest.json").write_text(json.dumps(manifest) + "\n", encoding="utf-8")
+    subprocess.run(["git", "init", "-q", repo], check=True)
+    subprocess.run(
+        ["git", "-C", repo, "remote", "add", "origin", "https://github.com/QuantStrategyLab/ResearchSignalContextPipelines.git"],
+        check=True,
+    )
+    subprocess.run(["git", "-C", repo, "add", "data/output"], check=True)
+    subprocess.run(
+        [
+            "git",
+            "-C",
+            repo,
+            "-c",
+            "user.name=Test",
+            "-c",
+            "user.email=test@example.invalid",
+            "commit",
+            "-qm",
+            "trusted fixture",
+        ],
+        check=True,
+    )
+    return signal
+
+
 def test_build_advisory_report_blocks_execution_and_allocation() -> None:
     report = build_advisory_report(
         as_of="2026-05-30",
@@ -61,13 +112,13 @@ def test_low_confidence_events_remain_verify_source_until_verified() -> None:
     assert by_symbol["EVT1"]["evidence_score"] > by_symbol["EVT4"]["evidence_score"]
 
 
-def test_ai_avoid_bias_defers_research_item() -> None:
+def test_ai_avoid_bias_defers_research_item(tmp_path: Path) -> None:
     report = build_advisory_report(
         as_of="2026-05-30",
         cadence="monthly",
         political_events_path=ROOT / "examples/political_events.example.csv",
         political_watchlist_path=ROOT / "examples/political_watchlist.example.csv",
-        ai_signal_path=ROOT / "examples/research_signal_context.example.json",
+        ai_signal_path=write_trusted_example_ai_signal(tmp_path),
     )
 
     by_symbol = {item["symbol"]: item for item in report["recommendations"]}
@@ -138,13 +189,13 @@ def test_mixed_confidence_recommendation_is_not_tier_one(tmp_path: Path) -> None
     assert rec["recommendation_tier"] == "tier_2"
 
 
-def test_long_horizon_window_is_measured_in_years() -> None:
+def test_long_horizon_window_is_measured_in_years(tmp_path: Path) -> None:
     report = build_advisory_report(
         as_of="2026-05-30",
         cadence="weekly",
         political_events_path=ROOT / "examples/political_events.example.csv",
         political_watchlist_path=ROOT / "examples/political_watchlist.example.csv",
-        ai_signal_path=ROOT / "examples/research_signal_context.example.json",
+        ai_signal_path=write_trusted_example_ai_signal(tmp_path),
     )
 
     by_symbol = {item["symbol"]: item for item in report["recommendations"]}
@@ -280,7 +331,7 @@ def test_report_manifest_records_contract_version_and_hashes(tmp_path: Path) ->
     assert manifest["artifacts"]["markdown"]["sha256"]
 
 
-def test_theme_bias_can_lift_static_watchlist_item_without_direct_symbol_bias(tmp_path: Path) -> None:
+def test_legacy_v1_theme_bias_is_untrusted_no_op(tmp_path: Path) -> None:
     events_path = tmp_path / "events.csv"
     events_path.write_text(
         "event_id,event_date,symbol,event_type,direction,confidence,source_url,notes\n",
@@ -339,11 +390,12 @@ def test_theme_bias_can_lift_static_watchlist_item_without_direct_symbol_bias(tm
 
     rec = report["recommendations"][0]
     assert rec["symbol"] == "MU"
-    assert rec["rating"] == "watch"
-    assert rec["evidence_score"] > 4
-    assert any("主题=hbm_memory" in reason for reason in rec["reasons"])
+    assert rec["rating"] == "monitor"
+    assert rec["evidence_score"] == 4
+    assert rec["ai_context"]["source"] == ""
+    assert report["summary"]["data_quality_warnings"] == ["ai_signal_provenance_untrusted"]
     assert report["summary"]["long_context_available"] is False
-    assert report["summary"]["long_context_missing_reason"] == "current_candidates_do_not_meet_long_context_gate"
+    assert report["summary"]["long_context_missing_reason"] == "ai_signal_not_available"
     assert "MU" not in report["summary"]["long_context_symbols"]
     assert report["final_decisions"]["horizon_action_buckets"]["long"]["watch"] == []