fix(reports): retain original authority across partial source reads

docs/product/periodic-reports.md

@@ -340,6 +340,20 @@ grant or changing frozen artifacts still blocks delivery, including after a
 partial or uncertain send. Existing provider readback decides which original
 parts remain unsent; scope growth never authorizes another model run or resend.
 
+Read coverage and authorization are separate. A partial source may omit an
+unreadable or oversized Goal without revoking the original report grant. New
+private editorial requests freeze the complete authorized Goal identities
+alongside their existing authorization digest. Delivery checks this immutable
+witness, including unread Goals; it never upgrades partial coverage to complete.
+Later authorized additions belong to a future report, while removal or identity
+replacement of an original authorized Goal still blocks publication.
+
+Older frozen requests remain immutable. When their readable Goal list does not
+reconstruct the original authorization digest, recovery requires the complete
+current authorized scope to match that original digest exactly. An absent
+identity witness never permits guessing which Goals were originally authorized;
+an ambiguous later scope change remains retained, without another model run.
+
 The existing idle Chat wake producer admits one calendar window and creates
 one native finite generation Todo for it. The original Chat worker and native
 thread author the report, using complete frozen source records and the owner's

loopx/capabilities/native_chat/steward_reports.py

@@ -184,12 +184,23 @@ def delivery_current(self, binding_id: str, window: dict[str, Any]) -> bool:
                 or not isinstance(frozen_ids, list) or not all(isinstance(value, str) for value in frozen_ids)
                 or current_ids is None or not set(frozen_ids).issubset(current_ids)):
             return False
-        # A new authorized Goal belongs to a future report. Recheck every frozen
-        # Goal's original identity and current grant, rather than treating scope
-        # growth as revocation or rewriting the already validated source/body.
-        current = manager_authorization_scope_id_for_registry(self.controller.registry_path,
-            frozen_ids, runtime_root=self.root, channel_id=session["channel_id"])
-        return current == expected["authorization_scope_id"]
+        def scope_id(ids):
+            return manager_authorization_scope_id_for_registry(self.controller.registry_path,
+                ids, runtime_root=self.root, channel_id=session["channel_id"])
+        if "authorization_goal_ids" in expected:
+            authority_ids = expected["authorization_goal_ids"]
+            if (not isinstance(authority_ids, list)
+                    or any(not isinstance(value, str) or not value for value in authority_ids)
+                    or not set(frozen_ids).issubset(authority_ids)
+                    or not set(authority_ids).issubset(current_ids)):
+                return False
+            return scope_id(authority_ids) == expected["authorization_scope_id"]
+        # Old requests froze the whole authorization hash, but only recorded
+        # readable Goals. Full-coverage history can still tolerate scope growth.
+        # Partial history without an identity witness requires the entire live
+        # scope to match exactly; never infer missing identities or rewrite it.
+        return (scope_id(frozen_ids) == expected["authorization_scope_id"]
+            or scope_id(current_ids) == expected["authorization_scope_id"])
 
     def health(self, binding_id: str) -> dict[str, Any]:
         """Content-free metadata, never a source/provider verification receipt."""
@@ -285,12 +296,19 @@ def prepare(self, request: dict[str, Any], adapter: Any, *, authorization_scope_
         if reader is None or not authorization_scope_id:
             raise ValueError("the original scoped report reader is unavailable")
         session = self.controller.store.load_session(request["actor"]["session_id"])
+        from ...chat_manager_context import manager_authorization_scope_id_for_registry
+        authority_ids = self.bindings.steward_scope(session)
+        if (authority_ids is None or manager_authorization_scope_id_for_registry(
+                self.controller.registry_path, authority_ids, runtime_root=self.root,
+                channel_id=session["channel_id"]) != authorization_scope_id):
+            raise ValueError("the original report authorization changed before collection")
         rules = self.managed._workspace(session) / "REPORTING.md"
         if rules.is_symlink():
             raise ValueError("the original reporting criteria changed source")
         reporting = rules.read_text(encoding="utf-8") if rules.exists() else "Use the existing overview-to-depth report contract and preserve all coverage gaps."
         frozen = prepare_private_report(root=self.root, window=window, read=lambda query: reader(TOOL_NAME, query),
-            authorization_scope_id=authorization_scope_id, reporting=reporting)
+            authorization_scope_id=authorization_scope_id, reporting=reporting,
+            authorization_goal_ids=authority_ids)
         return report_prompt(frozen, attempt_directory(self.root, window))
 
     def freeze(self, request: dict[str, Any], candidate: dict[str, Any]) -> dict[str, Any]:

loopx/capabilities/periodic_report/private_generation.py

@@ -32,7 +32,8 @@ def _read(path: Path) -> dict[str, Any]:
 
 
 def prepare_private_report(*, root: Path, window: dict[str, Any], read,
-                           authorization_scope_id: str, reporting: str) -> dict[str, Any]:
+                           authorization_scope_id: str, authorization_goal_ids: list[str],
+                           reporting: str) -> dict[str, Any]:
     """Freeze permitted evidence once; full records stay beside bounded fact headings."""
     from ..manager_context.report_source import collect_manager_report_source
 
@@ -50,6 +51,10 @@ def prepare_private_report(*, root: Path, window: dict[str, Any], read,
             return request
         source = collect_manager_report_source(read, {
             "window_start": window["start_at"], "window_end": window["due_at"]})
+        if (not isinstance(authorization_goal_ids, list)
+                or any(not isinstance(value, str) or not value for value in authorization_goal_ids)
+                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)
         atomic_write_json(folder / "source.json", source)
         facts = [{"fact_id": "portfolio_coverage", "title": "本期来源覆盖",
@@ -70,6 +75,9 @@ def prepare_private_report(*, root: Path, window: dict[str, Any], read,
             source_artifact=str(folder / "source.json"), source_revision=source["source_revision"],
             source_artifact_sha256=_canonical_digest(source), window=window,
             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.",

tests/test_native_steward_reports.py

@@ -185,6 +185,120 @@ def legacy_validator(*args, **kwargs):
     assert capture.read_bytes() == calls and bot.writes == sent
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("legacy", [False, True])
+@pytest.mark.parametrize("change", ["none", "growth", "remove_unread", "replace_unread"])
+def test_partial_read_coverage_is_not_the_original_report_authorization(steward, tmp_path, monkeypatch, provider, legacy, change):
+    from loopx.bootstrap import bootstrap_project
+    from loopx.capabilities.manager_context import report_source
+    from loopx.capabilities.native_chat import steward_reports as report_module
+    from loopx.cli_commands import turn as turn_command
+    from loopx.control_plane.turn_driver.journal_store import find_loopx_turn_key_by_settlement_identity, turn_journal_path
+    store, runtime, bot, transport, binding, capture, _ = steward
+    sid, actor, reports, _ = configured(steward, tmp_path, monkeypatch, provider)
+
+    def add_goal(gid):
+        project = tmp_path / gid
+        project.mkdir()
+        assert bootstrap_project(project=project, registry_path=runtime.registry_path,
+            runtime_root=runtime.coordination_runtime_root, goal_id=gid,
+            objective="Retain this independently authorized synthetic work.", domain="test", role="primary",
+            parent_goal_id=None, state_file=None, goal_doc=None, adapter_kind="generic_project_goal_v0",
+            adapter_status="connected", next_probe=None, spawn_allowed=False, max_children=0,
+            allowed_domains=[], write_scope=[], preserve_todos=True, force=False, dry_run=False,
+            sync_global=False)["ok"]
+
+    unread = "unread-authorized-goal"
+    add_goal(unread)
+    collect = report_source.collect_manager_report_source
+
+    def partial(read, query):
+        def source_io(arguments):
+            page = read(arguments)
+            if arguments["view"] == "portfolio":
+                # The actual scoped reader remains authoritative; simulate an
+                # oversized/unreadable row, not an authorization or success.
+                return {**page, "rows": [r for r in page["rows"] if r.get("goal_id") != unread], "unknown": True}
+            return page
+        return collect(source_io, query)
+
+    monkeypatch.setattr(report_source, "collect_manager_report_source", partial)
+    install_model(steward, tmp_path, sid)
+    prepare, command = report_module.prepare_private_report, turn_command.handle_turn_command
+
+    def old_prepare(**kwargs):
+        from loopx.capabilities.periodic_report.pending_intent import _canonical_digest
+        from loopx.registry import atomic_write_json
+        request = dict(prepare(**kwargs))
+        # Seed only the historical input format, before any model or generation.
+        # The real worker, validator, journal and sender still own the outcome.
+        folder = attempt_directory(kwargs["root"], kwargs["window"])
+        assert not (folder / "private-generation.json").exists()
+        request.pop("authorization_goal_ids", None)
+        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)
+        return request
+
+    def old_validator(*args, **kwargs):
+        kwargs.pop("native_task_validator", None)
+        return command(*args, **kwargs)
+
+    with monkeypatch.context() as old:
+        if legacy:
+            embedded_report_python(tmp_path, old)
+            old.setattr(report_module, "prepare_private_report", old_prepare)
+            old.setattr(turn_command, "handle_turn_command", old_validator)
+        original = pump_steward_monitors(runtime)[0]
+        done = finish(runtime, original)
+    assert done["status"] == "completed"
+    if legacy:
+        assert done["response"]["native_loopx"]["result_kind"] == "validation_failed"
+    window = read_cadence_journal(runtime_root=reports.root, goal_id=actor["goal_id"])["window"]
+    folder = attempt_directory(reports.root, window)
+    source = json.loads((folder / "source.json").read_text())
+    request = json.loads((folder / "editorial-request.json").read_text())
+    assert unread not in source["goal_ids"] and not source["coverage"]["complete"]
+    if legacy:
+        assert "authorization_goal_ids" not in request
+    frozen = [(folder / name).read_bytes() for name in ["report.md", "source.json", "editorial-request.json"]]
+    key = find_loopx_turn_key_by_settlement_identity(reports.root, goal_id=actor["goal_id"],
+        agent_id=actor["agent_id"], todo_id=original["loopx_request"]["todo_id"],
+        turn_instance_id=original["loopx_request"]["turn_instance_id"])
+    path = turn_journal_path(reports.root, goal_id=actor["goal_id"], turn_key=key)
+    journal, history, calls, writes = path.read_bytes(), store.load_turn(sid, original["turn_id"]), capture.read_bytes(), list(bot.writes)
+    if change == "growth":
+        add_goal("later-authorized-goal")
+    elif change in {"remove_unread", "replace_unread"}:
+        registry = json.loads(runtime.registry_path.read_text())
+        if change == "remove_unread":
+            registry["goals"] = [row for row in registry["goals"] if row["id"] != unread]
+        else:
+            next(row for row in registry["goals"] if row["id"] == unread)["creation_operation_id"] += ".replacement"
+        runtime.registry_path.write_text(json.dumps(registry))
+    allowed = change == "none" or change == "growth" and not legacy
+    assert reports.delivery_current(binding["binding_id"], window) is allowed
+    if not legacy and allowed:
+        assert request["authorization_goal_ids"] == sorted([actor["goal_id"], unread])
+    runtime._steward_monitor_next_check.clear()
+    assert not pump_steward_monitors(runtime)
+    assert capture.read_bytes() == calls and store.load_turn(sid, original["turn_id"]) == history
+    assert [(folder / name).read_bytes() for name in ["report.md", "source.json", "editorial-request.json"]] == frozen
+    if not allowed:
+        assert path.read_bytes() == journal
+        assert transport.steward_reports.reconcile() == 0 and bot.writes == writes
+        assert not (folder / "private-delivery.json").exists()
+        return
+    recovered = json.loads(path.read_text())
+    assert recovered["status"] == "committed" and recovered["turn_key"] == key
+    if legacy:
+        assert recovered["host_attempt_count"] == json.loads(journal)["host_attempt_count"]
+        assert recovered["recovery_audit"]["actual"]["host_invoked"] is False
+    assert transport.steward_reports.reconcile() == 1
+    assert reports.health(binding["binding_id"])["publication_verified"]
+    sent = list(bot.writes)
+    assert transport.steward_reports.reconcile() == 0 and bot.writes == sent and capture.read_bytes() == calls
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("result_kind", ["validated_progress", "validated_completion"])
 def test_original_due_report_freezes_once_and_returns_full_body_after_native_settlement(steward, tmp_path, monkeypatch, provider, result_kind):