diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts
index cc4a4a62c6..b83e54bf7f 100644
--- a/apps/presentation/dashboard/src/data/chat.ts
+++ b/apps/presentation/dashboard/src/data/chat.ts
@@ -2377,3 +2377,28 @@ export async function updateGoalOwnership(body: { goal_id: string; mode: Executi
     throw error;
   }
 }
+
+
+const privateConversationSchema = z.object({
+  binding_id: z.string(), app_ref: z.string(), context_kind: z.literal("project"),
+  project_ref: z.string(), project_title: z.string(), context_available: z.boolean(), executor_endpoint_id: z.string(),
+  grant: z.literal("workspace_read"), listener_status: z.string(),
+  pending_count: z.number().int(), recovery_count: z.number().int(),
+});
+const privateConversationsSchema = z.object({ok: z.literal(true), revision: z.number().int(),
+  connections: z.array(privateConversationSchema)});
+export type PrivateConversation = z.infer<typeof privateConversationSchema>;
+export async function fetchPrivateConversations() {
+  return privateConversationsSchema.parse(await requestJson<unknown>("/api/chat/lark/private-conversations"));
+}
+export async function connectPrivateConversation(appRef: string, projectRef: string, executor: string) {
+  return privateConversationsSchema.parse(await requestJson<unknown>("/api/chat/lark/private-conversations", {
+    method: "POST", headers: {"Content-Type": "application/json"},
+    body: JSON.stringify({app_ref: appRef, project_ref: projectRef, executor_endpoint_id: executor}),
+  }));
+}
+export async function disconnectPrivateConversation(bindingId: string, revision: number) {
+  return privateConversationsSchema.parse(await requestJson<unknown>("/api/chat/lark/private-conversations", {
+    method: "DELETE", headers: {"Content-Type": "application/json"}, body: JSON.stringify({binding_id: bindingId, revision}),
+  }));
+}
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/lark-settings-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/lark-settings-page.tsx
index 07ef1c6b8a..cade2c603e 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/lark-settings-page.tsx
+++ b/apps/presentation/dashboard/src/features/personal-workspace/lark-settings-page.tsx
@@ -1,3 +1,4 @@
+import {PrivateConversationPanel} from "./private-conversation-panel";
 import { useEffect, useMemo, useRef, useState } from "react";
 import {
   Bot,
@@ -539,6 +540,7 @@ export function LarkSettingsPage({
         </header>
       )}
 
+      <PrivateConversationPanel />
       <nav className="personal-lark-tabs" aria-label={t("lark.management")}>
         <button aria-current={tab === "apps" ? "page" : undefined} onClick={() => setTab("apps")} type="button">{t("lark.apps")} <span>{loading ? "…" : apps.length}</span></button>
         <button aria-current={tab === "connections" ? "page" : undefined} onClick={() => setTab("connections")} type="button">{t("lark.connections")} <span>{loading ? "…" : connections.length}</span></button>
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx b/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx
new file mode 100644
index 0000000000..066e903c50
--- /dev/null
+++ b/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx
@@ -0,0 +1,85 @@
+import {useEffect, useState} from "react";
+import {connectPrivateConversation, disconnectPrivateConversation, fetchPrivateConversations,
+  fetchChatProjects, fetchChatCapabilities, fetchLarkApps, type PrivateConversation,
+  type ChatProject, type LarkApp} from "../../data/chat";
+import {useWorkspaceI18n} from "./i18n";
+import "./private-conversation.css";
+
+export function PrivateConversationPanel() {
+  const {locale} = useWorkspaceI18n();
+  const zh = locale === "zh-CN";
+  const [apps, setApps] = useState<LarkApp[]>([]);
+  const [projects, setProjects] = useState<ChatProject[]>([]);
+  const [executors, setExecutors] = useState<string[]>([]);
+  const [rows, setRows] = useState<PrivateConversation[]>([]);
+  const [revision, setRevision] = useState(0);
+  const [app, setApp] = useState("");
+  const [project, setProject] = useState("");
+  const [executor, setExecutor] = useState("");
+  const [error, setError] = useState("");
+  const [busy, setBusy] = useState(false);
+
+  async function refresh() {
+    const state = await fetchPrivateConversations();
+    setRows(state.connections); setRevision(state.revision);
+  }
+  useEffect(() => {
+    let active = true;
+    Promise.all([fetchLarkApps(), fetchChatProjects(), fetchChatCapabilities(), fetchPrivateConversations()])
+      .then(([apps, projects, capabilities, state]) => {
+        if (!active) return;
+        setApps(apps.filter(app => app.ready && app.app_ref !== "default"));
+        setProjects(projects.projects);
+        const choices = (capabilities.adapters ?? []).filter(row => row.available).map(row => row.agent_id);
+        setExecutors(choices); setExecutor(choices.includes("codex") ? "codex" : choices[0] ?? "");
+        if (projects.projects.length === 1) setProject(projects.projects[0].project_ref);
+        setRows(state.connections); setRevision(state.revision);
+      }).catch(error => {if (active) setError(String(error));});
+    const timer = setInterval(() => {fetchPrivateConversations().then(state => {
+      if (active) {setRows(state.connections); setRevision(state.revision);}
+    }).catch(() => {});}, 5000);
+    return () => {active = false; clearInterval(timer);};
+  }, []);
+
+  function listenerLabel(state: string) {
+    const labels: Record<string, [string, string]> = {starting: ["启动中", "Starting"], listening: ["实时连接就绪", "Live connection ready"],
+      standby: ["等待已有监听服务", "Waiting for the existing listener"], retrying: ["重连中", "Reconnecting"],
+      stopped: ["已停止", "Stopped"], inactive: ["配置未启用", "Inactive"]};
+    return labels[state]?.[zh ? 0 : 1] ?? (zh ? "连接尚未确认" : "Connection unconfirmed");
+  }
+
+  async function act(operation: () => Promise<unknown>) {
+    setBusy(true); setError("");
+    try {await operation(); await refresh();} catch (error) {setError(String(error));}
+    finally {setBusy(false);}
+  }
+  return <section className="personal-detail-card personal-private-conversation" aria-label={zh ? "本人飞书私聊" : "Owner private Chat"}>
+    <h3>{zh ? "本人私聊 · 项目对话" : "Owner private Chat · Project conversation"}</h3>
+    <p>{zh ? "每个 App 单独核验登录本人，只读讨论所选工作区。普通私聊不会创建 Goal。" : "Verify the logged-in owner independently for each App. Discuss the selected workspace with a read grant; ordinary private Chat creates no Goal."}</p>
+    {rows.map(row => <article key={row.binding_id}>
+      <strong>{row.app_ref} · {row.context_available ? row.project_title : (zh ? "工作区不可用" : "Workspace unavailable")}</strong>
+      <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.recovery_count > 0 ? <p role="status">{zh ? "存在尚未确认的发送回执。服务会读取原回执恢复；不要重新发送同一任务。检查 App 登录、权限和原会话后刷新状态。" : "A send receipt is unconfirmed. The service reads the original receipt to recover; avoid resending the same task. Check this App login, permissions and original Session, then refresh status."}</p> : null}
+      {!row.context_available ? <p role="alert">{zh ? "工作区授权已失效；请恢复原工作区或重新选择。旧会话不会移到其它工作区。" : "The workspace grant is unavailable. Restore the original workspace or select a new one; the old Session will not move."}</p> : null}
+      <button disabled={busy} onClick={() => void act(() => disconnectPrivateConversation(row.binding_id, revision))} type="button">{zh ? "解绑" : "Disconnect"}</button>
+    </article>)}
+    {rows.length === 0 ? <p>{zh ? "尚未连接本人私聊。" : "No owner private Chat connected."}</p> : null}
+    <label>App<select aria-label={zh ? "私聊 App" : "Private Chat App"} value={app} disabled={busy} onChange={event => setApp(event.target.value)}>
+      <option value="">{zh ? "选择已验证 App" : "Select a verified App"}</option>
+      {apps.map(app => <option key={app.app_ref} value={app.app_ref}>{app.label} · {app.app_ref}</option>)}
+    </select></label>
+    <label>{zh ? "工作区" : "Workspace"}<select aria-label={zh ? "私聊工作区" : "Private Chat workspace"} value={project} disabled={busy} onChange={event => setProject(event.target.value)}>
+      <option value="">{zh ? "选择授权工作区" : "Select an authorized workspace"}</option>
+      {projects.map(project => <option key={project.project_ref} value={project.project_ref}>{project.title}</option>)}
+    </select></label>
+    <label>{zh ? "执行器" : "Executor"}<select aria-label={zh ? "私聊执行器" : "Private Chat executor"} value={executor} disabled={busy} onChange={event => setExecutor(event.target.value)}>
+      {executors.map(executor => <option key={executor} value={executor}>{executor}</option>)}
+    </select></label>
+    <div className="personal-detail-actions"><button disabled={busy || !app || !project || !executor} onClick={() => void act(() => connectPrivateConversation(app, project, executor))} type="button">
+      {busy ? (zh ? "正在核验" : "Verifying") : (zh ? "连接本人私聊" : "Connect owner private Chat")}</button>
+      <button disabled={busy} onClick={() => void act(refresh)} type="button">{zh ? "刷新状态" : "Refresh status"}</button></div>
+    <p>{zh ? "从手机发送文字开始；后续消息进入原会话队列。/status 查看状态，/stop 停止当前执行，/new 开启新会话。图片、文件会明确提示暂不支持。" : "Send text from your phone to begin; follow-ups queue in the same Session. /status checks state, /stop stops the current Turn, /new starts a new conversation. Images and files receive an explicit unsupported response."}</p>
+    {error ? <p role="alert">{error}</p> : null}
+  </section>;
+}
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/private-conversation.css b/apps/presentation/dashboard/src/features/personal-workspace/private-conversation.css
new file mode 100644
index 0000000000..30670e9a36
--- /dev/null
+++ b/apps/presentation/dashboard/src/features/personal-workspace/private-conversation.css
@@ -0,0 +1,20 @@
+.personal-private-conversation { display: grid; gap: 18px; }
+.personal-private-conversation > p { margin: 0; }
+.personal-private-conversation label { display: grid; gap: 8px; font-size: 13px; }
+.personal-private-conversation select {
+  width: 100%; min-height: 40px; padding: 10px 12px;
+  border: 1px solid var(--pw-line-strong); border-radius: 8px;
+  background: var(--pw-card); color: var(--pw-text); font: inherit;
+}
+.personal-private-conversation .personal-detail-actions { display: flex; flex-wrap: wrap; gap: 8px; }
+.personal-private-conversation button {
+  justify-self: start; min-height: 36px; padding: 8px 12px;
+  border: 1px solid var(--pw-line-strong); border-radius: 8px;
+  background: var(--pw-card); color: var(--pw-text); cursor: pointer; font-size: 13px;
+}
+.personal-private-conversation button:disabled { color: var(--pw-muted); cursor: default; opacity: .6; }
+.personal-private-conversation :is(button, select):focus-visible { outline: 2px solid var(--pw-blue); outline-offset: 2px; }
+.personal-private-conversation form { display: grid; gap: 12px; }
+.personal-private-conversation form button[type="submit"] { background: var(--pw-text); color: var(--pw-card); }
+.personal-private-conversation [role="alert"] { color: var(--pw-red); }
+.personal-private-conversation [role="status"]:empty { display: none; }
diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
index 06ce03f170..34dbbeabd1 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -30,11 +30,49 @@ the TypeScript owner decides context identity and scope. Native Codex resume ret
 the original upstream thread and workspace. HTTP/protocol fixtures qualify that
 continuity and denial behavior; they do not establish real model adoption.
 
-The initial grant is workspace reading for the local owner. Lark audience grants,
-ordinary private-message selection, durable inbound admission independent of
-terminal delivery, and installed/mobile acceptance remain open work in this RFC.
-The App scope does not qualify those journeys or authorize edits; a revoked grant
-keeps the history readable and blocks new messages until the host grants it again.
+The App grant is workspace reading for the local owner. The App scope alone does
+not qualify Lark private-message admission, installed or mobile journeys, or
+authorize edits. A revoked grant keeps the history readable and blocks new
+messages until the host grants it again. The independent Lark checkpoint below
+qualifies its source implementation separately.
+
+The source implementation also has an explicit owner-only Lark private-message
+binding. Settings → Lark selects an independently verified non-default App,
+a current host workspace and executor. Core owns the App-scoped owner/source
+identity, Session context and revocation; the provider stores no separate
+conversation authority. One machine/App lease covers both profile aliases and
+existing group listeners. Replies recheck the original source under that App;
+they do not require group-member scopes for private messages.
+
+Text admission persists the canonical Core Turn without waiting for its terminal
+answer. The existing bounded Session queue keeps follow-ups in FIFO order; another
+App can continue independently. Explicit `/status`, `/stop` and `/new` use the
+shared request owner, and a replay retains the original Session/Turn target.
+A lost admission correlation cannot move a request to a newer Session. Admission
+feedback and final delivery use separate durable provider intents; an ambiguous
+write is read back without blind resend. Unsupported media receives an explicit
+notice. This grant remains workspace reading, without a Goal, portfolio or peer
+execution authority.
+
+The packaged settings journey, source revocation/recovery, duplicate events,
+native queue/stop and two-App isolation have synthetic-provider regression and
+browser coverage. A native Codex source canary separately qualifies distinct
+upstream threads with independently observed real App identities. Neither is a
+live Lark/model/mobile result. Installed service qualification, phone journeys,
+registered Agent selection, real incremental/media/permission interactions and
+a separately granted long-running steward remain open acceptance. Runtime and
+permission-boundary changes require maintainer review before promotion.
+
+The private setup UI composes within Settings → Lark; the App workspace scope
+remains the sole ordinary local conversation entry. Legacy group profiles without
+a stored App identity retain their original profile-hash consumer lease. Requests
+without private bindings do not add provider authentication; configured aliases
+reuse one request-scoped verified identity observation. Lark HTTP composition
+resides in the extension, while the typed binding owner remains provider-neutral.
+
+Synthetic product previews: [desktop](../../assets/personal-workspace/private-project-conversations.png),
+[narrow](../../assets/personal-workspace/private-project-conversations-narrow.png),
+[revoked workspace](../../assets/personal-workspace/private-project-workspace-revoked.png).
 
 ## Decision: make the App the place where work conversations continue
 
diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
index fac2387a73..5dccae4bd4 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
@@ -25,10 +25,33 @@ grant 授权 peer delegation。
 store，TypeScript 负责上下文身份及范围。原生 Codex 恢复保留原 upstream thread
 和工作区；HTTP/协议 fixture 验证连续性与拒绝行为，不证明真实模型采用了上下文。
 
-首个 grant 仅面向本机 owner 的工作区读取。Lark 受众授权、普通私聊选择、独立于
-terminal delivery 的 durable 入站 admission，以及安装/手机验收仍是本 RFC 的未完成项。
-App 范围入口不代表这些旅程已通过，也不授权修改文件；grant 撤销后历史仍可读，
-新消息在宿主重新授权前被阻止。
+App grant 仅面向本机 owner 的工作区读取。App 范围入口本身不证明 Lark 私聊
+admission、安装或手机旅程已通过，也不授权修改文件；grant 撤销后历史仍可读，
+新消息在宿主重新授权前被阻止。下方独立检查点说明 Lark 的源码实现资格。
+
+本机 owner 的只读工作区入口现已接入独立核验的 Lark App 私聊绑定。既有设置 →
+Lark 页面选择一个非默认 App、当前可用工作区和宿主 executor，读回监听状态、待回复
+数量及恢复缺口；重新绑定不会把旧 Session 移到其它工作区。
+
+共享 typed Core 核验 App、本人、受众和当前工作区 grant，拥有稳定 Session、原生
+Turn FIFO 和确切 stop/new/status 目标。传输层保存独立的受理与最终投递意图，通过
+provider 读回确认回复；发生没有 receipt 的不确定写入时不盲目重发。两 App 复用
+现有服务及按 App 获取的 listener lease，不创建第二套队列、模型 runner 或隐藏 Goal。
+
+源码 canary 已验证打包页面的桌面/窄屏、撤权拒绝与恢复、慢会话期间另一 App 完成、
+后续消息持久排队、exact stop，以及重启后原会话恢复和已确认回复不重复发送。
+这些是合成 provider/协议验收；真实原生 Codex 另行验证独立线程与上下文隔离。
+真实 Lark 收发、安装候选、手机旅程、注册 Agent 选择、媒体/增量/权限回调，以及
+新管家的明确长期委托仍未验收。当前私聊绑定仅支持普通只读项目会话。
+
+私聊配置复用设置 → Lark；App 范围仍是本机普通对话的唯一入口。未存储 App 身份
+的旧群聊 profile 保留原 profile-hash 监听锁键。没有私聊绑定时不增加鉴权；有绑定
+时，同一请求内的别名检查与连接共用一次已验证身份观测。飞书 HTTP 组合位于
+extension，typed binding owner 继续保持 provider-neutral。
+
+![合成私聊工作区设置](../../assets/personal-workspace/private-project-conversations.png)
+![窄屏私聊设置](../../assets/personal-workspace/private-project-conversations-narrow.png)
+![工作区撤权读回](../../assets/personal-workspace/private-project-workspace-revoked.png)
 
 ## 决策：让 App 成为工作会话持续进行的地方
 
diff --git a/docs/assets/personal-workspace/private-project-conversations-narrow.png b/docs/assets/personal-workspace/private-project-conversations-narrow.png
new file mode 100644
index 0000000000..550edff047
Binary files /dev/null and b/docs/assets/personal-workspace/private-project-conversations-narrow.png differ
diff --git a/docs/assets/personal-workspace/private-project-conversations.png b/docs/assets/personal-workspace/private-project-conversations.png
new file mode 100644
index 0000000000..f5a9f9acd4
Binary files /dev/null and b/docs/assets/personal-workspace/private-project-conversations.png differ
diff --git a/docs/assets/personal-workspace/private-project-workspace-revoked.png b/docs/assets/personal-workspace/private-project-workspace-revoked.png
new file mode 100644
index 0000000000..a7ce17a66b
Binary files /dev/null and b/docs/assets/personal-workspace/private-project-workspace-revoked.png differ
diff --git a/loopx/capabilities/native_chat/conversation_bindings.py b/loopx/capabilities/native_chat/conversation_bindings.py
new file mode 100644
index 0000000000..6ff7abbcf2
--- /dev/null
+++ b/loopx/capabilities/native_chat/conversation_bindings.py
@@ -0,0 +1,92 @@
+"""Private IO for Core-owned conversation context and audience grants.
+
+Provider observations contain opaque App/owner references, not credentials.
+The typed owner validates configuration, replacement, revocation, and use.
+This file stores no conversation text and creates no Goal or model thread.
+"""
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+import uuid
+
+from ...chat_store import _atomic_write_json, _read_json
+from ...control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
+from ...file_lock import exclusive_file_lock
+
+
+class ChatConversationBindings:
+    def __init__(self, *, root: Path, project_contexts: Any,
+                 observe: Callable[[str], dict[str, Any]]) -> None:
+        self.path = root / "conversation-bindings.json"
+        self.projects = project_contexts
+        self.observe = observe
+
+    def read(self) -> dict[str, Any]:
+        if not self.path.exists():
+            return {"schema_version": "loopx_chat_conversation_bindings_v0", "revision": 0, "bindings": []}
+        # An unreadable existing grant is not an empty configuration to overwrite.
+        return _read_json(self.path)
+
+    @staticmethod
+    def _core(operation: str, params: dict[str, Any]) -> dict[str, Any]:
+        try:
+            result = effect_runtime_result(operation, params)
+        except EffectRuntimeRejected as exc:
+            raise ValueError(str(exc)) from exc
+        if not isinstance(result, dict):
+            raise ValueError("conversation authority returned an invalid result")
+        return result
+
+    def configure(self, *, transport_ref: str, project_ref: str,
+                  executor_endpoint_id: str) -> dict[str, Any]:
+        observation = self.observe(transport_ref)
+        candidate = {
+            "schema_version": "loopx_chat_conversation_binding_v0",
+            "binding_id": uuid.uuid4().hex[:24], "transport_ref": transport_ref,
+            "provider_ref": observation["provider_ref"], "operator_ref": observation["operator_ref"],
+            "context_kind": "project", "project_ref": project_ref,
+            "executor_endpoint_id": executor_endpoint_id, "grant": "workspace_read", "enabled": True,
+        }
+        with exclusive_file_lock(self.path, operation="configure_chat_conversation_binding"):
+            current = self.read()
+            result = self._core("collaboration.conversation.binding", {
+                "current": current, "expected_revision": current.get("revision"), "operation": "configure",
+                "binding": candidate, "observation": observation, "available_projects": self.projects.available(),
+            })
+            if result["changed"]:
+                _atomic_write_json(self.path, result["state"])
+            readback = self.read()
+            if readback != result["state"]:
+                raise OSError("conversation binding publication did not verify")
+            return next(row for row in readback["bindings"] if row["transport_ref"] == transport_ref)
+
+    def disconnect(self, binding_id: str, *, expected_revision: int) -> dict[str, Any]:
+        with exclusive_file_lock(self.path, operation="disconnect_chat_conversation_binding"):
+            result = self._core("collaboration.conversation.binding", {
+                "current": self.read(), "expected_revision": expected_revision,
+                "operation": "disconnect", "binding_id": binding_id,
+            })
+            if result["changed"]:
+                _atomic_write_json(self.path, result["state"])
+            if self.read() != result["state"]:
+                raise OSError("conversation binding revocation did not verify")
+            return dict(result["state"])
+
+    def resolve(self, *, binding_id: str, source_ref: str, sender_ref: str,
+                private_human_message: bool, session_context: dict[str, Any] | None = None) -> dict[str, Any]:
+        current = self.read()
+        row = next((item for item in current.get("bindings", []) if item.get("binding_id") == binding_id), None)
+        if row is None:
+            raise ValueError("conversation binding is no longer authorized")
+        return self._core("collaboration.conversation.bound_context", {
+            "current": current, "binding_id": binding_id, "source_ref": source_ref, "sender_ref": sender_ref,
+            "private_human_message": private_human_message, "observation": self.observe(row["transport_ref"]),
+            "available_projects": self.projects.available(),
+            **({"session_context": session_context} if session_context is not None else {}),
+        })
+
+    def session_context(self, saved: dict[str, Any]) -> dict[str, Any]:
+        return self.resolve(binding_id=saved["binding_id"], source_ref=saved["source_ref"],
+                            sender_ref=saved["operator_ref"], private_human_message=True, session_context=saved)
diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
new file mode 100644
index 0000000000..dff21378f7
--- /dev/null
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -0,0 +1,126 @@
+"""Native Chat admission/recovery IO for explicitly bound external sources.
+
+Sessions and Turns remain in ChatSessionStore. This journal only correlates a
+provider request with those canonical objects; it is not a second queue or model
+runner. Write it before admission so a crash can replay the same client identity.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from ...chat_store import _atomic_write_json, _read_json
+from ...file_lock import exclusive_file_lock
+
+
+class ChatExternalConversations:
+    def __init__(self, controller: Any) -> None:
+        self.controller = controller
+        self.bindings = controller.project_contexts.conversation_bindings
+        self.root = controller.store.root / "external-requests"
+
+    def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str,
+              message: str, command: str | None = None) -> dict[str, Any]:
+        import re
+        if not re.fullmatch(r"[a-f0-9]{24}", request_ref):
+            raise ValueError("invalid external request reference")
+        if command not in {None, "status", "new", "stop", "unsupported"}:
+            raise ValueError("unsupported external conversation command")
+        selected = self.bindings.resolve(binding_id=binding_id, **source)
+        path = self.root / f"{request_ref}.json"
+        with exclusive_file_lock(path, operation="admit_external_chat_request"):
+            expected = {"binding_id": binding_id, "source": source, "message": message, "command": command}
+            if path.exists():
+                row = _read_json(path)
+                if any(row.get(key) != value for key, value in expected.items()):
+                    raise ValueError("external request identity was reused with different content")
+                if row.get("status") != "prepared":
+                    return row
+            else:
+                row = {"schema_version": "loopx_chat_external_request_v0", "request_ref": request_ref,
+                       **expected, "status": "prepared", "session_id": None, "turn_id": None}
+                _atomic_write_json(path, row)
+            return self._admit_prepared(path, row, selected)
+
+    def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, Any]) -> dict[str, Any]:
+        controller = self.controller
+        current = controller.store.latest_session(goal_id=None,
+            agent_id=selected["binding"]["executor_endpoint_id"], channel_id=selected["channel_id"])
+        if row.get("session_id") and row["command"] is None:
+            current = controller.store.load_session(row["session_id"])
+            if current is None:
+                raise ValueError("the original request Session is unavailable")
+            client_id = f"external-{row['request_ref']}"
+            if controller.store.turn_for_client(current["session_id"], client_id) is not None:
+                # The canonical store validates exact replay before its closed
+                # Session check. Never move an accepted request to a new Session.
+                turn, _ = controller.store.create_queued_turn(current["session_id"],
+                    client_turn_id=client_id, message=row["message"], origin="lark")
+                row.update(status="accepted", turn_id=turn["turn_id"])
+                _atomic_write_json(path, row)
+                return row
+        from ...control_plane.effect_runtime import effect_runtime_result
+        plan = effect_runtime_result("collaboration.conversation.request", {"request": row, "current_session": current})
+        operation = plan["operation"]
+        if operation == "reply":
+            row.update(status="command_completed", session_id=plan["session_id"], response_code=plan["response_code"])
+        elif operation in {"new", "stop"}:
+            row.update(target_recorded=True, session_id=plan["session_id"], turn_id=plan["turn_id"])
+            _atomic_write_json(path, row)
+            if operation == "stop" and row["session_id"] and row["turn_id"]:
+                controller.interrupt_turn(session_id=row["session_id"], turn_id=row["turn_id"])
+            elif operation == "new" and row["session_id"]:
+                controller.close_session(row["session_id"])
+            row.update(status="command_completed", response_code=plan["response_code"])
+        else:
+            # Opening uses the shared route fence, canonical grant and native
+            # adapter. It never creates a Goal or waits for a terminal answer.
+            if current is None:
+                current, _ = controller.open_session(goal_id=None,
+                    agent_id=selected["binding"]["executor_endpoint_id"], work_dir=Path("."), objective="",
+                    mode="resume_latest", conversation_binding_id=row["binding_id"], source_context=row["source"])
+            row["session_id"] = current["session_id"]
+            _atomic_write_json(path, row)
+            try:
+                turn, _ = controller.enqueue_turn(session_id=current["session_id"],
+                    client_turn_id=plan["client_turn_id"], message=row["message"],
+                    work_dir=Path("."), objective="", origin="lark")
+                row.update(status="accepted", turn_id=turn["turn_id"])
+            except RuntimeError as exc:
+                if str(exc) != "session_queue_full":
+                    raise
+                row.update(status="rejected", response="队列已满，本条没有被受理；请稍后重新发送。")
+        _atomic_write_json(path, row)
+        return row
+
+    def pending(self) -> list[dict[str, Any]]:
+        return [_read_json(path) for path in sorted(self.root.glob("*.json"))]
+
+    def record_delivery(self, request_ref: str, *, session_id: str | None, turn_id: str | None) -> None:
+        # This transport receipt does not alter a canonical Turn or grant.
+        path = self.root / f"{request_ref}.json"
+        with exclusive_file_lock(path, operation="record_external_chat_delivery"):
+            row = _read_json(path)
+            if (row.get("session_id"), row.get("turn_id")) != (session_id, turn_id):
+                raise ValueError("delivery correlation changed")
+            row["delivery_verified"] = True
+            _atomic_write_json(path, row)
+
+    def recover(self) -> None:
+        for row in self.pending():
+            try:
+                if row.get("delivery_verified"):
+                    continue
+                self.bindings.resolve(binding_id=row["binding_id"], **row["source"])
+                if row["status"] == "prepared":
+                    self.admit(binding_id=row["binding_id"], source=row["source"], request_ref=row["request_ref"],
+                               message=row["message"], command=row["command"])
+                elif row["status"] == "accepted":
+                    session = self.controller.store.load_session(row["session_id"])
+                    if session and session.get("status") != "closed":
+                        context = self.controller.project_contexts.session_context(session)
+                        self.controller.resume_session_queue(session_id=row["session_id"],
+                            work_dir=context["project"], objective=context["objective"])
+            except (KeyError, ValueError, RuntimeError, OSError):
+                # Revoked/missing grants never fall back to a new model thread.
+                continue
diff --git a/loopx/capabilities/native_chat/project_context.py b/loopx/capabilities/native_chat/project_context.py
index 83dfed5fbe..0d5eec44f1 100644
--- a/loopx/capabilities/native_chat/project_context.py
+++ b/loopx/capabilities/native_chat/project_context.py
@@ -27,6 +27,7 @@ def __init__(self, roots: list[Path]) -> None:
         # Remember the owner's spelling as well as its initial canonical target.
         # A later symlink retarget must not redirect an accepted Session.
         self.roots = [(root.expanduser().absolute(), root.expanduser().resolve()) for root in roots]
+        self.conversation_bindings: Any | None = None
 
     def available(self) -> list[dict[str, str]]:
         contexts = {}
@@ -52,7 +53,12 @@ def session_context(self, session: dict[str, Any]) -> dict[str, Any]:
         saved = session.get("project_context")
         if not isinstance(saved, dict) or session.get("goal_id") is not None:
             raise ValueError("invalid ordinary project Session")
-        selected = self.resolve(str(saved.get("project_ref") or ""), session_context=saved)
+        if saved.get("audience") == "bound_owner":
+            if self.conversation_bindings is None:
+                raise ValueError("bound project conversation authority is unavailable")
+            selected = self.conversation_bindings.session_context(saved)
+        else:
+            selected = self.resolve(str(saved.get("project_ref") or ""), session_context=saved)
         if session.get("channel_id") != selected["channel_id"]:
             raise ValueError("project conversation channel mismatch")
         return {"project": Path(selected["context"]["workspace_path"]),
diff --git a/loopx/chat_lark_api.py b/loopx/chat_lark_api.py
index caca356ed5..15b3834263 100644
--- a/loopx/chat_lark_api.py
+++ b/loopx/chat_lark_api.py
@@ -633,6 +633,29 @@ def _lark_connect(self) -> None:
             # setting -- not a stored connection field or a request field --
             # decides which endpoint this connection runs on and which Session
             # it binds. The connection write below records the resolution.
+            private = getattr(getattr(getattr(self.server, "runtime_controller", None), "project_contexts", None), "conversation_bindings", None)
+            selected_profiles = [app_ref] if app_ref else [str(row.get("app_ref") or "") for row in body.get("agent_bindings", [])]
+            provider_runner = self._lark_runner()
+            if private is not None:
+                from .extensions.lark.conversation_identity import identity_ref
+                configured = private.read()["bindings"]
+                # Only active private bindings require the collision check.
+                if configured:
+                    cached_auth = {}
+                    original_runner = provider_runner
+                    def provider_runner(args, cwd, timeout):
+                        key = tuple(args)
+                        if tuple(args[-4:]) != ("auth", "status", "--verify", "--json"):
+                            return original_runner(args, cwd, timeout)
+                        if key not in cached_auth:
+                            cached_auth[key] = original_runner(args, cwd, timeout)
+                        return cached_auth[key]
+                for profile in selected_profiles if configured else []:
+                    if any(row["transport_ref"] == profile for row in configured):
+                        raise ValueError("this App owns a private listener; disconnect it first")
+                    identity = _app_identity_for_private_guard(profile, provider_runner, cli_bin)
+                    if identity and any(row["provider_ref"] == identity_ref(identity) for row in configured):
+                        raise ValueError("this App alias owns a private listener; disconnect it first")
             executor_endpoint_id = (
                 manager_executor_endpoint_default(
                     machine_defaults=steward_machine_defaults(
@@ -733,7 +756,7 @@ def _lark_connect(self) -> None:
                     getattr(self.server, "runtime_controller", None)
                 ),
                 "execute": body.get("execute") is True,
-                "runner": self._lark_runner(),
+                "runner": provider_runner,
                 "cli_bin": cli_bin,
             }
             if app_refs_by_agent is not None:
@@ -806,3 +829,16 @@ def _lark_disconnect(self) -> None:
             return
         self._refresh_lark_goal_topic_runtime()
         self._send_json(packet)
+
+
+def _app_identity_for_private_guard(profile: str, runner: Any, cli_bin: str) -> str:
+    from .extensions.lark.goal_channel_transport import APP_ID_PATTERN, call, json_payload, lark_args
+    result = call(runner, lark_args(cli_bin=cli_bin, profile=profile, tail=["auth", "status", "--verify", "--json"]))
+    payload = json_payload(result)
+    app_id = str(payload.get("appId") or "")
+    identities = payload.get("identities")
+    bot = identities.get("bot") if isinstance(identities, Mapping) else None
+    if (result.get("returncode") != 0 or not APP_ID_PATTERN.fullmatch(app_id)
+            or not isinstance(bot, Mapping) or bot.get("available") is not True or bot.get("verified") is not True):
+        raise ValueError("the App identity could not be verified")
+    return app_id
diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py
index 69d238e0d1..9936fab716 100644
--- a/loopx/chat_runtime.py
+++ b/loopx/chat_runtime.py
@@ -112,7 +112,7 @@ def start(
         *,
         codex_bin: str,
         work_dir: Path,
-        goal_id: str,
+        goal_id: str | None,
         objective: str,
         resume_thread_id: str | None = None,
         startup_timeout_sec: float = 30.0,
@@ -373,7 +373,7 @@ def _managed_upstream_mode(session: dict[str, Any]) -> str:
     @staticmethod
     def _session_objective(
         *,
-        goal_id: str,
+        goal_id: str | None,
         objective: str,
         history: list[dict[str, Any]] | None,
         project_coordination: bool = False,
@@ -562,13 +562,31 @@ def open_session(
         agent_goal_id: str | None = None,
         manager_executor_allocation: Mapping[str, Any] | None = None,
         project_ref: str | None = None,
+        conversation_binding_id: str | None = None,
+        source_context: dict[str, Any] | None = None,
     ) -> tuple[dict[str, Any], bool]:
         capability = next((item for item in self.capabilities() if item["agent_id"] == agent_id), None)
         if mode not in {"resume_latest", "new"}:
             raise ValueError("mode must be resume_latest or new")
         selected_channel = channel_id or f"goal.{goal_id}"
         project_context = None
-        if project_ref is not None:
+        if conversation_binding_id is not None:
+            if goal_id is not None or agent_goal_id is not None or project_ref is not None:
+                raise ValueError("bound project Chat cannot substitute a Goal or workspace")
+            bindings = self.project_contexts.conversation_bindings
+            if bindings is None or source_context is None:
+                raise ValueError("bound project conversation authority is unavailable")
+            selected = bindings.resolve(binding_id=conversation_binding_id, **source_context)
+            if agent_id != selected["binding"]["executor_endpoint_id"]:
+                raise ValueError("executor does not match the conversation grant")
+            if channel_id is not None and channel_id != selected["channel_id"]:
+                raise ValueError("bound conversation channel mismatch")
+            project_context, selected_channel = selected["context"], selected["channel_id"]
+            context = self.project_contexts.session_context({
+                "goal_id": None, "channel_id": selected_channel, "project_context": project_context,
+            })
+            work_dir, objective = context["project"], context["objective"]
+        elif project_ref is not None:
             if goal_id is not None or agent_goal_id is not None:
                 raise ValueError("ordinary project conversations cannot carry a Goal")
             selected = self.project_contexts.resolve(project_ref)
@@ -977,6 +995,8 @@ def submit_turn(
         if session is None:
             raise KeyError("chat session was not found")
         if session.get("project_context") is not None:
+            if session["project_context"].get("audience") == "bound_owner":
+                raise ValueError("bound project Chat requires its external source admission")
             context = self.project_contexts.session_context(session)
             work_dir, objective = context["project"], context["objective"]
             if loopx_execution:
@@ -1103,6 +1123,10 @@ def steer_active_turn(
         session = self.store.load_session(session_id)
         if session is None or session.get("status") == "closed":
             raise KeyError("chat session was not found")
+        if session.get("project_context") is not None:
+            if session["project_context"].get("audience") == "bound_owner":
+                raise ValueError("bound project steering requires its external source admission")
+            self.project_contexts.session_context(session)
         if session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED:
             session = require_current_attached_session(
                 store=self.store,
@@ -1237,8 +1261,8 @@ def enqueue_turn(
         if session is None or session.get("status") == "closed":
             raise KeyError("chat session was not found")
         if session.get("project_context") is not None:
-            if origin != "web":
-                raise ValueError("local project grant does not authorize an external audience")
+            if conversation_scope(session, origin=origin)["kind"] != "project_workspace":
+                raise ValueError("project grant does not authorize this external audience or source")
             context = self.project_contexts.session_context(session)
             work_dir, objective = context["project"], context["objective"]
         if session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED:
diff --git a/loopx/chat_server.py b/loopx/chat_server.py
index 3be6c84016..5a03392c42 100644
--- a/loopx/chat_server.py
+++ b/loopx/chat_server.py
@@ -48,6 +48,10 @@
 from .chat_session_open import open_chat_session
 from .chat_ssh_source_api import SshSourceRequestMixin
 from .chat_store import ChatSessionStore
+from .capabilities.native_chat.conversation_bindings import ChatConversationBindings
+from .extensions.lark.private_conversation_api import PrivateConversationRequestMixin, PRIVATE_CONVERSATIONS_PATH
+from .extensions.lark.conversation_identity import observe_lark_conversation_identity
+from .extensions.lark.private_conversations import LarkPrivateConversations
 from .chat_loopx_mode import handle_loopx_request
 from .capabilities.manager_context.roundtrip import project_chat_session_snapshot
 from .control_plane.goals.active_state_metadata import active_state_section_text
@@ -416,6 +420,7 @@ class ChatHTTPServer(ThreadingHTTPServer):
     lark_cli_resolution: LarkCliResolution
     lark_app_setup_manager: LarkAppSetupManager
     lark_goal_topic_runtime: LarkGoalTopicRuntimeService
+    lark_private_conversations: LarkPrivateConversations
     ssh_config_path: Path | None
     goal_subagent_configuration_enabled: bool
 
@@ -438,6 +443,7 @@ def server_close(self) -> None:
 
 
 class ChatRequestHandler(
+    PrivateConversationRequestMixin,
     CompletedTodoRequestMixin,
     AttachedSessionRequestMixin,
     SshSourceRequestMixin,
@@ -1409,6 +1415,7 @@ def do_GET(self) -> None:
             CHAT_LARK_APPS_PATH: self._lark_apps,
             CHAT_LARK_CHATS_PATH: self._lark_chats,
             CHAT_LARK_CONNECTIONS_PATH: self._lark_connections,
+            PRIVATE_CONVERSATIONS_PATH: self._private_conversations,
             CHAT_GOAL_CHANNEL_TARGETS_PATH: self._goal_channel_targets,
             **self._configuration_get_routes(),
             DEFAULT_CHAT_STATUS_PATH: self._status,
@@ -1456,6 +1463,7 @@ def do_POST(self) -> None:
             CHAT_GOAL_CHANNEL_CONFIGURE_PATH: self._goal_channel_configure,
             CHAT_LARK_APP_SETUPS_PATH: self._lark_setup_start,
             CHAT_LARK_CONNECTIONS_PATH: self._lark_connect,
+            PRIVATE_CONVERSATIONS_PATH: self._private_conversation_connect,
             **self._configuration_post_routes(),
             **self._ssh_source_post_routes(),
         }
@@ -1500,6 +1508,8 @@ def do_DELETE(self) -> None:
             return self._lark_setup_cancel(setup_parts[4])
         if path == CHAT_LARK_CONNECTIONS_PATH:
             return self._lark_disconnect()
+        if path == PRIVATE_CONVERSATIONS_PATH:
+            return self._private_conversation_disconnect()
         prefix = f"{CHAT_SESSIONS_PATH}/"
         if not path.startswith(prefix):
             return self._send_error("unknown path", status=404)
@@ -1597,6 +1607,21 @@ def serve_chat(
         idle_timeout_sec=idle_timeout_sec,
         hard_timeout_sec=hard_timeout_sec,
     )
+    server.runtime_controller.project_contexts.conversation_bindings = ChatConversationBindings(
+        root=server.chat_store.root, project_contexts=server.runtime_controller.project_contexts,
+        observe=lambda profile: observe_lark_conversation_identity(profile=profile, runner=server.lark_runner,
+            cli_bin=server.lark_cli_resolution.command or "lark-cli"))
+    private_transport = LarkPrivateConversations(controller=server.runtime_controller, runtime_root=runtime_root,
+        runner=server.lark_runner, cli_bin=server.lark_cli_resolution.command or "lark-cli")
+
+    server.lark_private_conversations = private_transport
+
+    def _lark_snapshot():
+        snapshot = build_lark_goal_topic_runtime_snapshot(registry_path=server.registry_path,
+            runtime_root_override=server.runtime_root_override)
+        snapshot["private_profiles"] = private_transport.profiles()
+        return snapshot
+
     server.action_service = ChatActionService(
         store=server.action_store,
         registry_path=resolved_registry_path,
@@ -1612,10 +1637,8 @@ def serve_chat(
         server.action_service.project_team_plan_preview
     )
     server.lark_goal_topic_runtime = LarkGoalTopicRuntimeService(
-        snapshot_provider=lambda: build_lark_goal_topic_runtime_snapshot(
-            registry_path=server.registry_path,
-            runtime_root_override=server.runtime_root_override,
-        ),
+        snapshot_provider=_lark_snapshot,
+        private_conversations=private_transport,
         runtime_root=runtime_root,
         runtime_controller=server.runtime_controller,
         action_service=server.action_service,
diff --git a/loopx/chat_store.py b/loopx/chat_store.py
index 0315537612..a87234f4da 100644
--- a/loopx/chat_store.py
+++ b/loopx/chat_store.py
@@ -273,9 +273,7 @@ def create_session(
         }
         if project_context is not None:
             from .control_plane.effect_runtime import effect_runtime_result
-            selected = effect_runtime_result("collaboration.project.context", {
-                "project_ref": project_context.get("project_ref"), "available": [project_context],
-            })
+            selected = effect_runtime_result("collaboration.project.session_identity", {"context": project_context})
             if goal_id is not None or goal_instance_id is not None or channel_id != selected["channel_id"] or normalized_mode != CHAT_SESSION_MODE_MANAGED:
                 raise ValueError("ordinary project Sessions require their exact channel and no Goal")
             project_context = selected["context"]
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
new file mode 100644
index 0000000000..60e7333439
--- /dev/null
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -0,0 +1,138 @@
+import type {JsonObject} from "../effect_program.ts";
+import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
+import {requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts";
+import {normalizeProjectContext} from "./conversation_scope.ts";
+
+/** Core owns the context and audience grant. A provider supplies verified,
+ * opaque identity observations; neither a message nor a model selects them.
+ * This contract intentionally carries no Goal and no provider credentials.
+ */
+const BINDING_SCHEMA = "loopx_chat_conversation_binding_v0";
+const SET_SCHEMA = "loopx_chat_conversation_bindings_v0";
+
+function ref(value: unknown, label: string): string {
+  const result = requireNonEmptyString(value, label);
+  if (!/^[a-f0-9]{24}$/.test(result)) throw new EffectRuntimeRequestError(`${label} is invalid`);
+  return result;
+}
+
+function token(value: unknown, label: string): string {
+  const result = requireNonEmptyString(value, label);
+  if (!/^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$/.test(result)) throw new EffectRuntimeRequestError(`${label} is invalid`);
+  return result;
+}
+
+function binding(value: unknown): JsonObject {
+  const row = requireJsonObject(value, "conversation binding");
+  if (row.schema_version !== BINDING_SCHEMA || row.context_kind !== "project"
+      || row.grant !== "workspace_read" || row.enabled !== true) {
+    throw new EffectRuntimeRequestError("unsupported conversation binding");
+  }
+  return {
+    schema_version: BINDING_SCHEMA, binding_id: ref(row.binding_id, "binding identity"),
+    transport_ref: token(row.transport_ref, "transport reference"),
+    provider_ref: ref(row.provider_ref, "provider identity"),
+    operator_ref: ref(row.operator_ref, "verified operator identity"),
+    context_kind: "project", project_ref: ref(row.project_ref, "workspace reference"),
+    executor_endpoint_id: token(row.executor_endpoint_id, "executor endpoint"),
+    grant: "workspace_read", enabled: true,
+  };
+}
+
+function state(value: unknown): {schema_version: string; revision: number; bindings: JsonObject[]} {
+  const row = requireJsonObject(value, "conversation bindings");
+  if (row.schema_version !== SET_SCHEMA || !Number.isSafeInteger(row.revision)
+      || Number(row.revision) < 0 || !Array.isArray(row.bindings)) {
+    throw new EffectRuntimeRequestError("invalid conversation bindings state");
+  }
+  const bindings = row.bindings.map(binding);
+  for (const field of ["binding_id", "transport_ref", "provider_ref"]) {
+    if (new Set(bindings.map(row => row[field])).size !== bindings.length) {
+      throw new EffectRuntimeRequestError("conversation listener ownership is ambiguous");
+    }
+  }
+  return {schema_version: SET_SCHEMA, revision: Number(row.revision), bindings};
+}
+
+function observed(row: JsonObject, value: unknown): void {
+  const observation = requireJsonObject(value, "provider identity observation");
+  if (observation.verified !== true || observation.transport_ref !== row.transport_ref
+      || observation.provider_ref !== row.provider_ref || observation.operator_ref !== row.operator_ref) {
+    throw new EffectRuntimeRequestError("provider or operator identity is not independently verified");
+  }
+}
+
+export function planConversationBinding(params: JsonObject): JsonObject {
+  const current = state(params.current);
+  if (params.expected_revision !== current.revision) throw new EffectRuntimeRequestError("conversation binding revision changed");
+  let rows: JsonObject[];
+  if (params.operation === "configure") {
+    const candidate = binding(params.binding);
+    observed(candidate, params.observation);
+    if (!Array.isArray(params.available_projects)
+        || !params.available_projects.map(normalizeProjectContext).some(row => row.project_ref === candidate.project_ref)) {
+      throw new EffectRuntimeRequestError("workspace grant is unavailable");
+    }
+    const previous = current.bindings.find(row => row.transport_ref === candidate.transport_ref);
+    // An exact repeat is idempotent. A changed context receives a new binding
+    // identity, so an existing Session can never silently move workspaces.
+    if (previous && JSON.stringify({...previous, binding_id: null}) === JSON.stringify({...candidate, binding_id: null})) {
+      return {changed: false, state: current};
+    }
+    if (previous?.binding_id === candidate.binding_id) throw new EffectRuntimeRequestError("changed context requires a new binding identity");
+    rows = [...current.bindings.filter(row => row.transport_ref !== candidate.transport_ref), candidate];
+  } else if (params.operation === "disconnect") {
+    const id = ref(params.binding_id, "binding identity");
+    rows = current.bindings.filter(row => row.binding_id !== id);
+    if (rows.length === current.bindings.length) return {changed: false, state: current};
+  } else throw new EffectRuntimeRequestError("unsupported conversation binding operation");
+  const next = state({...current, revision: current.revision + 1, bindings: rows});
+  return {changed: true, state: next};
+}
+
+export function resolveBoundConversation(params: JsonObject): JsonObject {
+  const current = state(params.current);
+  const id = ref(params.binding_id, "binding identity");
+  const row = current.bindings.find(row => row.binding_id === id);
+  if (!row) throw new EffectRuntimeRequestError("conversation binding is no longer authorized");
+  observed(row, params.observation);
+  const source = ref(params.source_ref, "source conversation identity");
+  if (params.sender_ref !== row.operator_ref || params.private_human_message !== true) {
+    throw new EffectRuntimeRequestError("message audience is outside the binding grant");
+  }
+  if (!Array.isArray(params.available_projects)) throw new EffectRuntimeRequestError("workspace grants unavailable");
+  const projects = params.available_projects.map(normalizeProjectContext).filter(project => project.project_ref === row.project_ref);
+  if (projects.length !== 1) throw new EffectRuntimeRequestError("workspace grant is unavailable or ambiguous");
+  const context = {...projects[0], audience: "bound_owner", binding_id: id, source_ref: source,
+    provider_ref: row.provider_ref, operator_ref: row.operator_ref};
+  if (params.session_context !== undefined && JSON.stringify(params.session_context) !== JSON.stringify(context)) {
+    throw new EffectRuntimeRequestError("bound Session context changed");
+  }
+  return {binding: row, context, channel_id: `project.external.${id}.${source}`};
+}
+
+/** Shared native command projection. Provider grammar carries the explicit
+ * command; Core binds its exact target before IO. Replays retain that target.
+ */
+export function planBoundConversationRequest(params: JsonObject): JsonObject {
+  const row = requireJsonObject(params.request, "external request");
+  const request = ref(row.request_ref, "external request identity");
+  const command = row.command;
+  if (![null, "status", "new", "stop", "unsupported"].includes(command as null | string)) {
+    throw new EffectRuntimeRequestError("unsupported external conversation command");
+  }
+  const current = params.current_session === null ? null : requireJsonObject(params.current_session, "current Session");
+  const target = row.target_recorded === true ? row : current;
+  const session = target?.session_id ?? null;
+  const turn = row.target_recorded === true ? row.turn_id ?? null : current?.active_turn_id ?? null;
+  if (command === "status" || command === "unsupported") {
+    return {operation: "reply", session_id: session, turn_id: null,
+      response_code: command === "unsupported" ? "unsupported_attachment" : !current ? "no_session"
+        : current.active_turn_id ? "active_session" : "ready_session"};
+  }
+  if (command === "stop" || command === "new") {
+    return {operation: command, session_id: session, turn_id: turn,
+      response_code: command === "new" ? "new_session" : turn ? "stop_requested" : "no_active_turn"};
+  }
+  return {operation: "admit_turn", client_turn_id: `external-${request}`, session_id: session, turn_id: null};
+}
diff --git a/loopx/control_plane/collaboration/conversation_scope.ts b/loopx/control_plane/collaboration/conversation_scope.ts
index ac706997ee..ccae68cd5a 100644
--- a/loopx/control_plane/collaboration/conversation_scope.ts
+++ b/loopx/control_plane/collaboration/conversation_scope.ts
@@ -6,18 +6,32 @@ export function normalizeProjectContext(value: unknown): Record<string, string>
   const ref = context.project_ref, workspace = context.workspace_path;
   if (typeof ref !== "string" || !/^[a-f0-9]{24}$/.test(ref)
       || typeof workspace !== "string" || !workspace || context.kind !== "project_workspace"
-      || context.audience !== "local_owner" || context.grant !== "workspace_read"
+      || !["local_owner", "bound_owner"].includes(String(context.audience)) || context.grant !== "workspace_read"
       || (workspace[0] !== "/" && !/^[A-Za-z]:[\\/]/.test(workspace))) {
     throw new Error("invalid project conversation context");
   }
-  return {kind: "project_workspace", project_ref: ref, workspace_path: workspace,
-    audience: "local_owner", grant: "workspace_read"};
+  const normalized: Record<string, string> = {kind: "project_workspace", project_ref: ref, workspace_path: workspace,
+    audience: String(context.audience), grant: "workspace_read"};
+  if (context.audience === "bound_owner") {
+    for (const field of ["binding_id", "source_ref", "provider_ref", "operator_ref"]) {
+      const value = context[field];
+      if (typeof value !== "string" || !/^[a-f0-9]{24}$/.test(value)) throw new Error("incomplete bound project identity");
+      normalized[field] = value;
+    }
+  }
+  return normalized;
+}
+
+export function projectConversationIdentity(input: Record<string, unknown>): Record<string, unknown> {
+  const context = normalizeProjectContext(input.context);
+  return {context, channel_id: context.audience === "local_owner"
+    ? `project.${context.project_ref}` : `project.external.${context.binding_id}.${context.source_ref}`};
 }
 
 type ConversationScope = Record<string, unknown> & (
   | {kind: "owner_portfolio"; goal_ids: null; private_conversation: true}
   | {kind: "owner_goal"; goal_ids: [string]; private_conversation: true}
-  | {kind: "project_workspace"; goal_ids: []; private_conversation: true}
+  | {kind: "project_workspace"; goal_ids: []; private_conversation: boolean}
   | {kind: "external_audience" | "unavailable"; goal_ids: []; private_conversation: false}
 );
 
@@ -27,12 +41,17 @@ type ConversationScope = Record<string, unknown> & (
 export function resolveConversationScope(input: Record<string, unknown>): ConversationScope {
   const channel = input.channel_id;
   const goal = input.goal_id;
-  if (goal === null && (input.origin === undefined || input.origin === "web")) {
+  if (goal === null) {
     try {
       const context = normalizeProjectContext(input.project_context);
-      if (channel === `project.${context.project_ref}`) {
+      if (context.audience === "local_owner" && (input.origin === undefined || input.origin === "web")
+          && channel === `project.${context.project_ref}`) {
         return {kind: "project_workspace", goal_ids: [], private_conversation: true};
       }
+      if (context.audience === "bound_owner" && (input.origin === undefined || input.origin === "lark")
+          && channel === `project.external.${context.binding_id}.${context.source_ref}`) {
+        return {kind: "project_workspace", goal_ids: [], private_conversation: false};
+      }
     } catch { /* Incomplete host identity grants no context. */ }
   }
   if (channel === "manager") {
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 5d3346197c..f5ec3b1a4a 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -577,6 +577,10 @@ export function createEffectRuntimeHandlers(
     ["collaboration.conversation.trigger", lazyHandler(() => import("./collaboration/conversation_trigger.ts"), ({resolveConversationTrigger}) => resolveConversationTrigger)],
     ["collaboration.conversation.scope", lazyHandler(() => import("./collaboration/conversation_scope.ts"), ({resolveConversationScope}) => resolveConversationScope)],
     ["collaboration.project.context", lazyHandler(() => import("./collaboration/project_conversation.ts"), ({resolveProjectConversation}) => resolveProjectConversation)],
+    ["collaboration.project.session_identity", lazyHandler(() => import("./collaboration/conversation_scope.ts"), ({projectConversationIdentity}) => projectConversationIdentity)],
+    ["collaboration.conversation.binding", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({planConversationBinding}) => planConversationBinding)],
+    ["collaboration.conversation.bound_context", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({resolveBoundConversation}) => resolveBoundConversation)],
+    ["collaboration.conversation.request", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({planBoundConversationRequest}) => planBoundConversationRequest)],
     ["collaboration.peer.context_access", lazyHandler(() => import("./collaboration/peer_context.ts"), ({requirePeerContextAccess}) => requirePeerContextAccess)],
     ["collaboration.source.recipients", lazyHandler(() => import("./collaboration/source_grants.ts"), ({resolveSourceRecipients}) => resolveSourceRecipients)],
     ["collaboration.source.configure_recipient", lazyHandler(() => import("./collaboration/source_grants.ts"), ({configureSourceRecipient}) => configureSourceRecipient)],
diff --git a/loopx/extensions/lark/conversation_identity.py b/loopx/extensions/lark/conversation_identity.py
new file mode 100644
index 0000000000..71db018cf4
--- /dev/null
+++ b/loopx/extensions/lark/conversation_identity.py
@@ -0,0 +1,61 @@
+"""Lark-specific identity observation for a Core-owned private conversation.
+
+Each non-default profile independently verifies its App and logged-in owner.
+Core receives only opaque, App-scoped references. Tokens remain in lark-cli.
+"""
+from __future__ import annotations
+
+import hashlib
+import re
+from collections.abc import Mapping
+from typing import Any
+
+from .goal_channel_transport import APP_ID_PATTERN, call, json_payload, lark_args
+from .identity_shapes import LARK_OPEN_ID_PATTERN as OPEN_ID_PATTERN
+from .goal_topic_connections import _profile_ref
+from .presentation.kanban import CommandRunner
+
+
+def identity_ref(*values: str) -> str:
+    return hashlib.sha256("\0".join(values).encode("utf-8")).hexdigest()[:24]
+
+
+def observe_lark_conversation_identity(*, profile: str, runner: CommandRunner,
+                                       cli_bin: str) -> dict[str, Any]:
+    profile = _profile_ref(profile)
+    if profile.casefold() == "default":
+        raise ValueError("private conversations require an explicit non-default App profile")
+    result = call(runner, lark_args(cli_bin=cli_bin, profile=profile,
+                                  tail=["auth", "status", "--verify", "--json"]))
+    payload = json_payload(result)
+    identities = payload.get("identities")
+    identities = identities if isinstance(identities, Mapping) else {}
+    bot, owner = identities.get("bot"), identities.get("user")
+    app_id = str(payload.get("appId") or "")
+    if result.get("returncode") != 0 or not APP_ID_PATTERN.fullmatch(app_id):
+        raise ValueError("the selected App identity could not be verified")
+    if not isinstance(bot, Mapping) or not isinstance(owner, Mapping):
+        raise ValueError("verify this App and its owner independently before binding private Chat")
+    if not all(row.get("available") is True and row.get("verified") is True
+               for row in [bot, owner]):
+        raise ValueError("verify this App and its owner independently before binding private Chat")
+    owner_id = str(owner.get("openId") or "")
+    if not OPEN_ID_PATTERN.fullmatch(owner_id):
+        raise ValueError("the selected App has no verified owner identity")
+    provider_ref = identity_ref(app_id)
+    return {"transport_ref": profile, "provider_ref": provider_ref,
+            "operator_ref": identity_ref(provider_ref, owner_id), "verified": True, "consumer_ref": hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32], "bot_display_name": str(bot.get("appName") or "")}
+
+
+def lark_private_source(*, provider_ref: str, event: Mapping[str, Any]) -> dict[str, Any]:
+    """Normalize provider provenance; Core still decides whether it is admitted."""
+    from .identity_shapes import LARK_CHAT_ID_PATTERN, LARK_MESSAGE_ID_PATTERN
+
+    chat = str(event.get("chat_id") or "")
+    sender = str(event.get("sender_id") or "")
+    message = str(event.get("message_id") or "")
+    if not (re.fullmatch(r"[a-f0-9]{24}", provider_ref) and LARK_CHAT_ID_PATTERN.fullmatch(chat)
+            and LARK_MESSAGE_ID_PATTERN.fullmatch(message) and OPEN_ID_PATTERN.fullmatch(sender)):
+        raise ValueError("incomplete private-message provenance")
+    return {"source_ref": identity_ref(provider_ref, chat, sender), "sender_ref": identity_ref(provider_ref, sender),
+            "private_human_message": event.get("chat_type") == "p2p" and event.get("sender_type") == "user"}
diff --git a/loopx/extensions/lark/goal_topic_runtime.py b/loopx/extensions/lark/goal_topic_runtime.py
index 51628b490a..4fba2c2507 100644
--- a/loopx/extensions/lark/goal_topic_runtime.py
+++ b/loopx/extensions/lark/goal_topic_runtime.py
@@ -105,13 +105,15 @@ def __init__(self, error_code: str, effect_receipt: Mapping[str, Any]) -> None:
 
 _EVENT_PROJECTION = (
     '{schema_version:"lark_event_inbox_event_v0",'
-    "event_id:(.event_id // .message_id // .id),"
-    "message_id:(.message_id // .id),"
-    "create_time:.create_time,content:.content,"
-    "sender_id:(.sender_id // .sender.id // .sender.sender_id "
-    "// .event.sender.sender_id // .event.sender.id),"
+    "event_id:(.event_id // .header.event_id // .message_id // .event.message.message_id // .id),"
+    "message_id:(.message_id // .message.message_id // .event.message.message_id // .id),"
+    "create_time:(.create_time // .message.create_time // .event.message.create_time),"
+    "content:(.content // .message.content // .event.message.content),"
+    "sender_id:(.sender_id.open_id? // .sender_id // .sender.id // .sender.sender_id.open_id? // .sender.sender_id "
+    "// .event.sender.sender_id.open_id? // .event.sender.sender_id // .event.sender.id),"
     "sender_type:(.sender_type // .sender.sender_type // .event.sender.sender_type),"
-    "chat_id:.chat_id,"
+    "chat_id:(.chat_id // .message.chat_id // .event.message.chat_id),chat_type:(.chat_type // .message.chat_type // .event.message.chat_type),"
+    "message_type:(.message_type // .msg_type // .message.message_type // .event.message.message_type),"
     "root_id:(.root_id // .message.root_id // .event.message.root_id),"
     "parent_id:(.parent_id // .reply_to // .message.parent_id // .message.reply_to "
     "// .event.message.parent_id // .event.message.reply_to),"
@@ -154,6 +156,10 @@ def _active_profile_configs(snapshot: Mapping[str, Any]) -> dict[str, dict[str,
                     "bot_app_id": str(identity.get("bot_app_id") or ""),
                 },
             )
+    for profile, config in dict(snapshot.get("private_profiles") or {}).items():
+        if profile in profiles:
+            raise ValueError("an App cannot own both private and group listeners")
+        profiles[profile] = dict(config)
     return profiles
 
 
@@ -403,6 +409,7 @@ def poll_lark_goal_topic_profile_once(
     provider_runner: Any = subprocess.run,
     reply_runner: CommandRunner = _default_simple_runner,
     proposal_deliverer: ProposalDeliverer | None = None,
+    private_admitter: Callable[[str, dict[str, object]], Mapping[str, object]] | None = None,
 ) -> dict[str, Any]:
     """Consume one bounded event batch for an App and reuse Inbox reply/ACK."""
 
@@ -443,6 +450,11 @@ def poll_lark_goal_topic_profile_once(
     event_statuses: list[str] = []
     event_reasons: list[str | None] = []
     for event in events:
+        if profile in dict(snapshot.get("private_profiles") or {}):
+            private_result = private_admitter(profile, dict(event)) if private_admitter else {"status": "private_admission_unavailable"}
+            event_statuses.append(str(private_result.get("status") or "unknown"))
+            event_reasons.append(None)
+            continue
         scoped, rejection = _profile_event_route(
             profile=profile,
             snapshot=snapshot,
@@ -498,6 +510,7 @@ def stream_lark_goal_topic_profile(
     reply_runner: CommandRunner = _default_simple_runner,
     health_sink: HealthSink | None = None,
     proposal_deliverer: ProposalDeliverer | None = None,
+    private_admitter: Callable[[str, dict[str, object]], Mapping[str, object]] | None = None,
     review_callback_handler: ReviewCallbackHandler | None = None,
 ) -> dict[str, Any]:
     """Keep one bounded long-lived CLI consumer attached between messages."""
@@ -606,6 +619,7 @@ def handle_event(line: str) -> None:
             provider_runner=provider_runner,
             reply_runner=reply_runner,
             proposal_deliverer=proposal_deliverer,
+            private_admitter=private_admitter,
         )
         with result_lock:
             if int(result.get("event_count") or 0) and not provider_ready:
diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index 35faf20c42..0eadb6e3e6 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -38,6 +38,7 @@ def __init__(
         runtime_root: str | Path,
         runtime_controller: Any,
         action_service: Any | None = None,
+        private_conversations: Any | None = None,
         profile_poller: ProfilePoller | None = None,
         manager_route_reconciler: ManagerRouteReconciler | None = None,
     ) -> None:
@@ -45,6 +46,7 @@ def __init__(
         self.runtime_root = Path(runtime_root).expanduser().resolve()
         self.runtime_controller = runtime_controller
         self.action_service = action_service
+        self.private_conversations = private_conversations
         self._profile_poller = profile_poller or self._poll_profile
         self.manager_route_reconciler = manager_route_reconciler
         self._lock = threading.Lock()
@@ -52,6 +54,7 @@ def __init__(
         self._health: dict[str, dict[str, Any]] = {}
         self._closed = threading.Event()
         self._startup_thread: threading.Thread | None = None
+        self._delivery_thread: threading.Thread | None = None
 
     def start(self) -> None:
         """Discover existing bindings without blocking the HTTP readiness path."""
@@ -65,6 +68,17 @@ def start(self) -> None:
                 daemon=True,
             )
             self._startup_thread.start()
+            if self.private_conversations is not None:
+                self._delivery_thread = threading.Thread(target=self._reconcile_private, name="loopx-lark-private-delivery", daemon=True)
+                self._delivery_thread.start()
+
+    def _reconcile_private(self) -> None:
+        while not self._closed.is_set():
+            try:
+                self.private_conversations.reconcile()
+            except Exception:
+                logging.getLogger(__name__).warning("Private Chat delivery reconciliation is pending")
+            self._closed.wait(1)
 
     def _refresh_on_start(self) -> None:
         while not self._closed.is_set():
@@ -125,7 +139,7 @@ def _profile_worker_fingerprint(
         )
         return (
             str(profile_config.get("cli_bin") or "lark-cli"),
-            str(profile_config.get("bot_app_id") or ""),
+            str(profile_config.get("bot_app_id") or profile_config.get("binding_id") or ""),
             callback_chats,
         )
 
@@ -232,8 +246,12 @@ def handle_review_callback(
                     # A bot App can have several local profile aliases and Chat
                     # servers can use different runtime roots or ports. The
                     # consumer lease must therefore be machine/App scoped.
-                    app_id = str(profile_config.get("bot_app_id") or profile)
-                    digest = hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32]
+                    app_id = str(profile_config.get("bot_app_id") or "")
+                    # Both paths lease the same App-scoped digest, so aliases
+                    # or a second server cannot start another consumer.
+                    digest = (hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32]
+                              if app_id else str(profile_config.get("consumer_ref")
+                              or hashlib.sha256(profile.encode("utf-8")).hexdigest()[:32]))
                     lease = Path.home() / ".loopx" / "lark-consumers" / digest
                     with try_exclusive_file_lock(
                         lease, operation="lark_event_consumer"
@@ -252,10 +270,11 @@ def handle_review_callback(
                             stop=stop,
                             runtime_root=self.runtime_root,
                             answer=answer,
+                            private_admitter=(self.private_conversations.admit if self.private_conversations is not None else None),
                             proposal_deliverer=deliver_proposals,
                             review_callback_handler=(
                                 handle_review_callback
-                                if self.action_service is not None
+                                if self.action_service is not None and profile not in dict(self.snapshot_provider().get("private_profiles") or {})
                                 else None
                             ),
                             health_sink=lambda update: self._update_health(
@@ -431,6 +450,8 @@ def close(self) -> None:
             stop.set()
         for _stop, thread, _fingerprint in workers:
             thread.join(timeout=3)
+        if self._delivery_thread is not None:
+            self._delivery_thread.join(timeout=3)
 
 
 __all__ = ["LarkGoalTopicRuntimeService"]
diff --git a/loopx/extensions/lark/inbox_reply.py b/loopx/extensions/lark/inbox_reply.py
index 69773f604d..a1790f1a03 100644
--- a/loopx/extensions/lark/inbox_reply.py
+++ b/loopx/extensions/lark/inbox_reply.py
@@ -282,6 +282,7 @@ def _deliver_lark_inbox_outbound(
     execute: bool = False,
     provider_preflight: bool = False,
     runner: CommandRunner = _default_runner,
+    source_membership_verifier: Callable[[], bool] | None = None,
     before_send: Callable[[str], Mapping[str, Any]] | None = None,
     delivery_attempt_recorder: Callable[[Mapping[str, str | None]], None] | None = None,
     short_message_limit: int | None = DEFAULT_LARK_TEXT_LIMIT,
@@ -399,21 +400,24 @@ def _deliver_lark_inbox_outbound(
             format_preflight_passed=True,
         )
 
-    membership = _call(
-        runner,
-        base
-        + [
-            "im",
-            "chats",
-            "get",
-            "--chat-id",
-            chat_id,
-            "--as",
-            "bot",
-            "--format",
-            "json",
-        ],
-    )
+    if source_membership_verifier is not None:
+        membership = {"returncode": 0 if source_membership_verifier() else 1}
+    else:
+        membership = _call(
+            runner,
+            base
+            + [
+                "im",
+                "chats",
+                "get",
+                "--chat-id",
+                chat_id,
+                "--as",
+                "bot",
+                "--format",
+                "json",
+            ],
+        )
     if membership.get("returncode") != 0:
         return _result(
             status="gate_required",
@@ -514,6 +518,7 @@ def _deliver_lark_inbox_outbound(
                     project=project, config_path=config_path, message_id=message_id,
                     text=text, content_format="text", execute=execute,
                     provider_preflight=provider_preflight, runner=runner, before_send=before_send,
+                    source_membership_verifier=source_membership_verifier,
                     delivery_attempt_recorder=delivery_attempt_recorder,
                 )
                 result.update(content_format="text", format_fallback="post_size_limit")
@@ -738,6 +743,7 @@ def reply_lark_event_inbox(
     execute: bool = False,
     provider_preflight: bool = False,
     runner: CommandRunner = _default_runner,
+    source_membership_verifier: Callable[[], bool] | None = None,
     before_send: Callable[[str], Mapping[str, Any]] | None = None,
     delivery_attempt_recorder: Callable[[Mapping[str, str | None]], None] | None = None,
     short_message_limit: int | None = DEFAULT_LARK_TEXT_LIMIT,
@@ -758,6 +764,7 @@ def reply_lark_event_inbox(
         execute=execute,
         provider_preflight=provider_preflight,
         runner=runner,
+        source_membership_verifier=source_membership_verifier,
         before_send=before_send,
         delivery_attempt_recorder=delivery_attempt_recorder,
         short_message_limit=short_message_limit,
@@ -776,6 +783,7 @@ def verify_lark_inbox_reply(
     text: str,
     attempt: Mapping[str, Any],
     runner: CommandRunner = _default_runner,
+    source_membership_verifier: Callable[[], bool] | None = None,
 ) -> dict[str, Any]:
     """Read back one prior Lark reply without sending another message."""
 
@@ -839,21 +847,24 @@ def verify_lark_inbox_reply(
             "reply_verified": False,
             "blocker": "provider_verification_unavailable",
         }
-    membership = _call(
-        runner,
-        [
-            *base,
-            "im",
-            "chats",
-            "get",
-            "--chat-id",
-            chat_id,
-            "--as",
-            "bot",
-            "--format",
-            "json",
-        ],
-    )
+    if source_membership_verifier is not None:
+        membership = {"returncode": 0 if source_membership_verifier() else 1}
+    else:
+        membership = _call(
+            runner,
+            [
+                *base,
+                "im",
+                "chats",
+                "get",
+                "--chat-id",
+                chat_id,
+                "--as",
+                "bot",
+                "--format",
+                "json",
+            ],
+        )
     if membership.get("returncode") != 0:
         return {
             "ok": False,
diff --git a/loopx/extensions/lark/private_conversation_api.py b/loopx/extensions/lark/private_conversation_api.py
new file mode 100644
index 0000000000..66f0b7c0b8
--- /dev/null
+++ b/loopx/extensions/lark/private_conversation_api.py
@@ -0,0 +1,67 @@
+"""Loopback setup companion for native, owner-only external project Chat."""
+from __future__ import annotations
+from pathlib import Path
+from typing import Any
+
+PRIVATE_CONVERSATIONS_PATH = "/api/chat/lark/private-conversations"
+
+
+class PrivateConversationRequestMixin:
+    server: Any
+
+    def _private_conversations(self) -> None:
+        bindings = self.server.runtime_controller.project_contexts.conversation_bindings
+        current = bindings.read()
+        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()
+        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"],
+             "project_ref": row["project_ref"], "context_available": row["project_ref"] in titles, "project_title": titles.get(row["project_ref"], "Unavailable workspace"),
+             "executor_endpoint_id": row["executor_endpoint_id"], "grant": row["grant"],
+             "listener_status": health.get(row["transport_ref"], {}).get("status", "starting"),
+             **deliveries.get(row["binding_id"], {"pending_count": 0, "recovery_count": 0})}
+            for row in current["bindings"]]})
+
+    def _private_conversation_connect(self) -> None:
+        from ...extensions.lark.goal_topic_runtime import _active_profile_configs
+        from ...chat_lark_api import build_lark_goal_topic_runtime_snapshot
+        try:
+            body = self._read_json()
+            if set(body) != {"app_ref", "project_ref", "executor_endpoint_id"}:
+                raise ValueError("select an App, authorized workspace and executor")
+            profile = str(body["app_ref"])
+            existing = _active_profile_configs(build_lark_goal_topic_runtime_snapshot(
+                registry_path=self.server.registry_path, runtime_root_override=self.server.runtime_root_override))
+            from ...extensions.lark.conversation_identity import identity_ref
+            observed = self.server.runtime_controller.project_contexts.conversation_bindings.observe(profile)
+            from ...chat_lark_api import _app_identity_for_private_guard
+            group_apps = [str(config.get("bot_app_id") or _app_identity_for_private_guard(
+                other_profile, self._lark_runner(), self._require_lark_cli()))
+                for other_profile, config in existing.items()]
+            if profile in existing or any(identity_ref(app) == observed["provider_ref"] for app in group_apps):
+                raise ValueError("this App already owns a group listener; disconnect it first")
+            endpoint = str(body["executor_endpoint_id"])
+            if not any(row["agent_id"] == endpoint and row["available"] for row in self.server.runtime_controller.capabilities()):
+                raise ValueError("the selected executor is unavailable")
+            self.server.runtime_controller.project_contexts.conversation_bindings.configure(
+                transport_ref=profile, project_ref=str(body["project_ref"]), executor_endpoint_id=endpoint)
+            self.server.lark_goal_topic_runtime.refresh()
+        except (ValueError, OSError, KeyError) as exc:
+            self._send_error(str(exc), status=400)
+            return
+        self._private_conversations()
+
+    def _private_conversation_disconnect(self) -> None:
+        try:
+            body = self._read_json()
+            if set(body) != {"binding_id", "revision"}:
+                raise ValueError("disconnect the exact current binding revision")
+            self.server.runtime_controller.project_contexts.conversation_bindings.disconnect(
+                str(body["binding_id"]), expected_revision=body["revision"])
+            self.server.lark_goal_topic_runtime.refresh()
+        except (ValueError, OSError, KeyError) as exc:
+            self._send_error(str(exc), status=400)
+            return
+        self._private_conversations()
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
new file mode 100644
index 0000000000..ee3912553b
--- /dev/null
+++ b/loopx/extensions/lark/private_conversations.py
@@ -0,0 +1,257 @@
+"""Private-message transport for native, Core-bound project Chat.
+
+Only provider provenance, Inbox presentation and delivery receipts live here.
+Core owns audience grants, canonical Sessions, durable Turns and stop/recovery.
+"""
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+
+from ...capabilities.native_chat.external_conversations import ChatExternalConversations
+from ...chat_store import _atomic_write_json, _read_json
+from ...file_lock import exclusive_file_lock
+from .conversation_identity import identity_ref, lark_private_source
+from .event_inbox import acknowledge_lark_event_inbox, ingest_lark_event_inbox
+from .goal_channel_transport import call, json_payload, lark_args
+from .inbox_reply import _message, reply_lark_event_inbox, verify_lark_inbox_reply
+
+
+class LarkPrivateConversations:
+    def __init__(self, *, controller: Any, runtime_root: Path, runner: Any, cli_bin: str) -> None:
+        self.core = ChatExternalConversations(controller)
+        self.bindings = self.core.bindings
+        self.runtime_root, self.runner, self.cli_bin = runtime_root, runner, cli_bin
+        self.root = controller.store.root / "lark-private-deliveries"
+
+    def profiles(self) -> dict[str, dict[str, str]]:
+        profiles = {}
+        for row in self.bindings.read()["bindings"]:
+            try:
+                observation = self.bindings.observe(row["transport_ref"])
+            except (ValueError, OSError):
+                # One App's expired login must not block the independently
+                # verified App from acquiring its own listener lease.
+                continue
+            if any(observation.get(field) != row[field] for field in ["provider_ref", "operator_ref"]):
+                continue
+            profiles[row["transport_ref"]] = {"cli_bin": self.cli_bin, "provider_ref": row["provider_ref"],
+                "binding_id": row["binding_id"], "consumer_ref": observation["consumer_ref"]}
+        return profiles
+
+    def health(self) -> dict[str, dict[str, int]]:
+        health: dict[str, dict[str, int]] = {}
+        for path in self.root.glob("*.json"):
+            row = _read_json(path)
+            entry = health.setdefault(row["binding_id"], {"pending_count": 0, "recovery_count": 0})
+            if row["status"] != "delivered":
+                entry["pending_count"] += 1
+                entry["recovery_count"] += int(any(phase.get("started") and not phase.get("verified")
+                                                   for phase in row["deliveries"].values()))
+        return health
+
+    def _binding(self, profile: str) -> dict[str, Any]:
+        return next(row for row in self.bindings.read()["bindings"] if row["transport_ref"] == profile)
+
+    def _source_message(self, record: dict[str, Any]) -> Mapping[str, Any] | None:
+        """Read the exact source under this App, then recheck the Core audience.
+
+        A p2p source read proves the destination without requiring group-member
+        scopes. A caller cannot opt into this path with an HTTP boolean.
+        """
+        event = record["event"]
+        try:
+            self.bindings.resolve(binding_id=record["binding_id"], **record["source"])
+            result = call(self.runner, lark_args(cli_bin=self.cli_bin, profile=record["profile"],
+                tail=["im", "+messages-mget", "--message-ids", event["message_id"],
+                      "--as", "bot", "--no-reactions", "--format", "json"]))
+            message = _message(json_payload(result), event["message_id"])
+            sender = message.get("sender") if isinstance(message, Mapping) else None
+            if (result.get("returncode") == 0 and message is not None
+                    and message.get("chat_id") == event["chat_id"] and isinstance(sender, Mapping)
+                    and sender.get("id") == event["sender_id"] and sender.get("sender_type") == "user"):
+                return message
+        except (KeyError, ValueError, OSError):
+            pass
+        return None
+
+    def _source_verified(self, record: dict[str, Any]) -> bool:
+        return self._source_message(record) is not None
+
+    def _inbox(self, record: dict[str, Any]) -> Path:
+        observation = self.bindings.observe(record["profile"])
+        scope = record["source"]["source_ref"]
+        path = self.runtime_root / ".loopx" / "config" / "private-chat" / f"{scope}.json"
+        _atomic_write_json(path, {
+            "schema_version": "lark_event_inbox_config_v0", "enabled": True,
+            "inbox_dir": f".loopx/inbox/private-chat/{scope}", "capture_scope": "configured_chat_all",
+            "reply": {"enabled": True, "sender_profile": record["profile"], "sender_identity": "bot",
+                "bot_display_name": observation["bot_display_name"], "chat_id": record["event"]["chat_id"],
+                "placement_policy": "source_context", "received_reaction_emoji": ""},
+        })
+        event = record["event"]
+        # Some unsupported attachment events have no rendered content. Keep a
+        # source-backed presentation placeholder so the explicit notice can reply.
+        content = event.get("content") or "附件（内容尚不支持）"
+        captured = ingest_lark_event_inbox(project=self.runtime_root, config_path=path,
+            events=[{**event, "content": content, "schema_version": "lark_event_inbox_event_v0"}], execute=True)
+        if captured["invalid_count"]:
+            raise ValueError("private source could not be captured")
+        return path
+
+    def admit(self, profile: str, event: dict[str, Any]) -> dict[str, Any]:
+        try:
+            binding = self._binding(profile)
+            source = lark_private_source(provider_ref=binding["provider_ref"], event=event)
+            self.bindings.resolve(binding_id=binding["binding_id"], **source)
+        except (KeyError, StopIteration, ValueError):
+            return {"status": "audience_rejected"}
+        request = identity_ref(binding["provider_ref"], event["message_id"])
+        path = self.root / f"{request}.json"
+        with exclusive_file_lock(path, operation="capture_private_chat_request"):
+            if path.exists():
+                record = _read_json(path)
+                # event_id can change on redelivery; canonical message content
+                # and audience cannot change underneath the stable message id.
+                if any(record["event"].get(key) != event.get(key) for key in
+                       ["message_id", "sender_id", "chat_id", "content", "message_type"]):
+                    return {"status": "source_conflict"}
+            else:
+                record = {"schema_version": "lark_private_chat_delivery_v0", "request_ref": request,
+                    "profile": profile, "binding_id": binding["binding_id"], "source": source,
+                    "event": event, "deliveries": {}, "status": "captured"}
+                _atomic_write_json(path, record)
+            source_message = self._source_message(record)
+            if source_message is None:
+                return {"status": "source_verification_failed"}
+            message_type = str(event.get("message_type") or "")
+            text = ""
+            if message_type == "text":
+                # lark-cli renders event and mget content as plain text. Read
+                # the full canonical message; do not decode a rendered event.
+                canonical_type = source_message.get("msg_type", source_message.get("message_type"))
+                if canonical_type != "text":
+                    return {"status": "source_conflict"}
+                content = source_message.get("content")
+                if isinstance(content, str):
+                    text = content
+                else:
+                    # Raw provider envelopes are supported by the existing
+                    # readback contract too; keep parsing isolated to body.
+                    try:
+                        raw = json.loads(source_message["body"]["content"])
+                        text = raw["text"]
+                    except (ValueError, KeyError, TypeError):
+                        return {"status": "invalid_text"}
+                    if not isinstance(text, str):
+                        return {"status": "invalid_text"}
+                if not text.strip():
+                    return {"status": "empty_text"}
+            command = {"/status": "status", "/new": "new", "/stop": "stop"}.get(text.strip())
+            if message_type != "text":
+                command = "unsupported"
+            try:
+                admitted = self.core.admit(binding_id=binding["binding_id"], source=source,
+                    request_ref=request, message=text, command=command)
+            except RuntimeError as exc:
+                if str(exc) != "session_queue_full":
+                    raise
+                record.update(status="rejected", response="队列已满，本条没有被受理；请稍后重试。")
+                _atomic_write_json(path, record)
+                return {"status": "queue_rejected"}
+            record.update(status=admitted["status"], session_id=admitted["session_id"],
+                          turn_id=admitted["turn_id"], response=admitted.get("response"), response_code=admitted.get("response_code"))
+            _atomic_write_json(path, record)
+            status = "durably_accepted" if admitted["status"] == "accepted" else (
+                "queue_rejected" if admitted["status"] == "rejected" else "command_recorded")
+            return {"status": status}
+
+    def _reply_runner(self, args: list[str]) -> Any:
+        return self.runner([self.cli_bin, *args[1:]], None, 30)
+
+    def _deliver(self, path: Path, record: dict[str, Any], phase: str, text: str) -> bool:
+        if not text:
+            return False
+        phase_state = record["deliveries"].get(phase, {})
+        if phase_state.get("verified"):
+            return True
+        if phase_state.get("text") not in (None, text):
+            raise ValueError("delivery content changed after an attempt")
+        config = self._inbox(record)
+        kwargs = dict(project=self.runtime_root, config_path=config,
+            message_id=record["event"]["message_id"], text=text, runner=self._reply_runner,
+            source_membership_verifier=lambda: self._source_verified(record))
+        if phase_state.get("attempt"):
+            result = verify_lark_inbox_reply(**kwargs, attempt=phase_state["attempt"])
+        elif phase_state.get("started"):
+            # A crash between provider write and its receipt is ambiguous. Keep
+            # the original source recoverable; never resend blindly.
+            return False
+        else:
+            def before_send(_intent: str) -> dict[str, bool]:
+                phase_state.update(text=text, started=True)
+                record["deliveries"][phase] = phase_state
+                _atomic_write_json(path, record)
+                return {"continue_delivery": True}
+
+            def attempt(value: Any) -> None:
+                phase_state["attempt"] = dict(value)
+                _atomic_write_json(path, record)
+
+            result = reply_lark_event_inbox(**kwargs, execute=True, before_send=before_send,
+                                           delivery_attempt_recorder=attempt, short_message_limit=None)
+        phase_state["verified"] = result.get("reply_verified") is True
+        phase_state["blocker"] = result.get("blocker")
+        record["deliveries"][phase] = phase_state
+        _atomic_write_json(path, record)
+        return bool(phase_state["verified"])
+
+    def reconcile(self) -> int:
+        self.core.recover()
+        # A crash after Core admission but before the transport correlation
+        # record is repaired by replaying the same message identity.
+        for pending in sorted(self.root.glob("*.json")):
+            row = _read_json(pending)
+            if row["status"] == "captured":
+                self.admit(row["profile"], row["event"])
+        delivered = 0
+        for path in sorted(self.root.glob("*.json")):
+            with exclusive_file_lock(path, operation="deliver_private_chat_request"):
+                record = _read_json(path)
+                if record["status"] in {"captured", "delivered"}:
+                    continue
+                try:
+                    self.bindings.resolve(binding_id=record["binding_id"], **record["source"])
+                    if record["status"] == "accepted":
+                        # Receipt follows persistent Core admission and is
+                        # independent of terminal execution and reply delivery.
+                        self._deliver(path, record, "admission", "已持久受理；若已有执行，本条会排队。可发送 /status、/stop 或 /new。")
+                        turn = self.core.controller.store.load_turn(record["session_id"], record["turn_id"])
+                        if not turn or turn["status"] not in {"completed", "failed", "interrupted", "expired"}:
+                            continue
+                        response = str((turn.get("response") or {}).get("message") or "") if turn["status"] == "completed" else (
+                            "本次执行已停止。" if turn["status"] == "interrupted" else
+                            "本次执行失败或已过期，原会话已保留；请发送 /status 后再决定是否重试。")
+                    else:
+                        response = _command_text(str(record.get("response_code") or "")) or str(record.get("response") or "")
+                    if self._deliver(path, record, "terminal", response):
+                        config = self._inbox(record)
+                        acknowledge_lark_event_inbox(project=self.runtime_root, config_path=config,
+                            message_ids=[record["event"]["message_id"]], execute=True)
+                        self.core.record_delivery(record["request_ref"], session_id=record.get("session_id"), turn_id=record.get("turn_id"))
+                        record["status"] = "delivered"
+                        _atomic_write_json(path, record)
+                        delivered += 1
+                except (KeyError, ValueError, OSError):
+                    # Keep pending receipts available for actionable recovery.
+                    continue
+        return delivered
+
+
+def _command_text(code: str) -> str:
+    return {"unsupported_attachment": "此入口目前只支持文字；图片或文件没有交给模型。请发送文字描述。",
+        "no_session": "尚无会话；发送文字即可开始。", "active_session": "正在执行；后续文字会进入同一会话队列。",
+        "ready_session": "会话已就绪，可继续发送文字。", "new_session": "已关闭此前会话；下一条文字将开启新会话。",
+        "stop_requested": "已请求停止这条消息对应的执行。", "no_active_turn": "当前没有正在执行的消息。"}.get(code, "")
diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json
index 3bc84e714b..a3fecd352d 100644
--- a/loopx/semantics/project_registry_io_manifest_v1.json
+++ b/loopx/semantics/project_registry_io_manifest_v1.json
@@ -431,7 +431,7 @@
     },
     {
       "site": "loopx/chat_server.py::<module>.ChatRequestHandler._goal_channel_extension_ready::codec_read:load_registry#1",
-      "line": 992,
+      "line": 998,
       "column": 24,
       "kind": "codec_read",
       "api": "load_registry",
@@ -439,7 +439,7 @@
     },
     {
       "site": "loopx/chat_server.py::<module>.ChatRequestHandler._registry_and_goal::codec_read:load_registry#1",
-      "line": 519,
+      "line": 525,
       "column": 20,
       "kind": "codec_read",
       "api": "load_registry",
@@ -447,7 +447,7 @@
     },
     {
       "site": "loopx/chat_server.py::<module>.serve_chat::codec_read:load_registry#1",
-      "line": 1546,
+      "line": 1556,
       "column": 16,
       "kind": "codec_read",
       "api": "load_registry",
@@ -455,7 +455,7 @@
     },
     {
       "site": "loopx/chat_server.py::<module>.serve_chat._wake_goal_context::codec_read:load_registry#1",
-      "line": 1635,
+      "line": 1658,
       "column": 20,
       "kind": "codec_read",
       "api": "load_registry",
diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts
new file mode 100644
index 0000000000..fe8f522896
--- /dev/null
+++ b/tests/control_plane_ts/conversation_binding.test.ts
@@ -0,0 +1,76 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import {planConversationBinding, resolveBoundConversation, planBoundConversationRequest} from "../../loopx/control_plane/collaboration/conversation_binding.ts";
+import {resolveConversationScope} from "../../loopx/control_plane/collaboration/conversation_scope.ts";
+
+const project = {kind: "project_workspace", project_ref: "a".repeat(24), workspace_path: "/authorized/notes",
+  audience: "local_owner", grant: "workspace_read"};
+const row = {schema_version: "loopx_chat_conversation_binding_v0", binding_id: "b".repeat(24),
+  transport_ref: "notes-app", provider_ref: "c".repeat(24), operator_ref: "d".repeat(24),
+  context_kind: "project", project_ref: project.project_ref, executor_endpoint_id: "codex",
+  grant: "workspace_read", enabled: true};
+const observation = {transport_ref: row.transport_ref, provider_ref: row.provider_ref,
+  operator_ref: row.operator_ref, verified: true};
+const current = {schema_version: "loopx_chat_conversation_bindings_v0", revision: 0, bindings: []};
+const request = {current, expected_revision: 0, operation: "configure", binding: row, observation,
+  available_projects: [project]};
+
+test("binding independently verifies the owner and does not create a Goal", () => {
+  const result = planConversationBinding(request);
+  assert.equal(result.changed, true);
+  const next = result.state as typeof current;
+  assert.equal(next.revision, 1);
+  assert.equal("goal_id" in next.bindings[0], false);
+  const repeat = planConversationBinding({...request, current: next, expected_revision: 1,
+    binding: {...row, binding_id: "e".repeat(24)}});
+  assert.equal(repeat.changed, false);
+  for (const proof of [{...observation, verified: false}, {...observation, operator_ref: "f".repeat(24)},
+    {...observation, provider_ref: "e".repeat(24)}]) {
+    assert.throws(() => planConversationBinding({...request, observation: proof}), /independently verified/);
+  }
+  assert.throws(() => planConversationBinding({...request, available_projects: []}), /workspace/);
+  assert.throws(() => planConversationBinding({...request, expected_revision: 1}), /revision/);
+});
+
+test("one App has one binding owner, and a context change cannot move an existing Session", () => {
+  const next = planConversationBinding(request).state as typeof current;
+  const second = {...row, transport_ref: "other-app", binding_id: "e".repeat(24)};
+  assert.throws(() => planConversationBinding({...request, current: next, expected_revision: 1,
+    binding: second, observation: {...observation, transport_ref: "other-app"}}), /ownership/);
+  const other = {...project, project_ref: "f".repeat(24), workspace_path: "/authorized/other"};
+  assert.throws(() => planConversationBinding({...request, current: next, expected_revision: 1,
+    binding: {...row, project_ref: other.project_ref}, available_projects: [other]}), /new binding identity/);
+});
+
+test("external project audience has its own Session identity without portfolio or peer scope", () => {
+  const next = planConversationBinding(request).state;
+  const use = {current: next, binding_id: row.binding_id, source_ref: "e".repeat(24),
+    sender_ref: row.operator_ref, private_human_message: true, observation, available_projects: [project]};
+  const selected = resolveBoundConversation(use);
+  assert.equal(selected.channel_id, `project.external.${row.binding_id}.${use.source_ref}`);
+  assert.deepEqual(resolveConversationScope({channel_id: selected.channel_id, project_context: selected.context,
+    goal_id: null, origin: "lark"}), {kind: "project_workspace", goal_ids: [], private_conversation: false});
+  for (const bad of [{...use, sender_ref: "f".repeat(24)}, {...use, private_human_message: false},
+    {...use, available_projects: []}, {...use, observation: {...observation, provider_ref: "f".repeat(24)}}]) {
+    assert.throws(() => resolveBoundConversation(bad));
+  }
+  assert.throws(() => resolveBoundConversation({...use, source_ref: "f".repeat(24), session_context: selected.context}), /context changed/);
+  const revoked = planConversationBinding({current: next, expected_revision: 1,
+    operation: "disconnect", binding_id: row.binding_id}).state;
+  assert.throws(() => resolveBoundConversation({...use, current: revoked}), /no longer authorized/);
+});
+
+
+test("native commands retain their recorded target across a new Session and redelivery", () => {
+  const request = {request_ref: "e".repeat(24), command: "stop", target_recorded: true,
+    session_id: "original", turn_id: "original-turn"};
+  const current = {session_id: "replacement", active_turn_id: "different-turn"};
+  assert.deepEqual(planBoundConversationRequest({request, current_session: current}), {
+    operation: "stop", session_id: "original", turn_id: "original-turn", response_code: "stop_requested"});
+  assert.deepEqual(planBoundConversationRequest({request: {...request, command: null}, current_session: current}), {
+    operation: "admit_turn", client_turn_id: `external-${request.request_ref}`, session_id: "original", turn_id: null});
+  assert.equal(planBoundConversationRequest({request: {...request, command: "unsupported"},
+    current_session: null}).response_code, "unsupported_attachment");
+  assert.throws(() => planBoundConversationRequest({request: {...request, command: "grant"}, current_session: current}),
+    /unsupported/);
+});
diff --git a/tests/extensions/test_lark_goal_topic_runtime.py b/tests/extensions/test_lark_goal_topic_runtime.py
index 974cec8eb0..59082b8228 100644
--- a/tests/extensions/test_lark_goal_topic_runtime.py
+++ b/tests/extensions/test_lark_goal_topic_runtime.py
@@ -2064,11 +2064,11 @@ def provider_runner(
     assert state["consume_args"][:3] == ["fake-lark", "--profile", "mew"]
     assert "event" in state["consume_args"]
     projection = state["consume_args"][state["consume_args"].index("--jq") + 1]
-    assert "message_id:(.message_id // .id)" in projection
+    assert "message_id:(.message_id // .message.message_id // .event.message.message_id // .id)" in projection
     assert "root_id:(.root_id // .message.root_id" in projection
     assert "parent_id:(.parent_id // .reply_to" in projection
     assert "thread_id" in projection
-    assert "sender_id:(.sender_id // .sender.id // .sender.sender_id" in projection
+    assert ".event.sender.sender_id.open_id?" in projection
     assert state["reply_text"] == "当前运行的是 LoopX 开发版。"
 
 
@@ -3567,3 +3567,24 @@ def answer(route: Any, text: str):
     finally:
         for session in store.list_sessions():
             controller.close_session(session["session_id"])
+
+
+def test_message_event_projection_preserves_flat_cli_and_raw_provider_provenance():
+    import shutil
+    from loopx.extensions.lark.goal_topic_runtime import _EVENT_PROJECTION
+    jq = shutil.which("jq")
+    if jq is None:
+        pytest.skip("jq runtime is unavailable")
+    expected = {"event_id": "event_fixture", "message_id": "om_fixture", "sender_id": "ou_fixture",
+        "sender_type": "user", "chat_id": "oc_fixture", "chat_type": "p2p", "message_type": "text"}
+    flat = {**expected, "content": "plain rendered text", "create_time": "123"}
+    nested = {"header": {"event_id": expected["event_id"]}, "event": {
+        "sender": {"sender_id": {"open_id": expected["sender_id"]}, "sender_type": "user"},
+        "message": {key: value for key, value in flat.items() if key not in {"event_id", "sender_id", "sender_type"}}}}
+    for payload in [flat, nested]:
+        completed = subprocess.run([jq, "-c", _EVENT_PROJECTION], input=json.dumps(payload), text=True,
+                                   capture_output=True, check=True)
+        row = json.loads(completed.stdout)
+        for key, value in expected.items():
+            assert row[key] == value
+        assert row["content"] == "plain rendered text"
diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py
new file mode 100644
index 0000000000..9dfcda123d
--- /dev/null
+++ b/tests/test_chat_conversation_bindings.py
@@ -0,0 +1,98 @@
+"""Bound external project Chat uses Core Session, queue, and exact stop."""
+import time
+
+import pytest
+
+from test_chat_ordinary_project import ordinary  # noqa: F401, F811
+from loopx.capabilities.native_chat.conversation_bindings import ChatConversationBindings
+
+
+def test_two_bound_audiences_continue_independently_and_use_core_queue_and_stop(ordinary):  # noqa: F811
+    store, runtime, contexts, _, _, _, workspace = ordinary
+    observations = {
+        "notes-app": {"transport_ref": "notes-app", "provider_ref": "c" * 24,
+                      "operator_ref": "d" * 24, "verified": True},
+        "other-app": {"transport_ref": "other-app", "provider_ref": "e" * 24,
+                      "operator_ref": "f" * 24, "verified": True},
+    }
+    bindings = ChatConversationBindings(root=store.root, project_contexts=contexts,
+                                        observe=lambda profile: observations[profile])
+    contexts.conversation_bindings = bindings
+    ref = contexts.available()[0]["project_ref"]
+    first = bindings.configure(transport_ref="notes-app", project_ref=ref, executor_endpoint_id="codex")
+    second = bindings.configure(transport_ref="other-app", project_ref=ref, executor_endpoint_id="codex")
+    assert bindings.configure(transport_ref="notes-app", project_ref=ref,
+                              executor_endpoint_id="codex")["binding_id"] == first["binding_id"]
+
+    def open_bound(row):
+        return runtime.open_session(goal_id=None, agent_id="codex", work_dir=workspace, objective="ignored",
+            mode="resume_latest", conversation_binding_id=row["binding_id"], source_context={
+                "source_ref": "a" * 24, "sender_ref": row["operator_ref"], "private_human_message": True})[0]
+
+    try:
+        one, two = open_bound(first), open_bound(second)
+        assert one["session_id"] != two["session_id"] and one["channel_id"] != two["channel_id"]
+        assert one["goal_id"] is None and two["goal_id"] is None
+        assert open_bound(first)["session_id"] == one["session_id"]
+        blocked, _ = runtime.enqueue_turn(session_id=one["session_id"], client_turn_id="slow", message="wait for interrupt",
+            work_dir=workspace, objective="ignored", origin="lark")
+        deadline = time.monotonic() + 10
+        while store.load_session(one["session_id"])["active_turn_id"] != blocked["turn_id"]:
+            assert time.monotonic() < deadline
+            time.sleep(.01)
+        started = time.monotonic()
+        queued, created = runtime.enqueue_turn(session_id=one["session_id"], client_turn_id="later", message="follow-up",
+            work_dir=workspace, objective="ignored", origin="lark")
+        assert created and time.monotonic() - started < 3
+        assert store.load_turn(one["session_id"], queued["turn_id"])["status"] == "queued"
+        assert store.turn_for_client(one["session_id"], "later")["turn_id"] == queued["turn_id"]
+        assert runtime.enqueue_turn(session_id=one["session_id"], client_turn_id="later", message="follow-up",
+            work_dir=workspace, objective="ignored", origin="lark")[1] is False
+        independent, _ = runtime.enqueue_turn(session_id=two["session_id"], client_turn_id="independent", message="other audience",
+            work_dir=workspace, objective="ignored", origin="lark")
+        assert runtime.wait_for_turn(session_id=two["session_id"], turn_id=independent["turn_id"], timeout_sec=10)["status"] == "completed"
+        assert store.load_turn(one["session_id"], blocked["turn_id"])["status"] not in {"completed", "interrupted"}
+        runtime.interrupt_turn(session_id=one["session_id"], turn_id=blocked["turn_id"])
+        assert runtime.wait_for_turn(session_id=one["session_id"], turn_id=queued["turn_id"], timeout_sec=10)["status"] == "completed"
+        assert store.load_turn(one["session_id"], blocked["turn_id"])["status"] == "interrupted"
+        assert "other audience" not in str(store.messages(one["session_id"]))
+        assert "follow-up" not in str(store.messages(two["session_id"]))
+        with pytest.raises(ValueError, match="source"):
+            runtime.submit_turn(session_id=one["session_id"], client_turn_id="web", message="wrong source",
+                                work_dir=workspace, objective="ignored")
+        bindings.disconnect(first["binding_id"], expected_revision=bindings.read()["revision"])
+        with pytest.raises(ValueError, match="no longer authorized"):
+            runtime.enqueue_turn(session_id=one["session_id"], client_turn_id="revoked", message="old grant",
+                                 work_dir=workspace, objective="ignored", origin="lark")
+        assert store.turn_for_client(one["session_id"], "revoked") is None
+    finally:
+        runtime.close()
+
+
+def test_binding_replacement_or_identity_change_cannot_move_existing_context(tmp_path):
+    from loopx.capabilities.native_chat.project_context import ChatProjectContexts
+    from loopx.chat_store import ChatSessionStore
+
+    a, b = tmp_path / "notes", tmp_path / "other"
+    a.mkdir()
+    b.mkdir()
+    contexts = ChatProjectContexts([a, b])
+    store = ChatSessionStore(tmp_path / "runtime")
+    observation = {"transport_ref": "notes-app", "provider_ref": "c" * 24,
+                   "operator_ref": "d" * 24, "verified": True}
+    bindings = ChatConversationBindings(root=store.root, project_contexts=contexts, observe=lambda _: observation)
+    old = bindings.configure(transport_ref="notes-app", project_ref=contexts.available()[0]["project_ref"], executor_endpoint_id="codex")
+    saved = bindings.resolve(binding_id=old["binding_id"], source_ref="a" * 24,
+                             sender_ref=observation["operator_ref"], private_human_message=True)["context"]
+    observation["operator_ref"] = "f" * 24
+    with pytest.raises(ValueError, match="independently verified"):
+        bindings.session_context(saved)
+    observation["operator_ref"] = "d" * 24
+    new = bindings.configure(transport_ref="notes-app", project_ref=contexts.available()[1]["project_ref"], executor_endpoint_id="codex")
+    assert new["binding_id"] != old["binding_id"]
+    with pytest.raises(ValueError, match="no longer authorized"):
+        bindings.session_context(saved)
+    with pytest.raises(ValueError, match="revision"):
+        bindings.disconnect(new["binding_id"], expected_revision=0)
+    assert bindings.read()["bindings"][0]["binding_id"] == new["binding_id"]
+    assert bindings.path.stat().st_mode & 0o777 == 0o600
diff --git a/tests/test_lark_private_compatibility.py b/tests/test_lark_private_compatibility.py
new file mode 100644
index 0000000000..96f17ea9d3
--- /dev/null
+++ b/tests/test_lark_private_compatibility.py
@@ -0,0 +1,124 @@
+"""Private configuration must preserve the existing group listener boundary."""
+import hashlib
+import http.client
+import json
+from pathlib import Path
+import subprocess
+import sys
+import threading
+import time
+from types import SimpleNamespace
+
+import pytest
+
+from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
+from loopx.extensions.lark.conversation_identity import identity_ref
+from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX cross-process flock contract")
+def test_legacy_consumer_stays_standby_until_original_profile_owner_exits(tmp_path, monkeypatch):
+    from loopx.extensions.lark import goal_topic_runtime as runtime
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    profile = "legacy-group"
+    lease = tmp_path / ".loopx/lark-consumers" / hashlib.sha256(profile.encode()).hexdigest()[:32]
+    lease.parent.mkdir(parents=True)
+    # This is the actual prior owner key and an independent process, not a
+    # synthetic result from the new key selection rule.
+    owner = subprocess.Popen([sys.executable, "-u", "-c",
+        "import fcntl,sys; f=open(sys.argv[1],'a'); fcntl.flock(f,fcntl.LOCK_EX); "
+        "print('owned',flush=True); sys.stdin.read()", str(lease.with_name(lease.name + ".lock"))],
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
+    assert owner.stdout.readline().strip() == "owned"
+    snapshot = {"target_payload": {"targets": {"group": {"enabled": True,
+        "channel": {"chat_id": "oc_public_fixture"},
+        "identity": {"sender_profile": profile, "cli_bin": "fake-lark"}}}},
+        "binding_payloads": {"goal": {"bindings": {"goal": {"goal_id": "goal", "enabled": True,
+            "provider": "lark", "target_ref": "group", "topic": {"root_message_id": "om_public_fixture"}}}}},
+        "private_profiles": {}, "goal_contexts": {}}
+    started = threading.Event()
+    monkeypatch.setattr(runtime, "stream_lark_goal_topic_profile",
+        lambda **_: started.set() or {"ok": True, "status": "stopped"})
+    service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: snapshot,
+        runtime_root=tmp_path / "runtime", runtime_controller=object())
+    stop = threading.Event()
+    worker = threading.Thread(target=service._poll_profile, args=(profile, stop))
+    try:
+        worker.start()
+        deadline = time.monotonic() + 2
+        while service.health_snapshot().get(profile, {}).get("status") != "standby" and time.monotonic() < deadline:
+            threading.Event().wait(.01)
+        assert service.health_snapshot()[profile]["status"] == "standby"
+        assert not started.is_set()
+        stop.set()
+        worker.join(2)
+        owner.stdin.close()
+        assert owner.wait(2) == 0
+        resumed_stop = threading.Event()
+        successor = threading.Thread(target=service._poll_profile, args=(profile, resumed_stop))
+        successor.start()
+        assert started.wait(2)
+        resumed_stop.set()
+        successor.join(2)
+    finally:
+        stop.set()
+        worker.join(2)
+        service.close()
+        if owner.poll() is None:
+            owner.stdin.close()
+            owner.wait(2)
+
+
+@pytest.mark.parametrize("private_rows", [[], [{"transport_ref": "other-app", "provider_ref": identity_ref("cli_other_fixture")}],
+    [{"transport_ref": "other-alias", "provider_ref": identity_ref("cli_public_fixture")}]] )
+def test_group_http_preview_preserves_auth_cost_and_checks_real_private_aliases(tmp_path, private_rows):
+    calls = []
+    registry = {"schema_version": "0.1", "goals": [{"id": "goal", "repo": str(tmp_path),
+        "coordination": {"registered_agents": ["codex"]}}]}
+    class Handler(ChatRequestHandler):
+        def _require_lark_cli(self):
+            return "fake-lark"
+        def _goal_channel_context(self, _goal):
+            return registry, tmp_path / "binding.json"
+        def _goal_channel_target_path(self):
+            return tmp_path / "targets.json"
+        def _refresh_lark_goal_topic_runtime(self):
+            pass
+        def _lark_runner(self):
+            def run(args, _cwd, _timeout):
+                calls.append(args)
+                if args[-4:] == ["auth", "status", "--verify", "--json"]:
+                    return {"returncode": 0, "stdout": json.dumps({"appId": "cli_public_fixture", "identities": {
+                        "bot": {"openId": "ou_public_fixture", "appName": "Fixture", "available": True, "verified": True}}})}
+                return {"returncode": 0, "stdout": "{}"}
+            return run
+    server = ChatHTTPServer(("127.0.0.1", 0), Handler)
+    server.verbose = False
+    server.runtime_controller = SimpleNamespace(close=lambda: None, project_contexts=SimpleNamespace(
+        conversation_bindings=SimpleNamespace(read=lambda: {"bindings": private_rows})))
+    worker = threading.Thread(target=server.serve_forever, daemon=True)
+    worker.start()
+    def post(body):
+        conn = http.client.HTTPConnection(*server.server_address, timeout=5)
+        try:
+            conn.request("POST", "/api/chat/lark/connections", json.dumps(body), {"Content-Type": "application/json"})
+            result = conn.getresponse()
+            return result.status, json.loads(result.read())
+        finally:
+            conn.close()
+    try:
+        assert post({"app_ref": "workspace-app"})[0] == 400
+        assert not calls  # malformed structural input must never reach auth
+        code, result = post({"app_ref": "workspace-app", "goal_id": "goal", "agent_id": "codex",
+            "chat_id": "oc_public_fixture", "chat_name": "Fixture", "execute": False})
+        auth = [args for args in calls if args[-4:] == ["auth", "status", "--verify", "--json"]]
+        assert len(auth) == 1, (code, result, calls)
+        if private_rows and private_rows[0]["provider_ref"] == identity_ref("cli_public_fixture"):
+            assert code == 400 and "private listener" in result["error"]
+        else:
+            assert code == 200 and result["ok"], result
+        assert not (tmp_path / "binding.json").exists()
+    finally:
+        server.shutdown()
+        server.server_close()
+        worker.join(2)
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
new file mode 100644
index 0000000000..d1ccdcd84d
--- /dev/null
+++ b/tests/test_lark_private_conversations.py
@@ -0,0 +1,261 @@
+"""Public synthetic provider/native Core journeys, never private credentials."""
+import json
+import time
+
+import pytest
+from test_chat_ordinary_project import ordinary  # noqa: F401
+
+from loopx.capabilities.native_chat.conversation_bindings import ChatConversationBindings
+from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations
+from loopx.extensions.lark.conversation_identity import observe_lark_conversation_identity
+from loopx.extensions.lark.private_conversations import LarkPrivateConversations
+from loopx.extensions.lark.goal_topic_runtime import poll_lark_goal_topic_profile_once
+
+
+class Provider:
+    def __init__(self):
+        self.messages = {}
+        self.calls = []
+        self.writes = []
+        self.verify_replies = True
+
+    def event(self, profile, name, text, kind="text"):
+        event = {"schema_version": "lark_event_inbox_event_v0", "event_id": f"event_{name}",
+            "message_id": f"om_{name}", "chat_id": f"oc_{profile.replace('-', '_')}", "chat_type": "p2p",
+            "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}
+        return event
+
+    def __call__(self, args, cwd=None, timeout=None):
+        self.calls.append(list(args))
+        profile = args[args.index("--profile") + 1]
+        if "auth" in args:
+            data = {"ok": True, "appId": f"cli_{profile.replace('-', '_')}", "identities": {
+                "bot": {"available": True, "verified": True, "appName": profile},
+                "user": {"available": True, "verified": True, "openId": f"ou_{profile.replace('-', '_')}"}}}
+        elif "+messages-mget" in args:
+            ref = args[args.index("--message-ids") + 1]
+            message = self.messages[ref]
+            if not self.verify_replies and ref.startswith("om_out_"):
+                message = {**message, "body": {"content": json.dumps({"text": "not yet visible"})}}
+            data = {"ok": True, "data": {"items": [message]}}
+        elif "+messages-send" in args:
+            text = args[args.index("--text") + 1]
+            content = json.dumps({"text": text})
+            if "--dry-run" in args:
+                data = {"ok": True, "api": [{"body": {"content": content}}]}
+            else:
+                ref = f"om_out_{len(self.writes)}"
+                self.writes.append((profile, text))
+                self.messages[ref] = {"message_id": ref, "body": {"content": content}}
+                data = {"ok": True, "data": {"message_id": ref}}
+        else:
+            pytest.fail(f"unnecessary provider operation: {args[2:5]}")
+        return {"returncode": 0, "stdout": json.dumps(data), "stderr": ""}
+
+
+def connect(fixture):
+    store, runtime, contexts, _, _, _, _ = fixture
+    provider = Provider()
+    bindings = ChatConversationBindings(root=store.root, project_contexts=contexts,
+        observe=lambda profile: observe_lark_conversation_identity(profile=profile, runner=provider, cli_bin="lark-cli"))
+    contexts.conversation_bindings = bindings
+    for profile in ["notes-app", "steward-app"]:
+        bindings.configure(transport_ref=profile, project_ref=contexts.available()[0]["project_ref"], executor_endpoint_id="codex")
+    return store, runtime, provider, LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent,
+                                                             runner=provider, cli_bin="lark-cli")
+
+
+def test_native_private_admission_queue_other_app_stop_and_verified_delivery(ordinary):  # noqa: F811
+    store, runtime, provider, transport = connect(ordinary)
+    try:
+        first = provider.event("notes-app", "slow", "wait for interrupt")
+        assert transport.admit("notes-app", first)["status"] == "durably_accepted"
+        canonical = ChatExternalConversations(runtime).pending()[0]
+        sid = canonical["session_id"]
+        deadline = time.monotonic() + 10
+        while store.load_session(sid).get("active_turn_id") != canonical["turn_id"]:
+            assert time.monotonic() < deadline
+            time.sleep(.01)
+        queued = provider.event("notes-app", "queued", "follow-up")
+        started = time.monotonic()
+        assert transport.admit("notes-app", queued)["status"] == "durably_accepted"
+        assert time.monotonic() - started < 3
+        assert len(store.queued_turns(sid)) == 1
+        assert transport.admit("notes-app", {**queued, "event_id": "redelivery"})["status"] == "durably_accepted"
+        assert len(store.queued_turns(sid)) == 1
+        other = provider.event("steward-app", "other", "different audience")
+        assert transport.admit("steward-app", other)["status"] == "durably_accepted"
+        other_row = next(row for row in transport.core.pending() if row["message"] == "different audience")
+        assert runtime.wait_for_turn(session_id=other_row["session_id"], turn_id=other_row["turn_id"], timeout_sec=10)["status"] == "completed"
+        transport.reconcile()
+        assert any(profile == "steward-app" and text == "Runtime response." for profile, text in provider.writes)
+        assert any(profile == "notes-app" and "持久受理" in text for profile, text in provider.writes)
+        stop = provider.event("notes-app", "stop", "/stop")
+        assert transport.admit("notes-app", stop)["status"] == "command_recorded"
+        queued_row = next(row for row in transport.core.pending() if row["message"] == "follow-up")
+        assert runtime.wait_for_turn(session_id=sid, turn_id=queued_row["turn_id"], timeout_sec=10)["status"] == "completed"
+        transport.reconcile()
+        before = len(provider.writes)
+        transport.reconcile()
+        assert len(provider.writes) == before
+        assert "different audience" not in str(store.messages(sid))
+        assert not any("chats" in call for call in provider.calls)
+        assert not any("schema_version" in text for _, text in provider.writes)
+        assert all(row["goal_id"] is None for row in store.list_sessions())
+    finally:
+        runtime.close()
+
+
+def test_source_rejection_attachment_notice_and_ambiguous_reply_readback(ordinary):  # noqa: F811
+    _, runtime, provider, transport = connect(ordinary)
+    try:
+        image = provider.event("notes-app", "image", "", kind="image")
+        assert transport.admit("notes-app", {**image, "sender_type": "app"})["status"] == "audience_rejected"
+        assert transport.admit("steward-app", image)["status"] == "audience_rejected"
+        assert transport.core.pending() == []
+        assert transport.admit("notes-app", image)["status"] == "command_recorded"
+        provider.verify_replies = False
+        assert transport.reconcile() == 0
+        assert len(provider.writes) == 1 and "图片或文件没有交给模型" in provider.writes[0][1]
+        assert transport.reconcile() == 0 and len(provider.writes) == 1
+        provider.verify_replies = True
+        assert transport.reconcile() == 1 and len(provider.writes) == 1
+        assert runtime.store.list_sessions() == []
+        with pytest.raises(ValueError, match="non-default"):
+            observe_lark_conversation_identity(profile="default", runner=provider, cli_bin="lark-cli")
+    finally:
+        runtime.close()
+
+
+def test_existing_consumer_dispatches_private_admission_without_waiting_for_model():
+    event = {"chat_type": "p2p", "message_type": "text", "message_id": "om_synthetic"}
+    admitted = []
+    result = poll_lark_goal_topic_profile_once(profile="notes-app", snapshot={"private_profiles": {
+        "notes-app": {"cli_bin": "lark-cli", "provider_ref": "a" * 24}}}, runtime_root=".",
+        answer=lambda *_: pytest.fail("ordinary private Chat must not call a Goal answer"),
+        consume_runner=lambda args: {"returncode": 0, "stdout": json.dumps(event)},
+        private_admitter=lambda profile, event: admitted.append((profile, event)) or {"status": "durably_accepted"})
+    assert result["event_statuses"] == ["durably_accepted"]
+    assert admitted == [("notes-app", event)]
+
+
+def test_new_session_replay_cannot_close_a_later_session(ordinary):  # noqa: F811
+    store, runtime, provider, transport = connect(ordinary)
+    try:
+        transport.admit("notes-app", provider.event("notes-app", "first", "first"))
+        first = next(row for row in transport.core.pending() if row["message"] == "first")
+        runtime.wait_for_turn(session_id=first["session_id"], turn_id=first["turn_id"], timeout_sec=10)
+        new = provider.event("notes-app", "new", "/new")
+        transport.admit("notes-app", new)
+        transport.admit("notes-app", provider.event("notes-app", "next", "next"))
+        second = next(row for row in transport.core.pending() if row["message"] == "next")
+        assert first["session_id"] != second["session_id"]
+        transport.admit("notes-app", new)
+        assert store.load_session(second["session_id"])["status"] != "closed"
+        # Simulate a lost correlation response after original Core admission.
+        # Its stable Turn must remain in the closed original Session.
+        original_path = transport.core.root / f"{first['request_ref']}.json"
+        original = json.loads(original_path.read_text())
+        original["status"] = "prepared"
+        original_path.write_text(json.dumps(original))
+        transport.core.recover()
+        recovered = json.loads(original_path.read_text())
+        assert recovered["session_id"] == first["session_id"] and recovered["turn_id"] == first["turn_id"]
+        assert store.turn_for_client(second["session_id"], f"external-{first['request_ref']}") is None
+        runtime.wait_for_turn(session_id=second["session_id"], turn_id=second["turn_id"], timeout_sec=10)
+    finally:
+        runtime.close()
+
+
+def test_private_setup_companion_verifies_each_app_and_redacts_subjects(ordinary, tmp_path):  # noqa: F811
+    from types import SimpleNamespace
+    from loopx.extensions.lark.private_conversation_api import PrivateConversationRequestMixin
+    _, runtime, _, transport = connect(ordinary)
+    registry = tmp_path / "fresh-registry.json"
+    registry.write_text(json.dumps({"schema_version": "0.1", "goals": []}))
+    updates = []
+    server = SimpleNamespace(runtime_controller=runtime, registry_path=registry, runtime_root_override=tmp_path,
+        lark_private_conversations=transport, lark_goal_topic_runtime=SimpleNamespace(
+            refresh=lambda: updates.append(True), health_snapshot=lambda: {}))
+
+    class Handler(PrivateConversationRequestMixin):
+        def __init__(self):
+            self.server = server
+            self.body = {}
+            self.result = None
+        def _read_json(self):
+            return self.body
+        def _send_json(self, value):
+            self.result = value
+        def _send_error(self, value, **kwargs):
+            self.result = {"error": value, **kwargs}
+
+    handler = Handler()
+    try:
+        handler._private_conversations()
+        assert len(handler.result["connections"]) == 2
+        assert all(row["context_available"] for row in handler.result["connections"])
+        encoded = json.dumps(handler.result)
+        assert all(marker not in encoded for marker in ["operator_ref", "provider_ref", "workspace_path", "ou_notes", "cli_notes"])
+        handler.body = {"app_ref": "notes-app", "project_ref": "f" * 24, "executor_endpoint_id": "codex"}
+        handler._private_conversation_connect()
+        assert handler.result["status"] == 400 and not updates
+        handler.body = {"binding_id": transport.bindings.read()["bindings"][0]["binding_id"], "revision": 0}
+        handler._private_conversation_disconnect()
+        assert handler.result["status"] == 400 and not updates
+        handler.body["revision"] = transport.bindings.read()["revision"]
+        handler._private_conversation_disconnect()
+        assert handler.result["ok"] and len(handler.result["connections"]) == 1 and updates == [True]
+    finally:
+        runtime.close()
+
+
+def test_rendered_cli_text_is_preserved_and_source_type_must_agree(ordinary):  # noqa: F811
+    store, runtime, provider, transport = connect(ordinary)
+    try:
+        literal = '{"text":"this is literal user text"}'
+        event = provider.event("notes-app", "rendered", literal)
+        assert transport.admit("notes-app", event)["status"] == "durably_accepted"
+        assert transport.core.pending()[0]["message"] == literal
+        mismatched = provider.event("notes-app", "wrong_type", "not a text message")
+        provider.messages[mismatched["message_id"]]["msg_type"] = "image"
+        assert transport.admit("notes-app", mismatched)["status"] == "source_conflict"
+        assert len(transport.core.pending()) == 1
+    finally:
+        runtime.close()
+
+
+def test_private_app_alias_guard_rejects_unknown_or_unverified_identity():
+    from loopx.chat_lark_api import _app_identity_for_private_guard
+    provider = Provider()
+    assert _app_identity_for_private_guard("notes-app", provider, "lark-cli") == "cli_notes_app"
+    for data in [{}, {"appId": "cli_notes_app"}, {"appId": "malformed", "identities": {"bot": {
+        "available": True, "verified": True}}}, {"appId": "cli_notes_app", "identities": {"bot": {
+        "available": True, "verified": False}}}]:
+        def unverified(*_args, **_kwargs):
+            return {"returncode": 0, "stdout": json.dumps(data), "stderr": ""}
+        with pytest.raises(ValueError, match="could not be verified"):
+            _app_identity_for_private_guard("notes-app", unverified, "lark-cli")
+
+
+def test_one_expired_app_identity_does_not_hide_the_other_listener(ordinary):  # noqa: F811
+    _, runtime, _, transport = connect(ordinary)
+    original = transport.bindings.observe
+    def observe(profile):
+        if profile == "steward-app":
+            raise ValueError("this App owner is no longer verified")
+        return original(profile)
+    transport.bindings.observe = observe
+    try:
+        assert set(transport.profiles()) == {"notes-app"}
+        transport.bindings.observe = lambda profile: {**original(profile), "operator_ref": "f" * 24}
+        assert transport.profiles() == {}
+        transport.bindings.observe = observe
+        rejected = {"message_id": "om_unknown", "chat_id": "oc_steward_app", "sender_id": "ou_steward_app",
+                    "chat_type": "p2p", "sender_type": "user", "message_type": "text", "content": "unverified"}
+        assert transport.admit("steward-app", rejected)["status"] == "audience_rejected"
+    finally:
+        runtime.close()