fix: isolate chat event loops and persist pending input

AGENTS.md

@@ -15,6 +15,7 @@
 ## Root Ownership
 
 - `agent.py` owns `Agent`, `AgentContext`, and loop data.
+- Each `AgentContext` runs agent work on its own named event-loop thread so a blocked chat cannot stall unrelated chats.
 - `Agent.hist_add_ai_response` owns Responses-API state advancement: it calls `_remember_llm_result_state` internally. Model turns pass an `LLMResult`; omitted results and legacy positional string IDs use the non-LLM sentinel. Callers must not invoke `_remember_llm_result_state` manually.
 - Prepared local Responses input may project intact native history through `helpers/responses_history.py`; retain original Chat/fallback messages, current protocol/extras, summaries and masking. Durable capability metadata stores only a stable-prefix digest for eligibility.
 - Native function calls use canonical call content for history and repeat comparison, even when accompanied by commentary. The canonical content passes through the normal history template/masking hook; provider output metadata remains intact.

agent.py

@@ -289,7 +289,7 @@ def run_task(
     ):
         if not self.task:
             self.task = DeferredTask(
-                thread_name=self.__class__.__name__,
+                thread_name=f"{self.__class__.__name__}-{self.id}",
             )
         self.task.start_task(func, *args, **kwargs)
         return self.task

api/message.py

@@ -1,7 +1,7 @@
 from agent import AgentContext, UserMessage
 from helpers.api import ApiHandler, Request, Response
 
-from helpers import files, extension, message_queue as mq
+from helpers import files, extension, message_queue as mq, persist_chat
 import os
 from helpers.security import safe_filename
 from helpers.defer import DeferredTask
@@ -65,7 +65,21 @@ async def communicate(self, input: dict, request: Request):
         # Store attachments in agent data
         # context.agent0.set_data("attachments", attachment_paths)
 
-        # Log to console and UI using helper function
-        mq.log_user_message(context, message, attachment_paths, message_id)
+        # Persist an inbox copy before dispatch. It stays on disk until the
+        # message loop completes and saves the now-drained queue, so a wedged
+        # or interrupted loop cannot swallow the user's input without a trace.
+        pending = mq.add(context, message, attachment_paths, item_id=message_id)
+        mq.log_user_message(
+            context, message, attachment_paths, message_id=pending["id"]
+        )
+        persist_chat.save_tmp_chat(context)
+        mq.pop_item(context, pending["id"])
 
-        return context.communicate(UserMessage(message=message, attachments=attachment_paths, id=message_id or "")), context
+        task = context.communicate(
+            UserMessage(
+                message=message,
+                attachments=attachment_paths,
+                id=pending["id"],
+            )
+        )
+        return task, context

api/message.py.dox.md

@@ -22,12 +22,13 @@
 - Update this file whenever request payloads, authentication or CSRF requirements, response shapes, route side effects, or WebSocket event contracts change.
 - `Message` is an `ApiHandler`.
 - `Message` defines `process(...)`.
+- UI messages are copied into the persisted message queue before agent dispatch. The in-memory queue entry is drained immediately, while the persisted copy remains recoverable until normal message-loop completion saves the drained state.
 - Observed side-effect areas: filesystem reads, filesystem writes, settings/state persistence, scheduler state.
 - Imported dependency areas include: `agent`, `helpers`, `helpers.api`, `helpers.defer`, `helpers.security`, `os`.
 
 ## Key Concepts
 
-- Important called helpers/classes observed in the source: `request.content_type.startswith`, `self.use_context`, `mq.log_user_message`, `self.communicate`, `self.respond`, `task.result`, `request.files.getlist`, `files.get_abs_path`, `request.get_json`, `extension.call_extensions_async`, `context.communicate`, `os.makedirs`, `UserMessage`, `safe_filename`, `attachment.save`, `context.get_agent`, `os.path.join`.
+- Important called helpers/classes observed in the source: `request.content_type.startswith`, `self.use_context`, `mq.add`, `mq.log_user_message`, `persist_chat.save_tmp_chat`, `mq.pop_item`, `self.communicate`, `self.respond`, `task.result`, `request.files.getlist`, `files.get_abs_path`, `request.get_json`, `extension.call_extensions_async`, `context.communicate`, `os.makedirs`, `UserMessage`, `safe_filename`, `attachment.save`, `context.get_agent`, `os.path.join`.
 - Keep request/response, tool, or helper semantics documented here at the same time as source changes.
 
 ## Work Guidance
@@ -39,6 +40,7 @@
 ## Verification
 
 - Run endpoint-specific or API/WebSocket tests for changed behavior; smoke-test browser callers when no focused test exists.
+- Run `pytest tests/test_message_durability.py` for the pre-dispatch persistence contract.
 - Related tests observed by source search:
   - `tests/email_parser_test.py`
   - `tests/rate_limiter_test.py`

tests/test_agent_context_event_loop_isolation.py

@@ -0,0 +1,52 @@
+import asyncio
+import sys
+import threading
+import types
+from pathlib import Path
+
+import pytest
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+# AgentContext.run_task does not use embeddings. Avoid importing the optional
+# sentence-transformers stack just to exercise this scheduler seam.
+sentence_transformers = types.ModuleType("sentence_transformers")
+sentence_transformers.SentenceTransformer = object
+sys.modules.setdefault("sentence_transformers", sentence_transformers)
+
+from agent import AgentContext
+
+
+def _bare_context(context_id: str) -> AgentContext:
+    context = AgentContext.__new__(AgentContext)
+    context.id = context_id
+    context.task = None
+    return context
+
+
+def test_blocked_chat_does_not_stall_another_chat():
+    blocked = _bare_context("blocked-chat")
+    healthy = _bare_context("healthy-chat")
+    blocker_started = threading.Event()
+    release_blocker = threading.Event()
+
+    async def block_event_loop():
+        blocker_started.set()
+        release_blocker.wait()
+
+    async def finish_immediately():
+        await asyncio.sleep(0)
+        return "healthy chat completed"
+
+    blocked_task = blocked.run_task(block_event_loop)
+    assert blocker_started.wait(timeout=1)
+
+    healthy_task = healthy.run_task(finish_immediately)
+    try:
+        assert healthy_task.result_sync(timeout=0.5) == "healthy chat completed"
+    finally:
+        release_blocker.set()
+        blocked_task.result_sync(timeout=1)

tests/test_message_durability.py

@@ -0,0 +1,103 @@
+import copy
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+sentence_transformers = types.ModuleType("sentence_transformers")
+sentence_transformers.SentenceTransformer = object
+sys.modules.setdefault("sentence_transformers", sentence_transformers)
+
+from api.message import Message
+from helpers import message_queue
+
+
+class _Request:
+    content_type = "application/json"
+
+    @staticmethod
+    def get_json():
+        return {
+            "text": "keep this message",
+            "context": "chat-1",
+            "message_id": "message-1",
+        }
+
+
+class _Log:
+    def log(self, **_kwargs):
+        return None
+
+
+class _Context:
+    def __init__(self):
+        self.id = "chat-1"
+        self.data = {}
+        self.output_data = {}
+        self.log = _Log()
+        self.communicated = []
+
+    def get_agent(self):
+        return object()
+
+    def get_data(self, key):
+        return self.data.get(key)
+
+    def set_data(self, key, value):
+        self.data[key] = value
+
+    def set_output_data(self, key, value):
+        self.output_data[key] = value
+
+    def communicate(self, message):
+        self.communicated.append(message)
+        return "task"
+
+
+@pytest.mark.asyncio
+async def test_user_message_is_persisted_as_pending_before_dispatch(monkeypatch):
+    from api import message as message_api
+
+    context = _Context()
+    handler = Message.__new__(Message)
+    handler.use_context = lambda _context_id: context
+    events = []
+    persisted_queue = []
+
+    async def no_extensions(*_args, **_kwargs):
+        return None
+
+    def save_chat(saved_context):
+        events.append("persisted")
+        persisted_queue.extend(copy.deepcopy(message_queue.get_queue(saved_context)))
+
+    def communicate(message):
+        events.append("dispatched")
+        context.communicated.append(message)
+        return "task"
+
+    monkeypatch.setattr(message_api.extension, "call_extensions_async", no_extensions)
+    monkeypatch.setattr(message_api.persist_chat, "save_tmp_chat", save_chat)
+    monkeypatch.setattr(context, "communicate", communicate)
+
+    task, returned_context = await handler.communicate({}, _Request())
+
+    assert task == "task"
+    assert returned_context is context
+    assert events == ["persisted", "dispatched"]
+    assert persisted_queue == [
+        {
+            "id": "message-1",
+            "seq": 1,
+            "text": "keep this message",
+            "attachments": [],
+        }
+    ]
+    assert message_queue.get_queue(context) == []
+    assert context.communicated[0].id == "message-1"