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 49dc23ed0a..9959e623bf 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -118,6 +118,12 @@ later answer. Core freezes the original Session and canonical candidate Turn
 identities in the existing request journal **before** native inventory or
 transport resume can fail. Native observation may only narrow that frozen set;
 a retry cannot add later completed/active Turns or switch to a newer Session.
+The existing Lark listener returns after this durable freeze; its existing reply
+workers recover the prepared request to observe and cancel native background
+commands. Slow or unavailable native control therefore does not hold inbound
+admission. This is scheduling only: it introduces no queue or completion state,
+does not defer an active Turn's exact interrupt, and sends no successful stop
+reply until native control and the original delivery checks have succeeded.
 A recovered legacy request without a recorded target has unknown original scope;
 Core rejects retargeting and preserves it unconfirmed. A fresh stop is a separate
 user request, never an automatic replacement for the unknown one.
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 759907f52d..19f87afc2f 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
@@ -81,6 +81,14 @@ listener、队列或模型 runner。一个 App 仍只有一个 binding owner，
 编辑和回执恢复复用现有投递路径，使用 `source_thread` 落点。解绑或原消息不可读后，
 即使执行完成，也不得继续发送结果。
 
+普通 Codex 话题空闲时，`/stop` 也覆盖同一原生 Session 中已完成 Turn 遗留的后台命令。
+Core 先在原请求日志固定 Session 与候选 Turn，原 listener 随即继续受理消息；现有
+回复工作池再恢复这条 prepared 请求，读取原生命令清单并取消原目标。恢复只能缩小
+固定候选，不能追停后来的 Turn 或新 Session。慢速或失败的原生控制不阻塞入站，
+也不提前回复“已停止”；没有固定目标的旧请求保持未知，活跃 Turn 仍沿原确切中断路径。
+这只调整现有 IO 调度，不新增队列、消费者或完成状态；实际安装、停止时延与完整社区
+准出仍需分别验收。
+
 群上下文强制使用 `workspace_only`，不受本人私聊 `host_default` 影响。只保留选定
 工作区的读写 grant，不提供个人 portfolio、直连 Agent 选择、全局 skills、MCP、
 shell profile 或继承的账号环境。原生宿主必须确认确切 permissions profile 和工作区
diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
index d53d8c2221..64f16e55a6 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -91,7 +91,7 @@ def _automation_cadence_observation(
     def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str,
               message: str, command: str | None = None,
               attachments: list[dict[str, Any]] | None = None,
-              origin: str = "lark") -> dict[str, Any]:
+              origin: str = "lark", defer_background_stop: bool = False) -> dict[str, Any]:
         import re
         if origin not in {"lark", "external"}:
             raise ValueError("unsupported external conversation origin")
@@ -106,36 +106,52 @@ def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str,
             if not isinstance(ref, str) or not re.fullmatch(r"[a-f0-9]{24}", ref):
                 raise ValueError("invalid external conversation source reference")
         path = self.root / f"{request_ref}.json"
-        with exclusive_file_lock(self.root / "source-fences" / f"{binding_id}.{source['source_ref']}.json", operation="route_external_chat_request"), exclusive_file_lock(path, operation="admit_external_chat_request"):
-            # Resolve after waiting for both fences, including exact replay.
-            # A lock-external provider probe cannot authorize the write and
-            # would duplicate this fresh authority check on every request.
-            selected = self.bindings.resolve(binding_id=binding_id, **source)
-            expected = {"binding_id": binding_id, "source": source, "message": message, "command": command,
-                        "attachments": normalize_chat_image_attachments(attachments) or None, "origin": origin}
-            recovering_request = path.exists()
-            if recovering_request:
-                row = _read_json(path)
-                # Existing Lark journals predate persisted transport origin.
-                if any((row.get(key, "lark") if key == "origin" else 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,
-                       "created_at": datetime.now(timezone.utc).isoformat()}
-                _atomic_write_json(path, row)
+        # A duplicate waits on its own request before taking the route fence;
+        # it must not hold up other messages while native cancellation runs.
+        with exclusive_file_lock(path, operation="admit_external_chat_request"):
+            with exclusive_file_lock(self.root / "source-fences" / f"{binding_id}.{source['source_ref']}.json", operation="route_external_chat_request"):
+                # Resolve after waiting for both fences, including exact replay.
+                # A lock-external provider probe cannot authorize the write and
+                # would duplicate this fresh authority check on every request.
+                selected = self.bindings.resolve(binding_id=binding_id, **source)
+                expected = {"binding_id": binding_id, "source": source, "message": message, "command": command,
+                            "attachments": normalize_chat_image_attachments(attachments) or None, "origin": origin}
+                recovering_request = path.exists()
+                if recovering_request:
+                    row = _read_json(path)
+                    # Existing Lark journals predate persisted transport origin.
+                    if any((row.get(key, "lark") if key == "origin" else 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,
+                           "created_at": datetime.now(timezone.utc).isoformat()}
+                    _atomic_write_json(path, row)
+                try:
+                    row = self._admit_prepared(path, row, selected, recovering_request=recovering_request)
+                except ValueError:
+                    self._reject_prepared(path, row)
+                    raise
+            # Native inventory/cancellation needs the frozen request fence,
+            # not the route fence. New messages retain their own fresh checks.
+            if (defer_background_stop or row["status"] != "prepared"
+                    or row.get("command") != "stop" or not row.get("background_turn_ids")):
+                return row
             try:
-                return self._admit_prepared(path, row, selected, recovering_request=recovering_request)
+                return self._finish_background_stop(path, row, selected)
             except ValueError:
-                # Invalid commands and definitive grant denials are terminal
-                # admission outcomes; recovery must not retry them forever.
-                row.update(status="rejected", response="原授权或命令格式不可用，本条未进入执行队列；请用 /agents 或 /project 查看可用入口。")
-                _atomic_write_json(path, row)
+                self._reject_prepared(path, row)
                 raise
 
+    @staticmethod
+    def _reject_prepared(path: Path, row: dict[str, Any]) -> None:
+        # Definitive command/grant denials must not retry indefinitely.
+        row.update(status="rejected", response="原授权或命令格式不可用，本条未进入执行队列；请用 /agents 或 /project 查看可用入口。")
+        _atomic_write_json(path, row)
+
     def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, Any],
                         *, recovering_request: bool = False) -> dict[str, Any]:
         controller = self.controller
@@ -264,17 +280,7 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
                 row["background_turn_ids"] = plan.get("background_turn_ids", [])
             _atomic_write_json(path, row)
             if operation == "stop" and row["background_turn_ids"]:
-                from .codex_background_control import completed_background_turns, interrupt_completed_background_turns
-                row["background_turn_ids"] = completed_background_turns(
-                    controller, row["session_id"], row["background_turn_ids"])
-                _atomic_write_json(path, row)
-                # Reuse Core's response projection after narrowing. No live
-                # terminals is idle; failed inventory never reaches this point.
-                plan = effect_runtime_result("collaboration.conversation.request", {
-                    "request": {key: value for key, value in row.items() if key != "attachments"},
-                    "current_session": current, "binding": selected["binding"], "agent_target": target})
-                if row["background_turn_ids"]:
-                    interrupt_completed_background_turns(controller, row["session_id"], row["background_turn_ids"])
+                return row  # Finish outside the route fence, under this request.
             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"]:
@@ -304,6 +310,26 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
         _atomic_write_json(path, row)
         return row
 
+    def _finish_background_stop(self, path: Path, row: dict[str, Any], selected: dict[str, Any]) -> dict[str, Any]:
+        from .codex_background_control import completed_background_turns, interrupt_completed_background_turns
+        from ...control_plane.effect_runtime import effect_runtime_result
+
+        controller = self.controller
+        row["background_turn_ids"] = completed_background_turns(
+            controller, row["session_id"], row["background_turn_ids"])
+        _atomic_write_json(path, row)
+        # Reuse the typed response after narrowing. No live terminals is idle;
+        # failed inventory never reaches this point. Later work is ineligible.
+        plan = effect_runtime_result("collaboration.conversation.request", {
+            "request": {key: value for key, value in row.items() if key != "attachments"},
+            "current_session": controller.store.load_session(row["session_id"]),
+            "binding": selected["binding"], "agent_target": row.get("agent_target")})
+        if row["background_turn_ids"]:
+            interrupt_completed_background_turns(controller, row["session_id"], row["background_turn_ids"])
+        row.update(status="command_completed", response_code=plan["response_code"])
+        _atomic_write_json(path, row)
+        return row
+
     def _record_steward_ingress(self, row: dict[str, Any], selected: dict[str, Any],
                                session: dict[str, Any], client_turn_id: str) -> None:
         if selected["binding"]["context_kind"] != "steward":
diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py
index 93b912dd0c..060e5e9258 100644
--- a/loopx/extensions/lark/goal_topic_runtime_service.py
+++ b/loopx/extensions/lark/goal_topic_runtime_service.py
@@ -5,6 +5,7 @@
 from collections.abc import Callable, Mapping
 from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime, timezone
+from functools import partial
 import hashlib
 import logging
 from pathlib import Path
@@ -311,7 +312,8 @@ 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),
+                            private_admitter=(partial(self.private_conversations.admit, defer_background_stop=True)
+                                if self.private_conversations is not None else None),
                             proposal_deliverer=deliver_proposals,
                             review_callback_handler=(
                                 handle_review_callback
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index 86d4d74cd0..c2c0cc7548 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -332,7 +332,8 @@ def _inbox(self, record: dict[str, Any], *, observation: dict[str, Any] | None =
             raise ValueError("private source could not be captured")
         return path
 
-    def admit(self, profile: str, event: dict[str, Any]) -> dict[str, Any]:
+    def admit(self, profile: str, event: dict[str, Any], *,
+              defer_background_stop: bool = False) -> dict[str, Any]:
         try:
             binding = self._binding(profile)
             source = lark_private_source(provider_ref=binding["provider_ref"], event=event)
@@ -425,7 +426,8 @@ def admit(self, profile: str, event: dict[str, Any]) -> dict[str, Any]:
                 command = "unsupported"
             try:
                 admitted = self.core.admit(binding_id=binding["binding_id"], source=source,
-                    request_ref=request, message=text, command=command, attachments=attachments)
+                    request_ref=request, message=text, command=command, attachments=attachments,
+                    defer_background_stop=defer_background_stop)
             except ValueError:
                 record.update(status="rejected", response=("群助手仅处理当前项目和话题，不能选择个人 Agent 会话；/status 查看状态，/help 查看用法。" if binding.get("audience") == "group" else "操作或原授权不可用；Agent 请先用 /agents 查看确切命令，/project 返回项目对话。新委托请使用 /delegate --tokens N 具体目标，确认或取消请使用原预览中的完整命令。"))
                 _atomic_write_json(path, record)
@@ -436,7 +438,10 @@ def admit(self, profile: str, event: dict[str, Any]) -> dict[str, Any]:
                 record.update(status="rejected", response="队列已满，本条没有被受理；请稍后重试。")
                 _atomic_write_json(path, record)
                 return {"status": "queue_rejected"}
-            record.update(status=admitted["status"], session_id=admitted["session_id"],
+            # Keep the existing recoverable capture until Core has performed
+            # the frozen stop. Older transport readers can recover it too.
+            record.update(status="captured" if admitted["status"] == "prepared" else 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 (
@@ -765,6 +770,23 @@ def reconcile_request(self, path: Path) -> int:
             # read its result before retrying admission from a stale snapshot.
             with exclusive_file_lock(path, operation="capture_private_chat_request"):
                 row = _read_json(path)
+                native_path = self.core.root / f"{row['request_ref']}.json"
+                if row["status"] == "captured" and native_path.exists():
+                    native = self.core.read_request(row["request_ref"])
+                    if (native.get("command") == "stop" and native.get("target_recorded") is True
+                            and all(native.get(key) == row.get(key)
+                                    for key in ("request_ref", "binding_id", "source"))):
+                        # This source was already admitted and target-frozen.
+                        # Core revalidates recovery; delivery below re-reads the
+                        # provider source. No second content admission is needed.
+                        self.core.recover(request_ref=row["request_ref"])
+                        native = self.core.read_request(row["request_ref"])
+                        if native["status"] == "prepared":
+                            return 0
+                        if native["status"] == "command_completed":
+                            row.update(**{key: native.get(key) for key in
+                                ("status", "session_id", "turn_id", "response", "response_code")})
+                            _atomic_write_json(path, row)
             if row["status"] == "captured":
                 self.admit(row["profile"], row["event"])
                 row = _read_json(path)
diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py
index a433b62e51..69a77ea6d7 100644
--- a/tests/test_chat_conversation_bindings.py
+++ b/tests/test_chat_conversation_bindings.py
@@ -19,6 +19,210 @@ def _wait_for_native_turn(store, session_id, turn_id):
         time.sleep(.01)
 
 
+def _listener_private_admitter(transport, runtime, monkeypatch):
+    """Use the shipped listener callback, without opening a provider stream."""
+    import threading
+    from pathlib import Path
+    from loopx.extensions.lark import goal_topic_runtime
+    from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService
+
+    stop, callbacks = threading.Event(), []
+    monkeypatch.setattr(Path, "home", lambda: runtime.store.root.parent / "listener-home")
+    def stream(**kwargs):
+        callbacks.append(kwargs["private_admitter"])
+        stop.set()
+        return {"status": "stopped"}
+    monkeypatch.setattr(goal_topic_runtime, "stream_lark_goal_topic_profile", stream)
+    service = LarkGoalTopicRuntimeService(
+        snapshot_provider=lambda: {"private_profiles": transport.profiles()},
+        runtime_root=transport.runtime_root, runtime_controller=runtime,
+        private_conversations=transport)
+    service._poll_profile("notes-app", stop)
+    assert len(callbacks) == 1
+    return callbacks[0]
+
+
+def _listener_background_stop_journey(fixture, monkeypatch, failure, *, owner=False):
+    import threading
+    from concurrent.futures import ThreadPoolExecutor
+    from test_chat_completed_background_stop import BackgroundWire, _completed_with_later_answer
+    from test_lark_private_conversations import connect, finish_group_turn
+    from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations
+    from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService
+    import loopx.extensions.lark.private_conversations as private
+
+    if owner:
+        store, runtime, provider, transport = connect(fixture)
+        root = provider.event("notes-app", "background-root", "Start bounded background work")
+        transport.admit("notes-app", root)
+        first = finish_group_turn(runtime, transport, message=root["content"])
+        sid = first["session_id"]
+        wire = BackgroundWire(runtime.adapters[sid].session,
+            store.load_turn(sid, first["turn_id"])["upstream_turn_id"], monkeypatch)
+        later = provider.event("notes-app", "later-answer", "Keep the corrected filename")
+        transport.admit("notes-app", later)
+        finish_group_turn(runtime, transport, message=later["content"])
+    else:
+        store, runtime, provider, transport, root, first, wire = _completed_with_later_answer(fixture, monkeypatch)
+    def event(name, text):
+        return (provider.event("notes-app", name, text) if owner else
+                provider.topic(name, text, root=root["message_id"]))
+    sid = first["session_id"]
+    completed = {row["turn_id"]: store.load_turn(sid, row["turn_id"])
+                 for row in transport.core.pending() if row.get("turn_id")}
+    admit = _listener_private_admitter(transport, runtime, monkeypatch)
+    stop = event("deferred-background-stop", "/stop")
+    service, worker = None, None
+    entered, release, duplicate_started = threading.Event(), threading.Event(), threading.Event()
+    callers = ThreadPoolExecutor(max_workers=2)
+    try:
+        wire.calls.clear()
+        if failure == "transport-journal":
+            write = private._atomic_write_json
+            def interrupted(path, value):
+                if path.parent == transport.root and value.get("status") == "captured" and value.get("session_id"):
+                    raise OSError("transport journal interrupted after Core freeze")
+                write(path, value)
+            with monkeypatch.context() as patch:
+                patch.setattr(private, "_atomic_write_json", interrupted)
+                with pytest.raises(OSError, match="after Core freeze"):
+                    admit("notes-app", stop)
+        else:
+            assert admit("notes-app", stop)["status"] == "command_recorded"
+        frozen = next(r for r in transport.core.pending() if r["command"] == "stop")
+        assert frozen["status"] == "prepared" and frozen["target_recorded"]
+        assert json.loads((transport.root / f"{frozen['request_ref']}.json").read_text())["status"] == "captured"
+        assert set(frozen["background_turn_ids"]) == set(completed)
+        assert not wire.calls and wire.children["owned"].poll() is None
+        if failure and failure != "transport-journal":
+            if failure == "identity":
+                provider.bot_ready = False
+            else:
+                wire.failure = failure
+            # Existing recovery retains the exact pending request on native
+            # failure. It must neither report success nor select another Turn.
+            transport.reconcile()
+            pending = transport.core.read_request(frozen["request_ref"])
+            assert pending["status"] == "prepared"
+            assert first["turn_id"] in pending["background_turn_ids"]
+            assert set(pending["background_turn_ids"]) <= set(frozen["background_turn_ids"])
+            assert wire.children["owned"].poll() is None
+            assert not any("停止" in text for _, text in provider.writes)
+            provider.bot_ready = True
+        # Reopen the journal owner as after a service restart. No new queue,
+        # reconstructed target or replacement model request is supplied.
+        transport.core = ChatExternalConversations(runtime)
+        wire.failure = None
+        write = wire.session._write
+        def delayed(packet):
+            if packet["method"] == "thread/backgroundTerminals/list" and not release.is_set():
+                # Block inside the native RPC, with its adapter lock held.
+                # New admission must persist without waiting for native IO.
+                entered.set()
+                assert release.wait(10)
+            return write(packet)
+        monkeypatch.setattr(wire.session, "_write", delayed)
+        service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {},
+            runtime_root=transport.runtime_root, runtime_controller=runtime,
+            private_conversations=transport)
+        worker = threading.Thread(target=service._reconcile_private)
+        worker.start()
+        assert entered.wait(5)
+        def duplicate():
+            duplicate_started.set()
+            return transport.core.admit(**{key: frozen[key] for key in
+                ["binding_id", "source", "request_ref", "message", "command", "attachments", "origin"]})
+        repeated = callers.submit(duplicate)
+        assert duplicate_started.wait(2)
+        newer_event = event("newer-during-background-stop", "wait for interrupt")
+        accepted = callers.submit(admit, "notes-app", newer_event)
+        assert accepted.result(timeout=5)["status"] == "durably_accepted"
+        newer = next(r for r in transport.core.pending() if r["message"] == newer_event["content"])
+        assert newer["session_id"] == sid
+        assert store.load_turn(sid, newer["turn_id"])["status"] in {"queued", "starting", "running"}
+        assert newer["turn_id"] not in frozen["background_turn_ids"]
+        assert not release.is_set() and not repeated.done()
+        assert wire.children["owned"].poll() is None
+        assert not any("停止" in text for _, text in provider.writes)
+        release.set()
+        assert repeated.result(timeout=10)["status"] == "command_completed"
+        _wait_for_native_turn(store, sid, newer["turn_id"])
+        newer_native = store.load_turn(sid, newer["turn_id"])["upstream_turn_id"]
+        deadline = time.monotonic() + 10
+        while True:
+            record = json.loads((transport.root / f"{frozen['request_ref']}.json").read_text())
+            if record["status"] == "delivered":
+                break
+            assert time.monotonic() < deadline, record
+            time.sleep(.01)
+        service._closed.set()
+        worker.join(10)
+        assert not worker.is_alive()
+        recovered = transport.core.read_request(frozen["request_ref"])
+        assert recovered["status"] == "command_completed"
+        assert recovered["background_turn_ids"] == [first["turn_id"]]
+        assert recovered["session_id"] == sid and recovered["turn_id"] is None
+        assert wire.children["owned"].poll() is not None and wire.children["unowned"].poll() is None
+        assert store.load_turn(sid, newer["turn_id"])["status"] == "running"
+        assert not any(method == "turn/interrupt" and params["turnId"] == newer_native
+                       for method, params in wire.calls)
+        assert all(store.load_turn(sid, key) == value for key, value in completed.items())
+        replies = list(provider.writes)
+        assert sum("停止" in text for _, text in replies) == 1
+        terminal = dict(record["deliveries"]["terminal"])
+        stop_reply = ("notes-app", terminal["text"])
+        assert provider.writes.count(stop_reply) == 1
+        if not owner:
+            assert any(topic == root["message_id"] and "停止" in text for _, topic, text in provider.topic_writes)
+        terminations = [call for call in wire.calls if call[0].endswith("/terminate")]
+        admit("notes-app", stop)
+        transport.reconcile()
+        # The newly admitted Turn may emit its own first progress now. Only
+        # the original stop's frozen reply/effect must remain unchanged.
+        assert provider.writes.count(stop_reply) == 1
+        assert json.loads((transport.root / f"{frozen['request_ref']}.json").read_text())["deliveries"]["terminal"] == terminal
+        assert [call for call in wire.calls if call[0].endswith("/terminate")] == terminations
+    finally:
+        release.set()
+        callers.shutdown(wait=True)
+        if service:
+            service._closed.set()
+        if worker:
+            worker.join(10)
+        wire.close()
+        runtime.close()
+
+
+@pytest.mark.parametrize("failure", [None, "thread/items/list", "thread/backgroundTerminals/terminate", "transport-journal", "identity"])
+def test_listener_background_stop_freezes_then_accepts_new_native_work_before_recovery(ordinary, monkeypatch, failure):  # noqa: F811
+    _listener_background_stop_journey(ordinary, monkeypatch, failure)
+
+
+def test_private_listener_background_stop_keeps_new_native_work_unblocked(ordinary, monkeypatch):  # noqa: F811
+    _listener_background_stop_journey(ordinary, monkeypatch, None, owner=True)
+
+
+def test_listener_active_stop_keeps_immediate_exact_interruption(ordinary, monkeypatch):  # noqa: F811
+    from test_lark_private_conversations import connect_group
+    import loopx.capabilities.native_chat.codex_background_control as background
+
+    store, runtime, provider, transport = connect_group(ordinary)
+    root = provider.topic("active-listener-stop", "wait for interrupt")
+    admit = _listener_private_admitter(transport, runtime, monkeypatch)
+    try:
+        admit("notes-app", root)
+        first = transport.core.pending()[0]
+        _wait_for_native_turn(store, first["session_id"], first["turn_id"])
+        monkeypatch.setattr(background, "completed_turn_ids", lambda *_: pytest.fail("active stop cannot wait on history"))
+        stop = provider.topic("stop-active-listener", "/stop", root=root["message_id"])
+        assert admit("notes-app", stop)["status"] == "command_recorded"
+        row = next(r for r in transport.core.pending() if r["command"] == "stop")
+        assert row["status"] == "command_completed" and row["turn_id"] == first["turn_id"]
+        assert runtime.wait_for_turn(session_id=first["session_id"], turn_id=first["turn_id"], timeout_sec=10)["status"] == "interrupted"
+    finally:
+        runtime.close()
+
+
 def test_binding_observation_is_opt_in_fresh_and_never_part_of_the_context(ordinary):  # noqa: F811
     store, runtime, contexts, _, _, _, _ = ordinary
     observation = {"transport_ref": "notes-app", "provider_ref": "c" * 24,