fix(lark): bind periodic report routes to the managed session

docs/product/periodic-reports.md

@@ -345,6 +345,13 @@ single-Goal report consumer cannot consume this private portfolio preset.
 This preset requires a calendar; its switch cannot select the generic
 stage-boundary reporting mode. Pause or cancel through the report actions.
 
+After `/new` and first preparation in the new Session, route discovery selects
+a verified source belonging to that exact managed actor's Session and private
+source. An older Session in the same chat cannot mask an eligible route, and
+a retained report route cannot fall back to another Session or audience.
+Provider, owner and original-message verification still apply. Owner Todo
+notifications retain their existing private-chat route selection.
+
 Portfolio evidence collection finishes before the governed driver captures its
 launch authority. A long read can then observe legitimate registry rewrites
 without immediately invalidating an earlier launch witness. The original

loopx/extensions/lark/owner_attention.py

@@ -91,7 +91,8 @@ def _binding(self, binding_id: str, expected_revision: int | None = None) -> dic
             raise ValueError("notifications require an authorized owner private steward")
         return row
 
-    def _route(self, binding: dict[str, Any], saved: dict[str, Any] | None = None) -> dict[str, Any]:
+    def _route(self, binding: dict[str, Any], saved: dict[str, Any] | None = None, *,
+               session_id: str | None = None, source_ref: str | None = None) -> dict[str, Any]:
         candidates = [saved] if saved is not None else [
             _read_json(path) for path in sorted(self.transport.root.glob("*.json"), reverse=True)
             if re.fullmatch(r"[a-f0-9]{24}", path.stem)]
@@ -111,6 +112,12 @@ def _route(self, binding: dict[str, Any], saved: dict[str, Any] | None = None) -
                     or native.get("origin", "lark") != "lark"
                     or native.get("status") in {"prepared", "rejected"}):
                 continue
+            # A report belongs to one prepared actor, even when this private
+            # chat has older Sessions. Core owns the Session correlation; a
+            # transport receipt's filename or copied session_id cannot select it.
+            if ((session_id is not None and native.get("session_id") != session_id)
+                    or (source_ref is not None and (native.get("source") or {}).get("source_ref") != source_ref)):
+                continue
             selected = self.transport.bindings.resolve(binding_id=binding["binding_id"],
                 **record["source"], include_observation=True)
             if not self.transport._source_message(record, selected=selected, render_content=False):

loopx/extensions/lark/steward_reports.py

@@ -28,8 +28,10 @@ def __init__(self, transport: Any) -> None:
         self.controller = transport.core.controller
         self.reports = StewardReports(self.controller)
 
-    def _original_route(self, binding: dict[str, Any], actor: dict[str, Any]) -> dict[str, Any]:
-        route = self.transport.owner_attention._route(binding)
+    def _original_route(self, binding: dict[str, Any], actor: dict[str, Any],
+                        saved: dict[str, Any] | None = None) -> dict[str, Any]:
+        route = self.transport.owner_attention._route(binding, saved,
+            session_id=actor["session_id"], source_ref=actor["source_ref"])
         native = self.transport.core.read_request(route["request_ref"])
         if native.get("session_id") != actor["session_id"] or route["source"]["source_ref"] != actor["source_ref"]:
             raise ValueError("the private source does not belong to the original reporting steward")
@@ -100,7 +102,7 @@ def current():
                         record = _read_json(path)
                         if record.get("publication_cursor"):
                             continue
-                        route = self.transport.owner_attention._route(binding, record["original_route"])
+                        route = self._original_route(binding, actor, record["original_route"])
                     else:
                         route = self._original_route(binding, actor)
                         original = _read_json(self.transport.root / (route["request_ref"] + ".json"))

tests/test_lark_steward_report_route.py

@@ -0,0 +1,116 @@
+"""Report routes retain the prepared actor across private Chat session changes."""
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from test_native_steward_managed_mode import original, storage
+from test_native_steward_private import steward as steward_fixture, finish
+from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+from loopx.capabilities.native_chat.steward_reports import StewardReports
+from loopx.extensions.lark.conversation_identity import identity_ref
+
+steward = steward_fixture
+
+
+@pytest.fixture(params=["file", "sqlite"])
+def rotated_steward(steward, tmp_path, monkeypatch, request):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    store, runtime, provider, transport, binding, capture, _ = steward
+    # This real receipt sorts before both the /new command and current message.
+    # Selection must use the actor's identity, never receipt-name chronology.
+    old_event = provider.event("steward-app", "old-1", "An earlier independent conversation.")
+    old_ref = identity_ref(binding["provider_ref"], old_event["message_id"])
+    assert transport.admit("steward-app", old_event)["status"] == "durably_accepted"
+    old = transport.core.read_request(old_ref)
+    assert finish(runtime, old)["status"] == "completed"
+    command = provider.event("steward-app", "rotate-review", "/new")
+    transport.admit("steward-app", command)
+    reset = transport.core.read_request(identity_ref(binding["provider_ref"], command["message_id"]))
+    assert reset["status"] == "command_completed" and reset["response_code"] == "new_session"
+    sid, managed = original(steward)
+    assert sid != old["session_id"]
+    current_ref = identity_ref(binding["provider_ref"], "om_original")
+    assert old_ref > max(current_ref, reset["request_ref"])
+    storage(runtime, request.param)
+    assert not managed.health(binding["binding_id"])["prepared"]
+    preview = managed.prepare(binding["binding_id"], session_id=sid)["plan"]
+    actor = managed.prepare(binding["binding_id"], session_id=sid, execute=True,
+        expected_plan_id=preview["plan_id"])["actor"]
+    binding = next(row for row in transport.bindings.read()["bindings"]
+        if row["binding_id"] == binding["binding_id"])
+    schedule = {"schema_version": "periodic_report_schedule_v0", "schedule_id": "daily-report",
+        "timezone": "UTC", "rrule": "FREQ=DAILY;BYHOUR=9;BYMINUTE=0", "start_date": "2999-01-01"}
+    return SimpleNamespace(store=store, runtime=runtime, provider=provider, transport=transport,
+        binding=binding, actor=actor, capture=capture, reports=StewardReports(runtime),
+        old=old, current_ref=current_ref, schedule=schedule)
+
+
+def test_report_preview_and_apply_find_original_actor_after_new_session(rotated_steward):
+    o = rotated_steward
+    sid = o.actor["session_id"]
+    before = o.runtime.registry_path.read_bytes(), o.store.messages(sid), o.capture.read_bytes(), list(o.provider.writes)
+    preview = o.reports.configure(o.binding["binding_id"], operation="enable", schedule=o.schedule)
+    assert not preview["applied"]
+    assert before == (o.runtime.registry_path.read_bytes(), o.store.messages(sid),
+        o.capture.read_bytes(), list(o.provider.writes))
+    applied = o.reports.configure(o.binding["binding_id"], operation="enable", schedule=o.schedule,
+        execute=True, expected_plan_id=preview["plan"]["plan_id"])
+    assert applied["applied"] and applied["report"]["enabled"]
+    assert not applied["model_invoked"] and not applied["external_writes_performed"]
+    assert o.reports.subscription(o.binding["binding_id"])["route_ref"] == o.reports.route_ref(o.binding, o.actor)
+    route = o.transport.steward_reports._original_route(o.binding, o.actor)
+    assert route["request_ref"] == o.current_ref and route["source"]["source_ref"] == o.actor["source_ref"]
+    assert o.transport.core.read_request(route["request_ref"])["session_id"] == sid
+    # The general owner-notification caller keeps its original route behavior.
+    assert o.transport.owner_attention._route(o.binding)["request_ref"] == o.old["request_ref"]
+    assert before[1:] == (o.store.messages(sid), o.capture.read_bytes(), list(o.provider.writes))
+
+
+@pytest.mark.parametrize("field,value", [("session_id", "0" * 32), ("source_ref", "0" * 24)])
+def test_report_route_never_falls_back_to_another_actor_or_source(rotated_steward, field, value):
+    o = rotated_steward
+    before = o.runtime.registry_path.read_bytes(), o.capture.read_bytes(), list(o.provider.writes)
+    with pytest.raises(ValueError):
+        o.transport.steward_reports.route_proof(o.binding, {**o.actor, field: value})
+    assert before == (o.runtime.registry_path.read_bytes(), o.capture.read_bytes(), list(o.provider.writes))
+
+
+@pytest.mark.parametrize("fault", ["wrong_app", "wrong_owner", "source_changed", "source_missing"])
+def test_matching_actor_still_requires_current_provider_and_source(rotated_steward, fault):
+    o = rotated_steward
+    if fault == "wrong_app":
+        o.provider.profile_apps["steward-app"] = "cli_foreign"
+    elif fault == "wrong_owner":
+        o.provider.owner_ready = False
+    elif fault == "source_changed":
+        o.provider.messages["om_original"]["chat_id"] = "oc_foreign"
+    else:
+        (o.transport.root / f"{o.current_ref}.json").unlink()
+    before = o.runtime.registry_path.read_bytes(), o.capture.read_bytes(), list(o.provider.writes)
+    with pytest.raises(ValueError):
+        o.reports.configure(o.binding["binding_id"], operation="enable", schedule=o.schedule)
+    assert before == (o.runtime.registry_path.read_bytes(), o.capture.read_bytes(), list(o.provider.writes))
+
+
+def test_report_route_uses_canonical_session_not_transport_copy(rotated_steward):
+    o = rotated_steward
+    path = o.transport.root / f"{o.current_ref}.json"
+    record = json.loads(path.read_text())
+    record["session_id"] = o.old["session_id"]
+    path.write_text(json.dumps(record))
+    route = o.transport.steward_reports._original_route(o.binding, o.actor)
+    assert route["request_ref"] == o.current_ref
+    assert o.transport.core.read_request(route["request_ref"])["session_id"] == o.actor["session_id"]
+
+
+def test_retained_report_route_cannot_fall_back_after_session_mismatch(rotated_steward):
+    o = rotated_steward
+    route = o.transport.steward_reports._original_route(o.binding, o.actor)
+    assert o.transport.steward_reports._original_route(o.binding, o.actor, route) == route
+    old_route = o.transport.owner_attention._route(o.binding)
+    before = o.capture.read_bytes(), list(o.provider.writes)
+    # A valid current route exists, but it cannot replace a frozen old route.
+    with pytest.raises(ValueError):
+        o.transport.steward_reports._original_route(o.binding, o.actor, old_route)
+    assert before == (o.capture.read_bytes(), list(o.provider.writes))