diff --git a/docs/assignment6-compat.md b/docs/assignment6-compat.md index 65c2e7c..91c432d 100644 --- a/docs/assignment6-compat.md +++ b/docs/assignment6-compat.md @@ -23,7 +23,7 @@ | `set_info` 字段值 | 任意字符串原样存 | 按注册表类型解析（bool/int 严格），无法解析拒绝 | 保留 空/False/0/未知 区分，杜绝 `bool("0")` 陷阱 | | 图表单键与业务键 | FastGPT state 为裸 dict，无映射概念 | 建立双向映射（作业 6）：表单六键（`has_injury` 等）<-> acd 业务键（`ywrysw` 等）；`bjrjs`/`ywmtc` 无表单来源，仅 `set_info` 可写 | 图的表单语义与 get_info 业务分组需要同一份事实 | | LLM 调用 | set/get 走辅助对话 + 事后删记录 | **零 LLM 调用**（测试 spy 保证） | 成本与延迟；删除 race 消失 | -| 重复 `delete_session` | FastGPT SDK 行为未固定 | **幂等成功**（业务状态删除天然幂等；checkpoint 清理重复触发但缺失静默容忍） | 行为固定为测试 | +| 重复 `delete_session` | FastGPT SDK 行为未固定 | **幂等成功**（checkpoint 正常返回或适配器明确报告不存在才可继续；存储故障返回业务码 500） | 不再将清理失败当成删除成功 | ## 图表单键 <-> 业务键映射（作业 6 建立） @@ -99,6 +99,16 @@ 取消流时先关闭内部生成器、取消图任务，再释放会话锁。 - needFormUpdate 只决定对外是否返回差量。图提交时通过内部回调同步 repository，关闭差量显示或消费者在提交后断开不会丢失业务字段。 +- 删除在同会话锁内先清理 checkpoint，再删除业务字段。只有存储正常返回， + 或适配器明确抛出 CheckpointNotFoundError 时才视为 checkpoint 清理成功； + 连接、超时、一般 RuntimeError/KeyError 等故障必须上报，业务字段保留供重试。 + 缺少 checkpointer 或不支持删除操作也返回失败。优先使用 adelete_thread， + 仅在没有异步接口时兼容同步 delete_thread。若业务存储删除本身失败， + 同样返回失败；此顺序不构成跨存储事务，可能需要再次重试完成清理。 +- SSE 错误按 error 事件的 code 区分：鉴权 401、限流 429、其余 500， + 使用与非流式相同的固定文案，不回传底层异常原文。首包前和输出途中 + 都使用同一分类；错误是终止事件，随后不会发送 done。已经建立的 + SSE HTTP 响应保持 200，调用方检查事件内的业务码。 - SDK 自动重试关闭，由应用层统一控制；默认首次尝试加 3 次重试， 每次尝试总超时 60 秒，退避 1、2、4 秒。单次 LLM 调用理论预算 4×60+1+2+4=247 秒（另有响应关闭的清理时间）。FastGPT SDK 的 @@ -114,4 +124,7 @@ - `test/services/test_app_service.py`：四接口组合（set→get→chat 注入→ 差量写回→分组读→删除→无泄漏），set/get/delete 零模型调用 spy， 重复删除幂等。 -- `test/api/test_assignment6_endpoints.py`：HTTP 层验收演示链路。 +- `test/api/test_assignment6_endpoints.py`：HTTP 层验收演示、删除失败保留数据及重试成功链路。 +- `test/services/test_delete_failure_semantics.py`：存储故障、明确不存在、取消、清理能力缺失与重试。 +- `test/api/test_stream_error_codes.py`：实际 LangGraphBackend 过 HTTP 协议层， + 验证首包前/输出途中 401、429、500 分类、文案与非流式一致、错误后无 done。 diff --git a/src/api/endpoints.py b/src/api/endpoints.py index 2e4e129..fb95f41 100644 --- a/src/api/endpoints.py +++ b/src/api/endpoints.py @@ -415,7 +415,13 @@ async def chat( f"duration_ms={(time.perf_counter() - stream_started_at) * 1000:.1f} " f"error={e}" ) - yield create_sse_event("error", {"msg": str(e), "code": "500"}) + if isinstance(e, ChatBackendAuthenticationError): + code, msg = "401", "认证失败" + elif isinstance(e, ChatBackendRateLimitError): + code, msg = "429", "请求过于频繁，请稍后重试" + else: + code, msg = "500", "大模型服务器无响应" + yield create_sse_event("error", {"msg": msg, "code": code}) return StreamingResponse(event_generator(), media_type="text/event-stream") diff --git a/src/core/fastgpt_client.py b/src/core/fastgpt_client.py index 4fd9e92..300631f 100644 --- a/src/core/fastgpt_client.py +++ b/src/core/fastgpt_client.py @@ -71,14 +71,19 @@ async def lifespan(app: FastAPI): async def delete_chat_session(thread_id: str) -> None: - """langgraph 模式下删除会话：清理图 checkpoint（对照计划作业 6 的 - 完整语义——业务字段、事件、路由持久化清理在接入 repository 后补齐）。 - """ + """删除图 checkpoint；缺少清理能力或存储出错必须报告失败。""" if _langgraph_graph is None: raise RuntimeError("LangGraph backend not initialized") checkpointer = getattr(_langgraph_graph, "checkpointer", None) - if checkpointer is not None and hasattr(checkpointer, "delete_thread"): - checkpointer.delete_thread(thread_id) + async_delete = getattr(checkpointer, "adelete_thread", None) + if callable(async_delete): + await async_delete(thread_id) + return + sync_delete = getattr(checkpointer, "delete_thread", None) + if callable(sync_delete): + sync_delete(thread_id) + return + raise RuntimeError("LangGraph checkpointer does not support thread deletion") def get_fastgpt_client() -> AsyncChatClient: diff --git a/src/services/app_service.py b/src/services/app_service.py index 478ad95..ff68778 100644 --- a/src/services/app_service.py +++ b/src/services/app_service.py @@ -31,7 +31,7 @@ from .fields_registry import ( parse_field_value, to_external_string, ) -from .repository import SessionRepository +from .repository import CheckpointNotFoundError, SessionRepository from .session_locks import SessionLocks # set_info 对未知/受保护 key 的拒绝码（与现有“入参不规范”文案对齐） @@ -192,16 +192,16 @@ class AccidentAppService: （图生命周期归 lifespan 管理，本层不持有图引用）。 与会话锁互斥：删除不与在途 chat/set/get 交错。 释放锁引用时保留等待者使用的锁，最后一个使用者退出后回收。 - repository 删除幂等；checkpoint 缺失时静默（幂等语义一致）。 + checkpoint 清理成功（或明确不存在）后才删除业务字段。 + 其他异常向上传播，由 endpoint 返回失败，保留业务字段供重试。 """ async with self._locks.hold(session_id): - self._repository.delete(session_id) if drop_checkpoint is not None: try: await drop_checkpoint(session_id) - except Exception: - # checkpoint 已不存在视为删除成功的一部分（幂等） + except CheckpointNotFoundError: pass + self._repository.delete(session_id) __all__ = ["AccidentAppService", "SET_INFO_INVALID_KEY_MSG"] diff --git a/src/services/repository.py b/src/services/repository.py index 5de875a..6b6ebb8 100644 --- a/src/services/repository.py +++ b/src/services/repository.py @@ -14,6 +14,10 @@ from typing import Protocol +class CheckpointNotFoundError(Exception): + """存储适配器明确确认线程不存在；不能用一般存储异常代替。""" + + class SessionRepository(Protocol): """业务字段存储接口（作业 7 由 PostgreSQL 实现替换）。""" @@ -49,4 +53,4 @@ class InMemorySessionRepository: self._store.pop(session_id, None) -__all__ = ["SessionRepository", "InMemorySessionRepository"] +__all__ = ["CheckpointNotFoundError", "SessionRepository", "InMemorySessionRepository"] diff --git a/test/api/test_assignment6_endpoints.py b/test/api/test_assignment6_endpoints.py index 133622e..ed7a1b3 100644 --- a/test/api/test_assignment6_endpoints.py +++ b/test/api/test_assignment6_endpoints.py @@ -104,3 +104,43 @@ def test_assignment6_acceptance_flow(client): "text": "新对话", "clientMode": "direct", }) assert response.json()["code"] == "200" + + +def test_delete_checkpoint_failure_returns_error_and_retry_cleans_session(client, monkeypatch): + from unittest.mock import AsyncMock + from src.core import fastgpt_client as wiring + + session = "delete-retry" + payload = {"sessionId": session, "timeStamp": TS} + assert client.post("/set_info", json={ + **payload, "key": "hphm1", "value": "test-plate", + }).json()["code"] == "200" + assert client.post("/chat", json={ + **payload, "text": "新对话", + }).json()["nextStageCode"] == "1001" + saver = wiring._langgraph_graph.checkpointer + config = {"configurable": {"thread_id": session}} + assert saver.get_tuple(config) is not None + original_delete = saver.adelete_thread + failing_delete = AsyncMock(side_effect=ConnectionError("private storage detail")) + monkeypatch.setattr(saver, "adelete_thread", failing_delete) + + response = client.request("DELETE", "/delete_session", json=payload) + assert response.json()["code"] == "500" + assert response.json()["msg"] == "删除会话失败" + assert "private storage detail" not in response.text + failing_delete.assert_awaited_once_with(session) + assert saver.get_tuple(config) is not None + assert json.loads(client.post("/get_info", json={ + **payload, "key": "hphm1", + }).json()["value"]) == "test-plate" + + monkeypatch.setattr(saver, "adelete_thread", original_delete) + assert client.request("DELETE", "/delete_session", json=payload).json()["code"] == "200" + assert saver.get_tuple(config) is None + assert json.loads(client.post("/get_info", json={ + **payload, "key": "hphm1", + }).json()["value"]) == "" + assert client.post("/chat", json={ + **payload, "text": "新对话", + }).json()["nextStageCode"] == "1001" diff --git a/test/api/test_stream_error_codes.py b/test/api/test_stream_error_codes.py new file mode 100644 index 0000000..b185b8b --- /dev/null +++ b/test/api/test_stream_error_codes.py @@ -0,0 +1,76 @@ +import json + +import httpx +import pytest +from fastapi import FastAPI + +from src.agent.collect import validated_reply +from src.agent.llm import LLMAPIError, LLMAuthError, LLMRateLimitError +from src.api.endpoints import router +from src.backends.langgraph import LangGraphBackend +from src.core.config import Config +from src.core.fastgpt_client import get_chat_backend + + +class FailingLLM: + def __init__(self, error_type, partial): + self.error_type = error_type + self.partial = partial + + async def complete(self, *args, **kwargs): + raise self.error_type('private provider detail') + + async def stream_tokens(self, *args, **kwargs): + if self.partial: + yield '1002' + yield 'partial reply' + raise self.error_type('private provider detail') + + +class Graph: + async def ainvoke(self, data, *, config, context): + await validated_reply(context.llm, 'system', 'user', '1002') + raise AssertionError('failed turn must not complete') + + +def sse_events(body): + events = [] + for block in body.strip().split('\n\n'): + lines = block.splitlines() + kind = next(line[7:] for line in lines if line.startswith('event: ')) + data = json.loads(next(line[6:] for line in lines if line.startswith('data: '))) + events.append((kind, data)) + return events + + +@pytest.mark.asyncio +@pytest.mark.parametrize('partial', [False, True], ids=['before-text', 'after-text']) +@pytest.mark.parametrize('error_type,code,message', [ + (LLMAuthError, '401', '认证失败'), + (LLMRateLimitError, '429', '请求过于频繁，请稍后重试'), + (LLMAPIError, '500', '大模型服务器无响应'), +]) +async def test_http_sse_errors_match_non_stream_codes_and_terminate( + monkeypatch, partial, error_type, code, message, +): + # 独立协议测试 app：实际 LangGraphBackend/LLM 错误映射，不启动真实 provider。 + monkeypatch.setattr(Config, 'AGENT_BACKEND', 'fastgpt') + app = FastAPI() + app.include_router(router) + backend = LangGraphBackend(graph=Graph(), llm=FailingLLM(error_type, partial)) + app.dependency_overrides[get_chat_backend] = lambda: backend + payload = {'sessionId': 'error-audit', 'timeStamp': '20261022120000', 'text': 'hello'} + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test') as client: + response = await client.post('/chat?stream=true', json=payload) + assert response.status_code == 200 # 错误分类在 SSE data.code，不改已发送的 HTTP 头 + assert response.headers['content-type'].startswith('text/event-stream') + events = sse_events(response.text) + assert events[-1] == ('error', {'code': code, 'msg': message}) + assert sum(kind == 'error' for kind, _ in events) == 1 + assert all(kind != 'done' for kind, _ in events) + assert any(kind == 'text_delta' for kind, _ in events) is partial + assert 'private provider detail' not in response.text + + result = await client.post('/chat', json=payload) + assert result.json()['code'] == code + assert result.json()['msg'] == message diff --git a/test/services/test_app_service.py b/test/services/test_app_service.py index a58cbab..78bd85f 100644 --- a/test/services/test_app_service.py +++ b/test/services/test_app_service.py @@ -211,8 +211,10 @@ async def test_delete_clears_business_state_and_is_idempotent(): @pytest.mark.asyncio async def test_delete_tolerates_missing_checkpoint(): + from src.services.repository import CheckpointNotFoundError + async def missing(_session_id): - raise RuntimeError("no such thread") + raise CheckpointNotFoundError("no such thread") service, _ = make_service() await service.set_info("s1", "xm1", "王五") diff --git a/test/services/test_delete_failure_semantics.py b/test/services/test_delete_failure_semantics.py new file mode 100644 index 0000000..27b89e5 --- /dev/null +++ b/test/services/test_delete_failure_semantics.py @@ -0,0 +1,99 @@ +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from src.core import fastgpt_client as wiring +from src.services.app_service import AccidentAppService +from src.services.repository import CheckpointNotFoundError, InMemorySessionRepository + + +@pytest.mark.asyncio +@pytest.mark.parametrize('error', [ + ConnectionError('storage offline'), TimeoutError('storage timeout'), + RuntimeError('no such thread'), KeyError('storage schema'), +]) +async def test_checkpoint_failure_keeps_business_fields_and_allows_retry(error): + repo = InMemorySessionRepository() + service = AccidentAppService(SimpleNamespace(), repo) + await service.set_info('s', 'hphm1', 'test-plate') + drop = AsyncMock(side_effect=error) + with pytest.raises(type(error)): + await service.delete_session('s', drop) + assert await service.get_info('s', 'hphm1') == '"test-plate"' + assert not service._locks._entries + drop.side_effect = None + await service.delete_session('s', drop) + assert await service.get_info('s', 'hphm1') == '""' + + +@pytest.mark.asyncio +async def test_explicit_missing_checkpoint_is_idempotent(): + service = AccidentAppService(SimpleNamespace(), InMemorySessionRepository()) + await service.set_info('s', 'hphm1', 'test-plate') + drop = AsyncMock(side_effect=CheckpointNotFoundError()) + await service.delete_session('s', drop) + await service.delete_session('s', drop) + assert await service.get_info('s', 'hphm1') == '""' + assert drop.await_count == 2 + + +@pytest.mark.asyncio +async def test_cancelled_checkpoint_cleanup_keeps_business_fields(): + service = AccidentAppService(SimpleNamespace(), InMemorySessionRepository()) + await service.set_info('s', 'hphm1', 'test-plate') + started = asyncio.Event() + async def drop(_): + started.set() + await asyncio.Future() + task = asyncio.create_task(service.delete_session('s', drop)) + await asyncio.wait_for(started.wait(), 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert await service.get_info('s', 'hphm1') == '"test-plate"' + assert not service._locks._entries + + +@pytest.mark.asyncio +async def test_business_delete_failure_is_not_reported_as_success(): + class FailingRepository(InMemorySessionRepository): + def delete(self, _): + raise ConnectionError('business storage offline') + service = AccidentAppService(SimpleNamespace(), FailingRepository()) + await service.set_info('s', 'hphm1', 'test-plate') + drop = AsyncMock() + with pytest.raises(ConnectionError): + await service.delete_session('s', drop) + drop.assert_awaited_once_with('s') + assert await service.get_info('s', 'hphm1') == '"test-plate"' + + +@pytest.mark.asyncio +async def test_checkpoint_cleanup_uses_async_api_and_propagates_errors(monkeypatch): + saver = SimpleNamespace(adelete_thread=AsyncMock(side_effect=ConnectionError()), delete_thread=Mock()) + monkeypatch.setattr(wiring, '_langgraph_graph', SimpleNamespace(checkpointer=saver)) + with pytest.raises(ConnectionError): + await wiring.delete_chat_session('s') + saver.adelete_thread.assert_awaited_once_with('s') + saver.delete_thread.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('graph', [None, SimpleNamespace(), SimpleNamespace(checkpointer=object())]) +async def test_missing_cleanup_capability_fails(monkeypatch, graph): + monkeypatch.setattr(wiring, '_langgraph_graph', graph) + with pytest.raises(RuntimeError): + await wiring.delete_chat_session('s') + + +@pytest.mark.asyncio +async def test_sync_checkpoint_cleanup_is_supported_and_errors_propagate(monkeypatch): + saver = SimpleNamespace(delete_thread=Mock()) + monkeypatch.setattr(wiring, '_langgraph_graph', SimpleNamespace(checkpointer=saver)) + await wiring.delete_chat_session('s') + saver.delete_thread.assert_called_once_with('s') + saver.delete_thread.side_effect = ConnectionError() + with pytest.raises(ConnectionError): + await wiring.delete_chat_session('s')