diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 2ed0e6e562..f5f5196287 100644 --- a/docs/product/periodic-reports.md +++ b/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 diff --git a/loopx/capabilities/periodic_report/private_generation.py b/loopx/capabilities/periodic_report/private_generation.py index 30d4546964..eed0af955d 100644 --- a/loopx/capabilities/periodic_report/private_generation.py +++ b/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"]}, diff --git a/tests/capabilities/test_periodic_report_private_generation.py b/tests/capabilities/test_periodic_report_private_generation.py index 2d3a1833f4..369dc6d4cc 100644 --- a/tests/capabilities/test_periodic_report_private_generation.py +++ b/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: diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index 4f7897c587..168c9e4e52 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -51,6 +51,32 @@ def configured_local_report(steward, tmp_path, monkeypatch, provider="sqlite", s return session, ManagedSessionReports(runtime), config, task["todo_id"] +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_local_report_rejects_source_substitution_before_original_model(steward, tmp_path, monkeypatch, provider): + from loopx.capabilities.periodic_report.pending_intent import _canonical_digest + from loopx.registry import atomic_write_json + + session, reports, _, _ = configured_local_report(steward, tmp_path, monkeypatch, provider) + store, _, bot, _, _, capture, _ = steward + sid = session["session_id"] + reports.frontier(sid) + _, window = reports.current_window(sid) + request = {"actor": reports.selected(sid)["actor"], "report_window_id": window["window_id"]} + messages, calls, writes = store.messages(sid), capture.read_bytes(), list(bot.writes) + reports.prepare(request, None) + folder = attempt_directory(reports.root, window) + original_source = (folder / "source.json").read_bytes() + editorial = json.loads((folder / "editorial-request.json").read_text()) + editorial["facts"][0]["source_ref"] = "https://example.org/not-in-the-original-goal" + editorial["request_digest"] = _canonical_digest({k: v for k, v in editorial.items() if k != "request_digest"}) + atomic_write_json(folder / "editorial-request.json", editorial) + with pytest.raises(ValueError, match="facts do not match"): + reports.prepare(request, None) + assert (folder / "source.json").read_bytes() == original_source + assert not (folder / "private-generation.json").exists() + assert store.messages(sid) == messages and capture.read_bytes() == calls and bot.writes == writes + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_local_agent_calendar_report_uses_original_worker_and_verified_core_return(steward, tmp_path, monkeypatch, provider): from loopx.capabilities.native_chat.steward_monitor import StewardMonitor