fix(chat): stop owned background commands after completed answers

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

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)

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"]:

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()

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};
 }

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. |