From dc5a1de904ad5b4872a0fd8449e965f181b11aa7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 05:42:25 +0800 Subject: [PATCH] fix(chat): freeze stop scope before native inventory Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 14 +- .../native_chat/codex_background_control.py | 18 ++- .../native_chat/external_conversations.py | 29 ++-- .../collaboration/conversation_binding.ts | 5 + .../references/repair-patterns.md | 2 +- .../conversation_binding.test.ts | 14 ++ tests/test_chat_completed_background_stop.py | 132 +++++++++++++++++- 7 files changed, 200 insertions(+), 14 deletions(-) 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 9fa6213421..f8ba9a47ba 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -111,13 +111,21 @@ 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. +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. +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. 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 +does not select newer work. If `/new` has deliberately closed the original +Session, its pending stop stays unconfirmed; recovery does not reopen it or +borrow control of the replacement Session. 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, diff --git a/loopx/capabilities/native_chat/codex_background_control.py b/loopx/capabilities/native_chat/codex_background_control.py index 05ab00dc66..1cea60ad5b 100644 --- a/loopx/capabilities/native_chat/codex_background_control.py +++ b/loopx/capabilities/native_chat/codex_background_control.py @@ -24,7 +24,8 @@ def _adapter(controller: Any, session: dict[str, Any]) -> Any: return adapter -def completed_background_turns(controller: Any, session_id: str) -> list[str]: +def completed_turn_ids(controller: Any, session_id: str) -> list[str]: + """Canonical candidates only: never start/resume a native transport here.""" 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 @@ -38,6 +39,21 @@ def completed_background_turns(controller: Any, session_id: str) -> list[str]: 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"] + return list(identities) + + +def completed_background_turns(controller: Any, session_id: str, turn_ids: list[str]) -> list[str]: + """Narrow an already frozen canonical set; later Turns are never eligible.""" + with controller._session_adapter_lock(session_id): + session = controller.store.load_session(session_id) + if not session: + raise ValueError("the original background command Session is unavailable") + identities = {} + 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") + identities[turn_id] = turn["upstream_turn_id"] if not identities: return [] adapter = _adapter(controller, session) diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 9872739c8b..d53d8c2221 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -113,7 +113,8 @@ def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str, 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} - if path.exists(): + 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 @@ -127,7 +128,7 @@ def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str, "created_at": datetime.now(timezone.utc).isoformat()} _atomic_write_json(path, row) try: - return self._admit_prepared(path, row, selected) + return self._admit_prepared(path, row, selected, recovering_request=recovering_request) except ValueError: # Invalid commands and definitive grant denials are terminal # admission outcomes; recovery must not retry them forever. @@ -135,7 +136,8 @@ def admit(self, *, binding_id: str, source: dict[str, Any], request_ref: str, _atomic_write_json(path, row) raise - def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, Any]) -> dict[str, Any]: + def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, Any], + *, recovering_request: bool = False) -> dict[str, Any]: controller = self.controller self.bindings.ensure_delivery_scope(selected) if not row.get("routing_recorded"): @@ -187,9 +189,10 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A 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"]) + # Freeze canonical candidates before any fallible native inventory + # or transport resume. Historical IO must not gate active stopping. + from .codex_background_control import completed_turn_ids + observations["background_turn_ids"] = completed_turn_ids(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"], @@ -204,6 +207,7 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A plan = effect_runtime_result("collaboration.conversation.request", { "request": {key: value for key, value in row.items() if key != "attachments"}, "attachment_count": len(row.get("attachments") or []), + "recovering_request": recovering_request, "current_session": current, "binding": selected["binding"], "agent_target": target, **observations}) operation = plan["operation"] @@ -260,8 +264,17 @@ 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 interrupt_completed_background_turns - interrupt_completed_background_turns(controller, row["session_id"], 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"]) 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/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 3e21b3c6d9..ae4784d83d 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -386,6 +386,11 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject { response_code: "unsupported_attachment"}; } if (command === "stop" || command === "new") { + if (command === "stop" && params.recovering_request === true && row.target_recorded !== true) { + // Older prepared journals may predate freezing before inventory. Their + // original scope is unknown; current work cannot reconstruct that intent. + throw new EffectRuntimeRequestError("the original stop scope was not recorded"); + } 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"); diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md index b51d560667..2110567b77 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.` 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, 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. | +| `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 the original Session and canonical candidate Turn identities before the first fallible native inventory/resume, narrow only within that set, retain targets on retry and verify native terminal disappearance. First-inventory failure followed by new active/completed work or a new Session must not retarget the old stop. Reject recovery of legacy unfrozen requests whose original scope is unknown; do not infer it from current work. 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/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index b469d148a5..3f55951879 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -16,6 +16,20 @@ const current = {schema_version: "loopx_chat_conversation_bindings_v0", revision const request = {current, expected_revision: 0, operation: "configure", binding: row, observation, available_projects: [project]}; +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"}; + assert.throws(() => planBoundConversationRequest({request, current_session: current, + recovering_request: true, background_turn_ids: ["new-background"]}), /original stop scope was not recorded/); + assert.equal(planBoundConversationRequest({request, current_session: current, + recovering_request: false}).turn_id, "new-active"); + assert.deepEqual(planBoundConversationRequest({request: {...request, target_recorded: true, + session_id: "original", turn_id: null, background_turn_ids: ["old"]}, + current_session: current, recovering_request: true, background_turn_ids: ["new-background"]}), { + operation: "stop", session_id: "original", turn_id: null, + background_turn_ids: ["old"], response_code: "stop_requested"}); +}); + test("group topics require an observed App membership and cannot inherit private recipients or portfolio", () => { const group = "1".repeat(24), topic = "2".repeat(24); const configure = {...request, binding: {...row, audience: "group", group_refs: [group]}, diff --git a/tests/test_chat_completed_background_stop.py b/tests/test_chat_completed_background_stop.py index 76757626de..90d0fe9aba 100644 --- a/tests/test_chat_completed_background_stop.py +++ b/tests/test_chat_completed_background_stop.py @@ -160,7 +160,9 @@ def test_failed_background_stop_preserves_frozen_targets_and_never_confirms_deli 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["target_recorded"] and row["session_id"] == first["session_id"] + assert first["turn_id"] in row["background_turn_ids"] + # Unknown inventory preserves canonical candidates, never idle. 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 @@ -217,3 +219,131 @@ def test_core_replay_keeps_original_background_targets_instead_of_new_observatio "background_turn_ids": ["new-background"]}) assert result == {"operation": "stop", "session_id": "original", "turn_id": None, "background_turn_ids": ["old"], "response_code": "stop_requested"} + + +@pytest.mark.parametrize("failure", ["thread/items/list", "thread/backgroundTerminals/list"]) +@pytest.mark.parametrize("later", ["active", "completed", "new_session", "resume"]) +@pytest.mark.parametrize("legacy", [False, True]) +def test_first_inventory_failure_cannot_expand_retried_stop(ordinary, monkeypatch, failure, later, legacy): # noqa: F811 + store, runtime, provider, transport, root, first, wire = _completed_with_later_answer(ordinary, monkeypatch) + sid, tid = first["session_id"], first["turn_id"] + completed_before = {m["turn_id"]: store.load_turn(sid, m["turn_id"]) + for m in store.messages(sid) if m.get("turn_id")} + native = completed_before[tid]["upstream_turn_id"] + stop = provider.topic("stop-before-inventory-fault", "/stop", root=root["message_id"]) + original_start = runtime._start_adapter + resumed = [] + interrupted_native = set() + + def install_history(session): + original = session._write + + def write(packet): + method, params = packet["method"], packet["params"] + if method == "thread/turns/list": + # Include ALL original and later native Turns, not a fixture + # that could hide a wrongly selected newer active target. + turns = {store.load_turn(sid, m["turn_id"])["upstream_turn_id"]: + store.load_turn(sid, m["turn_id"]) + for m in store.messages(sid) if m.get("turn_id")} + result = {"data": [{"id": key, "status": "interrupted" if key in interrupted_native + else "inProgress" if value["status"] in {"running", "interrupting"} + else value["status"]} for key, value in turns.items()]} + elif method == "thread/items/list" and params["turnId"] == newer_native: + result = {"data": [{"turnId": newer_native, "item": { + "type": "commandExecution", "id": "future-command", "status": "completed"}}]} + else: + if method == "turn/interrupt": + interrupted_native.add(params["turnId"]) + return original(packet) + session.messages.put({"id": packet["id"], "result": result}) + + monkeypatch.setattr(session, "_write", write) + + def start(**kwargs): + adapter = original_start(**kwargs) + if kwargs.get("resume_thread_id") == store.load_session(sid)["upstream_thread_id"]: + resumed.append(BackgroundWire(adapter.session, native, monkeypatch, children=wire.children)) + install_history(adapter.session) + return adapter + + try: + wire.failure = failure + with pytest.raises(CodexChatAgentError): + transport.admit("notes-app", stop) + prepared = next(r for r in transport.core.pending() if r["command"] == "stop") + assert prepared["status"] == "prepared" and wire.children["owned"].poll() is None + if legacy: + from loopx.chat_store import _atomic_write_json + # Persist the exact old schema's pre-inventory failure shape. An + # upgrade has no original targets to recover from this journal. + prepared.update(session_id=None, turn_id=None) + prepared.pop("target_recorded", None) + prepared.pop("background_turn_ids", None) + _atomic_write_json(transport.core.root / f"{prepared['request_ref']}.json", prepared) + if later == "new_session": + transport.admit("notes-app", provider.topic("new-after-stop-fault", "/new", root=root["message_id"])) + message = provider.topic("later-work-after-stop-fault", + "wait for interrupt" if later in {"active", "new_session"} else "Another answer", + root=root["message_id"]) + transport.admit("notes-app", message) + newer = next(r for r in transport.core.pending() if r["message"] == message["content"]) + newer_sid, newer_tid = newer["session_id"], newer["turn_id"] + if later in {"completed", "resume"}: + # Keep the failed control request pending until the explicit retry + # below; normal delivery reconciliation also retries such requests. + assert runtime.wait_for_turn(session_id=newer_sid, turn_id=newer_tid, timeout_sec=10)["status"] == "completed" + deadline = time.monotonic() + 10 + while not ((newer_turn := store.load_turn(newer_sid, newer_tid)).get("upstream_turn_id") + and newer_turn["status"] in {"running", "completed"}): + assert time.monotonic() < deadline + time.sleep(.01) + newer_native = newer_turn["upstream_turn_id"] + wire.children["future"] = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) + if sid in runtime.adapters: + install_history(runtime.adapters[sid].session) + if later == "resume": + runtime.adapters.pop(sid).close_session() + monkeypatch.setattr(runtime, "_start_adapter", start) + wire.failure = None + if legacy: + with pytest.raises(EffectRuntimeRejected, match="original stop scope was not recorded"): + transport.admit("notes-app", stop) + row = next(r for r in transport.core.pending() if r["request_ref"] == prepared["request_ref"]) + assert row["status"] == "prepared" and not row.get("target_recorded") + assert row["session_id"] is None and row["turn_id"] is None + assert all(child.poll() is None for child in wire.children.values()) + assert store.load_turn(newer_sid, newer_tid) == newer_turn + assert not any("停止" in text for _, _, text in provider.topic_writes) + return + if later == "new_session": + # /new deliberately closed the original Session. Its pending stop + # remains unconfirmed; cancellation cannot reopen it or move to the + # newer Session just to turn this failure into a successful reply. + with pytest.raises(KeyError, match="chat session was not found"): + transport.admit("notes-app", stop) + row = next(r for r in transport.core.pending() if r["request_ref"] == prepared["request_ref"]) + assert row["status"] == "prepared" and row["session_id"] == sid and row["turn_id"] is None + assert row["background_turn_ids"] == prepared["background_turn_ids"] + assert wire.children["owned"].poll() is None + assert wire.children["future"].poll() is None and wire.children["unowned"].poll() is None + assert store.load_turn(newer_sid, newer_tid) == newer_turn + assert not any("停止" in text for _, _, text in provider.topic_writes) + return + assert transport.admit("notes-app", stop)["status"] == "command_recorded" + row = next(r for r in transport.core.pending() if r["request_ref"] == prepared["request_ref"]) + assert row["session_id"] == sid and row["turn_id"] is None + assert row["background_turn_ids"] == [tid] + assert wire.children["owned"].poll() is not None + assert wire.children["future"].poll() is None and wire.children["unowned"].poll() is None + assert store.load_turn(newer_sid, newer_tid) == newer_turn + assert all(store.load_turn(sid, key) == value for key, value in completed_before.items()) + assert prepared["target_recorded"] and prepared["session_id"] == sid + assert set(prepared["background_turn_ids"]) == set(completed_before) + if later == "resume": + assert len(resumed) == 1 + transport.reconcile() + assert any(topic == root["message_id"] and "停止" in text for _, topic, text in provider.topic_writes) + finally: + wire.close() + runtime.close()