fix(reports): bind editorial facts to frozen sources

docs/product/periodic-reports.md

@@ -871,6 +871,19 @@ reader calls, 4,000 evidence rows and 4 MiB of evidence data, plus bounded sourc
 coverage metadata; exhaustion is partial coverage,
 not success or zero progress. These are IO bounds, not a model budget.
 
+New private and managed-Session editorial requests carry the local artifact
+format marker `source_fact_binding=frozen_source_v1`, also retained in the
+frozen source so stripping it from the request cannot select legacy behavior.
+Cache reuse, generation
+and independent generation validation derive the complete ordered fact list
+again from the unchanged frozen source, including its revision. A recomputed
+request digest cannot substitute an unsupported URL, summary or incomplete fact
+set. The generation receipt retains the same marker. Historical windows without
+it keep their original validation contract and bytes; they are not retroactively
+qualified or regenerated. This is source-to-fact consistency, not proof of a
+claim, public disclosure permission or URL reachability. Opaque references stay
+opaque when no permitted accessible source has been supplied.
+
 This source can also be read through the original Chat dynamic tool. Reading
 alone does not activate a subscription or deliver a report. The opt-in original
 steward report path binds it to the admitted calendar window, native generation

loopx/capabilities/periodic_report/private_generation.py

@@ -31,6 +31,43 @@ def _read(path: Path) -> dict[str, Any]:
     return value
 
 
+def _source_facts(source: dict[str, Any], *, local: bool) -> list[dict[str, Any]]:
+    """Derive editorial references from the same frozen source on every read."""
+    facts = [{"fact_id": "local_coverage" if local else "portfolio_coverage",
+        "title": "本期来源范围" if local else "本期来源覆盖",
+        "summary": ("当前原 Goal 的完整原生任务读取；任务状态在收集时观察，按完成时间区分本期成果和历史记录。来源声明及任务完成不证明外部业务已独立验收。"
+            if local else source["source_result"]["sections"][0]["items"][0]["summary"]),
+        "status": "read" if local else "partial" if source["gaps"] else "read",
+        "content_kind": "coverage", "source_ref": source["source_revision"]}]
+    for index, row in enumerate(source["rows"]):
+        encoded = json.dumps(row if local else row["evidence"], ensure_ascii=False, sort_keys=True)
+        facts.append({"fact_id": f"local_{index}" if local else f"portfolio_{index}",
+            "title": (str(row.get("text") or row["todo_id"]) if local else
+                f"{row.get('goal_id') or 'portfolio'} / {row['kind']}")[:240],
+            "summary": encoded if len(encoded) <= 1000 else encoded[:940] + (
+                "…（完整记录见冻结来源同一 todo_id）" if local else "…（摘录；完整记录见冻结来源的同一 source_ref）"),
+            "status": "recorded" if local else "unknown" if row["kind"] == "coverage_gap" else "recorded",
+            "source_ref": "todo:" + row["todo_id"] if local else row["source_ref"],
+            "content_kind": "coverage"})
+    return facts
+
+
+def _verify_source_facts(request: dict[str, Any], source: dict[str, Any]) -> None:
+    # Old windows retain their original input and validation contract. The
+    # marker belongs to this frozen IO format, not a new source/access policy.
+    if "source_fact_binding" not in request and "source_fact_binding" not in source:
+        return
+    if (request.get("source_fact_binding") != "frozen_source_v1"
+            or source.get("source_fact_binding") != "frozen_source_v1"
+            or request.get("destination") not in {
+                "original_managed_session_full_body", "original_owner_private_chat_full_body"}
+            or request.get("source_artifact_sha256") != _canonical_digest(source)
+            or request.get("source_revision") != source.get("source_revision")
+            or request.get("facts") != _source_facts(source,
+                local=request["destination"] == "original_managed_session_full_body")):
+        raise ValueError("editorial facts do not match the frozen report source")
+
+
 def prepare_private_report(*, root: Path, window: dict[str, Any], read,
                            authorization_scope_id: str, authorization_goal_ids: list[str],
                            reporting: str) -> dict[str, Any]:
@@ -48,6 +85,7 @@ def prepare_private_report(*, root: Path, window: dict[str, Any], read,
                     or request.get("request_digest") != _canonical_digest(
                         {k: v for k, v in request.items() if k != "request_digest"})):
                 raise ValueError("the frozen report belongs to another scope or window")
+            _verify_source_facts(request, _read(folder / "source.json"))
             return request
         source = collect_manager_report_source(read, {
             "window_start": window["start_at"], "window_end": window["due_at"]})
@@ -56,24 +94,16 @@ def prepare_private_report(*, root: Path, window: dict[str, Any], read,
                 or not set(source["goal_ids"]).issubset(authorization_goal_ids)):
             raise ValueError("report evidence is outside its frozen authorization")
         folder.mkdir(parents=True, exist_ok=True)
+        source = {**source, "source_fact_binding": "frozen_source_v1"}
         atomic_write_json(folder / "source.json", source)
-        facts = [{"fact_id": "portfolio_coverage", "title": "本期来源覆盖",
-            "summary": source["source_result"]["sections"][0]["items"][0]["summary"],
-            "status": "partial" if source["gaps"] else "read", "content_kind": "coverage",
-            "source_ref": source["source_revision"]}]
-        for index, row in enumerate(source["rows"]):
-            encoded = json.dumps(row["evidence"], ensure_ascii=False, sort_keys=True)
-            facts.append({"fact_id": f"portfolio_{index}", "title": f"{row.get('goal_id') or 'portfolio'} / {row['kind']}"[:240],
-                "summary": encoded if len(encoded) <= 1000 else encoded[:940] + "…（摘录；完整记录见冻结来源的同一 source_ref）",
-                "status": "unknown" if row["kind"] == "coverage_gap" else "recorded",
-                "source_ref": row["source_ref"], "content_kind": "coverage"})
         request = _build_editorial_request(intent=cadence_intent(window), runtime_root=root,
             goal_id=window["goal_id"], agent_id=window["agent_id"],
-            completed_at=window["due_at"], facts=facts)
+            completed_at=window["due_at"], facts=_source_facts(source, local=False))
         request.pop("request_digest")
         request.update(authorization_scope_id=authorization_scope_id, reporting=reporting,
             source_artifact=str(folder / "source.json"), source_revision=source["source_revision"],
             source_artifact_sha256=_canonical_digest(source), window=window,
+            source_fact_binding="frozen_source_v1",
             coverage=source["coverage"], destination="original_owner_private_chat_full_body")
         # A partial portfolio read is not the authorization universe. Keep
         # this original identity witness inside the existing frozen digest.
@@ -140,25 +170,21 @@ def prepare_session_report(*, root: Path, window: dict[str, Any], actor: dict[st
                     or frozen.get("request_digest") != _canonical_digest(
                         {k: v for k, v in frozen.items() if k != "request_digest"})):
                 raise ValueError("the frozen report belongs to another Session or window")
+            _verify_source_facts(frozen, _read(folder / "source.json"))
             return frozen
         source = read()
         rows = source["rows"]
         if len(rows) > 4000 or len(json.dumps(source, ensure_ascii=False).encode()) > 4 * 1024 * 1024:
             raise ValueError("the complete local report source exceeds its existing evidence bound")
-        facts = [{"fact_id": "local_coverage", "title": "本期来源范围",
-            "summary": "当前原 Goal 的完整原生任务读取；任务状态在收集时观察，按完成时间区分本期成果和历史记录。来源声明及任务完成不证明外部业务已独立验收。",
-            "status": "read", "content_kind": "coverage", "source_ref": source["source_revision"]}]
-        for index, row in enumerate(rows):
-            encoded = json.dumps(row, ensure_ascii=False, sort_keys=True)
-            facts.append({"fact_id": f"local_{index}", "title": str(row.get("text") or row["todo_id"])[:240],
-                "summary": encoded if len(encoded) <= 1000 else encoded[:940] + "…（完整记录见冻结来源同一 todo_id）",
-                "status": "recorded", "content_kind": "coverage", "source_ref": "todo:" + row["todo_id"]})
+        source = {**source, "source_fact_binding": "frozen_source_v1"}
         request = _build_editorial_request(intent=cadence_intent(window), runtime_root=root,
-            goal_id=window["goal_id"], agent_id=window["agent_id"], completed_at=window["due_at"], facts=facts)
+            goal_id=window["goal_id"], agent_id=window["agent_id"], completed_at=window["due_at"],
+            facts=_source_facts(source, local=True))
         request.pop("request_digest")
         request.update(managed_session_actor=actor, reporting=reporting, window=window,
             source_artifact=str(folder / "source.json"), source_revision=source["source_revision"],
-            source_artifact_sha256=_canonical_digest(source), destination="original_managed_session_full_body")
+            source_artifact_sha256=_canonical_digest(source), destination="original_managed_session_full_body",
+            source_fact_binding="frozen_source_v1")
         request["narrative_contract"]["requirements"] += [
             "Use only this original Goal's frozen evidence and permitted artifacts; no portfolio or other Goal access is granted.",
             "Current Task status is observed at collection time, not reconstructed as of the period end.",
@@ -187,6 +213,7 @@ def freeze_private_report(*, root: Path, window: dict[str, Any], response: dict[
     folder = attempt_directory(root, window)
     with exclusive_file_lock(folder / "generation.lock", operation="freeze_private_report"):
         request = _read(folder / "editorial-request.json")
+        _verify_source_facts(request, _read(folder / "source.json"))
         response_path = folder / "editorial-response.json"
         if response_path.exists() and _read(response_path) != response:
             raise ValueError("a report window already has a different authored response")
@@ -220,6 +247,8 @@ def freeze_private_report(*, root: Path, window: dict[str, Any], response: dict[
             "bundle_sha256": _canonical_digest(bundle), "publication_candidate_id": candidate["candidate_id"],
             "publication_candidate_sha256": _canonical_digest(candidate),
             "generation_complete": True, "delivery_verified": False}
+        if "source_fact_binding" in request:
+            receipt["source_fact_binding"] = request["source_fact_binding"]
         atomic_write_json(folder / "private-generation.json", receipt)
         validate_private_generation(folder)
         return receipt
@@ -239,8 +268,12 @@ def validate_private_generation(folder: Path) -> dict[str, Any]:
     if request["request_digest"] != _canonical_digest({k: v for k, v in request.items() if k != "request_digest"}):
         raise ValueError("frozen editorial input changed")
     authored = _load_editorial_response(request=request, response_path=folder / "editorial-response.json")
-    if request["source_artifact_sha256"] != _canonical_digest(_read(folder / "source.json")):
+    source = _read(folder / "source.json")
+    if request["source_artifact_sha256"] != _canonical_digest(source):
         raise ValueError("frozen portfolio evidence changed")
+    if request.get("source_fact_binding") != receipt.get("source_fact_binding"):
+        raise ValueError("frozen report source binding changed")
+    _verify_source_facts(request, source)
     window = validate_cadence_window(request["window"])
     document = build_periodic_report_document(title=authored["title"], generated_at=receipt["generated_at"],
         period_window={"start_at": window["start_at"], "end_at": window["due_at"]},

tests/capabilities/test_periodic_report_private_generation.py

@@ -23,6 +23,8 @@
 from loopx.capabilities.periodic_report.private_generation import (
     attempt_directory,
     freeze_private_report,
+    prepare_private_report,
+    prepare_session_report,
     validate_private_generation,
 )
 from loopx.registry import atomic_write_json
@@ -161,6 +163,119 @@ def _prepared_report(tmp_path: Path):
     return root, window, response
 
 
+def _source_prepared_report(tmp_path: Path, *, local: bool):
+    _, window, response = _prepared_report(tmp_path / "template")
+    root = tmp_path / "bound-runtime"
+    source = {"rows": [{"todo_id": "todo_source", "text": "保留原始证据与未完成工作", "status": "open"}]}
+    source["source_revision"] = _canonical_digest(source)
+
+    def read(arguments):
+        rows = ([{"goal_id": "encoding-goal", "activation_state": "active"}]
+            if arguments["view"] == "portfolio" else source["rows"]
+            if arguments["view"] == "todos" else [])
+        return {"ok": True, "rows": rows, "matched": len(rows), "next_offset": None,
+            "source": {"source_revision": source["source_revision"], "status": "read"}}
+
+    def prepare():
+        if local:
+            return prepare_session_report(root=root, window=window,
+                actor={"session_id": "original-session"}, read=lambda: source,
+                reporting="保留实际来源。")
+        return prepare_private_report(root=root, window=window, read=read,
+            authorization_scope_id="scope-example", authorization_goal_ids=["encoding-goal"],
+            reporting="保留实际来源。")
+
+    request = prepare()
+    response["request_digest"] = request["request_digest"]
+    for section in response["sections"]:
+        for item in section["items"]:
+            item["source_ref"] = request["facts"][0]["source_ref"]
+    return root, window, response, prepare
+
+
+@pytest.mark.parametrize("local", [False, True])
+def test_source_bound_report_keeps_complete_facts_and_original_artifacts(tmp_path, local):
+    root, window, response, prepare = _source_prepared_report(tmp_path, local=local)
+    folder = attempt_directory(root, window)
+    before = {name: (folder / name).read_bytes() for name in ["source.json", "editorial-request.json"]}
+    request = prepare()
+    assert len(request["facts"]) == len(json.loads(before["source.json"])["rows"]) + 1
+    receipt = freeze_private_report(root=root, window=window, response=response)
+    assert validate_private_generation(folder) == receipt
+    assert {name: (folder / name).read_bytes() for name in before} == before
+    bundle = json.loads((folder / "generation-bundle.json").read_text())
+    assert all(item["source_ref"] == request["facts"][0]["source_ref"]
+        for section in bundle["document"]["sections"] for item in section["items"])
+
+
+@pytest.mark.parametrize("local", [False, True])
+@pytest.mark.parametrize("entrypoint", ["prepare", "freeze"])
+@pytest.mark.parametrize("substitution", ["url", "summary", "omission", "revision", "binding_removed"])
+def test_source_bound_report_rejects_rehashed_input_substitution(tmp_path, local, entrypoint, substitution):
+    root, window, response, prepare = _source_prepared_report(tmp_path, local=local)
+    folder = attempt_directory(root, window)
+    original_source = (folder / "source.json").read_bytes()
+    request = json.loads((folder / "editorial-request.json").read_text())
+    if substitution == "url":
+        request["facts"][0]["source_ref"] = "https://example.org/not-in-the-frozen-evidence"
+    elif substitution == "summary":
+        request["facts"][0]["summary"] = "原始来源没有提供的成功结论。"
+    elif substitution == "omission":
+        request["facts"].pop()
+    elif substitution == "revision":
+        request["source_revision"] = "sha256:" + "e" * 64
+    else:
+        request.pop("source_fact_binding", None)
+        request["facts"][0]["source_ref"] = "https://example.org/removed-input-binding"
+    request["request_digest"] = _canonical_digest({k: v for k, v in request.items() if k != "request_digest"})
+    atomic_write_json(folder / "editorial-request.json", request)
+    response["request_digest"] = request["request_digest"]
+    for section in response["sections"]:
+        for item in section["items"]:
+            item["source_ref"] = request["facts"][0]["source_ref"]
+    with pytest.raises(ValueError, match="facts do not match"):
+        if entrypoint == "prepare":
+            prepare()
+        else:
+            freeze_private_report(root=root, window=window, response=response)
+    assert (folder / "source.json").read_bytes() == original_source
+    assert not (folder / "editorial-response.json").exists()
+    assert not (folder / "private-generation.json").exists()
+    assert not (folder / "report.md").exists()
+
+
+@pytest.mark.parametrize("local", [False, True])
+@pytest.mark.parametrize("remove_binding", [False, True])
+def test_generation_validator_rejects_rehashed_facts_with_self_consistent_receipts(tmp_path, local, remove_binding):
+    from loopx.capabilities.periodic_report.incremental import build_periodic_report_publication_candidate
+    from loopx.capabilities.periodic_report.post_writeback_hook import evaluate_periodic_report_trigger_evaluation_intent
+
+    root, window, response, _ = _source_prepared_report(tmp_path, local=local)
+    folder = attempt_directory(root, window)
+    receipt = freeze_private_report(root=root, window=window, response=response)
+    request = json.loads((folder / "editorial-request.json").read_text())
+    request["facts"][0]["summary"] = "替换了原始来源不支持的事实。"
+    if remove_binding:
+        request.pop("source_fact_binding", None)
+        receipt.pop("source_fact_binding", None)
+    request["request_digest"] = _canonical_digest({k: v for k, v in request.items() if k != "request_digest"})
+    response["request_digest"] = request["request_digest"]
+    candidate = build_periodic_report_publication_candidate(goal_id=window["goal_id"],
+        agent_id=window["agent_id"], generation_id=receipt["generation_id"],
+        trigger_receipt=evaluate_periodic_report_trigger_evaluation_intent(cadence_intent(window)),
+        facts=request["facts"], baseline=None)
+    # Every existing self-digest is consistent; the unchanged independent
+    # source remains the oracle, rather than another hash of substituted facts.
+    atomic_write_json(folder / "editorial-request.json", request)
+    atomic_write_json(folder / "editorial-response.json", response)
+    atomic_write_json(folder / "publication-candidate.json", candidate)
+    atomic_write_json(folder / "private-generation.json", {**receipt,
+        "request_digest": request["request_digest"], "publication_candidate_id": candidate["candidate_id"],
+        "publication_candidate_sha256": _canonical_digest(candidate)})
+    with pytest.raises(ValueError, match="facts do not match"):
+        validate_private_generation(folder)
+
+
 def test_private_report_artifacts_use_canonical_utf8_on_windows_text_mode(
     tmp_path: Path, monkeypatch
 ) -> None: