From 357000fcbe8d4da81bf708c786e0d9b6b64c3588 Mon Sep 17 00:00:00 2001 From: fcsguyxcw <3056887696@qq.com> Date: Wed, 9 Sep 2026 08:35:29 +0800 Subject: [PATCH] fix: isolate chat loops and persist pending input --- AGENTS.md | 1 + agent.py | 2 +- api/message.py | 22 +++- api/message.py.dox.md | 4 +- ...test_agent_context_event_loop_isolation.py | 52 +++++++++ tests/test_message_durability.py | 103 ++++++++++++++++++ 6 files changed, 178 insertions(+), 6 deletions(-) create mode 100644 tests/test_agent_context_event_loop_isolation.py create mode 100644 tests/test_message_durability.py diff --git a/AGENTS.md b/AGENTS.md index ff274ae6f7..30422f663d 100644 --- a/AGENTS.md +++ b/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. diff --git a/agent.py b/agent.py index 2e558bd154..24a8a1235a 100644 --- a/agent.py +++ b/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 diff --git a/api/message.py b/api/message.py index 6ea66027ed..8ddf3be56d 100644 --- a/api/message.py +++ b/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 diff --git a/api/message.py.dox.md b/api/message.py.dox.md index 9ec63fc71f..93da3d5e26 100644 --- a/api/message.py.dox.md +++ b/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` diff --git a/tests/test_agent_context_event_loop_isolation.py b/tests/test_agent_context_event_loop_isolation.py new file mode 100644 index 0000000000..6b4983349d --- /dev/null +++ b/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) diff --git a/tests/test_message_durability.py b/tests/test_message_durability.py new file mode 100644 index 0000000000..e2ba87ea3f --- /dev/null +++ b/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"