fix(chat): prevent provider input from entering after disconnect

loopx/capabilities/native_chat/conversation_bindings.py

@@ -213,16 +213,21 @@ def _resolve_observed(self, current, row, observation, source):
         })
 
     @contextmanager
-    def local_input(self, saved: dict[str, Any]):
+    def input_context(self, saved: dict[str, Any], *, origin: str):
         """Observe outside the publication fence, then retain it through enqueue.
 
-        This is the existing loopback operator's continuation of a private
-        binding. Saved fields select an identity; only fresh Core checks grant it.
+        Saved fields select an identity; fresh Core checks grant it. Local Web
+        continuation keeps its actual origin and never becomes a provider event.
         """
-        source = {"source_ref": saved["source_ref"], "sender_ref": saved["operator_ref"],
-                  "private_human_message": False, "origin": "web", "session_context": saved}
+        if origin == "web":
+            source = {"source_ref": saved["source_ref"], "sender_ref": saved["operator_ref"],
+                      "private_human_message": False, "origin": "web", "session_context": saved}
+        elif origin in {"lark", "external"}:
+            source = self._session_source(saved)
+        else:
+            raise ValueError("unsupported bound conversation input origin")
         observed = self.resolve(binding_id=saved["binding_id"], include_observation=True, **source)
-        with exclusive_file_lock(self.path, operation="admit_local_bound_input"):
+        with exclusive_file_lock(self.path, operation="admit_bound_input"):
             current = self.read()
             if next((row for row in current.get("bindings", [])
                      if row.get("binding_id") == saved["binding_id"]), None) != observed["binding"]:
@@ -301,14 +306,17 @@ def ensure_delivery_scope(self, selected: dict[str, Any], *, if_absent: bool = T
             channel=selected["channel_id"], local_delivery_scope=binding["goal_scope"],
             sender_id=binding["operator_ref"], execute=execute, if_absent=if_absent)
 
-    def session_context(self, saved: dict[str, Any]) -> dict[str, Any]:
+    @staticmethod
+    def _session_source(saved: dict[str, Any]) -> dict[str, Any]:
         if saved.get("audience") == "bound_group":
-            return self.resolve(binding_id=saved["binding_id"], source_ref=saved["source_ref"],
-                sender_ref=saved["operator_ref"], private_human_message=False,
-                group_ref=saved["group_ref"], topic_ref=saved["topic_ref"], group_human_message=True,
-                session_context=saved)
-        return self.resolve(binding_id=saved["binding_id"], source_ref=saved["source_ref"],
-                            sender_ref=saved["operator_ref"], private_human_message=True, session_context=saved)
+            return {"source_ref": saved["source_ref"], "sender_ref": saved["operator_ref"],
+                    "private_human_message": False, "group_ref": saved["group_ref"],
+                    "topic_ref": saved["topic_ref"], "group_human_message": True, "session_context": saved}
+        return {"source_ref": saved["source_ref"], "sender_ref": saved["operator_ref"],
+                "private_human_message": True, "session_context": saved}
+
+    def session_context(self, saved: dict[str, Any]) -> dict[str, Any]:
+        return self.resolve(binding_id=saved["binding_id"], **self._session_source(saved))
 
     def manager_runtime_owner(self, saved: dict[str, Any], channel_id: str) -> bool:
         """Fresh Core audience proof; saved Session text is never authority."""
@@ -390,11 +398,18 @@ def agent_candidates(self, binding_id: str) -> list[dict[str, Any]]:
 
     def change_agent_target(self, *, binding_id: str, expected_revision: int,
                             session_id: str | None = None, target_ref: str | None = None) -> dict[str, Any]:
+        row = next((item for item in self.read()["bindings"] if item["binding_id"] == binding_id), None)
+        if row is None:
+            raise ValueError("conversation binding is no longer authorized")
+        # Provider IO must not prevent disconnect from publishing revocation.
+        # Recheck the exact binding under its fence before using this observation.
+        observation = self.observe(row["transport_ref"])
         with exclusive_file_lock(self.path, operation="configure_conversation_agent_target"):
             current = self.read()
-            row = next(item for item in current["bindings"] if item["binding_id"] == binding_id)
+            if next((item for item in current["bindings"] if item["binding_id"] == binding_id), None) != row:
+                raise ValueError("conversation binding changed during identity verification")
             params = {"current": current, "expected_revision": expected_revision, "binding_id": binding_id,
-                      "observation": self.observe(row["transport_ref"]), "available_projects": self.projects.available()}
+                      "observation": observation, "available_projects": self.projects.available()}
             if session_id is not None:
                 observed = self.agent_observation(session_id)
                 target = {key: observed["session"].get(key) for key in ["session_id", "goal_id", "goal_instance_id",

loopx/capabilities/native_chat/queued_admission.py

@@ -15,10 +15,12 @@ def enqueue_turn(controller, *, session_id: str, client_turn_id: str, message: s
     if session is None or session.get("status") == "closed":
         raise KeyError("chat session was not found")
     bound = session.get("steward_context") or session.get("project_context") or {}
-    if origin == "web" and bound.get("audience") in {"bound_owner", "bound_group"}:
+    if bound.get("audience") in {"bound_owner", "bound_group"}:
         if external_agent_target is not None:
             raise ValueError("Agent target requires its original attached Session")
-        return _enqueue_local_bound(controller, session, client_turn_id, message, attachments)
+        if origin != "web" and work_dir is None:
+            raise ValueError("the conversation's runtime context is required")
+        return _enqueue_bound(controller, session, client_turn_id, message, attachments, origin)
     if work_dir is None:
         raise ValueError("the conversation's runtime context is required")
     if session.get("steward_context") is not None:
@@ -61,13 +63,13 @@ def enqueue_turn(controller, *, session_id: str, client_turn_id: str, message: s
     return turn, created
 
 
-def _enqueue_local_bound(controller, session, client_turn_id, message, attachments):
+def _enqueue_bound(controller, session, client_turn_id, message, attachments, origin):
     contexts = controller.project_contexts
     bindings = contexts.conversation_bindings
     if bindings is None:
         raise ValueError("bound conversation authority is unavailable")
     saved = session.get("steward_context") or session["project_context"]
-    with bindings.local_input(saved) as selected:
+    with bindings.input_context(saved, origin=origin) as selected:
         fresh = controller.store.load_session(session["session_id"])
         if fresh is None or fresh.get("status") == "closed":
             raise KeyError("chat session was not found")
@@ -78,11 +80,11 @@ def _enqueue_local_bound(controller, session, client_turn_id, message, attachmen
         if fresh.get("project_context") is not None:
             project_context_policy.validate_project_executor_scope(str(fresh["agent_id"]),
                 fresh["project_context"], controller.capabilities)
-        else:
+        elif origin == "web":
             bindings.ensure_delivery_scope(selected)
             _record_local_ingress(controller.coordination_runtime_root, fresh, client_turn_id, message)
         turn, created = controller.store.create_queued_turn(session["session_id"],
-            client_turn_id=client_turn_id, message=message, attachments=attachments, origin="web")
+            client_turn_id=client_turn_id, message=message, attachments=attachments, origin=origin)
     # Dispatch after releasing the publication fence. The worker independently
     # rechecks current scope before effects, including revocation after enqueue.
     controller.resume_session_queue(session_id=session["session_id"],

tests/control_plane/test_bound_owner_web_input.py

@@ -367,6 +367,84 @@ def revoke():
     assert turn["status"] in {"interrupted", "cancelled"}
 
 
+@pytest.mark.parametrize("kind", ["project", "steward"])
+@pytest.mark.parametrize("origin", ["lark", "external"])
+@pytest.mark.parametrize("phase", ["identity", "commit"])
+def test_provider_input_orders_revocation_around_identity_and_queue(bound_chat, monkeypatch, kind, origin, phase):
+    from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations
+
+    store, runtime, bindings, observed, _, open_bound, _, _ = bound_chat
+    binding, session = open_bound(kind)
+    monkeypatch.setattr(runtime, "resume_session_queue", lambda **_: None)
+    entered, release, revoked, queueing = (threading.Event() for _ in range(4))
+    results, failures, revoke_failures = [], [], []
+    core = ChatExternalConversations(runtime)
+    original = bindings.observe if phase == "identity" else store.create_queued_turn
+    enqueue = runtime.enqueue_turn
+
+    def enqueue_provider(**kwargs):
+        queueing.set()
+        return enqueue(**kwargs)
+
+    monkeypatch.setattr(runtime, "enqueue_turn", enqueue_provider)
+
+    def gated(*args, **kwargs):
+        # The identity observation is complete before its simulated IO returns;
+        # queue publication has not happened while its commit is paused.
+        value = original(*args, **kwargs) if phase == "identity" else None
+        if phase == "identity" and not queueing.is_set():
+            return value
+        entered.set()
+        assert release.wait(5)
+        return value if phase == "identity" else original(*args, **kwargs)
+
+    monkeypatch.setattr(bindings if phase == "identity" else store,
+                        "observe" if phase == "identity" else "create_queued_turn", gated)
+
+    def send():
+        try:
+            results.append(core.admit(binding_id=binding["binding_id"],
+                source={"source_ref": "a" * 24, "sender_ref": observed["operator_ref"],
+                        "private_human_message": True},
+                request_ref="b" * 24, message="Preserve the original request", origin=origin))
+        except Exception as exc:
+            failures.append(exc)
+
+    def revoke():
+        try:
+            bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"])
+        except Exception as exc:
+            revoke_failures.append(exc)
+        finally:
+            revoked.set()
+
+    sender, revoker = threading.Thread(target=send), threading.Thread(target=revoke)
+    sender.start()
+    try:
+        assert entered.wait(5)
+        revoker.start()
+        if phase == "identity":
+            assert revoked.wait(3), "provider IO blocked revocation"
+        else:
+            assert not revoked.wait(0.15), "revocation crossed canonical queue publication"
+    finally:
+        release.set()
+        sender.join(timeout=10)
+        if revoker.ident is not None:
+            revoker.join(timeout=10)
+    assert not sender.is_alive() and not revoker.is_alive()
+    assert revoked.is_set() and not revoke_failures
+    assert store.load_session(session["session_id"])["upstream_thread_id"] == session["upstream_thread_id"]
+    assert store.queued_turns(session["session_id"]) == []
+    if phase == "identity":
+        assert not results and len(failures) == 1 and isinstance(failures[0], ValueError), failures
+    else:
+        assert not failures and results[0]["status"] == "accepted", failures
+        turn = store.load_turn(session["session_id"], results[0]["turn_id"])
+        assert turn["status"] in {"interrupted", "cancelled"}
+        assert turn["message"] == "Preserve the original request" and turn["origin"] == origin
+
+
 def test_provider_and_local_followups_keep_busy_fifo_and_exact_replay(bound_chat):
     store, runtime, _, _, post, open_bound, _, _ = bound_chat
     _, session = open_bound()

tests/test_lark_private_agents.py

@@ -17,10 +17,10 @@
 )
 
 
-def target(fixture):
+def target(fixture, registry=None):
     store, runtime, provider, transport = connect(fixture)
     workspace = fixture[-1]
-    registry = _registry()
+    registry = _registry() if registry is None else registry
     registry["goals"][0]["repo"] = str(workspace)
     runtime.registry_path.write_text(json.dumps(registry))
     packet = bind_attached_agent_session(store=store, registry=registry, registry_path=runtime.registry_path,
@@ -477,3 +477,82 @@ def request(method, url, body=None, origin=None):
         server.server_close()
         thread.join(timeout=2)
         runtime.close()
+
+
+@pytest.mark.parametrize("provider_backend", ["file", "sqlite"])
+@pytest.mark.parametrize("operation", ["grant", "revoke"])
+def test_http_target_identity_probe_does_not_block_disconnect(ordinary, monkeypatch, provider_backend, operation):  # noqa: F811
+    import http.client
+    import threading
+    from concurrent.futures import ThreadPoolExecutor
+    from control_plane.canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture
+    from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
+
+    authority = ordinary[-1].parent / "authority"
+    authority.mkdir()
+    isolate_sqlite_runtime(authority, monkeypatch)
+    registry_path, _, state = promoted_create_fixture(authority, goal_id=GOAL_ID,
+        provider=provider_backend, registered_agents=[AGENT_ID])
+    registry = json.loads(registry_path.read_text())
+    registry["goals"][0]["state_file"] = str(state)
+    registry["goals"][0]["coordination"]["thread_agent_bindings"] = (
+        _registry()["goals"][0]["coordination"]["thread_agent_bindings"])
+    store, runtime, _, transport, bid, sid, grant = target(ordinary, registry)
+    bindings = transport.bindings
+    if operation == "grant":
+        bindings.change_agent_target(binding_id=bid, expected_revision=bindings.read()["revision"],
+                                     target_ref=grant["target_ref"])
+    revision = bindings.read()["revision"]
+    entered, release = threading.Event(), threading.Event()
+    observe = bindings.observe
+
+    def slow_identity(*args, **kwargs):
+        result = observe(*args, **kwargs)
+        entered.set()
+        assert release.wait(10), "the isolated provider probe was not released"
+        return result
+
+    monkeypatch.setattr(bindings, "observe", slow_identity)
+    server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
+    server.runtime_controller = runtime
+    server.lark_goal_topic_runtime = SimpleNamespace(health_snapshot=lambda: {}, refresh=lambda: None, close=lambda: None)
+    server.lark_private_conversations, server.verbose = transport, False
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    path = "/api/chat/lark/private-conversations"
+
+    def request(method, url, body):
+        connection = http.client.HTTPConnection(*server.server_address, timeout=15)
+        try:
+            connection.request(method, url, json.dumps(body), headers={"Content-Type": "application/json"})
+            response = connection.getresponse()
+            return response.status, json.loads(response.read())
+        finally:
+            connection.close()
+
+    try:
+        with ThreadPoolExecutor(max_workers=2) as workers:
+            try:
+                body = {"binding_id": bid, "revision": revision,
+                    **({"session_id": sid} if operation == "grant" else {"target_ref": grant["target_ref"]})}
+                editing = workers.submit(request, "POST", path + "/agent-targets", body)
+                assert entered.wait(5), "target configuration did not reach provider verification"
+                disconnecting = workers.submit(request, "DELETE", path, {"binding_id": bid, "revision": revision})
+                code, snapshot = disconnecting.result(timeout=3)
+                assert code == 200, snapshot
+                assert not release.is_set()
+                assert all(row["binding_id"] != bid for row in bindings.read()["bindings"])
+                revoked = bindings.read()
+            finally:
+                release.set()
+            code, error = editing.result(timeout=5)
+            assert code == 400 and "binding" in error["error"], error
+        assert bindings.read() == revoked
+        assert store.load_session(sid)["upstream_thread_id"] == HOST_SESSION_ID
+        assert store.queued_turns(sid) == []
+    finally:
+        release.set()
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)
+        runtime.close()