feat: bind v6 advisory artifacts to input digest

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

docs/advisory_contract.md

@@ -19,6 +19,43 @@ final_decisions: object, optional
 policy: object
 ```
 
+The current CLI and scheduled workflows remain explicit v5 producers. Readers
+accept both v5 and v6, but a producer must opt in to v6 only after it can emit
+all v6 provenance fields. No v6 field is silently added to, or accepted on, a
+v5 report.
+
+### V6 dual-read migration contract
+
+A v6 report uses `contract_version = model_recommendations.v6` and adds these
+required top-level fields:
+
+```text
+reference_time: exclusive start of the UTC day after as_of
+expires_at: generated_at + 7 days
+input_digest: 64 lowercase hexadecimal SHA-256 characters
+freshness.ai_signal
+freshness.theme_momentum
+```
+
+`freshness` is evaluated relative to the report's fixed `reference_time`, not
+the reader's wall clock. Declared source context and freshness state must agree;
+malformed, future, stale, or expired context fails closed. The checked-in
+contract fixture is `tests/fixtures/advisory_report_v6.json`.
+
+For a v6 producer, `input_digest` is the aggregate content identity of the
+complete logical input set. The producer computes the SHA-256 of canonical JSON
+whose keys are the sorted logical source roles and whose values are each source
+file's lowercase SHA-256, or `null` when that optional role is absent. Canonical
+JSON uses UTF-8, sorted keys, and separators `(',', ':')`. This binds the output
+to source contents instead of mutable file paths. Changing this algorithm
+requires a later schema version.
+
+Readers validate the digest shape and the full report artifact digest. A reader
+that also holds the original inputs may recompute `input_digest`; a report alone
+cannot prove that its producer hashed honest source bytes. V6 manifests repeat
+the report's exact `input_digest` so downstream lineage checks do not need to
+infer it from `source_artifacts` paths.
+
 ## Policy
 
 The policy block allows non-personalized intelligent-advisory research output but keeps account
@@ -285,6 +322,9 @@ policy
 generated_at
 ```
 
+For a v6 report, the manifest additionally carries the exact top-level
+`input_digest`. V5 manifests remain unchanged and omit this field.
+
 ## Source Mode
 
 `summary.source_mode` is:

src/quant_advisor_research/artifacts.py

@@ -120,4 +120,6 @@ def write_report_manifest(
         "policy": dict(on_disk_report.get("policy") or {}),
         "generated_at": datetime.now(timezone.utc).isoformat(),
     }
+    if on_disk_report.get("schema_version") == "6":
+        payload["input_digest"] = on_disk_report["input_digest"]
     return write_json(manifest_path, payload)

src/quant_advisor_research/contracts.py

@@ -129,6 +129,12 @@ def _require_contract_datetime(value: Any, name: str) -> dt.datetime:
         raise AdvisoryValidationError(f"{name}_invalid") from exc
 
 
+def _require_sha256_digest(value: Any, name: str) -> str:
+    if not isinstance(value, str) or re.fullmatch(r"[0-9a-f]{64}", value) is None:
+        raise AdvisoryValidationError(f"{name}_invalid")
+    return value
+
+
 def _validate_v6_freshness(
     freshness: Any,
     *,
@@ -255,13 +261,13 @@ def validate_advisory_report(payload: Mapping[str, Any]) -> None:
     if actual_contract is not None and actual_contract != expected_contract:
         raise AdvisoryValidationError("contract_version_mismatch")
     if schema_version == "5":
-        v6_only_keys = {"reference_time", "expires_at", "freshness"} & set(payload)
+        v6_only_keys = {"reference_time", "expires_at", "freshness", "input_digest"} & set(payload)
         if v6_only_keys:
             raise AdvisoryValidationError("v6_fields_on_v5")
     elif schema_version == "6":
         if actual_contract != "model_recommendations.v6":
             raise AdvisoryValidationError("v6_contract_version_required")
-        for key in ("reference_time", "expires_at", "freshness"):
+        for key in ("reference_time", "expires_at", "freshness", "input_digest"):
             if key not in payload:
                 raise AdvisoryValidationError("v6_fields_incomplete")
     else:
@@ -280,6 +286,7 @@ def validate_advisory_report(payload: Mapping[str, Any]) -> None:
             raise AdvisoryValidationError("generated_at_before_reference")
         if expires_at != report_generated_at + dt.timedelta(days=REPORT_EXPIRY_DAYS):
             raise AdvisoryValidationError("expires_at_mismatch")
+        _require_sha256_digest(payload["input_digest"], "input_digest")
         _validate_v6_freshness(
             payload["freshness"],
             report_as_of=report_as_of,

src/quant_advisor_research/publisher.py

@@ -111,7 +111,7 @@ def _is_compatibility_warning(value: Any) -> bool:
 def report_content_fingerprint(report: dict[str, Any]) -> str:
     ignored_top_level_keys = {
         "as_of", "generated_at", "reference_time", "expires_at", "schema_version",
-        "contract_version", "source_artifacts",
+        "contract_version", "source_artifacts", "input_digest",
     }
     normalized = {key: value for key, value in report.items() if key not in ignored_top_level_keys}
     freshness = normalized.get("freshness")

tests/fixtures/advisory_report_v6.json

@@ -0,0 +1,41 @@
+{
+  "as_of": "2026-08-22",
+  "audience_scope": "non_personalized_model_research",
+  "cadence": "daily",
+  "contract_version": "model_recommendations.v6",
+  "expires_at": "2026-08-30T00:00:00Z",
+  "freshness": {
+    "ai_signal": {
+      "present": false,
+      "reason": "not_provided",
+      "valid": false
+    },
+    "theme_momentum": {
+      "present": false,
+      "reason": "not_provided",
+      "valid": false
+    }
+  },
+  "generated_at": "2026-08-23T00:00:00Z",
+  "input_digest": "f9b98eb8d22f5ee5c403480c797757240afd76e117f4f7edf157dc330a87d014",
+  "mode": "model_recommendations",
+  "policy": {
+    "account_specific_advice_allowed": false,
+    "downstream_use": "Asset idea research observation only; do not route to portfolio allocation or broker execution.",
+    "execution_allowed": false,
+    "non_personalized_recommendations_allowed": true,
+    "personalized_advice_allowed": false,
+    "portfolio_allocation_allowed": false
+  },
+  "recommendations": [],
+  "reference_time": "2026-08-23T00:00:00Z",
+  "schema_version": "6",
+  "source_artifacts": {
+    "ai_signal": "",
+    "market_confirmation": "",
+    "political_events": "",
+    "political_watchlist": "",
+    "theme_momentum": ""
+  },
+  "summary": {}
+}

tests/test_artifact_integrity.py

@@ -25,6 +25,7 @@
 
 
 ROOT = Path(__file__).resolve().parents[1]
+V6_INPUT_DIGEST = "a" * 64
 
 
 def build_report(*, schema_version: str = "5") -> dict:
@@ -43,6 +44,7 @@ def build_report(*, schema_version: str = "5") -> dict:
                 "contract_version": "model_recommendations.v6",
                 "reference_time": reference_time.isoformat().replace("+00:00", "Z"),
                 "expires_at": (generated_at + dt.timedelta(days=7)).isoformat().replace("+00:00", "Z"),
+                "input_digest": V6_INPUT_DIGEST,
                 "freshness": {
                     "ai_signal": {"present": False, "valid": False, "reason": "not_provided"},
                     "theme_momentum": {"present": False, "valid": False, "reason": "not_provided"},
@@ -96,6 +98,14 @@ def test_v6_time_metadata_changes_artifact_digest_but_not_semantic_digest(field:
     assert artifact_integrity_digest(changed) != artifact_integrity_digest(report)
 
 
+def test_v6_input_digest_changes_artifact_digest_but_not_semantic_digest() -> None:
+    report = build_report(schema_version="6")
+    changed = dict(report, input_digest="b" * 64)
+
+    assert report_content_fingerprint(changed) == report_content_fingerprint(report)
+    assert artifact_integrity_digest(changed) != artifact_integrity_digest(report)
+
+
 def test_schema_version_change_is_reflected_in_artifact_digest() -> None:
     v5 = build_report()
     v6 = build_report(schema_version="6")

tests/test_publisher_period_redesign.py

@@ -21,6 +21,7 @@
 
 
 ROOT = Path(__file__).resolve().parents[1]
+V6_INPUT_DIGEST = "a" * 64
 
 
 def build_v5(as_of: str = "2026-06-20") -> dict:
@@ -41,6 +42,7 @@ def build_v6(as_of: str = "2026-06-20") -> dict:
             "reference_time": canonical_reference_time(dt.date.fromisoformat(as_of)).isoformat().replace("+00:00", "Z"),
             "generated_at": "2026-06-21T12:00:00.123456Z",
             "expires_at": "2026-06-28T12:00:00.123456Z",
+            "input_digest": V6_INPUT_DIGEST,
             "freshness": {
                 "ai_signal": {"present": False, "valid": False, "reason": "not_provided"},
                 "theme_momentum": {"present": False, "valid": False, "reason": "not_provided"},

tests/test_publisher_redesign_stage2a.py

@@ -15,6 +15,7 @@
 
 
 ROOT = Path(__file__).resolve().parents[1]
+V6_INPUT_DIGEST = "a" * 64
 
 
 def build_v5() -> dict:
@@ -35,6 +36,7 @@ def build_v6() -> dict:
             "reference_time": canonical_reference_time(dt.date(2026, 5, 30)).isoformat().replace("+00:00", "Z"),
             "generated_at": "2026-05-31T12:00:00.123456Z",
             "expires_at": "2026-06-07T12:00:00.123456Z",
+            "input_digest": V6_INPUT_DIGEST,
             "freshness": {
                 "ai_signal": {"present": False, "valid": False, "reason": "not_provided"},
                 "theme_momentum": {"present": False, "valid": False, "reason": "not_provided"},
@@ -72,6 +74,37 @@ def test_v5_legacy_missing_marker_remains_readable() -> None:
     validate_advisory_report(report)
 
 
+@pytest.mark.parametrize("input_digest", [None, "", "a" * 63, "A" * 64, "sha256:" + "a" * 64, True])
+def test_v6_input_digest_is_required_lowercase_sha256(input_digest: object) -> None:
+    report = build_v6()
+    if input_digest is None:
+        report.pop("input_digest")
+        expected = "v6_fields_incomplete"
+    else:
+        report["input_digest"] = input_digest
+        expected = "input_digest_invalid"
+    with pytest.raises(AdvisoryValidationError, match=expected):
+        validate_advisory_report(report)
+
+
+def test_v5_does_not_silently_accept_v6_input_digest() -> None:
+    report = build_v5()
+    report["input_digest"] = V6_INPUT_DIGEST
+    with pytest.raises(AdvisoryValidationError, match="v6_fields_on_v5"):
+        validate_advisory_report(report)
+
+
+def test_checked_in_v6_contract_fixture_is_readable() -> None:
+    fixture = json.loads((ROOT / "tests/fixtures/advisory_report_v6.json").read_text(encoding="utf-8"))
+    empty_input_set = {
+        name: None
+        for name in ("political_events", "political_watchlist", "ai_signal", "theme_momentum", "market_confirmation")
+    }
+    canonical = json.dumps(empty_input_set, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    assert fixture["input_digest"] == hashlib.sha256(canonical).hexdigest()
+    validate_advisory_report(fixture)
+
+
 @pytest.mark.parametrize(
     ("field", "value"),
     [
@@ -280,7 +313,12 @@ def test_manifest_version_is_report_derived_without_global_v6_contamination(tmp_
             markdown_path=markdown_path,
             manifest_path=tmp_path / f"{name}.manifest.json",
         )
-        assert json.loads(manifest.read_text(encoding="utf-8"))["contract_version"] == expected
+        payload = json.loads(manifest.read_text(encoding="utf-8"))
+        assert payload["contract_version"] == expected
+        if name == "v6":
+            assert payload["input_digest"] == V6_INPUT_DIGEST
+        else:
+            assert "input_digest" not in payload
 
 
 def test_manifest_rejects_incomplete_v6_but_keeps_v5_legacy_omission(tmp_path: Path) -> None: