From d7ab751b55408d9b489910b99c75ecde6b68396c Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 21:50:27 +0800 Subject: [PATCH] test(lark): locate background requests by source identity Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_chat_completed_background_stop.py | 5 +++-- tests/test_chat_conversation_bindings.py | 9 +++++---- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/tests/test_chat_completed_background_stop.py b/tests/test_chat_completed_background_stop.py index 40498c05c7..bd0da672b4 100644 --- a/tests/test_chat_completed_background_stop.py +++ b/tests/test_chat_completed_background_stop.py @@ -9,6 +9,7 @@ from test_lark_private_conversations import connect_group, finish_group_turn from loopx.chat_agent import CodexChatAgentError from loopx.control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result +from loopx.extensions.lark.conversation_identity import identity_ref class BackgroundWire: @@ -164,7 +165,7 @@ def write(packet): try: active = provider.topic("active-after-history", "wait for interrupt", root=root["message_id"]) transport.admit("notes-app", active) - row = next(r for r in transport.core.pending() if r["message"] == active["content"]) + row = transport.core.read_request(identity_ref(transport._binding("notes-app")["provider_ref"], active["message_id"])) deadline = time.monotonic() + 10 while not (store.load_turn(sid, row["turn_id"])["status"] == "running" and store.load_turn(sid, row["turn_id"]).get("upstream_turn_id")): @@ -328,7 +329,7 @@ def start(**kwargs): "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 = transport.core.read_request(identity_ref(transport._binding("notes-app")["provider_ref"], message["message_id"])) 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 diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index 9ee1085dfb..60b52e21ed 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/tests/test_chat_conversation_bindings.py @@ -7,6 +7,7 @@ from test_chat_ordinary_project import ordinary # noqa: F401, F811 from test_native_steward_private import steward # noqa: F401 from loopx.capabilities.native_chat.conversation_bindings import ChatConversationBindings +from loopx.extensions.lark.conversation_identity import identity_ref def _wait_for_native_turn(store, session_id, turn_id): @@ -137,7 +138,7 @@ def duplicate(): newer_event = event("newer-during-background-stop", "wait for interrupt") accepted = callers.submit(admit, "notes-app", newer_event) assert accepted.result(timeout=5)["status"] == "durably_accepted" - newer = next(r for r in transport.core.pending() if r["message"] == newer_event["content"]) + newer = transport.core.read_request(identity_ref(transport._binding("notes-app")["provider_ref"], newer_event["message_id"])) assert newer["session_id"] == sid assert store.load_turn(sid, newer["turn_id"])["status"] in {"queued", "starting", "running"} assert newer["turn_id"] not in frozen["background_turn_ids"] @@ -374,12 +375,12 @@ def test_group_disconnect_stops_running_and_queued_turns_without_stopping_other_ workspace = ordinary[-1] root = provider.topic("root", "first request" if resumed else "wait for interrupt") assert transport.admit("notes-app", root)["status"] == "durably_accepted" - first = next(row for row in transport.core.pending() if row["message"] == root["content"]) + first = transport.core.read_request(identity_ref(transport._binding("notes-app")["provider_ref"], root["message_id"])) if resumed: runtime.wait_for_turn(session_id=first["session_id"], turn_id=first["turn_id"], timeout_sec=10) followup = provider.topic("running", "wait for interrupt", root=root["message_id"]) assert transport.admit("notes-app", followup)["status"] == "durably_accepted" - active = next(row for row in transport.core.pending() if row["message"] == followup["content"]) + active = transport.core.read_request(identity_ref(transport._binding("notes-app")["provider_ref"], followup["message_id"])) assert active["session_id"] == first["session_id"] else: active = first @@ -387,7 +388,7 @@ def test_group_disconnect_stops_running_and_queued_turns_without_stopping_other_ _wait_for_native_turn(store, sid, tid) queued_event = provider.topic("queued", "must never run", root=root["message_id"]) assert transport.admit("notes-app", queued_event)["status"] == "durably_accepted" - queued = next(row for row in transport.core.pending() if row["message"] == queued_event["content"]) + queued = transport.core.read_request(identity_ref(transport._binding("notes-app")["provider_ref"], queued_event["message_id"])) assert store.load_turn(sid, queued["turn_id"])["status"] == "queued" peer, _ = runtime.open_session(goal_id=None, agent_id="codex", work_dir=workspace, objective="", mode="new", project_ref=ordinary[2].available()[0]["project_ref"])