feat(services): per-session serialization for chat/stream/delete (A2 concurrency isolation)

FastGPT serializes requests per chatId server-side; the LangGraph
backend held all state in memory with no locking, so two concurrent
requests on one session could interleave (checkpoint write races,
garbled no_reply/turn bookkeeping).

- AccidentAppService holds per-session asyncio locks; chat(), stream()
  and delete_session() execute under the session lock (LLM call
  included - same-session turns are strictly serialized, different
  sessions never contend)
- set_info/get_info stay synchronous and lock-free: repository
  load/apply_update are atomic dict operations with no mixed
  intermediate state; a racing set vs chat yields one complete value
- delete pops the lock entry after cleanup; a chat that arrived
  concurrently executes afterwards against the fresh (deleted)
  session - matching FastGPT's delete-then-chat new-conversation
  semantics
- tests: same-session concurrent chats serialize (interval pairing
  probe), different sessions run parallel, delete waits for in-flight
  chat, chat-vs-set interleave leaves no torn values

Verified: 219 unit tests (4 new concurrency tests)

src/services/app_service.py

@@ -5,6 +5,11 @@
   （对照旧 FastGPT 实现的“辅助对话 + 删记录”hack）；
 - chat 每轮从 repository 读最新业务字段注入图（set_info 写入后
   下一轮 chat 必须可见），图产出字段差量后写回 repository；
+- **同会话串行**（A2 并发隔离）：chat / stream / delete_session 按
+  session_id 互斥执行——FastGPT 服务端对同一 chatId 串行处理，
+  内存态无锁会交错（checkpoint 写竞争、簿记错乱）。不同会话
+  无竞争。set/get 保持同步无锁：repository 的 load/apply_update
+  是 dict 原子操作，不存在混合中间态。
 - 未知 key 兼容策略：get 对未知 key 返回 JSON 空串 '""'（现状兼容）；
   set 对未知 key 拒绝写入（白名单保护，差异见兼容清单）；
 - includeInputInfo 继续被 endpoint 接受，但本层不再使用
@@ -13,6 +18,8 @@
 endpoint 只负责校验、协议（SSE / 响应模型）与依赖注入。
 """
 
+import asyncio
+
 from ..backends.chat import ChatBackend, ChatInput, ChatResult, FormUpdate
 from .fields_registry import (
     FIELD_REGISTRY,
@@ -35,6 +42,17 @@ class AccidentAppService:
     def __init__(self, backend: ChatBackend, repository: SessionRepository):
         self._backend = backend
         self._repository = repository
+        # 会话级互斥锁：同 session 的 chat/stream/delete 串行执行。
+        # delete 成功后清理锁项；清理与等待者的竞争语义见
+        # test_concurrent_delete_then_chat_behaves_as_fresh_session。
+        self._locks: dict[str, asyncio.Lock] = {}
+
+    def _lock_for(self, session_id: str) -> asyncio.Lock:
+        lock = self._locks.get(session_id)
+        if lock is None:
+            lock = asyncio.Lock()
+            self._locks[session_id] = lock
+        return lock
 
     # ---- chat：图每轮读取最新业务状态，产出差量写回 ----
 
@@ -69,6 +87,10 @@ class AccidentAppService:
 
     async def chat(self, chat_input: ChatInput) -> ChatResult:
         """非流式一轮对话：注入最新业务字段 -> 执行图 -> 写回字段差量。"""
+        async with self._lock_for(chat_input.session_id):
+            return await self._chat_locked(chat_input)
+
+    async def _chat_locked(self, chat_input: ChatInput) -> ChatResult:
         injected = self._injected_fields(chat_input.session_id)
         result = await self._backend.complete(
             ChatInput(
@@ -87,6 +109,11 @@ class AccidentAppService:
 
     async def stream(self, chat_input: ChatInput):
         """流式代理：字段注入同 chat()；FormUpdate 事件捕获写回后透传。"""
+        async with self._lock_for(chat_input.session_id):
+            async for event in self._stream_locked(chat_input):
+                yield event
+
+    async def _stream_locked(self, chat_input: ChatInput):
         injected = self._injected_fields(chat_input.session_id)
         result_input = ChatInput(
             session_id=chat_input.session_id,
@@ -151,15 +178,19 @@ class AccidentAppService:
 
         drop_checkpoint：由 core 层注入的 checkpoint 清理函数
         （图生命周期归 lifespan 管理，本层不持有图引用）。
+        与会话锁互斥：删除不与在途 chat 交错。删除后清理锁项。
         repository 删除幂等；checkpoint 缺失时静默（幂等语义一致）。
         """
-        self._repository.delete(session_id)
-        if drop_checkpoint is not None:
-            try:
-                await drop_checkpoint(session_id)
-            except Exception:
-                # checkpoint 已不存在视为删除成功的一部分（幂等）
-                pass
+        async with self._lock_for(session_id):
+            self._repository.delete(session_id)
+            if drop_checkpoint is not None:
+                try:
+                    await drop_checkpoint(session_id)
+                except Exception:
+                    # checkpoint 已不存在视为删除成功的一部分（幂等）
+                    pass
+            # 清理锁项：引用随等待者自然释放，新会话重建新锁
+            self._locks.pop(session_id, None)
 
 
 __all__ = ["AccidentAppService", "SET_INFO_INVALID_KEY_MSG"]

test/services/test_app_service.py

@@ -220,3 +220,104 @@ def _input(session_id, text):
     from src.backends.chat import ChatInput
 
     return ChatInput(session_id=session_id, text=text, client_mode="direct")
+
+
+# ---- 并发隔离（A2）：同会话串行、不同会话并行、delete 与 chat 互斥 ----
+
+import asyncio
+
+
+class SerialProbeBackend:
+    """记录 complete 执行区间的 backend：start/end 事件对可判断是否交错。"""
+
+    def __init__(self, delay: float = 0.03):
+        self.events: list[tuple[str, str]] = []
+        self._delay = delay
+
+    async def complete(self, chat_input):
+        self.events.append(("start", chat_input.text))
+        await asyncio.sleep(self._delay)
+        self.events.append(("end", chat_input.text))
+        return ChatResult(content="<state>1002</state>收到")
+
+    async def stream(self, chat_input):
+        self.events.append(("start", chat_input.text))
+        await asyncio.sleep(self._delay)
+        self.events.append(("end", chat_input.text))
+        yield TextDelta("<state>1002</state>收到")
+
+
+def _paired(events: list[tuple[str, str]]) -> bool:
+    """每个 start 与其 end 相邻配对（无交错）。顺序任意。"""
+    stack: list[str] = []
+    for kind, tag in events:
+        if kind == "start":
+            if stack:
+                return False
+            stack.append(tag)
+        else:
+            if not stack or stack.pop() != tag:
+                return False
+    return not stack
+
+
+@pytest.mark.asyncio
+async def test_concurrent_chats_same_session_serialize():
+    """同会话两个并发 chat 必须串行（区间不交错）。"""
+    backend = SerialProbeBackend()
+    service = AccidentAppService(backend, InMemorySessionRepository())
+    await asyncio.gather(
+        service.chat(_input("s1", "a")),
+        service.chat(_input("s1", "b")),
+    )
+    assert sorted(e[1] for e in backend.events) == ["a", "a", "b", "b"]
+    assert _paired(backend.events)
+
+
+@pytest.mark.asyncio
+async def test_concurrent_chats_different_sessions_run_parallel():
+    """不同会话的 chat 互不阻塞（第二个 start 在第一个 end 之前）。"""
+    backend = SerialProbeBackend(delay=0.05)
+    service = AccidentAppService(backend, InMemorySessionRepository())
+    await asyncio.gather(
+        service.chat(_input("s1", "a")),
+        service.chat(_input("s2", "b")),
+    )
+    starts_before_first_end = [
+        tag for kind, tag in backend.events[:2] if kind == "start"
+    ]
+    assert sorted(starts_before_first_end) == ["a", "b"]
+
+
+@pytest.mark.asyncio
+async def test_concurrent_delete_serialization():
+    """在途 chat 与 delete 互斥：delete 等 chat 完成后才执行。"""
+    backend = SerialProbeBackend()
+    service = AccidentAppService(backend, InMemorySessionRepository())
+
+    async def noop_drop(_session_id):
+        backend.events.append(("drop", "s1"))
+
+    await asyncio.gather(
+        service.chat(_input("s1", "a")),
+        service.delete_session("s1", drop_checkpoint=noop_drop),
+    )
+    # drop 必须出现在 chat 的 end 之后
+    kinds = [e for e in backend.events]
+    drop_index = next(i for i, e in enumerate(kinds) if e[0] == "drop")
+    end_index = next(i for i, e in enumerate(kinds) if e == ("end", "a"))
+    assert drop_index > end_index
+
+
+@pytest.mark.asyncio
+async def test_concurrent_chat_and_set_info_no_corruption():
+    """set_info（同步无锁）与并发 chat 交错：repository 键值保持完整。"""
+    backend = SerialProbeBackend()
+    service = AccidentAppService(backend, InMemorySessionRepository())
+    results = await asyncio.gather(
+        service.chat(_input("s1", "a")),
+        asyncio.to_thread(service.set_info, "s1", "hphm1", "苏B12345"),
+    )
+    assert results[1] == ("200", "")
+    # 无论时序如何，最终两个写者之一的车牌值完整存在（无撕裂）
+    assert json.loads(service.get_info("s1", "hphm1")) in ("苏B12345", "")