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 ba7209b463..0bd34ac03f 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -109,6 +109,20 @@ target that topic. Admission feedback, streamed posts, terminal edits and receip
 recovery use the existing delivery path with `source_thread` placement. A revoked
 binding or unreadable source prevents later result delivery.
 
+When an ordinary Codex topic has no active Turn, `/stop` also covers background
+commands left by completed Turns in that same canonical Session, even after a
+later answer. Core freezes those completed Turn identities before cancellation.
+An active Turn keeps its existing exact stop path; unavailable historical item
+inventory cannot prevent that interruption or widen its target to earlier work.
+Native command-item and terminal readback scopes the effects; unknown terminals,
+other Topics and completed answer receipts remain intact. Replaying the stop
+does not select newer work. Unavailable control or incomplete readback keeps the
+original stop unconfirmed rather than reporting an idle topic. This extends the
+existing Lark command; active-Turn HTTP/frontend control and attached or native
+Goal hosts retain their separate contracts. Synthetic provider/Core and real
+child-process regressions qualify this source slice; installed host adoption,
+live stop timing and the full community release pack remain separate exits.
+
 Disconnecting an external binding, or replacing its identity/context/grant,
 publishes the typed revocation before applying native exact Turn interruption.
 Its queued work is stopped before its active Turn, across the binding's topics
diff --git a/loopx/capabilities/native_chat/codex_background_control.py b/loopx/capabilities/native_chat/codex_background_control.py
new file mode 100644
index 0000000000..05ab00dc66
--- /dev/null
+++ b/loopx/capabilities/native_chat/codex_background_control.py
@@ -0,0 +1,63 @@
+"""Codex terminal observations/effects for Core's frozen conversation stop.
+
+Chat's canonical history supplies the eligible Turn identities. Native item
+and terminal inventories prove which of those Turns still own processes. This
+does not change completed Turn receipts or enumerate another native thread.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from ...chat_store import CHAT_SESSION_MODE_ATTACHED, TERMINAL_TURN_STATES
+from ...chat_codex_goal import UPSTREAM_MODE as CODEX_GOAL_CHAT_MODE
+
+
+def _adapter(controller: Any, session: dict[str, Any]) -> Any:
+    from ...chat_runtime import CodexAppServerAdapter
+
+    adapter = controller._ensure_adapter_locked(session, work_dir=Path("."), objective="")
+    if not isinstance(adapter, CodexAppServerAdapter) or adapter.goal_driver is not None:
+        raise ValueError("completed command control requires ordinary native Codex Chat")
+    if adapter.upstream_thread_id != session.get("upstream_thread_id"):
+        raise ValueError("background control belongs to another native thread")
+    return adapter
+
+
+def completed_background_turns(controller: Any, session_id: str) -> list[str]:
+    with controller._session_adapter_lock(session_id):
+        session = controller.store.load_session(session_id)
+        if (not session or session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED
+                or session.get("agent_id") != "codex" or session.get("upstream_mode") == CODEX_GOAL_CHAT_MODE):
+            return []
+        identities = {}
+        for message in controller.store.messages(session_id):
+            turn_id = message.get("turn_id")
+            if not turn_id or turn_id in identities:
+                continue
+            turn = controller.store.load_turn(session_id, turn_id)
+            if turn and turn.get("status") in TERMINAL_TURN_STATES and turn.get("upstream_turn_id"):
+                identities[turn_id] = turn["upstream_turn_id"]
+        if not identities:
+            return []
+        adapter = _adapter(controller, session)
+        live = set(adapter.session.background_turn_ids(list(identities.values())))
+        return [turn_id for turn_id, upstream in identities.items() if upstream in live]
+
+
+def interrupt_completed_background_turns(controller: Any, session_id: str, turn_ids: list[str]) -> None:
+    with controller._session_adapter_lock(session_id):
+        session = controller.store.load_session(session_id)
+        if not session or session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED:
+            raise ValueError("the original background command Session is unavailable")
+        upstream_ids = []
+        for turn_id in turn_ids:
+            turn = controller.store.load_turn(session_id, turn_id)
+            if not turn or turn.get("status") not in TERMINAL_TURN_STATES or not turn.get("upstream_turn_id"):
+                raise ValueError("the original completed command Turn is unavailable")
+            upstream_ids.append(turn["upstream_turn_id"])
+        adapter = _adapter(controller, session)
+        for upstream in upstream_ids:
+            # The existing interrupt owner also handles already terminal Turns
+            # and verifies that only their command items lose live terminals.
+            adapter.interrupt_turn(upstream)
diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
index 65c30de70c..cd087fcee5 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -185,6 +185,11 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
                 return row
         from ...control_plane.effect_runtime import effect_runtime_result
         observations = {}
+        if (row["command"] == "stop" and current and not current.get("active_turn_id")
+                and not row.get("target_recorded") and not target):
+            # Historical inventory must not gate the existing active-Turn stop.
+            from .codex_background_control import completed_background_turns
+            observations["background_turn_ids"] = completed_background_turns(controller, current["session_id"])
         if row["command"] in {"status", "help"}:
             active_id = current.get("active_turn_id") if current else None
             observations = {"context": selected["context"],
@@ -251,7 +256,12 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
                 row["status_snapshot"] = plan["status_snapshot"]
         elif operation in {"new", "stop"}:
             row.update(target_recorded=True, session_id=plan["session_id"], turn_id=plan["turn_id"])
+            if operation == "stop":
+                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 interrupt_completed_background_turns
+                interrupt_completed_background_turns(controller, row["session_id"], row["background_turn_ids"])
             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"]:
diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py
index 45ef9b772b..7adc4b7665 100644
--- a/loopx/chat_agent.py
+++ b/loopx/chat_agent.py
@@ -1340,25 +1340,11 @@ def interrupt(self, turn_id: str | None = None) -> None:
             # an acknowledged interrupt alone still allows late command items.
             self._control_wait(deadline)
 
-        with self._command_items_lock:
-            owned_items = set(self._command_items.get(selected_turn_id, set()))
-        for entry in self._control_items("thread/items/list", {"turnId": selected_turn_id}, deadline):
-            item = entry.get("item")
-            if entry.get("turnId") != selected_turn_id or not isinstance(item, dict):
-                raise self._runtime_error("Codex returned items outside the selected Turn.")
-            if item.get("type") == "commandExecution":
-                item_id = item.get("id")
-                if not isinstance(item_id, str) or not item_id:
-                    raise self._runtime_error("Codex returned an unbound command item.")
-                # Completed command items can still own a background terminal.
-                owned_items.add(item_id)
+        owned_items = self._command_item_ids(selected_turn_id, deadline)
 
         def owned_terminals() -> list[dict[str, Any]]:
-            terminals = list(self._control_items("thread/backgroundTerminals/list", {}, deadline))
-            for terminal in terminals:
-                if not isinstance(terminal.get("itemId"), str) or not isinstance(terminal.get("processId"), str):
-                    raise self._runtime_error("Codex returned an unbound background terminal.")
-            return [terminal for terminal in terminals if terminal["itemId"] in owned_items]
+            return [terminal for terminal in self._background_terminals(deadline)
+                    if terminal["itemId"] in owned_items]
 
         for terminal in owned_terminals():
             result = self._request("thread/backgroundTerminals/terminate", {
@@ -1371,6 +1357,38 @@ def owned_terminals() -> list[dict[str, Any]]:
         while owned_terminals():
             self._control_wait(deadline)
 
+    def background_turn_ids(self, turn_ids: list[str]) -> list[str]:
+        """Observe live terminals owned by the supplied canonical native Turns."""
+        deadline = time.monotonic() + self.response_timeout_sec
+        live_items = {terminal["itemId"] for terminal in self._background_terminals(deadline)}
+        if not live_items:
+            return []
+        return [turn_id for turn_id in dict.fromkeys(turn_ids)
+                if live_items & self._command_item_ids(turn_id, deadline)]
+
+    def _command_item_ids(self, turn_id: str, deadline: float) -> set[str]:
+        with self._command_items_lock:
+            owned_items = set(self._command_items.get(turn_id, set()))
+        for entry in self._control_items("thread/items/list", {"turnId": turn_id}, deadline):
+            item = entry.get("item")
+            if entry.get("turnId") != turn_id or not isinstance(item, dict):
+                raise self._runtime_error("Codex returned items outside the selected Turn.")
+            if item.get("type") == "commandExecution":
+                item_id = item.get("id")
+                if not isinstance(item_id, str) or not item_id:
+                    raise self._runtime_error("Codex returned an unbound command item.")
+                # Completed command items can still own a background terminal.
+                owned_items.add(item_id)
+        return owned_items
+
+    def _background_terminals(self, deadline: float) -> list[dict[str, Any]]:
+        terminals = list(self._control_items("thread/backgroundTerminals/list", {}, deadline))
+        for terminal in terminals:
+            if any(not isinstance(terminal.get(key), str) or not terminal[key]
+                   for key in ("itemId", "processId")):
+                raise self._runtime_error("Codex returned an unbound background terminal.")
+        return terminals
+
     def _control_items(self, method: str, params: dict[str, Any], deadline: float) -> Iterator[dict[str, Any]]:
         cursor = None
         seen_cursors: set[str] = set()
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index 39ea366ae7..c456cd0f39 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -386,8 +386,16 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       response_code: "unsupported_attachment"};
   }
   if (command === "stop" || command === "new") {
+    const rawBackground = command === "stop"
+      ? (row.target_recorded === true ? row.background_turn_ids : params.background_turn_ids) ?? [] : [];
+    if (!Array.isArray(rawBackground)) throw new EffectRuntimeRequestError("invalid background Turn observation");
+    const background = rawBackground.map(value => sessionIdentity(value, "completed background Turn"));
+    if (new Set(background).size !== background.length || (background.length && (!session || params.agent_target))) {
+      throw new EffectRuntimeRequestError("background Turns require the original ordinary Session");
+    }
     return {operation: command, session_id: session, turn_id: turn,
-      response_code: command === "new" ? "new_session" : turn ? "stop_requested" : "no_active_turn"};
+      ...(background.length ? {background_turn_ids: background} : {}),
+      response_code: command === "new" ? "new_session" : turn || background.length ? "stop_requested" : "no_active_turn"};
   }
   return {operation: "admit_turn", client_turn_id: `external-${request}`, session_id: session, turn_id: null};
 }
diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md
index 80c40230ea..b51d560667 100644
--- a/skills/loopx-self-repair/references/repair-patterns.md
+++ b/skills/loopx-self-repair/references/repair-patterns.md
@@ -66,7 +66,7 @@ teaches a reusable control-plane lesson.
 | `opaque_revision_observation_gap` | An event-gated read model returns a business snapshot and opaque source revision that each look valid, yet a source change during the read is never refreshed; runtime-only events may also consume the cursor past an unchecked business candidate. | Pre/post source revisions, exact Session cursor, candidate and runtime event order, returned snapshot, expected binding, and business-reader call counts. | The snapshot and revision were captured in different source generations, or runtime delivery advanced the cursor before source reconciliation. | Capture the binding-scoped revision before and after the business read and publish only a stable pair; retry or fail closed on drift. Reconcile source candidates before runtime-only delivery, bind the exact expected identity into the revision, reject cursors beyond the exact Session log, and cover mid-read replacement plus simultaneous candidate/runtime races. |
 | `task_lease_generation_aba` | A lease is released or expires, the same todo is acquired again, and an older execution key/version can renew, release, or complete the new owner generation; a remote provider may show healthy monotonic storage generations while the LoopX fence repeats. | First and second acquire payloads, terminal lease record, full `(idempotency_key, version, lease_epoch)` fences, stale writeback result, provider generation, and release/re-acquire crash boundary. | Terminal release deleted the only per-todo generation record, so acquire rebuilt version 1; key-only mutation also made the CAS version optional and let execution-key reuse bypass no-ABA. | Keep provider generation opaque. Under the existing per-goal lock, atomically retain one inactive terminal lease record; advance authority-owned `lease_epoch` only for a new ownership generation and keep CAS `version` monotonic across every mutation. Require the current key/version pair for renew, transfer, release, and terminal writeback, reject immediate reuse of a retired key, and cover local CLI plus a live provider chain. |
 | `task_session_projection_gap` | The UI says work is running without an active Turn, marks an active follow-up or failed resume completed because an earlier answer exists, or opens an unrelated task Session. | Goal state, Todo id, Session `channel_id`, active Turn, current Session status, historical answers, and typed-action receipt resource ids. | Generic activity or historical output overrode current execution facts, or Sessions were joined without the Todo id. | Keep active Turns and `resume_failed` ahead of historical completion through `taskSessionStatus`; use that result for run status, Session status, activity and progress. Join task Sessions by Todo id and create or resume the dedicated `task.<todo_id>` Session before immediate monitor work. Cover active follow-ups, failed resumes, retained history and multiple Todos under one Goal. |
-| `chat_turn_terminal_transition_race` | A completed Chat Turn is later marked interrupted, contradictory terminal events coexist, a queued interrupt reaches another active Turn, or managed/attached terminal status appears before transcript and Session closeout. | Turn status and response, terminal event order, transcript messages, active Turn id, adapter interrupt target, completion mode, and in-memory cancellation state. | Completion and interruption read stale state before independently writing terminal status and side effects, or one completion mode bypasses the shared nonterminal closeout boundary. | Compare-and-set each Turn transition under the existing Turn file lock, scope adapter interruption to the active Turn, reserve every completion mode in a nonterminal state, replay mode-specific closeout idempotently, and publish terminal status last. Cover completion-wins, queued cancellation, managed and attached terminal visibility, and restart recovery deterministically. |
+| `chat_turn_terminal_transition_race` | A completed Chat Turn is later marked interrupted, contradictory terminal events coexist, a queued interrupt reaches another active Turn, managed/attached terminal status precedes transcript closeout, or `/stop` reports no active Turn while an earlier completed Turn's background command still runs. | Turn status/response, terminal events, transcript, recorded stop target, native command-item and background-terminal inventories, completion mode and cancellation state. | Completion/interruption race on stale state, a completion mode bypasses shared closeout, or stop routing mistakes a completed answer for proof that its commands terminated. | Compare-and-set Turn transitions, reserve completion nonterminally, replay closeout idempotently and publish terminal status last. Preserve completed receipts. Keep recorded active-Turn stopping independent of historical inventory. For an idle Session, scope cancellation to independently verified background commands of its canonical completed Turns; freeze identities before effects, retain targets on retry and verify native terminal disappearance. Never cancel unowned terminals or retarget newer work; unavailable inventory is not idle. Cover completion-wins, queued cancellation, completed background work after later answers, managed/attached visibility and restart recovery. |
 | `attached_session_expired_queue_close_gap` | An idle attached Session cannot close because an expired queued Turn remains stored as `queued`, even though that Turn is no longer eligible for dispatch. | Session status and active Turn id, queued Turn `expires_at` and status, close error code, and Turn event history. | Queue expiry was settled only by the claim path, while close treated every persisted queued row as live work. | Under the existing Session and file locks, share one expiry-settlement rule between claim and close. Persist expired Turns as `timed_out` with `session_queue_expired`, emit `turn.failed`, reject close only for live queued Turns, and cover both live and expired queues. |
 | `author_owned_blocking_review_unrepresentable` | An author-owned PR receives a complete exact-head blocking review, but GitHub rejects self-`REQUEST_CHANGES` and the PR queue continues to classify the `COMMENTED` fallback as missing or malformed. | Published review state/body/title, exact review commit, machine English verdict, five required Chinese headings, and `review_conclusion.invalid_reasons`. | The review contract defined an author-owned no-blocker fallback only, so the GitHub identity restriction made a blocking conclusion impossible to encode; loose English verdict prose could also disagree with formal state. | Define separate exact `COMMENTED` titles for author-owned approval and request-changes outcomes, require a line-starting typed English verdict, validate title/verdict/state alignment, read the rendered review back at the exact head, and cover approval, blocker, untitled, and mismatched-state cases. |
 | `state_projection_gap` | `quota should-run` says `should_run=true` or `must_attempt=true`, but user and agent open todo counts are both zero while `Next Action` still contains executable work. | quota payload, active state `## Agent Todo`, `## User Todo`, `Next Action`, `refresh-state` warnings. | Active state prose has drifted away from machine todo projection. | Expand executable prose into `- [ ]` agent todos, add refresh-state warning/smoke if projection missed it. |
diff --git a/tests/test_chat_completed_background_stop.py b/tests/test_chat_completed_background_stop.py
new file mode 100644
index 0000000000..76757626de
--- /dev/null
+++ b/tests/test_chat_completed_background_stop.py
@@ -0,0 +1,219 @@
+"""Original-topic stop after a final answer; synthetic wire, real child processes."""
+import subprocess
+import sys
+import time
+
+import pytest
+
+from test_chat_ordinary_project import ordinary  # noqa: F401
+from test_lark_private_conversations import connect_group, finish_group_turn
+from loopx.chat_agent import CodexChatAgentError
+from loopx.control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
+
+
+class BackgroundWire:
+    def __init__(self, session, turn_id, monkeypatch, *, children=None):
+        self.session, self.turn_id = session, turn_id
+        self.calls, self.failure = [], None
+        self.children = children if children is not None else {
+            name: subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
+            for name in ("owned", "unowned")}
+        original = session._write
+
+        def write(packet):
+            method, params = packet["method"], packet["params"]
+            self.calls.append((method, params))
+            if method == self.failure:
+                session.messages.put({"id": packet["id"], "error": {"message": "control unavailable"}})
+                return
+            if method == "thread/backgroundTerminals/list":
+                result = {"data": [{"itemId": name + "-command", "processId": str(child.pid)}
+                                   for name, child in self.children.items() if child.poll() is None]}
+            elif method == "thread/items/list":
+                result = {"data": [{"turnId": turn_id, "item": {
+                    "type": "commandExecution", "id": "owned-command", "status": "completed"}}]
+                    if params["turnId"] == turn_id else []}
+            elif method == "thread/turns/list":
+                result = {"data": [{"id": turn_id, "status": "completed"}]}
+            elif method == "thread/backgroundTerminals/terminate":
+                child = next(p for p in self.children.values() if str(p.pid) == params["processId"])
+                assert child is self.children["owned"], "foreign terminal must survive"
+                child.terminate()
+                child.wait(timeout=5)
+                result = {"terminated": True}
+            else:
+                return original(packet)
+            session.messages.put({"id": packet["id"], "result": result})
+
+        monkeypatch.setattr(session, "_write", write)
+
+    def close(self):
+        for child in self.children.values():
+            if child.poll() is None:
+                child.terminate()
+            child.wait(timeout=5)
+
+
+def _completed_with_later_answer(fixture, monkeypatch):
+    store, runtime, provider, transport = connect_group(fixture)
+    root = provider.topic("background-root", "Start bounded background work")
+    assert transport.admit("notes-app", root)["status"] == "durably_accepted"
+    first = finish_group_turn(runtime, transport, message=root["content"])
+    sid, tid = first["session_id"], first["turn_id"]
+    native_id = store.load_turn(sid, tid)["upstream_turn_id"]
+    wire = BackgroundWire(runtime.adapters[sid].session, native_id, monkeypatch)
+    later = provider.topic("later-answer", "Keep the corrected filename", root=root["message_id"])
+    assert transport.admit("notes-app", later)["status"] == "durably_accepted"
+    finish_group_turn(runtime, transport, message=later["content"])
+    assert store.load_session(sid)["active_turn_id"] is None
+    return store, runtime, provider, transport, root, first, wire
+
+
+def test_idle_topic_stop_finds_earlier_completed_commands_and_keeps_receipts_and_peers(ordinary, monkeypatch):  # noqa: F811
+    store, runtime, provider, transport, root, first, wire = _completed_with_later_answer(ordinary, monkeypatch)
+    sid, tid = first["session_id"], first["turn_id"]
+    before = store.load_turn(sid, tid)
+    peer = provider.topic("independent-topic", "Independent answer")
+    try:
+        transport.admit("notes-app", peer)
+        other = finish_group_turn(runtime, transport, message=peer["content"])
+        peer_before = store.load_session(other["session_id"])
+        stop = provider.topic("stop-background", "/stop", root=root["message_id"])
+        assert transport.admit("notes-app", stop)["status"] == "command_recorded"
+        assert wire.children["owned"].poll() is not None
+        assert wire.children["unowned"].poll() is None
+        row = next(r for r in transport.core.pending() if r["request_ref"] != first["request_ref"]
+                   and r["command"] == "stop")
+        assert row["background_turn_ids"] == [tid] and row["turn_id"] is None
+        assert row["response_code"] == "stop_requested"
+        assert store.load_turn(sid, tid) == before  # completion is not rewritten as interruption
+        assert store.load_session(other["session_id"]) == peer_before
+        transport.reconcile()
+        assert any(topic == root["message_id"] and "停止" in text for _, topic, text in provider.topic_writes)
+        effects = sum(method.endswith("/terminate") for method, _ in wire.calls)
+        transport.admit("notes-app", stop)
+        assert sum(method.endswith("/terminate") for method, _ in wire.calls) == effects
+    finally:
+        wire.close()
+        runtime.close()
+
+
+def test_active_topic_stop_is_not_gated_by_unrelated_completed_history(ordinary, monkeypatch):  # noqa: F811
+    store, runtime, provider, transport = connect_group(ordinary)
+    root = provider.topic("completed-history", "First completed answer")
+    transport.admit("notes-app", root)
+    first = finish_group_turn(runtime, transport, message=root["content"])
+    sid = first["session_id"]
+    completed = store.load_turn(sid, first["turn_id"])
+    old_native = completed["upstream_turn_id"]
+    foreign = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
+    session = runtime.adapters[sid].session
+    original, calls = session._write, []
+
+    def write(packet):
+        method, params = packet["method"], packet["params"]
+        calls.append((method, params))
+        if method == "thread/backgroundTerminals/list":
+            result = {"data": [{"itemId": "foreign-command", "processId": str(foreign.pid)}]}
+        elif method == "thread/items/list" and params.get("turnId") == old_native:
+            session.messages.put({"id": packet["id"], "error": {"message": "old history unavailable"}})
+            return
+        else:
+            return original(packet)
+        session.messages.put({"id": packet["id"], "result": result})
+
+    monkeypatch.setattr(session, "_write", write)
+    try:
+        active = provider.topic("active-after-history", "wait for interrupt", root=root["message_id"])
+        transport.admit("notes-app", active)
+        row = next(r for r in transport.core.pending() if r["message"] == active["content"])
+        deadline = time.monotonic() + 10
+        while not (store.load_turn(sid, row["turn_id"])["status"] == "running"
+                   and store.load_turn(sid, row["turn_id"]).get("upstream_turn_id")):
+            assert time.monotonic() < deadline
+            time.sleep(.01)
+        native = store.load_turn(sid, row["turn_id"])["upstream_turn_id"]
+        stop = provider.topic("stop-active-with-history", "/stop", root=root["message_id"])
+        assert transport.admit("notes-app", stop)["status"] == "command_recorded"
+        assert runtime.wait_for_turn(session_id=sid, turn_id=row["turn_id"], timeout_sec=10)["status"] == "interrupted"
+        assert any(m == "turn/interrupt" and p["turnId"] == native for m, p in calls)
+        assert not any(m == "thread/items/list" and p["turnId"] == old_native for m, p in calls)
+        assert foreign.poll() is None and store.load_turn(sid, first["turn_id"]) == completed
+    finally:
+        monkeypatch.setattr(session, "_write", original)
+        runtime.close()
+        if foreign.poll() is None:
+            foreign.terminate()
+        foreign.wait(timeout=5)
+
+
+@pytest.mark.parametrize("failure", ["thread/backgroundTerminals/list", "thread/items/list",
+                                    "thread/backgroundTerminals/terminate"])
+def test_failed_background_stop_preserves_frozen_targets_and_never_confirms_delivery(ordinary, monkeypatch, failure):  # noqa: F811
+    store, runtime, provider, transport, root, first, wire = _completed_with_later_answer(ordinary, monkeypatch)
+    stop = provider.topic("retry-stop", "/stop", root=root["message_id"])
+    wire.failure = failure
+    try:
+        with pytest.raises(CodexChatAgentError, match="rejected"):
+            transport.admit("notes-app", stop)
+        row = next(r for r in transport.core.pending() if r["command"] == "stop")
+        if failure.endswith("/terminate"):
+            assert row["target_recorded"] and row["background_turn_ids"] == [first["turn_id"]]
+        else:
+            assert not row.get("target_recorded")  # unavailable observation is not an empty inventory
+        assert row["status"] != "command_completed" and wire.children["owned"].poll() is None
+        assert not any("停止" in text for _, _, text in provider.topic_writes)
+        wire.failure = None
+        transport.admit("notes-app", stop)
+        assert wire.children["owned"].poll() is not None and wire.children["unowned"].poll() is None
+    finally:
+        wire.close()
+        runtime.close()
+
+
+def test_completed_background_stop_resumes_original_transport_without_starting_another_turn(ordinary, monkeypatch):  # noqa: F811
+    store, runtime, provider, transport, root, first, wire = _completed_with_later_answer(ordinary, monkeypatch)
+    sid = first["session_id"]
+    native_id = store.load_turn(sid, first["turn_id"])["upstream_turn_id"]
+    thread_id = store.load_session(sid)["upstream_thread_id"]
+    original_start = runtime._start_adapter
+    resumed = []
+
+    def start(**kwargs):
+        assert kwargs["resume_thread_id"] == thread_id
+        adapter = original_start(**kwargs)
+        resumed.append(BackgroundWire(adapter.session, native_id, monkeypatch, children=wire.children))
+        return adapter
+
+    monkeypatch.setattr(runtime, "_start_adapter", start)
+    runtime.adapters.pop(sid).close_session()
+    before = store.messages(sid)
+    try:
+        stop = provider.topic("stop-after-transport-close", "/stop", root=root["message_id"])
+        assert transport.admit("notes-app", stop)["status"] == "command_recorded"
+        assert len(resumed) == 1 and wire.children["owned"].poll() is not None
+        assert wire.children["unowned"].poll() is None
+        assert store.messages(sid) == before and store.load_session(sid)["upstream_thread_id"] == thread_id
+        assert not any(method == "turn/start" for method, _ in resumed[0].calls)
+    finally:
+        wire.close()
+        runtime.close()
+
+
+@pytest.mark.parametrize("background", [[""], ["same", "same"], "not-an-array", [False]])
+def test_core_rejects_unbound_or_ambiguous_background_targets(background):
+    with pytest.raises(EffectRuntimeRejected):
+        effect_runtime_result("collaboration.conversation.request", {
+            "request": {"request_ref": "a" * 24, "command": "stop"},
+            "binding": {"audience": "group"}, "current_session": {"session_id": "original", "active_turn_id": None},
+            "background_turn_ids": background})
+
+
+def test_core_replay_keeps_original_background_targets_instead_of_new_observation():
+    result = effect_runtime_result("collaboration.conversation.request", {
+        "request": {"request_ref": "a" * 24, "command": "stop", "target_recorded": True,
+                    "session_id": "original", "turn_id": None, "background_turn_ids": ["old"]},
+        "binding": {"audience": "group"}, "current_session": {"session_id": "new", "active_turn_id": "new-active"},
+        "background_turn_ids": ["new-background"]})
+    assert result == {"operation": "stop", "session_id": "original", "turn_id": None,
+                      "background_turn_ids": ["old"], "response_code": "stop_requested"}