fix(reports): preserve recorded citations in frozen portfolio inputs

docs/product/periodic-reports.md

@@ -16,6 +16,17 @@ This is presentation validation, not a claim that a URL is reachable, public or
 authorized for another audience. Source collection, frozen evidence, redaction
 and delivery authorization retain those responsibilities.
 
+Exact authorized portfolio delivery reads retain eligible references recorded in
+the original Turn's path evidence, alongside their digest identities. New
+private portfolio inputs bind those references to the same frozen source and
+make them available to the report author as **recorded, unverified** citations.
+Local paths, credential-bearing and malformed reference values remain withheld;
+the source reader never fetches a URL. A recorded address proves neither public
+access nor permission to publish to a different audience. The original scope,
+source gaps and unread-artifact status remain authoritative; an opaque digest
+does not supply a missing URL. Existing v1 frozen inputs retain their original
+fact set and bytes. Ordinary managed Session inputs retain their existing scope.
+
 Text reports include the same authored overview values, labels and details as
 the HTML artifact. When that overview is present, it replaces the generated
 title-only opening; reports without highlights retain their compiled summary.

loopx/capabilities/periodic_report/private_generation.py

@@ -49,6 +49,16 @@ def _source_facts(source: dict[str, Any], *, local: bool) -> list[dict[str, Any]
             "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"})
+        if not local and source.get("source_fact_binding") == "frozen_source_v2":
+            for citation_index, citation in enumerate(
+                    row["evidence"].get("recorded_details", {}).get("recorded_source_refs", [])):
+                facts.append({"fact_id": f"portfolio_{index}_source_{citation_index}",
+                    "title": f"{row.get('goal_id') or 'portfolio'} / 原记录出处（未读取）"[:240],
+                    "summary": f"原记录 {row['evidence'].get('source_ref', row['source_ref'])} 提供的出处；"
+                        f"冻结记录 {row['source_ref']}；标识 {citation['evidence_ref']}。"
+                        "仅限本次原授权受众；未读取引用产物，可达性和公开许可均未核验，不证明业务完成。",
+                    "status": "unverified", "source_ref": citation["source_ref"],
+                    "content_kind": "coverage"})
     return facts
 
 
@@ -57,8 +67,8 @@ def _verify_source_facts(request: dict[str, Any], source: dict[str, Any]) -> Non
     # 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"
+    if (request.get("source_fact_binding") not in {"frozen_source_v1", "frozen_source_v2"}
+            or source.get("source_fact_binding") != request.get("source_fact_binding")
             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)
@@ -94,7 +104,7 @@ 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"}
+        source = {**source, "source_fact_binding": "frozen_source_v2"}
         atomic_write_json(folder / "source.json", source)
         request = _build_editorial_request(intent=cadence_intent(window), runtime_root=root,
             goal_id=window["goal_id"], agent_id=window["agent_id"],
@@ -103,14 +113,15 @@ def prepare_private_report(*, root: Path, window: dict[str, Any], read,
         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",
+            source_fact_binding="frozen_source_v2",
             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.
         request["authorization_goal_ids"] = sorted(set(authorization_goal_ids))
         request["narrative_contract"]["requirements"] += [
             "Follow REPORTING and cover every authorized Goal, including stopped and unreadable Goals.",
             "Read full frozen source records for substantive claims; excerpts and Agent claims are not independent verification.",
+            "Recorded source references are scoped provenance, not proof of access, public status or publication permission. Keep unverified references explicit; never infer a URL from a digest or reuse private sources for another audience.",
             "Distinguish implementation, merged source, installed adoption and actual original-message delivery.",
             "Describe current and relevant completed tasks, counterevidence, source gaps and concrete next steps.",
             "A due unchanged window still needs a substantive report; observations or CI counts are not business progress.",

loopx/chat_manager_history.py

@@ -6,11 +6,14 @@
 import hashlib
 import json
 from pathlib import Path
+from urllib.parse import urlsplit
 
 from .chat_manager_details import _text
 from .control_plane.runtime.run_context_retention import goal_semantic_history_from_runs
 from .control_plane.work_items.delivery_outcome import MATERIAL_DELIVERY_OUTCOMES, PROGRESS_DELIVERY_OUTCOMES
 from .history import STATUS_NEUTRAL_CLASSIFICATIONS, load_index_snapshot
+from .public_safe_text import PUBLIC_SAFE_REFERENCE_PATTERN, find_private_text_match
+from .control_plane.runtime.public_safety import validate_public_safe_value
 
 
 def delivery_history_window(window_start=None, window_end=None):
@@ -66,6 +69,26 @@ def _recorded_details(run, *, complete=False):
         "sha256:" + hashlib.sha256(item.encode()).hexdigest()
         for item in (evidence if complete else evidence[:8])
     ] if valid_evidence else []
+    # Exact reads already belong to the original scoped reader. Preserve the
+    # recorded provenance too; a digest cannot be reversed into a citation.
+    # Shape/disclosure checks are not public-access or publication permission.
+    recorded_refs = delta.get("evidence_refs")
+    citations = []
+    if complete and isinstance(recorded_refs, list):
+        for reference in recorded_refs:
+            if (not isinstance(reference, str) or not PUBLIC_SAFE_REFERENCE_PATTERN.fullmatch(reference)
+                    or find_private_text_match(reference)):
+                continue
+            try:
+                validate_public_safe_value(reference)
+                address = urlsplit(reference)
+                if address.scheme not in {"http", "https"} or not address.hostname:
+                    continue
+                address.port
+            except ValueError:
+                continue
+            citations.append({"source_ref": reference,
+                "evidence_ref": "sha256:" + hashlib.sha256(reference.encode()).hexdigest()})
     return {
         "source": "core_run_index",
         "verification": "recorded_claim_not_independent_verification",
@@ -83,6 +106,7 @@ def _recorded_details(run, *, complete=False):
             ],
         },
         "evidence_refs": refs,
+        **({"recorded_source_refs": citations} if complete else {}),
         "evidence_coverage": {
             "status": "read" if valid_evidence else "missing_or_invalid",
             "known": len(evidence) if valid_evidence else None,

tests/capabilities/test_periodic_report_private_generation.py

@@ -208,6 +208,66 @@ def test_source_bound_report_keeps_complete_facts_and_original_artifacts(tmp_pat
         for section in bundle["document"]["sections"] for item in section["items"])
 
 
+@pytest.mark.parametrize("legacy", [False, True])
+def test_recorded_citation_reaches_frozen_facts_and_both_artifacts_without_public_access_claim(
+        tmp_path, legacy):
+    from loopx.capabilities.manager_context.report_source import collect_manager_report_source
+    from loopx.chat_manager_history import _recorded_details
+
+    root, window, response = _prepared_report(tmp_path / "template")
+    root = tmp_path / "citation-runtime"
+    reference = "https://example.org/results/42"
+    details = _recorded_details({"agent_vision": {"path_delta": {"evidence_refs": [reference]}}}, complete=True)
+
+    def read(arguments):
+        rows = ([{"goal_id": "encoding-goal"}] if arguments["view"] == "portfolio" else
+            [{"source_ref": "sha256:" + "d" * 64, "recorded_details": details}]
+            if arguments["view"] == "deliveries" else [])
+        return {"ok": True, "rows": rows, "matched": len(rows), "next_offset": None,
+            "source": {"status": "read", "source_revision": "sha256:" + "c" * 64}}
+
+    request = prepare_private_report(root=root, window=window, read=read,
+        authorization_scope_id="scope-example", authorization_goal_ids=["encoding-goal"], reporting="保留原出处。")
+    folder = attempt_directory(root, window)
+    # The shipped v1 input has no additional citation facts. Specify that
+    # compatibility delta directly rather than deriving it from a v1 helper.
+    if legacy:
+        source = json.loads((folder / "source.json").read_text())
+        request["facts"] = [fact for fact in request["facts"] if fact["source_ref"] != reference]
+        source["source_fact_binding"] = request["source_fact_binding"] = "frozen_source_v1"
+        request["source_artifact_sha256"] = _canonical_digest(source)
+        request["request_digest"] = _canonical_digest({k: v for k, v in request.items() if k != "request_digest"})
+        atomic_write_json(folder / "source.json", source)
+        atomic_write_json(folder / "editorial-request.json", request)
+    citation_facts = [fact for fact in request["facts"] if fact["source_ref"] == reference]
+    assert len(citation_facts) == (0 if legacy else 1)
+    if not legacy:
+        assert citation_facts[0]["status"] == "unverified"
+        assert "可达性和公开许可均未核验" in citation_facts[0]["summary"]
+        assert "sha256:" + "d" * 64 in citation_facts[0]["summary"]
+    response["request_digest"] = request["request_digest"]
+    for section in response["sections"]:
+        for item in section["items"]:
+            item["source_ref"] = request["facts"][0]["source_ref"] if legacy else reference
+    receipt = freeze_private_report(root=root, window=window, response=response)
+    before = {path.name: path.read_bytes() for path in folder.iterdir() if path.suffix in {".json", ".md"}}
+    assert validate_private_generation(folder) == receipt
+    assert prepare_private_report(root=root, window=window, read=lambda _: pytest.fail("frozen input reread"),
+        authorization_scope_id="scope-example", authorization_goal_ids=["encoding-goal"],
+        reporting="保留原出处。") == request
+    assert {name: (folder / name).read_bytes() for name in before} == before
+    bundle = json.loads((folder / "generation-bundle.json").read_text())
+    artifacts = {item["renderer_kind"]: item["content"] for item in bundle["artifacts"]}
+    assert (f"]({reference})" in artifacts["markdown"]) is not legacy
+    assert (f'href="{reference}"' in artifacts["html"]) is not legacy
+    snapshot = collect_manager_report_source(read, {"window_start": window["start_at"], "window_end": window["due_at"]})
+    assert not snapshot["coverage"]["referenced_artifacts_read"]
+    # A different audience cannot adopt the same frozen input.
+    with pytest.raises(ValueError, match="another scope"):
+        prepare_private_report(root=root, window=window, read=read,
+            authorization_scope_id="other-scope", authorization_goal_ids=["encoding-goal"], reporting="保留原出处。")
+
+
 @pytest.mark.parametrize("local", [False, True])
 @pytest.mark.parametrize("entrypoint", ["prepare", "freeze"])
 @pytest.mark.parametrize("substitution", ["url", "summary", "omission", "revision", "binding_removed"])

tests/test_manager_report_source.py

@@ -1,4 +1,5 @@
 """One original scoped reader supplies complete, fixed-window report evidence."""
+import hashlib
 import json
 from copy import deepcopy
 
@@ -202,7 +203,8 @@ def test_actual_canonical_sources_join_a_completed_task_and_fixed_boundaries(tmp
     assert todo["status"] == "done"
     path = runtime / "goals/goal-a/runs/index.jsonl"
     path.parent.mkdir(parents=True, exist_ok=True)
-    run = {"goal_id": "goal-a", "todo_id": todo["todo_id"], "delivery_outcome": "outcome_progress"}
+    run = {"goal_id": "goal-a", "todo_id": todo["todo_id"], "delivery_outcome": "outcome_progress",
+        "agent_vision": {"path_delta": {"evidence_refs": ["https://example.org/results/42"]}}}
     path.write_text("".join(json.dumps({**run, "generated_at": timestamp,
         "delivery_outcome": "outcome_gap" if i == 0 else "outcome_progress"}) + "\n" for i, timestamp in enumerate((
         "2026-01-01T09:00:00+08:00", "2026-01-02T08:59:00+08:00", "2026-01-02T09:00:00+08:00"))))
@@ -217,6 +219,29 @@ def test_actual_canonical_sources_join_a_completed_task_and_fixed_boundaries(tmp
     linked = [row["evidence"] for row in snapshot["rows"] if row["kind"] == "delivery_task"]
     assert len(linked) == 1 and linked[0]["todo_id"] == todo["todo_id"] and linked[0]["status"] == "done"
     assert deliveries[0]["recorded_details"]["verification"] == "recorded_claim_not_independent_verification"
+    # The exact reader supplies the actual recorded reference alongside its
+    # digest; it has not read the remote artifact or qualified public access.
+    assert deliveries[0]["recorded_details"]["recorded_source_refs"] == [{
+        "source_ref": "https://example.org/results/42",
+        "evidence_ref": "sha256:" + hashlib.sha256(b"https://example.org/results/42").hexdigest()}]
     ordinary = tool.read(TOOL_NAME, {"view": "deliveries", "goal_id": "goal-a", **WINDOW})
     assert ordinary["matched"] == 1 and ordinary["rows"][0]["outcome"] == "outcome_progress"
     assert list_goal_todos(registry_path=registry, runtime_root_arg=str(runtime), goal_id="goal-a")["todos"][0] == todo
+
+
+@pytest.mark.parametrize("reference", [
+    "/Users/fixture/private.txt", "file:///tmp/private", "../.local/private.json",
+    "https://user:pass@example.org/file", "https://example.org/file?key=hidden",
+    "https://docs" + ".internal/design", "javascript:alert", "https://example.org/\nprivate",
+    "https://example.org:invalid/file", "https://#missing-host",
+])
+def test_exact_delivery_reference_disclosure_does_not_promote_private_or_unsafe_values(reference):
+    from loopx.chat_manager_history import _recorded_details
+
+    run = {"agent_vision": {"path_delta": {"evidence_refs": [reference]}},
+        "progress_observation": {"evidence_ids": ["evidence:original"]}}
+    details = _recorded_details(run, complete=True)
+    assert details["recorded_source_refs"] == []
+    assert len(details["evidence_refs"]) == 1
+    assert details["artifact_read_status"] == "not_read"
+    assert "recorded_source_refs" not in _recorded_details(run)