fix(reports): retain recorded Task citations in managed reports

docs/product/periodic-reports.md

@@ -25,7 +25,23 @@ 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.
+fact set and bytes. New ordinary managed Session inputs use the same frozen
+citation format for a native Task whose evidence field is one eligible HTTP(S)
+reference. The original Task reference and complete source remain available;
+the address is explicitly unread and unverified. Prose, notes and compound
+evidence strings are not searched for links. This adds no source read, fetch,
+publication route or audience, and existing local v1 windows remain unchanged.
+
+The packaged local-Session journey below uses a disposable SQLite backend and
+synthetic model response: normal configuration preview/apply, one native report
+Turn, its complete original conversation return, then the existing answer
+reader. The six source links retain the recorded Task address in both formats.
+The fixed synthetic prose qualifies transport and reading mechanics, not live
+editorial quality or the address's availability. No layout or editor changed.
+
+| Packaged answer reader | Narrow answer reader |
+| --- | --- |
+| ![A recorded Task address in the full report](../assets/personal-workspace/steward-reports/local-citations/packaged-reader-desktop.png) | ![The same source link and report at 390 pixels](../assets/personal-workspace/steward-reports/local-citations/packaged-reader-narrow.png) |
 
 Text reports include the same authored overview values, labels and details as
 the HTML artifact. When that overview is present, it replaces the generated

loopx/capabilities/periodic_report/private_generation.py

@@ -33,6 +33,8 @@ def _read(path: Path) -> dict[str, Any]:
 
 def _source_facts(source: dict[str, Any], *, local: bool) -> list[dict[str, Any]]:
     """Derive editorial references from the same frozen source on every read."""
+    from ...chat_manager_history import _recorded_web_references
+
     facts = [{"fact_id": "local_coverage" if local else "portfolio_coverage",
         "title": "本期来源范围" if local else "本期来源覆盖",
         "summary": ("当前原 Goal 的完整原生任务读取；任务状态在收集时观察，按完成时间区分本期成果和历史记录。来源声明及任务完成不证明外部业务已独立验收。"
@@ -49,13 +51,20 @@ 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']}。"
+        if source.get("source_fact_binding") == "frozen_source_v2":
+            # A native Task evidence field can itself be one exact reference.
+            # Do not extract links from prose, notes or compound references.
+            citations = (_recorded_web_references([row.get("evidence")]) if local else
+                row["evidence"].get("recorded_details", {}).get("recorded_source_refs", []))
+            record_ref = "todo:" + row["todo_id"] if local else row["evidence"].get("source_ref", row["source_ref"])
+            frozen_ref = "todo:" + row["todo_id"] if local else row["source_ref"]
+            for citation_index, citation in enumerate(citations):
+                facts.append({"fact_id": (f"local_{index}_source_{citation_index}" if local else
+                        f"portfolio_{index}_source_{citation_index}"),
+                    "title": ("原任务出处（未读取）" if local else
+                        f"{row.get('goal_id') or 'portfolio'} / 原记录出处（未读取）")[:240],
+                    "summary": f"原记录 {record_ref} 提供的出处；"
+                        f"冻结记录 {frozen_ref}；标识 {citation['evidence_ref']}。"
                         "仅限本次原授权受众；未读取引用产物，可达性和公开许可均未核验，不证明业务完成。",
                     "status": "unverified", "source_ref": citation["source_ref"],
                     "content_kind": "coverage"})
@@ -187,15 +196,15 @@ def prepare_session_report(*, root: Path, window: dict[str, Any], actor: dict[st
         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")
-        source = {**source, "source_fact_binding": "frozen_source_v1"}
+        source = {**source, "source_fact_binding": "frozen_source_v2"}
         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=_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_fact_binding="frozen_source_v1")
+            source_fact_binding="frozen_source_v2")
         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.",

loopx/chat_manager_history.py

@@ -33,6 +33,28 @@ def delivery_history_window(window_start=None, window_end=None):
     return start.astimezone(timezone.utc), end.astimezone(timezone.utc)
 
 
+def _recorded_web_references(references):
+    """Retain exact eligible addresses as unread provenance, without fetching."""
+    citations = []
+    if not isinstance(references, list):
+        return citations
+    for reference in references:
+        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 citations
+
+
 def _recorded_details(run, *, complete=False):
     """Hydrate existing Core fields; never dereference artifact paths or URLs."""
     observation = run.get("progress_observation")
@@ -72,23 +94,7 @@ def _recorded_details(run, *, complete=False):
     # 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()})
+    citations = _recorded_web_references(delta.get("evidence_refs")) if complete else []
     return {
         "source": "core_run_index",
         "verification": "recorded_claim_not_independent_verification",

tests/capabilities/test_periodic_report_private_generation.py

@@ -163,10 +163,12 @@ def _prepared_report(tmp_path: Path):
     return root, window, response
 
 
-def _source_prepared_report(tmp_path: Path, *, local: bool):
+def _source_prepared_report(tmp_path: Path, *, local: bool, evidence=None):
     _, window, response = _prepared_report(tmp_path / "template")
     root = tmp_path / "bound-runtime"
     source = {"rows": [{"todo_id": "todo_source", "text": "保留原始证据与未完成工作", "status": "open"}]}
+    if evidence is not None:
+        source["rows"][0]["evidence"] = evidence
     source["source_revision"] = _canonical_digest(source)
 
     def read(arguments):
@@ -193,6 +195,55 @@ def prepare():
     return root, window, response, prepare
 
 
+@pytest.mark.parametrize("legacy", [False, True])
+def test_local_task_citation_reaches_frozen_artifacts_without_rewriting_legacy(tmp_path, legacy):
+    reference = "https://example.org/results/42"
+    root, window, response, prepare = _source_prepared_report(tmp_path, local=True, evidence=reference)
+    folder = attempt_directory(root, window)
+    request = json.loads((folder / "editorial-request.json").read_text())
+    if legacy:
+        source = json.loads((folder / "source.json").read_text())
+        source["source_fact_binding"] = request["source_fact_binding"] = "frozen_source_v1"
+        request["facts"] = [fact for fact in request["facts"] if fact["source_ref"] != reference]
+        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)
+    citations = [fact for fact in request["facts"] if fact["source_ref"] == reference]
+    assert len(citations) == (0 if legacy else 1)
+    if not legacy:
+        assert citations[0]["status"] == "unverified"
+        assert "todo:todo_source" in citations[0]["summary"]
+        assert "可达性和公开许可均未核验" in citations[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() == request
+    assert {name: (folder / name).read_bytes() for name in before} == before
+    bundle = json.loads((folder / "generation-bundle.json").read_text())
+    assert all((reference in artifact["content"]) != legacy for artifact in bundle["artifacts"])
+
+
+@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",
+    "See https://example.org/results/42", "https://example.org/one | https://example.org/two",
+])
+def test_local_task_citation_does_not_guess_or_promote_unsafe_evidence(tmp_path, reference):
+    root, window, response, prepare = _source_prepared_report(tmp_path, local=True, evidence=reference)
+    request = prepare()
+    assert {fact["source_ref"] for fact in request["facts"]} == {
+        request["source_revision"], "todo:todo_source"}
+    receipt = freeze_private_report(root=root, window=window, response=response)
+    assert validate_private_generation(attempt_directory(root, window)) == receipt
+
+
 @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)

tests/test_native_steward_reports.py

@@ -78,15 +78,24 @@ def test_local_report_rejects_source_substitution_before_original_model(steward,
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_local_agent_calendar_report_uses_original_worker_and_verified_core_return(steward, tmp_path, monkeypatch, provider):
+@pytest.mark.parametrize("recorded_reference", [None, "https://example.org/review/42"])
+def test_local_agent_calendar_report_uses_original_worker_and_verified_core_return(steward, tmp_path, monkeypatch, provider, recorded_reference):
     from loopx.capabilities.native_chat.steward_monitor import StewardMonitor
     from loopx.capabilities.native_chat.steward_managed_mode import ManagedSessionTurn
     session, reports, _, task_id = configured_local_report(steward, tmp_path, monkeypatch, provider)
     store, runtime, bot, _, _, capture, _ = steward
     sid = session["session_id"]
+    if recorded_reference:
+        from test_native_steward_turn_host import cli
+        cli(runtime, "todo", "update", "--goal-id", session["goal_id"], "--todo-id", task_id,
+            "--agent-id", session["managed_agent_id"], "--evidence", recorded_reference)
     with monkeypatch.context() as changed_source:
         changed_source.setattr(inspect, "getsource", lambda *_: "not the imported model callable")
         install_model(steward, tmp_path, sid)
+    if recorded_reference:
+        helper = tmp_path / "report_model.py"
+        helper.write_text(helper.read_text().replace('ref = request["facts"][0]["source_ref"]',
+            "ref = " + repr(recorded_reference)))
     # A host's default codec must not decide how frozen Chinese artifacts read.
     real_read_text = Path.read_text
     def host_codec_read_text(path, *args, **kwargs):
@@ -111,9 +120,16 @@ def host_codec_read_text(path, *args, **kwargs):
     assert frozen["destination"] == "original_managed_session_full_body"
     assert source["goal_id"] == session["goal_id"] and source["complete_native_task_read"]
     assert task_id in {row["todo_id"] for row in source["rows"]}
+    if recorded_reference:
+        citation = next(fact for fact in frozen["facts"] if fact["source_ref"] == recorded_reference)
+        assert citation["status"] == "unverified" and "todo:" + task_id in citation["summary"]
     assert frozen["actual_work_window"]["end_at"] == window["due_at"]
     assert datetime.fromisoformat(window["due_at"]) - datetime.fromisoformat(window["start_at"]) == timedelta(days=1)
     body = (folder / "report.md").read_text(encoding="utf-8")
+    if recorded_reference:
+        assert f"]({recorded_reference})" in body
+        bundle = json.loads((folder / "generation-bundle.json").read_text(encoding="utf-8"))
+        assert all(recorded_reference in artifact["content"] for artifact in bundle["artifacts"])
     messages = store.messages(sid)
     assert messages[:len(before)] == before
     returned = [row for row in messages if row["role"] == "agent" and row["text"] == body]