From 000e4ca3dcf3e39875a8f9911d31423b398dad22 Mon Sep 17 00:00:00 2001 From: song Date: Mon, 26 Oct 2026 10:57:52 +0800 Subject: [PATCH] fix(lark): select report routes for the prepared steward session Filter original-route candidates by the managed actor's canonical Session and source before selection. Preserve provider checks, saved-route identity and default owner notification routing, with File/SQLite regressions. Related to #7179. Signed-off-by: Codex --- docs/product/periodic-reports.md | 7 ++ loopx/extensions/lark/owner_attention.py | 9 +- loopx/extensions/lark/steward_reports.py | 8 +- tests/test_lark_steward_report_route.py | 116 +++++++++++++++++++++++ 4 files changed, 136 insertions(+), 4 deletions(-) create mode 100644 tests/test_lark_steward_report_route.py diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 12950c510e..d5362380a5 100644 --- a/docs/product/periodic-reports.md +++ b/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 diff --git a/loopx/extensions/lark/owner_attention.py b/loopx/extensions/lark/owner_attention.py index f39759ae74..49ba8bc426 100644 --- a/loopx/extensions/lark/owner_attention.py +++ b/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): diff --git a/loopx/extensions/lark/steward_reports.py b/loopx/extensions/lark/steward_reports.py index 38ae605a5d..fed02e40c8 100644 --- a/loopx/extensions/lark/steward_reports.py +++ b/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")) diff --git a/tests/test_lark_steward_report_route.py b/tests/test_lark_steward_report_route.py new file mode 100644 index 0000000000..9dcaf099fc --- /dev/null +++ b/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))