diff --git a/src/services/app_service.py b/src/services/app_service.py
index 2f0f93b..7ccfd33 100644
--- a/src/services/app_service.py
+++ b/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"]
diff --git a/test/services/test_app_service.py b/test/services/test_app_service.py
index 81720ca..ae17de5 100644
--- a/test/services/test_app_service.py
+++ b/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="1002收到")
+
+ 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("1002收到")
+
+
+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", "")