fix(api): report checkpoint deletion failures and classify SSE errors

Delete checkpoints before removing business fields. Suppress only an
explicit CheckpointNotFoundError; propagate storage failures and
cancellation so the endpoint cannot report false success. Prefer async
checkpoint deletion and fail clearly when cleanup is unsupported.

Classify SSE error events as 401 for authentication, 429 for rate limits,
and 500 for other failures, using the same fixed messages as non-stream
responses. Errors terminate the stream without a success done event or
provider exception text in the response.

Validation: 286 offline tests passed (19 added), including failed-delete
HTTP retry with preserved business/checkpoint state, explicit missing
threads, cancellation, missing cleanup capability, and HTTP SSE errors
before and after text output.

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。

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")
 

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 的
+    """删除图 checkpoint；缺少清理能力或存储出错必须报告失败。"""
-    完整语义——业务字段、事件、路由持久化清理在接入 repository 后补齐）。
-    """
     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"):
+    async_delete = getattr(checkpointer, "adelete_thread", None)
-        checkpointer.delete_thread(thread_id)
+    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:

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:
+                except CheckpointNotFoundError:
-                    # checkpoint 已不存在视为删除成功的一部分（幂等）
                     pass
+            self._repository.delete(session_id)
 
 
 __all__ = ["AccidentAppService", "SET_INFO_INVALID_KEY_MSG"]

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"]

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"

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 '<state>1002</state>'
+            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

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", "王五")

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')