loopx-agent · Oct 18, 2026 · Oct 18, 2026 · Oct 18, 2026
@@ -2515,6 +2515,9 @@ const privateConversationSchema = z.object({
   goal_scope: z.enum(["all_registered", "selected"]).nullable().optional(),
   audience: z.enum(["owner", "group"]).default("owner"), group_count: z.number().int().default(0),
   pending_count: z.number().int(), recovery_count: z.number().int(),
+  ingress: z.object({status: z.enum(["unobserved", "verified", "gap", "unavailable"]),
+    last_checked_at: z.number().nullable(), last_complete_at: z.number().nullable(),
+    source_gap_count: z.number().int().nonnegative(), without_reply_count: z.number().int().nonnegative()}).optional(),
   owner_attention: z.discriminatedUnion("available", [
     z.object({available: z.literal(true), enabled: z.boolean(), revision: z.number().int(),
       last_checked_at: z.number().nullable(), error_code: z.string().nullable(),

@@ -103,6 +103,17 @@ export function PrivateConversationPanel() {
       {row.audience === "group" ? <p>{zh ? `群话题助手 · 已授权 ${row.group_count} 个群 · 独立工作区隔离` : `Group topic assistant · ${row.group_count} authorized groups · Isolated workspace`}</p> : null}
       <p>{row.executor_endpoint_id} · {zh ? "监听状态" : "Listener"}: {listenerLabel(row.listener_status)}</p>
       <p>{zh ? `待处理或回复：${row.pending_count}` : `Pending execution or reply: ${row.pending_count}`}</p>
+      {row.ingress ? <p role={row.ingress.status === "gap" || row.ingress.status === "unavailable" ? "alert" : "status"}>
+        {row.ingress.status === "gap"
+          ? (zh ? `${row.ingress.source_gap_count} 条原消息缺少收件记录，需核对；历史任务不会自动重放。`
+            : `${row.ingress.source_gap_count} source messages lack receipt records and need reconciliation. Historical tasks are not replayed.`)
+          : row.ingress.status === "unavailable"
+            ? (zh ? "原消息核对暂不可用；监听就绪不代表完整收件。" : "Source reconciliation is unavailable; a live connection does not prove complete receipt.")
+            : row.ingress.status === "verified" && row.ingress.last_complete_at !== null
+              ? (zh ? `原消息已核对 · ${new Date(row.ingress.last_complete_at * 1000).toLocaleTimeString(locale)}`
+                : `Sources reconciled · ${new Date(row.ingress.last_complete_at * 1000).toLocaleTimeString(locale)}`)
+              : (zh ? "原消息核对待运行。" : "Source reconciliation has not run yet.")}
+      </p> : null}
       {row.context_kind === "steward" && row.audience === "owner" ? <div>
         {row.owner_attention?.available ? <>
           <label style={{display: "flex", alignItems: "center", gap: 8, minHeight: 44}}><input type="checkbox" style={{width: "auto", margin: 0, flexShrink: 0}} checked={row.owner_attention.enabled} disabled={busy}

@@ -1058,6 +1058,28 @@ not scan and dispatch captured history. Consecutive concise requests and a
 correction must retain their own object, constraints and return lineage. Transport
 fixtures do not qualify receiver adoption or actual repair and merge.
 
+Owner-private Lark connections also compare bounded provider history with native
+capture receipts in the existing background delivery scanner. This default
+observation starts within one minute and repeats one minute after completion;
+it uses the same verified App and an already-admitted owner destination. The
+initial lookback is bounded to a day; later complete windows overlap by a minute.
+Each pass reads at most 20 pages with a five-second history-page request budget;
+identity verification retains its existing timeout.
+Incomplete, failed, stale or changed-binding reads cannot advance coverage.
+The private receipt retains every previously observed gap; absence from a later
+window does not close it. Group topics and other providers retain their own
+coverage contracts.
+
+The existing connection API and Lark settings show source coverage separately
+from live-listener readiness and pending execution. Its `ingress` observation
+uses `unobserved`, `verified`, `gap` and `unavailable`, with timestamps and counts,
+without source identifiers or bodies. An observed Bot reply does not establish
+native capture, a final answer or requester adoption. A newly observed unanswered
+gap may stop only this service's existing consumer once; the existing App lease
+and refresh lifecycle own its replacement. History never enters Core admission,
+replies, retries or failed-Turn recovery. Reconnection is a live-path mitigation,
+not proof that a missing historical request ran or that provider loss was fixed.
+
 The shared conversation intake resolves intent and verifies decision-relevant
 facts before delegation. A currently satisfied outcome returns its evidence
 without duplicate work or another protected-action proposal; a historical record

@@ -765,6 +765,22 @@ message，仅以该 App 身份下载属于这条消息的资源，再将 PNG/JPE
 原 Session、重复投递、持久重启与下载中撤权；真实 provider/model 验收另行记录。
 本增量不新增 Session authority、queue、worker 或默认关闭的功能开关。
 
+### 飞书私聊原消息核对
+
+本人私聊的默认后台同时核对飞书原消息与原生收件记录，复用既有投递扫描器、
+已核验 App 和已经准入的本人目的地；不新增消费者。首次在一分钟内开始，完成后
+一分钟再查；初次最多回看一天，后续完整窗口重叠一分钟，每轮最多 20 页、每页
+历史请求预算五秒，身份核验保留既有超时。分页未完、失败、过期或绑定变化均不推进覆盖时间。已发现缺口
+保留在私有回执中，不因后来窗口缺少该消息而消失；群话题与其他 provider 维持
+各自覆盖契约。
+
+原连接 API 与 Lark 设置将 `ingress` 原消息核对状态和实时监听、待执行数量分开，
+只展示未核对、已核对、有缺口或暂不可用，以及时间和计数，不返回原消息 ID 或正文。
+看到 Bot 回复不代表原生收件、最终答复或本人采纳。新发现且没有原 App 回复的缺口
+至多触发一次本服务自有消费者重连，由既有 App 租约与刷新生命周期启动替代者。
+历史消息不进入 Core 准入、发送、重试或失败 Turn 恢复；重连只缓解后续实时入口，
+不能证明历史请求已经执行或 provider 漏消息根因已修复。
+
 ### 飞书进度编辑预算与最终结果恢复
 
 私聊与管家委托复用已持久化的 Turn 事件更新同一条进度消息；完整结果仍由 Core

@@ -61,10 +61,26 @@ export const privateStewardScopeScenario = {
       await page.getByTestId("personal-goal-home").waitFor();
       await settings();
       await panel.getByText("LoopX 管家 · 全部已注册工作 · 150 个 Goal", {exact: true}).waitFor();
+      // The same populated settings journey must distinguish a ready stream
+      // from source coverage. Neither a gap nor failed observation is empty.
+      row.ingress = {status: "gap", last_checked_at: 1000, last_complete_at: 995,
+        source_gap_count: 1, without_reply_count: 1};
+      await page.reload(); await page.getByTestId("personal-goal-home").waitFor(); await settings();
+      await panel.getByRole("alert").filter({hasText: "1 条原消息缺少收件记录"}).waitFor();
+      await panel.getByText("待处理或回复：0", {exact: true}).waitFor();
+      await page.screenshot({path: resolve(outputDir, "private-ingress-gap-desktop.png"), animations: "disabled"});
       await page.setViewportSize({width: 390, height: 844});
       await panel.scrollIntoViewIfNeeded();
       await page.screenshot({path: resolve(outputDir, "private-steward-scope-mobile.png"), animations: "disabled"});
       if (await page.evaluate(() => document.documentElement.scrollWidth > innerWidth)) throw new Error("Scope readback overflows");
+      row.ingress.status = "unavailable";
+      await page.reload(); await page.getByTestId("personal-goal-home").waitFor(); await settings();
+      await panel.getByRole("alert").filter({hasText: "原消息核对暂不可用"}).waitFor();
+      await page.screenshot({path: resolve(outputDir, "private-ingress-unavailable-mobile.png"), animations: "disabled"});
+      row.ingress = {...row.ingress, status: "verified", source_gap_count: 0, without_reply_count: 0};
+      await page.reload(); await page.getByTestId("personal-goal-home").waitFor(); await settings();
+      await panel.getByRole("status").filter({hasText: "原消息已核对"}).waitFor();
+      if (await panel.getByRole("alert").count() || writes.length !== 2) throw new Error("Source readback must clear stale warnings without creating another write");
       const expectedErrors = context.errors.filter(error => error === "Failed to load resource: the server responded with a status of 500 (Internal Server Error)");
       if (expectedErrors.length !== 1 || context.errors.length !== 1) throw new Error(context.errors.join(" | "));
       return {coverageEntries: await context.close(), note: "Failed scope publication retains selected scope and allows keyboard retry; success clears the failure and survives reload; desktop/mobile, synthetic API paired with native backend fault regressions"};

@@ -87,6 +87,8 @@ def _reconcile_private(self) -> None:
         attempted: dict[Path, int] = {}
         attention: Future[int] | None = None
         attention_due = 0.0
+        ingress: Future[list[str]] | None = None
+        ingress_due = time.monotonic() + 60
         wakeup_scope = getattr(self.runtime_controller, "turn_release_wakeup", None)
         # Older/custom controllers retain polling. One local hint wakes the
         # existing scanner; it is neither a queue nor a delivery receipt.
@@ -96,6 +98,21 @@ def _reconcile_private(self) -> None:
                 # Clear BEFORE scanning: an arrival during the scan must also
                 # wake the following wait, rather than disappear in that gap.
                 self._delivery_wakeup.clear()
+                if ingress is not None and ingress.done():
+                    try:
+                        for profile in ingress.result():
+                            # Stop only this service's existing live worker.
+                            # Its finally/refresh path releases the App lease
+                            # before starting its replacement; Core work and
+                            # historical messages are never replayed.
+                            with self._lock:
+                                worker = self._workers.get(profile)
+                                if worker is not None:
+                                    worker[0].set()
+                    except Exception:
+                        logging.getLogger(__name__).warning("Private Chat source observation is unavailable")
+                    ingress = None
+                    ingress_due = time.monotonic() + 60
                 if attention is not None and attention.done():
                     try:
                         attention.result()
@@ -117,15 +134,21 @@ def _reconcile_private(self) -> None:
                     owner_attention = getattr(transport, "owner_attention", None)
                     attention_ready = (owner_attention is not None and attention is None
                         and time.monotonic() >= attention_due and owner_attention.has_enabled())
+                    observer = getattr(transport, "ingress", None)
+                    ingress_ready = (observer is not None and ingress is None
+                        and time.monotonic() >= ingress_due)
                     for path in sorted(paths, key=lambda path: attempted.get(path, 0)):
-                        if len(pending) + int(attention is not None or attention_ready) >= 4 or self._closed.is_set():
+                        if (len(pending) + int(attention is not None or attention_ready)
+                                + int(ingress is not None or ingress_ready) >= 4 or self._closed.is_set()):
                             break
                         if path in pending:
                             continue
                         pending[path] = pool.submit(transport.reconcile_request, path)
                         attempted[path] = attempted.get(path, 0) + 1
-                    if attention_ready and len(pending) < 4 and not self._closed.is_set():
+                    if attention_ready and len(pending) + int(ingress is not None or ingress_ready) < 4 and not self._closed.is_set():
                         attention = pool.submit(owner_attention.reconcile)
+                    if ingress_ready and len(pending) + int(attention is not None) < 4 and not self._closed.is_set():
+                        ingress = pool.submit(observer.observe, cancelled=self._closed.is_set)
                 except Exception:
                     logging.getLogger(__name__).warning("Private Chat delivery discovery is pending")
                 self._delivery_wakeup.wait(1)

@@ -37,6 +37,8 @@ def _private_conversations(self) -> None:
         titles = {row["project_ref"]: Path(row["workspace_path"]).name
                   for row in self.server.runtime_controller.project_contexts.available()}
         health = self.server.lark_goal_topic_runtime.health_snapshot()
+        composed = getattr(self.server, "conversation_transports", None)
+        lark_profiles = set(health) - set(composed.transports if composed is not None else {})
         deliveries = self.server.lark_private_conversations.health()
         self._send_json({"ok": True, "revision": current["revision"], "connections": [
             {"binding_id": row["binding_id"], "app_ref": row["transport_ref"], "context_kind": row["context_kind"],
@@ -49,6 +51,10 @@ def _private_conversations(self) -> None:
                 for item in bindings.agent_candidates(row["binding_id"])],
              "agent_targets": row.get("agent_targets", []),
              "listener_status": self._private_listener_status(row["transport_ref"], health),
+             **({"ingress": self.server.lark_private_conversations.ingress.health(row["binding_id"])}
+                if row.get("audience", "owner") == "owner"
+                and row["transport_ref"] in lark_profiles
+                and hasattr(self.server.lark_private_conversations, "ingress") else {}),
              **({"owner_attention": self.server.lark_private_conversations.owner_attention.health(row["binding_id"])}
                 if row["context_kind"] == "steward" and row.get("audience", "owner") == "owner" else {}),
              **deliveries.get(row["binding_id"], {"pending_count": 0, "recovery_count": 0})}

@@ -53,6 +53,8 @@ def __init__(self, *, controller: Any, runtime_root: Path, runner: Any, cli_bin:
         self.root = controller.store.root / "lark-private-deliveries"
         from .owner_attention import LarkOwnerAttention
         self.owner_attention = LarkOwnerAttention(self)
+        from .private_ingress import PrivateIngressObserver
+        self.ingress = PrivateIngressObserver(self)
 
     def profiles(self) -> dict[str, dict[str, str]]:
         """Discover current App configuration, never authorize a conversation.