From cc17db7ad9a9bbbbe3bcbcdf8d8b3b77dee91862 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 15:43:22 +0800 Subject: [PATCH] fix(chat): share source observation during stop recovery reads Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/conversation_bindings.py | 6 +- tests/test_lark_private_conversations.py | 75 +++++++++++++++++++ 2 files changed, 80 insertions(+), 1 deletion(-) diff --git a/loopx/capabilities/native_chat/conversation_bindings.py b/loopx/capabilities/native_chat/conversation_bindings.py index 01b2d82ccd..dda5e5aff5 100644 --- a/loopx/capabilities/native_chat/conversation_bindings.py +++ b/loopx/capabilities/native_chat/conversation_bindings.py @@ -14,7 +14,9 @@ import hashlib from ...chat_store import TERMINAL_TURN_STATES, _atomic_write_json, _read_json -from ...control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result +from ...control_plane.effect_runtime import ( + EffectRuntimeRejected, effect_runtime_request_scope, effect_runtime_result, +) from ...file_lock import exclusive_file_lock if TYPE_CHECKING: @@ -530,12 +532,14 @@ def resolve_agent_target(self, selected: dict[str, Any], target: dict[str, Any]) "target_observation": self.agent_observation(target["session_id"])}) + @effect_runtime_request_scope() def stop_recoveries(self, current: dict[str, Any]) -> list[dict[str, Any]]: """Project revoked execution from canonical Turns, never recreate a grant. Session and external-target context retain the exact binding identity across restart. No second stop ledger or provider diagnostics are needed. A recovery disappears only when the existing Turn owner settles it. + Share the runtime source observation, while reading each queue afresh. """ if self.controller is None: return [] diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index c778b8cf5d..2151db554b 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -3179,6 +3179,81 @@ def request(path, body=None): runtime.close() +def test_http_stop_recovery_reuses_source_revision_without_reusing_turn_state(ordinary, monkeypatch): # noqa: F811 + from loopx.chat_server import ChatHTTPServer, ChatRequestHandler + from loopx.control_plane import effect_runtime + + store, runtime, provider, transport = connect(ordinary) + binding = transport._binding("notes-app") + sessions = [store.create_session(goal_id="example-goal", agent_id="codex", + adapter_kind="codex_app_server", upstream_thread_id=f"idle-{index}") for index in range(8)] + sid = sessions[0]["session_id"] + revoked = "a" * 24 + def queue(name, binding_id): + return store.create_queued_turn(sid, client_turn_id=name, message=name, origin="external", + external_agent_target={"context": {"binding_id": binding_id, + "project_ref": binding["project_ref"]}, "target": {"session_id": sid}})[0] + owed = queue("revoked-queued-work", revoked) + authorized = queue("authorized-queued-work", binding["binding_id"]) + original_binding = transport.bindings.path.read_bytes() + original_turns = [store._turn_path(sid, turn["turn_id"]).read_bytes() for turn in [owed, authorized]] + native_before = ordinary[4].read_bytes() if ordinary[4].exists() else None + revision_reads = [] + observed = threading.local() + fingerprint = effect_runtime._runtime_fingerprint + stop_recoveries = ChatConversationBindings.stop_recoveries + + def counted_fingerprint(): + if getattr(observed, "stop_recovery", False): + revision_reads.append(True) + return fingerprint() + + def observed_stop_recoveries(self, current): + observed.stop_recovery = True + try: + return stop_recoveries(self, current) + finally: + observed.stop_recovery = False + + monkeypatch.setattr(effect_runtime, "_runtime_fingerprint", counted_fingerprint) + monkeypatch.setattr(ChatConversationBindings, "stop_recoveries", observed_stop_recoveries) + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.chat_store, server.runtime_controller, server.verbose = store, runtime, False + server.lark_private_conversations = transport + server.lark_goal_topic_runtime = SimpleNamespace(health_snapshot=lambda: {}, close=lambda: None) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + def snapshot(): + revision_reads.clear() + connection = http.client.HTTPConnection(*server.server_address, timeout=15) + try: + connection.request("GET", "/api/chat/lark/private-conversations") + response = connection.getresponse() + assert response.status == 200 + result = json.loads(response.read()) + assert len(revision_reads) <= 1 # At most one source inventory, rather than one per Session. + return result + finally: + connection.close() + + try: + assert snapshot()["stop_recoveries"] == [{"binding_id": revoked, + "project_ref": binding["project_ref"], "project_title": ordinary[-1].name, "pending_count": 1}] + assert [store._turn_path(sid, turn["turn_id"]).read_bytes() for turn in [owed, authorized]] == original_turns + store.update_turn(sid, owed["turn_id"], status="interrupted") + assert snapshot()["stop_recoveries"] == [] # A later read observes settlement immediately. + assert store.load_turn(sid, authorized["turn_id"])["status"] == "queued" + assert transport.bindings.path.read_bytes() == original_binding + assert (ordinary[4].read_bytes() if ordinary[4].exists() else None) == native_before + assert not provider.writes + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + runtime.close() + + def test_http_listener_health_belongs_to_each_bound_transport(ordinary): # noqa: F811 from loopx.chat_server import ChatHTTPServer, ChatRequestHandler from loopx.capabilities.native_chat.transports import ChatConversationTransports