test: qualify report source-history and delivery owners together

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) |

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,

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()

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