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 0f8c17b30e..029cfafd99 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -742,7 +742,22 @@ origin, owner or provider claims. The TS binding/source owners decide admission
 and provenance; Python observes identity and commits under the binding fence.
 Steward ingress uses the registered coordination root and existing source policy,
 separately from execution grants. File/SQLite HTTP regressions cover full text,
-replay, FIFO and revocation races. This is a Core/API prerequisite: packaged
+replay, FIFO and revocation races. The existing session-list endpoint also accepts
+`GET /api/chat/sessions?conversation_binding_id=<binding_id>` to discover existing
+private Sessions through that same fresh binding owner. It returns the existing
+public Session summaries, retaining a project's null Goal and the original
+Session/channel; it does not create or resume a thread, enqueue work or grant
+execution. Group, unavailable and changed sources refuse; closed Sessions are
+not offered as current targets and their history remains stored. An empty,
+duplicate or mixed selector cannot fall back to the ordinary unfiltered list.
+The filtered response includes the matching `conversation_binding_id`; callers
+must require that match before adopting its targets. Older services may ignore
+the query and return an ordinary list without it, which does not qualify this
+discovery operation.
+Provider observation stays outside the publication fence, and ordinary session
+listing adds no provider IO. The product selector must use this verified original
+identity rather than equating an ordinary workspace channel with a bound Chat.
+This is a Core/API prerequisite: packaged
 frontend discovery, live shared-Lark adoption, project-recipient collaboration
 and original-conversation result return still require their existing acceptance.
 
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 e5feb0bf41..2c23c408eb 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
@@ -479,6 +479,15 @@ CLI/SSH 导出使用 `goal-portfolio --manager-view todos --goal-id GOAL --todo-
 TS binding/source owner 决定准入和来源关联，Python 观测身份并在 binding 发布锁内落队列。
 管家 ingress 沿注册的 coordination root 和现有 source policy，执行授权仍独立。
 真实 File/SQLite HTTP 回归覆盖完整文本、重放、FIFO 和撤权竞态。
+现有会话列表也支持 `GET /api/chat/sessions?conversation_binding_id=<binding_id>`，
+由同一 binding owner 重新核验后发现已经存在的本人私聊 Session。
+返回既有公开 Session 摘要，保留项目的空 Goal 和原 Session/channel；不创建或恢复线程、
+不排入工作，也不授予执行权限。群、不可用或已经变化的来源明确拒绝；关闭会话不再作为
+当前目标提供，历史继续保存。空、重复或混合筛选不能退回普通的未筛选列表。
+筛选响应包含相同的 `conversation_binding_id`；调用方采用目标前必须核对一致。
+旧服务可能忽略这个 query，返回不带该字段的普通列表；这不算已经完成此发现操作。
+Provider 观测在发布锁外完成；普通会话列表不增加 provider IO。产品选择器须采用
+这个已核验的原会话身份，不能把普通工作区 channel 当作已经绑定的 Chat。
 此项是 Core/API 前置：打包前端发现、真实共享飞书会话采用、项目接收者协作与原会话
 结果返回仍须完成各自现有验收，不以合成测试宣称整条用户旅程完成。
 
diff --git a/loopx/capabilities/native_chat/conversation_bindings.py b/loopx/capabilities/native_chat/conversation_bindings.py
index 76760e0536..8782849943 100644
--- a/loopx/capabilities/native_chat/conversation_bindings.py
+++ b/loopx/capabilities/native_chat/conversation_bindings.py
@@ -229,6 +229,46 @@ def local_input(self, saved: dict[str, Any]):
                 raise ValueError("conversation binding changed before input admission")
             yield self._resolve_observed(current, observed["binding"], observed["observation"], source)
 
+    def local_sessions(self, binding_id: str) -> list[dict[str, Any]]:
+        """Read original targets; metadata hints never replace Core's grant.
+
+        Provider IO and store inventory stay outside the publication fence.
+        Discovery neither restores a thread nor creates a Session or Turn.
+        """
+        if self.controller is None:
+            raise ValueError("the bound conversation store is unavailable")
+        row = next((item for item in self.read()["bindings"] if item["binding_id"] == binding_id), None)
+        if row is None:
+            raise ValueError("conversation binding is no longer authorized")
+        observation = dict(self.observe(row["transport_ref"]))
+        store = self.controller.store
+        sessions: list[dict[str, Any]] = []
+        for summary in store.list_sessions():
+            session = store.load_session(summary["session_id"])
+            # Narrow the IO payload only. Core independently validates every
+            # matching context, audience and canonical Session identity.
+            if session and any(isinstance(session.get(key), dict)
+                    and session[key].get("binding_id") == binding_id
+                    for key in ("project_context", "steward_context")):
+                sessions.append(session)
+        with exclusive_file_lock(self.path, operation="read_local_bound_sessions"):
+            current = self.read()
+            if next((item for item in current["bindings"] if item["binding_id"] == binding_id), None) != row:
+                raise ValueError("conversation binding changed during identity verification")
+            result = self._core("collaboration.conversation.local_sessions", {
+                "current": current, "binding_id": binding_id, "observation": observation,
+                "available_projects": self.projects.available(),
+                "sessions": [{key: session.get(key) for key in (
+                    "session_id", "goal_id", "agent_id", "executor_endpoint_id", "channel_id",
+                    "session_mode", "status", "project_context", "steward_context")}
+                    for session in sessions],
+                **({"available_goal_ids": self.goal_scope_ids(row)} if row.get("goal_scope") == "all_registered" else {}),
+            })
+            # Build the lookup only after Core has rejected duplicate identity;
+            # an IO dictionary must not silently turn ambiguity into one target.
+            by_id = {session["session_id"]: session for session in sessions}
+            return [store.public_session(by_id[sid]) for sid in result["session_ids"]]
+
     def goal_scope_ids(self, binding: dict[str, Any]) -> list[str]:
         if binding.get("goal_scope") != "all_registered":
             return list(binding.get("goal_ids", []))
diff --git a/loopx/chat_server.py b/loopx/chat_server.py
index 77a5ff98d5..a7a5399df1 100644
--- a/loopx/chat_server.py
+++ b/loopx/chat_server.py
@@ -946,7 +946,23 @@ def _session_snapshot(self, session_id: str) -> None:
             self._send_error("chat session was not found", status=404)
 
     def _list_sessions(self) -> None:
-        query = parse_qs(urlparse(self.path).query)
+        query = parse_qs(urlparse(self.path).query, keep_blank_values=True)
+        if "conversation_binding_id" in query:
+            if set(query) != {"conversation_binding_id"} or len(query["conversation_binding_id"]) != 1 or not query["conversation_binding_id"][0]:
+                return self._send_error("select one exact conversation binding without other Session filters", status=400)
+            bindings = self.server.runtime_controller.project_contexts.conversation_bindings
+            try:
+                if bindings is None:
+                    raise ValueError("conversation bindings unavailable")
+                sessions = bindings.local_sessions(query["conversation_binding_id"][0])
+            except (ValueError, OSError, KeyError):
+                # A failed provider/store read may contain private diagnostics.
+                # Refuse this selector; never fall back to an unfiltered list.
+                return self._send_error("Original private conversation is unavailable; verify its App, owner and workspace binding.", status=409)
+            # Older services ignore this new query. A matching selector in the
+            # response lets callers reject their ordinary, unfiltered result.
+            return self._send_json({"ok": True, "schema_version": "loopx_chat_session_list_v1",
+                                    "conversation_binding_id": query["conversation_binding_id"][0], "sessions": sessions})
         goal_id = _compact_text((query.get("goal_id") or [""])[0], limit=160) or None
         agent_id = _compact_text((query.get("agent_id") or [""])[0], limit=80) or None
         channel_id = _compact_text((query.get("channel_id") or [""])[0], limit=160) or None
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index 57119e2c79..7db60a86be 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -279,6 +279,16 @@ export function planConversationBinding(params: JsonObject): JsonObject {
   return {changed: true, state: next};
 }
 
+function boundProject(params: JsonObject, row: JsonObject): Record<string, string> {
+  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");
+  if (row.grant === "workspace_write" && projects[0].grant !== "workspace_write") {
+    throw new EffectRuntimeRequestError("workspace write grant is no longer available");
+  }
+  return projects[0];
+}
+
 export function resolveBoundConversation(params: JsonObject): JsonObject {
   const current = state(params.current);
   const id = ref(params.binding_id, "binding identity");
@@ -307,15 +317,10 @@ export function resolveBoundConversation(params: JsonObject): JsonObject {
       || params.group_ref !== undefined || params.topic_ref !== undefined || params.group_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");
-  if (row.grant === "workspace_write" && projects[0].grant !== "workspace_write") {
-    throw new EffectRuntimeRequestError("workspace write grant is no longer available");
-  }
+  const project = boundProject(params, row);
   const goals = row.context_kind === "steward" && row.goal_scope === "all_registered"
     ? normalizeStewardGoalScope(params.available_goal_ids) : row.goal_ids;
-  const rawContext = {...projects[0], ...topic, audience: group ? "bound_group" : "bound_owner", binding_id: id, source_ref: source,
+  const rawContext = {...project, ...topic, audience: group ? "bound_group" : "bound_owner", binding_id: id, source_ref: source,
     provider_ref: row.provider_ref, operator_ref: row.operator_ref, grant: row.grant,
     ...(row.context_kind === "steward" ? {kind: "bound_steward", grant: "portfolio_read", goal_ids: goals} : {})};
   const context = group ? normalizeProjectContext(rawContext) : rawContext;
@@ -333,11 +338,44 @@ export function resolveBoundConversation(params: JsonObject): JsonObject {
   // verified owner in this private source may inherit an explicit machine
   // owner grant. Portfolio scope and project write grants do not imply it.
   return {binding: row, context, owner_manager_audience: row.context_kind === "steward"
-      && projects[0].filesystem_scope !== "workspace_only",
+      && project.filesystem_scope !== "workspace_only",
     channel_id: row.context_kind === "steward"
     ? `manager.external.native.${id}.${source}` : `project.external.${id}.${source}`};
 }
 
+/** Discover existing private targets from canonical metadata. This is a read
+ * projection: the next input still needs fresh source and execution admission.
+ */
+export function listLocalBoundSessions(params: JsonObject): JsonObject {
+  const current = state(params.current);
+  const id = ref(params.binding_id, "binding identity");
+  const row = current.bindings.find(item => item.binding_id === id);
+  if (!row) throw new EffectRuntimeRequestError("conversation binding is no longer authorized");
+  observed(row, params.observation);
+  if (row.audience === "group") throw new EffectRuntimeRequestError("local discovery requires a private binding");
+  boundProject(params, row);
+  if (!Array.isArray(params.sessions)) throw new EffectRuntimeRequestError("canonical Session observations unavailable");
+  const ids: string[] = [];
+  for (const value of params.sessions) {
+    const session = requireJsonObject(value, "canonical Session");
+    const contexts = [session.project_context, session.steward_context];
+    if (!contexts.some(value => value && typeof value === "object" && !Array.isArray(value)
+        && (value as JsonObject).binding_id === id) || session.status === "closed") continue;
+    const steward = row.context_kind === "steward";
+    const saved = requireJsonObject(steward ? session.steward_context : session.project_context, "bound Session context");
+    const selected = resolveBoundConversation({...params, source_ref: saved.source_ref,
+      sender_ref: row.operator_ref, private_human_message: false, origin: "web", session_context: saved});
+    if (session.channel_id !== selected.channel_id || session.goal_id !== (steward ? "loopx-manager" : null)
+        || session.agent_id !== row.executor_endpoint_id || session.executor_endpoint_id !== row.executor_endpoint_id
+        || session.session_mode !== "managed_runtime" || (steward ? session.project_context : session.steward_context) != null) {
+      throw new EffectRuntimeRequestError("canonical bound Session identity changed");
+    }
+    ids.push(sessionIdentity(session.session_id, "bound Session identity"));
+  }
+  if (new Set(ids).size !== ids.length) throw new EffectRuntimeRequestError("canonical bound Session identity is ambiguous");
+  return {session_ids: ids};
+}
+
 /** Shared native command projection. Provider grammar carries the explicit
  * command; Core binds its exact target before IO. Replays retain that target.
  */
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 6319ad93db..5f25e1c69f 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -617,6 +617,7 @@ export function createEffectRuntimeHandlers(
     ["collaboration.steward.authorize_creation", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({authorizeStewardCreation}) => authorizeStewardCreation)],
     ["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.local_sessions", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({listLocalBoundSessions}) => listLocalBoundSessions)],
     ["collaboration.conversation.request", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({planBoundConversationRequest}) => planBoundConversationRequest)],
     ["collaboration.conversation.agent_target", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({resolveConversationAgentTarget}) => resolveConversationAgentTarget)],
     ["collaboration.peer.context_access", lazyHandler(() => import("./collaboration/peer_context.ts"), ({requirePeerContextAccess}) => requirePeerContextAccess)],
diff --git a/tests/control_plane/test_bound_owner_web_input.py b/tests/control_plane/test_bound_owner_web_input.py
index 0fc6d4a4b7..aba4de8782 100644
--- a/tests/control_plane/test_bound_owner_web_input.py
+++ b/tests/control_plane/test_bound_owner_web_input.py
@@ -74,6 +74,7 @@ def open_bound(kind="project"):
             source_context={"source_ref": "a" * 24, "sender_ref": observed["operator_ref"], "private_human_message": True})
         return binding, session
 
+    post.server_address = server.server_address
     yield store, runtime, bindings, observed, post, open_bound, capture, canonical
     server.shutdown()
     server.server_close()
@@ -81,6 +82,116 @@ def open_bound(kind="project"):
     runtime.close()
 
 
+def _bound_sessions(post, query):
+    connection = http.client.HTTPConnection(*post.server_address, timeout=15)
+    try:
+        connection.request("GET", "/api/chat/sessions?" + query)
+        response = connection.getresponse()
+        return response.status, json.loads(response.read())
+    finally:
+        connection.close()
+
+
+@pytest.mark.parametrize("kind", ["project", "steward"])
+def test_http_binding_discovery_selects_original_session_without_starting_work(bound_chat, kind):
+    store, runtime, _, _, post, open_bound, capture, _ = bound_chat
+    binding, original = open_bound(kind)
+    other, _ = runtime.open_session(goal_id=None, agent_id="codex", objective="", mode="new",
+        work_dir=Path(runtime.project_contexts.available()[0]["workspace_path"]),
+        project_ref=binding["project_ref"])
+    before = store.load_session(original["session_id"])
+    requests_before = capture.read_bytes()
+    status, listed = _bound_sessions(post, "conversation_binding_id=" + binding["binding_id"])
+    assert status == 200, listed
+    assert listed["conversation_binding_id"] == binding["binding_id"]
+    assert [row["session_id"] for row in listed["sessions"]] == [original["session_id"]]
+    assert listed["sessions"][0]["goal_id"] == ("loopx-manager" if kind == "steward" else None)
+    assert listed["sessions"][0]["session_id"] != other["session_id"]
+    assert store.load_session(original["session_id"]) == before
+    assert capture.read_bytes() == requests_before
+    status, ordinary = _bound_sessions(post, "")
+    assert status == 200 and {row["session_id"] for row in ordinary["sessions"]} == {
+        original["session_id"], other["session_id"]}
+    assert "conversation_binding_id" not in ordinary
+    # The discovery result is an existing identity, not a new execution grant.
+    selected = listed["sessions"][0]
+    status, accepted = post(selected, {"message": "Continue the discovered original conversation.",
+                                     "client_turn_id": "discovered"})
+    assert status == 202, accepted
+    assert runtime.wait_for_turn(session_id=selected["session_id"], turn_id=accepted["turn_id"],
+                                 timeout_sec=15)["status"] == "completed"
+    assert store.load_session(selected["session_id"])["upstream_thread_id"] == original["upstream_thread_id"]
+
+
+@pytest.mark.parametrize("query", ["conversation_binding_id=", "conversation_binding_id=x&conversation_binding_id=y",
+                                  "conversation_binding_id=x&goal_id=goal-a"])
+def test_http_binding_discovery_rejects_ambiguous_selector_without_unfiltered_fallback(bound_chat, query):
+    _, _, _, _, post, open_bound, _, _ = bound_chat
+    open_bound()
+    status, denied = _bound_sessions(post, query)
+    assert status == 400 and "sessions" not in denied
+
+
+@pytest.mark.parametrize("mutation", ["operator_ref", "verified", "workspace", "revoked", "closed"])
+def test_http_binding_discovery_rechecks_private_source_and_retains_closed_history(bound_chat, mutation):
+    store, _, bindings, observed, post, open_bound, capture, canonical = bound_chat
+    binding, session = open_bound()
+    if mutation == "operator_ref":
+        observed[mutation] = "f" * 24
+    elif mutation == "verified":
+        observed[mutation] = False
+    elif mutation == "workspace":
+        Path(session["project_context"]["workspace_path"]).rename(canonical.parent / "removed-workspace")
+    elif mutation == "revoked":
+        bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"])
+    else:
+        store.update_session(session["session_id"], status="closed")
+    before = capture.read_bytes()
+    status, result = _bound_sessions(post, "conversation_binding_id=" + binding["binding_id"])
+    if mutation == "closed":
+        assert status == 200 and result["sessions"] == []
+        assert store.load_session(session["session_id"])["status"] == "closed"
+    else:
+        assert status == 409 and "sessions" not in result
+    assert capture.read_bytes() == before
+
+
+def test_http_binding_discovery_does_not_hold_publication_fence_during_provider_io(bound_chat, monkeypatch):
+    _, _, bindings, observed, post, open_bound, _, _ = bound_chat
+    binding, _ = open_bound()
+    verifying, release = threading.Event(), threading.Event()
+    result = []
+
+    def observe(_):
+        snapshot = dict(observed)
+        verifying.set()
+        assert release.wait(5)
+        return snapshot
+
+    monkeypatch.setattr(bindings, "_observe_owner", observe)
+    reader = threading.Thread(target=lambda: result.append(_bound_sessions(post,
+        "conversation_binding_id=" + binding["binding_id"])))
+    reader.start()
+    try:
+        assert verifying.wait(5)
+        bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"])
+    finally:
+        release.set()
+        reader.join(timeout=10)
+    assert not reader.is_alive() and result[0][0] == 409
+    assert "sessions" not in result[0][1]
+
+
+def test_http_binding_discovery_preserves_duplicate_observations_for_the_typed_owner(bound_chat, monkeypatch):
+    store, _, _, _, post, open_bound, _, _ = bound_chat
+    binding, _ = open_bound()
+    observed = store.list_sessions()
+    # Inject an ambiguous storage read, never duplicate/corrupt a live store.
+    monkeypatch.setattr(store, "list_sessions", lambda **_: observed + observed)
+    status, denied = _bound_sessions(post, "conversation_binding_id=" + binding["binding_id"])
+    assert status == 409 and "sessions" not in denied
+
+
 def test_http_local_owner_continues_private_project_thread_and_queue(bound_chat):
     store, runtime, _, _, post, open_bound, capture, _ = bound_chat
     _, session = open_bound()
@@ -131,6 +242,8 @@ def test_http_local_input_cannot_inherit_a_group_topic(bound_chat):
     status, denied = post(session, {"message": "not a verified group event", "client_turn_id": "group-web"})
     assert status == 400 and denied["delivery_state"] == "not_delivered"
     assert store.queued_turns(session["session_id"]) == []
+    status, denied = _bound_sessions(post, "conversation_binding_id=" + binding["binding_id"])
+    assert status == 409 and "sessions" not in denied
 
 
 def test_http_bound_steward_keeps_web_provenance_at_coordination_root(bound_chat, monkeypatch):
diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts
index 85a9b571a8..c982083440 100644
--- a/tests/control_plane_ts/conversation_binding.test.ts
+++ b/tests/control_plane_ts/conversation_binding.test.ts
@@ -1,7 +1,8 @@
 import assert from "node:assert/strict";
 import test from "node:test";
 import {planConversationBinding, resolveBoundConversation, planBoundConversationRequest,
-  stewardCommand, authorizeStewardCreation, resolveConversationAgentTarget} from "../../loopx/control_plane/collaboration/conversation_binding.ts";
+  stewardCommand, authorizeStewardCreation, resolveConversationAgentTarget,
+  listLocalBoundSessions} from "../../loopx/control_plane/collaboration/conversation_binding.ts";
 import {resolveConversationScope, projectConversationIdentity} from "../../loopx/control_plane/collaboration/conversation_scope.ts";
 
 const project = {kind: "project_workspace", project_ref: "a".repeat(24), workspace_path: "/authorized/notes",
@@ -16,6 +17,56 @@ const current = {schema_version: "loopx_chat_conversation_bindings_v0", revision
 const request = {current, expected_revision: 0, operation: "configure", binding: row, observation,
   available_projects: [project]};
 
+for (const kind of ["project", "steward"]) {
+  const selected = kind === "steward" ? {...row, context_kind: kind, grant: "portfolio_read", goal_ids: []} : row;
+  const configured = planConversationBinding({...request, binding: selected}).state;
+  const lookup = {current: configured, binding_id: row.binding_id, observation, available_projects: [project]};
+  const context = resolveBoundConversation({...lookup, source_ref: "e".repeat(24),
+    sender_ref: row.operator_ref, private_human_message: true}).context;
+  const session = {session_id: "original", goal_id: kind === "steward" ? "loopx-manager" : null,
+    agent_id: "codex", executor_endpoint_id: "codex", session_mode: "managed_runtime", status: "ready",
+    channel_id: `${kind === "steward" ? "manager.external.native" : "project.external"}.${row.binding_id}.${"e".repeat(24)}`,
+    ...(kind === "steward" ? {steward_context: context} : {project_context: context})};
+
+  test(`${kind} discovery reuses verified saved identity and excludes unrelated or closed Sessions`, () => {
+    const snapshots = [session, {...session, session_id: "historical", status: "closed"},
+      {session_id: "local", goal_id: null, channel_id: `project.${project.project_ref}`}];
+    const before = structuredClone(snapshots);
+    assert.deepEqual(listLocalBoundSessions({...lookup, sessions: snapshots}), {session_ids: ["original"]});
+    assert.deepEqual(snapshots, before);
+    assert.deepEqual(listLocalBoundSessions({...lookup, sessions: []}), {session_ids: []});
+    assert.throws(() => listLocalBoundSessions({...lookup, sessions: [session, session]}), /ambiguous/);
+  });
+
+  test(`${kind} discovery refuses changed canonical role, executor, channel or saved source`, () => {
+    for (const change of [{goal_id: "another-goal"}, {channel_id: "manager"}, {agent_id: "other"},
+      {executor_endpoint_id: "other"}, {session_mode: "attached_host"},
+      {[kind === "steward" ? "project_context" : "steward_context"]: context},
+      {[kind === "steward" ? "steward_context" : "project_context"]:
+        {...context as Record<string, unknown>, operator_ref: "f".repeat(24)}},
+      {[kind === "steward" ? "steward_context" : "project_context"]:
+        {...context as Record<string, unknown>, source_ref: "f".repeat(24)}}]) {
+      assert.throws(() => listLocalBoundSessions({...lookup, sessions: [{...session, ...change}]}));
+    }
+    for (const change of [{current}, {available_projects: []},
+      {observation: {...observation, verified: false}},
+      {observation: {...observation, provider_ref: "f".repeat(24)}},
+      {observation: {...observation, operator_ref: "f".repeat(24)}},
+      {observation: {...observation, transport_ref: "another-app"}}]) {
+      // An empty inventory is not permission to bypass current source checks.
+      assert.throws(() => listLocalBoundSessions({...lookup, ...change, sessions: []}));
+    }
+  });
+}
+
+test("local discovery never promotes a group binding to a private recipient", () => {
+  const group = "1".repeat(24);
+  const configured = planConversationBinding({...request,
+    binding: {...row, audience: "group", group_refs: [group]}, available_group_refs: [group]}).state;
+  assert.throws(() => listLocalBoundSessions({current: configured, binding_id: row.binding_id,
+    observation, available_projects: [project], sessions: []}), /private binding/);
+});
+
 test("an unfrozen recovered stop cannot borrow current cancellation scope", () => {
   const request = {request_ref: "e".repeat(24), command: "stop"};
   const current = {session_id: "new-session", active_turn_id: "new-active"};
diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py
index 05b6c72c3b..36c160f242 100644
--- a/tests/test_chat_conversation_bindings.py
+++ b/tests/test_chat_conversation_bindings.py
@@ -42,7 +42,10 @@ def stream(**kwargs):
         return {"status": "stopped"}
     monkeypatch.setattr(goal_topic_runtime, "stream_lark_goal_topic_profile", stream)
     service = LarkGoalTopicRuntimeService(
-        snapshot_provider=lambda: {"private_profiles": transport.profiles()},
+        # Capture one listener callback. Remove its synthetic profile when the
+        # stream stops, so the real finally/refresh path cannot start replacement
+        # consumers that outlive this helper and race later tests.
+        snapshot_provider=lambda: {} if stop.is_set() else {"private_profiles": transport.profiles()},
         runtime_root=transport.runtime_root, runtime_controller=runtime,
         private_conversations=transport)
     service._poll_profile("notes-app", stop)