diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 8a1ae9edde..9a0ba85b1a 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -39,6 +39,16 @@ 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. +The public browser fixture also keeps the original private outbound owner +running through its periodic source-history check. Its shared synthetic +provider returns the stored messages with source-bound pagination and time +windows; it does not replace history with an empty successful response. +The observation includes worker liveness, history-page count and ingress +coverage so a background exception cannot be mistaken for a completed journey. +File and SQLite regressions retain all 55 synthetic uncaptured sources without +admitting or replying to them. These qualify test mechanics; live delivery, +account switching and an unknown-effect failed production window remain separate. + | 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) | diff --git a/examples/personal-workspace-browser/steward_report_fixture.py b/examples/personal-workspace-browser/steward_report_fixture.py index 539070fa1d..ad4b625b2c 100644 --- a/examples/personal-workspace-browser/steward_report_fixture.py +++ b/examples/personal-workspace-browser/steward_report_fixture.py @@ -139,6 +139,9 @@ def fail_report(_adapter, _prompt, event_sink): "model_call_count": sum(json.loads(line).get("method") == "turn/start" for line in capture.read_bytes()[len(before_calls):].splitlines()), "provider_writes": len(provider.writes), + "private_outbound_alive": delivery.is_alive() if delivery else None, + "private_ingress": transport.ingress.health(binding["binding_id"]), + "provider_history_pages": sum("+chat-messages-list" in call for call in provider.calls), **({"managed_source_present": artifact.exists(), "managed_session": store.load_session(managed_session["session_id"])} if args.managed_session else {}), **({"managed_report": True, diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 8a1cea6127..62bc95203a 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -5,6 +5,7 @@ import http.client import threading from concurrent.futures import ThreadPoolExecutor +from datetime import datetime from types import SimpleNamespace import pytest @@ -41,7 +42,8 @@ def event(self, profile, name, text, kind="text"): "sender_type": "user", "sender_id": f"ou_{profile.replace('-', '_')}", "message_type": kind, "content": text} self.messages[event["message_id"]] = {"message_id": event["message_id"], "chat_id": event["chat_id"], - "sender": {"id": event["sender_id"], "sender_type": "user"}, "msg_type": kind, "content": text} + "sender": {"id": event["sender_id"], "sender_type": "user"}, "msg_type": kind, "content": text, + "create_time": str(int(time.time() * 1000))} return event def topic(self, name, text, *, root=None, chat="oc_community", sender="ou_member", addressed=True): @@ -71,6 +73,37 @@ def __call__(self, args, cwd=None, timeout=None): elif "+chat-list" in args: data = {"ok": True, "data": {"chats": [{"chat_id": chat, "name": title} for chat, title in [("oc_community", "Community trial"), ("oc_second", "Second trial")]]}} + elif "+chat-messages-list" in args: + # Exercise the existing ingress owner against actual synthetic + # history, including its time windows and continuation tokens. + chat = args[args.index("--chat-id") + 1] + if chat != f"oc_{profile.replace('-', '_')}": + raise ValueError("synthetic private history belongs to another profile") + start = args[args.index("--start") + 1] if "--start" in args else None + end = args[args.index("--end") + 1] + order = args[args.index("--order") + 1] + if order not in {"asc", "desc"}: + raise ValueError("unsupported synthetic history order") + lower = datetime.fromisoformat(start).timestamp() * 1000 if start else float("-inf") + upper = datetime.fromisoformat(end).timestamp() * 1000 + query = [profile, chat, start, end, order] + offset = 0 + if "--page-token" in args: + cursor = json.loads(args[args.index("--page-token") + 1]) + if (not isinstance(cursor, list) or len(cursor) != 6 or cursor[:5] != query + or type(cursor[5]) is not int or cursor[5] < 0): + raise ValueError("synthetic history cursor does not match its source window") + offset = cursor[5] + size = int(args[args.index("--page-size") + 1]) + if not 1 <= size <= 50: + raise ValueError("invalid synthetic history page size") + rows = [dict(row) for row in tuple(self.messages.values()) + if row["chat_id"] == chat and lower <= int(row["create_time"]) <= upper] + rows.sort(key=lambda row: (int(row["create_time"]), row["message_id"]), reverse=order == "desc") + page = rows[offset:offset + size] + more = offset + size < len(rows) + data = {"ok": True, "identity": "bot", "data": {"messages": page, "has_more": more, + "page_token": json.dumps([*query, offset + size]) if more else None}} elif "api" in args and "GET" in args: path = args[args.index("GET") + 1] assert path.startswith("/open-apis/im/v1/messages/") @@ -113,7 +146,8 @@ def __call__(self, args, cwd=None, timeout=None): target = self.messages[args[args.index("--message-id") + 1]] if "+messages-reply" in args else None self.messages[ref] = {"message_id": ref, "msg_type": kind, "body": {"content": content}, "chat_id": target["chat_id"] if target else args[args.index("--chat-id") + 1], - "sender": {"id": self.profile_apps[profile], "sender_type": "app"}} + "sender": {"id": self.profile_apps[profile], "sender_type": "app"}, + "create_time": str(int(time.time() * 1000))} if target: assert "--reply-in-thread" in args root = target.get("root_id") or target["message_id"] @@ -148,6 +182,31 @@ def __call__(self, args, cwd=None, timeout=None): return {"returncode": 0, "stdout": json.dumps(data), "stderr": ""} +def test_synthetic_private_history_preserves_window_pages_and_audience(): + provider = Provider() + for profile in ("notes-app", "steward-app"): + for index in range(5): + event = provider.event(profile, f"{profile}-{index}", f"source {index}") + provider.messages[event["message_id"]]["create_time"] = str((100 + index) * 1000) + args = ["lark-cli", "--profile", "steward-app", "im", "+chat-messages-list", "--as", "bot", + "--chat-id", "oc_steward_app", "--start", "1970-01-01T00:01:41+00:00", + "--end", "1970-01-01T00:01:43+00:00", "--order", "asc", "--page-size", "2"] + first = json.loads(provider(args)["stdout"])["data"] + assert [row["message_id"] for row in first["messages"]] == ["om_steward-app-1", "om_steward-app-2"] + assert first["has_more"] + second = json.loads(provider([*args, "--page-token", first["page_token"]])["stdout"])["data"] + assert [row["message_id"] for row in second["messages"]] == ["om_steward-app-3"] + assert not second["has_more"] and second["page_token"] is None + wrong_window = list(args) + wrong_window[wrong_window.index("--end") + 1] = "1970-01-01T00:01:44+00:00" + with pytest.raises(ValueError, match="cursor does not match"): + provider([*wrong_window, "--page-token", first["page_token"]]) + foreign = list(args) + foreign[foreign.index("--chat-id") + 1] = "oc_notes_app" + with pytest.raises(ValueError, match="another profile"): + provider(foreign) + + def connect(fixture): store, runtime, contexts, _, _, _, _ = fixture provider = Provider() diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index 93c12b5a02..5523518ccf 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -1023,6 +1023,9 @@ def test_original_due_report_freezes_once_and_returns_full_body_after_native_set sid, actor, reports, _ = configured(steward, tmp_path, monkeypatch, provider) install_model(steward, tmp_path, sid, result_kind=result_kind) calls, writes, history = capture.read_bytes(), list(bot.writes), store.messages(sid) + assert transport.ingress.observe() == [] + assert transport.ingress.health(binding["binding_id"])["status"] == "verified" + assert (capture.read_bytes(), bot.writes) == (calls, writes) thread = store.load_session(sid)["upstream_thread_id"] turns = pump_steward_monitors(runtime) assert len(turns) == 1 and turns[0]["session_id"] == sid @@ -1058,12 +1061,36 @@ def test_original_due_report_freezes_once_and_returns_full_body_after_native_set assert proof["verified"] and proof["text"] == body.strip() assert len(body.encode()) > 40000 # Exercise the original multipart full-body sender. after = capture.read_bytes(), list(bot.writes) + assert transport.ingress.observe() == [] + assert transport.ingress.health(binding["binding_id"])["status"] == "verified" runtime._steward_monitor_next_check.clear() assert not pump_steward_monitors(runtime) assert transport.steward_reports.reconcile() == 0 assert after == (capture.read_bytes(), bot.writes) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_report_fixture_ingress_reads_all_pages_without_admitting_missed_business( + steward, tmp_path, monkeypatch, provider): + import time + + store, runtime, bot, transport, binding, capture, _ = steward + sid, _, _, _ = configured(steward, tmp_path, monkeypatch, provider) + for index in range(55): + event = bot.event("steward-app", f"missed-{index}", f"Uncaptured private source {index}") + bot.messages[event["message_id"]]["create_time"] = str(int((time.time() - 30) * 1000)) + before = store.messages(sid), capture.read_bytes(), list(bot.writes) + assert transport.ingress.observe() == ["steward-app"] + health = transport.ingress.health(binding["binding_id"]) + assert health["status"] == "gap" and health["source_gap_count"] == health["without_reply_count"] == 55 + assert transport.ingress.observe() == [] # Observation does not replay an effect or reconnect twice. + assert before == (store.messages(sid), capture.read_bytes(), bot.writes) + pages = [args for args in bot.calls if "+chat-messages-list" in args] + assert any("--page-token" in args for args in pages) + receipt = transport.ingress._path(binding["binding_id"]).read_text() + assert "Uncaptured private source" not in receipt + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_report_collects_evidence_before_capturing_native_launch_authority(steward, tmp_path, monkeypatch, provider): store, runtime, _, transport, binding, capture, _ = steward