fix(chat): defer historical native probes during active work

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -1847,6 +1847,10 @@ card cannot substitute for final-message readback.
 
 ### Recovering a native result after Core failure
 
+Historical native-result discovery waits while the same Core Session is busy
+or has an active Turn, leaving shared reply workers available for current
+progress. Already verified result metadata keeps its existing correlation owner.
+
 An Antigravity owner request can finish natively while Core retains a transport
 failure. The original reply worker may adopt that existing answer, without
 admitting another Turn, sending native input or cancelling an invocation. The

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -1210,6 +1210,9 @@ reaction 清理未完成时，也只恢复原结果 proof，不能再次发送
 
 ### Core 失败后的原生结果恢复
 
+同一 Core Session 忙碌或仍有活跃 Turn 时，历史原生结果暂不进入读回发现，避免
+占用当前进度所需的共享回复 worker。已核验结果的元数据仍由原关联修复路径负责。
+
 Antigravity 原生端可能已经完成，而 Core 仍保留 transport 失败。原回复 worker
 可以采用既有答复，不准入另一个 Turn，不发送原生输入，也不取消执行。会话的
 typed owner 核验原私聊管家 binding、来源、失败 Turn 和当前授权，并排除附件、

loopx/control_plane/collaboration/conversation_binding.ts

@@ -606,6 +606,9 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
     if (adopted?.delivery_verified === true) {
       return planNativeMetadataReview(recovery, row, current, failed, adopted, version);
     }
+    // A fresh native result requires an idle, stable trajectory. Do not spend
+    // shared reply workers probing older results while this Session is busy.
+    if (current.active_turn_id != null || current.status !== "ready") return no;
     if (row.context_recovery != null) return no;
     const candidate = {...no, operation: "adopt_native_result", recovery_candidate: true,
       recover_native_result: true, failed_turn_id: failed.turn_id, session_id: current.session_id};

tests/control_plane_ts/conversation_binding.test.ts

@@ -138,7 +138,9 @@ test("native terminal recovery adopts a result without Turn admission or lost fa
     {failed_turn: {...failed, error_code: "host_gate"}}, {failed_turn: {...failed, status: "interrupted"}}]) {
     assert.equal(planBoundConversationRequest({...packet, context_recovery: {...packet.context_recovery, ...change}}).recovery_eligible, false);
   }
-  for (const change of [{status: "closed"}, {session_mode: "attached_host"}, {agent_id: "codex"}]) {
+  for (const change of [{status: "closed"}, {session_mode: "attached_host"}, {agent_id: "codex"},
+    {status: "busy", active_turn_id: "current"}, {status: "ready", active_turn_id: "current"},
+    {status: "busy", active_turn_id: null}]) {
     assert.equal(planBoundConversationRequest({...packet, current_session: {...session, ...change}}).recovery_candidate, false);
   }
   assert.throws(() => planBoundConversationRequest({...packet,

tests/test_lark_private_conversations.py

@@ -328,6 +328,70 @@ def crash(*args, **kwargs):
         fixture.close()
 
 
+def test_active_original_reply_does_not_wait_for_older_native_result_probes(tmp_path, language_server, monkeypatch):  # noqa: F811
+    from concurrent.futures import ThreadPoolExecutor
+    from test_native_steward_private import steward
+    from loopx.capabilities.native_chat.antigravity_adapter import AntigravityAdapter
+    from loopx.chat_agent import CodexChatAgentError
+    import time
+
+    monkeypatch.setenv("LOOPX_MANAGER_MODEL", "gemini-4-argon-high")
+    monkeypatch.setenv("LOOPX_MANAGER_REASONING_EFFORT", "high")
+    fixture = steward.__wrapped__(tmp_path)
+    store, runtime, provider, transport, binding, *_ = next(fixture)
+    try:
+        binding = transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"],
+            executor_endpoint_id="antigravity", context_kind="steward")
+        start = AntigravityAdapter.start_turn
+
+        def lose_commit(adapter, message, event_sink):
+            start(adapter, message, event_sink)
+            raise CodexChatAgentError("Synthetic lost Core commit", gate=None, error_code="antigravity_rpc_error")
+
+        monkeypatch.setattr(AntigravityAdapter, "start_turn", lose_commit)
+        first = provider.event("steward-app", "older-result", "Older original request")
+        transport.admit("steward-app", first)
+        ref = identity_ref(binding["provider_ref"], first["message_id"])
+        row = transport.core.read_request(ref)
+        assert runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=15)["status"] == "failed"
+        assert transport.reconcile_request(transport.root / f"{ref}.json") == 1
+        assert transport.core.context_recovery_candidate(ref)  # Idle discovery is preserved.
+        monkeypatch.setattr(AntigravityAdapter, "start_turn", start)
+        from loopx.capabilities.native_chat import antigravity_adapter
+        post = antigravity_adapter._rpc_post
+
+        def keep_new_request_running(url, token, method, payload, **kwargs):
+            result = post(url, token, method, payload, **kwargs)
+            if method == "SendUserCascadeMessage" and language_server["sent_count"] == 2:
+                language_server["mode"] = "running"
+            return result
+
+        monkeypatch.setattr(antigravity_adapter, "_rpc_post", keep_new_request_running)
+        later = provider.event("steward-app", "current-work", "Current original work")
+        transport.admit("steward-app", later)
+        current = transport.core.read_request(identity_ref(binding["provider_ref"], later["message_id"]))
+        deadline = time.monotonic() + 5
+        while not store.load_turn(current["session_id"], current["turn_id"]).get("upstream_turn_id"):
+            assert time.monotonic() < deadline
+            time.sleep(0.01)
+        assert not transport.core.context_recovery_candidate(ref)
+        monkeypatch.setattr(AntigravityAdapter, "completed_result_observation", staticmethod(
+            lambda **_: pytest.fail("Historical native IO while current original work is active")))
+        assert transport.reconcile_request(transport.root / f"{ref}.json") == 0
+        with ThreadPoolExecutor(max_workers=1) as replies:
+            try:
+                replies.submit(transport.reconcile_request, transport.root / f"{current['request_ref']}.json").result(timeout=2)
+                record = json.loads((transport.root / f"{current['request_ref']}.json").read_text())
+                assert record["deliveries"]["progress"]["verified"]
+                assert store.load_turn(current["session_id"], current["turn_id"])["status"] in {"starting", "running"}
+            finally:
+                language_server["mode"] = "complete"
+        assert language_server["sent_count"] == 2
+    finally:
+        language_server["mode"] = "complete"
+        fixture.close()
+
+
 @pytest.mark.parametrize("fault", [None, "wrong-target", "changed-brief", "other-action", "still-invalid",
     "revoked", "new-input", "stop", "unknown-send", "post-enqueue-crash", "reservation-crash", "revoke-during-review",
     "tool-call", "tool-step", "mixed-execution", "missing-stop"])