From 2bf8e510994a6782d8e0e0e6d0019614ffe20336 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 3 Nov 2026 03:36:05 +0800 Subject: [PATCH] fix(chat): defer historical native probes during active work Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 4 ++ ...p-conversation-and-async-inbox-v0.zh-CN.md | 3 + .../collaboration/conversation_binding.ts | 3 + .../conversation_binding.test.ts | 4 +- tests/test_lark_private_conversations.py | 64 +++++++++++++++++++ 5 files changed, 77 insertions(+), 1 deletion(-) 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 9d17d4b38e..a43ab16626 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/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 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 408d623a5a..60c04c8c77 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 @@ -1210,6 +1210,9 @@ reaction 清理未完成时，也只恢复原结果 proof，不能再次发送 ### Core 失败后的原生结果恢复 +同一 Core Session 忙碌或仍有活跃 Turn 时，历史原生结果暂不进入读回发现，避免 +占用当前进度所需的共享回复 worker。已核验结果的元数据仍由原关联修复路径负责。 + Antigravity 原生端可能已经完成，而 Core 仍保留 transport 失败。原回复 worker 可以采用既有答复，不准入另一个 Turn，不发送原生输入，也不取消执行。会话的 typed owner 核验原私聊管家 binding、来源、失败 Turn 和当前授权，并排除附件、 diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 6a838a0650..928407ae5e 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/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}; diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 0b0657aacc..167ef8fa13 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/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, diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 9e2fe69e6a..04dd76655a 100644 --- a/tests/test_lark_private_conversations.py +++ b/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"])