diff --git a/docs/assignment6-compat.md b/docs/assignment6-compat.md
new file mode 100644
index 0000000..057ad98
--- /dev/null
+++ b/docs/assignment6-compat.md
@@ -0,0 +1,44 @@
+# 作业 6 兼容差异清单：LangGraph 业务状态 vs FastGPT 现状
+
+> 范围：`/set_info`、`/get_info`、`/delete_session`、`/chat` 四个接口在
+> `AGENT_BACKEND=langgraph` 下接入应用服务 + repository 后的行为差异。
+> FastGPT 模式（`AGENT_BACKEND=fastgpt`）保留旧实现，不受影响。
+
+## 接口响应契约（保持不变）
+
+| 项 | 现状（FastGPT） | LangGraph（作业 6） |
+|---|---|---|
+| 请求/响应模型 | `ProcessRequest/Response_*` | 相同（endpoint 未改模型） |
+| `get_info` 分组名 | `all` / `acdinfo` / `acdhuman1` / `acdhuman2` | 相同 |
+| `get_info` 布尔外部表示 | `True→"1"`，`False→"0"`，未知→`""` | 相同（`to_external_string`） |
+| `get_info` 未知 key | 返回 JSON 空串 `'""'` | **保持** `'""'`（现状兼容） |
+| `includeInputInfo` 字段 | 触发 FastGPT 辅助对话 | 继续接受但**忽略**——新后端零辅助对话 |
+
+## 有意的行为差异（新后端收紧）
+
+| 项 | FastGPT 现状 | LangGraph | 理由 |
+|---|---|---|---|
+| `set_info` 未知 key | 任意 key 写入 state 变量 | **拒绝**（500 + "入参不规范：未知或受保护的字段 key"） | 白名单保护；任意写入会污染图输入。上线前需审计旧调用方是否依赖任意 key 写入 |
+| `set_info` 图内部键（`stage_code` 等） | 可写（state 是裸 dict） | **拒绝**（`INTERNAL_FIELD_KEYS` 保护） | 内部状态只允许图自己推进 |
+| `set_info` 字段值 | 任意字符串原样存 | 按注册表类型解析（bool/int 严格），无法解析拒绝 | 保留 空/False/0/未知 区分，杜绝 `bool("0")` 陷阱 |
+| LLM 调用 | set/get 走辅助对话 + 事后删记录 | **零 LLM 调用**（测试 spy 保证） | 成本与延迟；删除 race 消失 |
+| 重复 `delete_session` | FastGPT SDK 行为未固定 | **幂等成功**（业务状态 + checkpoint 均静默容忍缺失） | 行为固定为测试 |
+
+## 待审计项（作业 6-7 要求的后续工作）
+
+- [ ] 审计现有调用方是否存在对 `set_info` 任意 key 写入的依赖
+      （如动态自定义键）。存在则在注册表补登记或提供逃生口。
+- [ ] 流式 `chat` 的 `formUpdate` 差量写回 repository：
+      骨架阶段流式原样透传，差量写回仅在非流式 `chat()` 生效；
+      完整版随 FormUpdate 事件捕获补齐（见 `app_service.stream` 注释）。
+- [ ] `get_info` 返回值中的敏感字段（证件号/手机号）回显边界：
+      注册表已记录 `SensitiveLevel`，访问控制随作业 9 落地。
+
+## 验证
+
+- `test/services/test_fields_registry.py`：注册表完整性、类型/敏感级别、
+  外部表示、`bool("0")` 防线。
+- `test/services/test_app_service.py`：四接口组合（set→get→chat 注入→
+  差量写回→分组读→删除→无泄漏），set/get/delete 零模型调用 spy，
+  重复删除幂等。
+- `test/api/test_assignment6_endpoints.py`：HTTP 层验收演示链路。
diff --git a/src/api/endpoints.py b/src/api/endpoints.py
index 3d70cad..3c5bf25 100644
--- a/src/api/endpoints.py
+++ b/src/api/endpoints.py
@@ -11,7 +11,12 @@ from ..backends.chat import (
     FormUpdate,
     TextDelta,
 )
-from ..core.fastgpt_client import get_chat_backend, get_fastgpt_client
+from ..core.fastgpt_client import (
+    delete_chat_session,
+    get_app_service,
+    get_chat_backend,
+    get_fastgpt_client,
+)
 from ..core.config import Config
 from ..utils.text_chunker import SentenceTextChunker, SentenceTextChunkerConfig
 from loguru import logger
@@ -131,7 +136,11 @@ async def chat(
     stream: bool = False,
     backend: ChatBackend = Depends(get_chat_backend)
 ):
-    """Handle chat completion request."""
+    """Handle chat completion request.
+
+    作业 6：langgraph 模式由应用服务承载业务状态同步
+    （repository 字段注入 + 差量写回），endpoint 仍负责协议解析。
+    """
     json_data = request.model_dump()
     client_mode = json_data['clientMode']
     need_form_update = json_data.get('needFormUpdate', False)
@@ -142,6 +151,14 @@ async def chat(
         need_form_update=need_form_update,
         client_mode=client_mode,
     )
+    # 应用服务：langgraph 模式下包装字段注入/写回；fastgpt 模式透传。
+    # 单测直调函数（未走 lifespan）时回退 backend 直调。
+    app_service = None
+    if Config.AGENT_BACKEND == "langgraph":
+        try:
+            app_service = get_app_service()
+        except RuntimeError:
+            app_service = None
     request_started_at = time.perf_counter()
     logger.info(
         "Chat request received "
@@ -250,7 +267,11 @@ async def chat(
                     state_filter_buffer = ""
                     return cleaned
                 
-                async for event in backend.stream(chat_input):
+                async for event in (
+                    app_service.stream(chat_input)
+                    if app_service is not None
+                    else backend.stream(chat_input)
+                ):
                     try:
                         if not first_event_logged:
                             first_event_logged = True
@@ -370,7 +391,10 @@ async def chat(
         return StreamingResponse(event_generator(), media_type="text/event-stream")
 
     try:
-        result = await backend.complete(chat_input)
+        if app_service is not None:
+            result = await app_service.chat(chat_input)
+        else:
+            result = await backend.complete(chat_input)
         logger.info(
             "Chat backend non-stream response received "
             f"sessionId={json_data['sessionId']} "
@@ -485,10 +509,28 @@ async def chat(
 @router.post("/set_info", response_model=ProcessResponse_set)
 async def set_info(
     request: ProcessRequest_set,
-    client: AsyncChatClient = Depends(get_fastgpt_client),
+    client=None,   # fastgpt 模式使用；注解省略（FastAPI 会解析参数注解）
 ):
-    """Set information in chat state."""
+    """Set information in chat state.
+
+    作业 6：langgraph 模式走应用服务 + repository——零 LLM 调用、
+    零辅助对话；includeInputInfo 继续接受但不再触发辅助对话。
+    fastgpt 模式保留旧实现。
+    """
     json_data = request.model_dump()
+
+    if Config.AGENT_BACKEND == "langgraph":
+        code, msg = get_app_service().set_info(
+            json_data['sessionId'], json_data['key'], json_data['value']
+        )
+        return ProcessResponse_set(
+            sessionId=json_data['sessionId'],
+            timeStamp=json_data['timeStamp'],
+            code=code,
+            msg=msg or None,
+        )
+
+    client = client or get_fastgpt_client()
     set_info_payload = {'key': json_data['key'], 'value': json_data['value']}
     set_info_input = format_set_info_input(
         set_info_payload,
@@ -577,10 +619,25 @@ async def set_info(
 @router.post("/get_info", response_model=ProcessResponse_get)
 async def get_info(
     request: ProcessRequest_get,
-    client: AsyncChatClient = Depends(get_fastgpt_client),
+    client=None,   # fastgpt 模式使用；注解省略（FastAPI 会解析参数注解）
 ):
-    """Get information from chat state."""
+    """Get information from chat state.
+
+    作业 6：langgraph 模式走应用服务 + repository——零 LLM 调用；
+    分组与单键读法兼容现有契约，未知 key 返回 JSON 空串。
+    """
     json_data = request.model_dump()
+
+    if Config.AGENT_BACKEND == "langgraph":
+        return ProcessResponse_get(
+            sessionId=json_data['sessionId'],
+            timeStamp=json_data['timeStamp'],
+            value=get_app_service().get_info(json_data['sessionId'], json_data['key']),
+            code="200",
+            msg="",
+        )
+
+    client = client or get_fastgpt_client()
     get_info_payload = {'key': json_data['key']}
     get_info_input = format_set_info_input(
         get_info_payload,
@@ -700,9 +757,10 @@ async def delete_session(
         )
 
     if Config.AGENT_BACKEND == "langgraph":
-        from ..core.fastgpt_client import delete_chat_session
         try:
-            await delete_chat_session(chat_id)
+            await get_app_service().delete_session(
+                chat_id, drop_checkpoint=delete_chat_session
+            )
             return ProcessResponse_delete_session(
                 sessionId=json_data['sessionId'],
                 timeStamp=json_data['timeStamp'],
diff --git a/src/backends/chat.py b/src/backends/chat.py
index 18533ea..4c43c06 100644
--- a/src/backends/chat.py
+++ b/src/backends/chat.py
@@ -13,6 +13,10 @@ class ChatInput:
     text: str
     need_form_update: bool = False
     client_mode: str = ""
+    # 作业 6：应用服务注入的最新业务字段（acd 进 accident_fields，
+    # 人员进 person_fields）。LangGraph 后端合并进图输入；
+    # FastGPT 后端忽略该字段。
+    injected_fields: dict | None = None
 
 
 @dataclass(frozen=True)
diff --git a/src/backends/langgraph.py b/src/backends/langgraph.py
index 5ea057c..c9ae6a0 100644
--- a/src/backends/langgraph.py
+++ b/src/backends/langgraph.py
@@ -49,12 +49,17 @@ class LangGraphBackend(ChatBackend):
         - 本阶段不处理 ``need_form_update`` 的真实表单差量
           （作业 6 起实现）。
         """
+        graph_input: dict = {
+            "session_id": chat_input.session_id,
+            "input_text": chat_input.text,
+            "client_mode": chat_input.client_mode,
+        }
+        # 作业 6：应用服务注入的最新业务字段覆盖本轮输入，
+        # 使 set_info 写入后本轮 chat 立即可见（输入覆盖 checkpoint 同键）。
+        if chat_input.injected_fields:
+            graph_input.update(chat_input.injected_fields)
         result = await self._graph.ainvoke(
-            {
-                "session_id": chat_input.session_id,
-                "input_text": chat_input.text,
-                "client_mode": chat_input.client_mode,
-            },
+            graph_input,
             config={"configurable": {"thread_id": chat_input.session_id}},
             context=GraphDeps(llm=self._llm),
         )
diff --git a/src/core/fastgpt_client.py b/src/core/fastgpt_client.py
index fdd307a..9695256 100644
--- a/src/core/fastgpt_client.py
+++ b/src/core/fastgpt_client.py
@@ -6,12 +6,15 @@ from ..agent.graph import build_graph
 from ..backends.chat import ChatBackend
 from ..backends.fastgpt import FastGPTBackend
 from ..backends.langgraph import LangGraphBackend
+from ..services.app_service import AccidentAppService
+from ..services.repository import InMemorySessionRepository
 from .config import Config
 
 
 _fastgpt_client: AsyncChatClient | None = None
 _chat_backend: ChatBackend | None = None
 _langgraph_graph = None
+_app_service: AccidentAppService | None = None
 
 
 @asynccontextmanager
@@ -25,9 +28,13 @@ async def lifespan(app: FastAPI):
         # TODO(作业 2): 若语音仍启用，需在此单独初始化并校验 FastGPT 配置。
         from ..agent.llm import build_llm_service
 
-        global _langgraph_graph
+        global _langgraph_graph, _app_service
         _langgraph_graph = build_graph()
         _chat_backend = LangGraphBackend(_langgraph_graph, llm=build_llm_service())
+        # 作业 6：应用服务 + 内存 repository（作业 7 换 PostgreSQL）。
+        _app_service = AccidentAppService(
+            _chat_backend, InMemorySessionRepository()
+        )
     else:
         _fastgpt_client = AsyncChatClient(
             api_key=Config.FASTGPT_API_KEY,
@@ -45,6 +52,7 @@ async def lifespan(app: FastAPI):
     finally:
         _chat_backend = None
         _langgraph_graph = None
+        _app_service = None
         if _fastgpt_client:
             await _fastgpt_client.__aexit__(None, None, None)
         _fastgpt_client = None
@@ -68,6 +76,13 @@ def get_fastgpt_client() -> AsyncChatClient:
     return _fastgpt_client
 
 
+def get_app_service() -> AccidentAppService:
+    """Get the initialized application service（langgraph 模式）。"""
+    if _app_service is None:
+        raise RuntimeError("Application service not initialized")
+    return _app_service
+
+
 def get_chat_backend() -> ChatBackend:
     """Get the backend-neutral chat service."""
     if _chat_backend is None:
diff --git a/src/services/__init__.py b/src/services/__init__.py
new file mode 100644
index 0000000..fef5e97
--- /dev/null
+++ b/src/services/__init__.py
@@ -0,0 +1,5 @@
+"""应用服务层（作业 6）：应用服务、字段注册表、会话 repository。
+
+endpoint 只处理校验、协议（SSE/响应模型）和依赖注入；
+业务状态读写、字段白名单与分组语义集中在本层。
+"""
diff --git a/src/services/app_service.py b/src/services/app_service.py
new file mode 100644
index 0000000..cde348c
--- /dev/null
+++ b/src/services/app_service.py
@@ -0,0 +1,152 @@
+"""事故处理应用服务（作业 6-1）：四业务接口共享同一份业务状态。
+
+职责边界（对应计划作业 6）：
+- set_info / get_info / delete_session **不调用 LLM、不创建辅助对话**
+  （对照旧 FastGPT 实现的“辅助对话 + 删记录”hack）；
+- chat 每轮从 repository 读最新业务字段注入图（set_info 写入后
+  下一轮 chat 必须可见），图产出字段差量后写回 repository；
+- 未知 key 兼容策略：get 对未知 key 返回 JSON 空串 '""'（现状兼容）；
+  set 对未知 key 拒绝写入（白名单保护，差异见兼容清单）；
+- includeInputInfo 继续被 endpoint 接受，但本层不再使用
+  ——新后端无需触发辅助对话。
+
+endpoint 只负责校验、协议（SSE / 响应模型）与依赖注入。
+"""
+
+from ..backends.chat import ChatBackend, ChatInput, ChatResult
+from .fields_registry import (
+    FIELD_REGISTRY,
+    GROUP_KEYS,
+    FieldGroup,
+    parse_field_value,
+    to_external_string,
+)
+from .repository import SessionRepository
+
+# set_info 对未知/受保护 key 的拒绝码（与现有“入参不规范”文案对齐）
+SET_INFO_INVALID_KEY_MSG = "入参不规范：未知或受保护的字段 key"
+
+
+class AccidentAppService:
+    """四个业务接口的应用服务（langgraph 后端）。"""
+
+    def __init__(self, backend: ChatBackend, repository: SessionRepository):
+        self._backend = backend
+        self._repository = repository
+
+    # ---- chat：图每轮读取最新业务状态，产出差量写回 ----
+
+    def _injected_fields(self, session_id: str) -> dict:
+        """repository 业务字段 -> 图输入的分组注入（acd 进 accident_fields，
+        两组人员进 person_fields，与 AgentState 的键结构对齐）。"""
+        business = self._repository.load(session_id)
+        if not business:
+            return {}
+        injected: dict = {"accident_fields": {}, "person_fields": {}}
+        for key, value in business.items():
+            spec = FIELD_REGISTRY.get(key)
+            if spec is None:
+                continue
+            if spec.group is FieldGroup.ACD:
+                injected["accident_fields"][key] = value
+            else:
+                injected["person_fields"][key] = value
+        return injected
+
+    async def chat(self, chat_input: ChatInput) -> ChatResult:
+        """非流式一轮对话：注入最新业务字段 -> 执行图 -> 写回字段差量。"""
+        injected = self._injected_fields(chat_input.session_id)
+        result = await self._backend.complete(
+            ChatInput(
+                session_id=chat_input.session_id,
+                text=chat_input.text,
+                need_form_update=chat_input.need_form_update,
+                client_mode=chat_input.client_mode,
+                injected_fields=injected or None,
+            )
+        )
+        # 图产出的字段差量（turn_patch / 完整字段）写回 repository，
+        # 使 set/get_info 与后续轮次看到同一事实来源
+        patch = result.form_update if isinstance(result.form_update, dict) else {}
+        if patch:
+            self._repository.apply_update(chat_input.session_id, patch)
+        return result
+
+    async def stream(self, chat_input: ChatInput):
+        """流式代理：字段注入与写回语义同 chat()，事件流原样透传。"""
+        injected = self._injected_fields(chat_input.session_id)
+        result_input = ChatInput(
+            session_id=chat_input.session_id,
+            text=chat_input.text,
+            need_form_update=chat_input.need_form_update,
+            client_mode=chat_input.client_mode,
+            injected_fields=injected or None,
+        )
+        # 骨架阶段原样透传事件流；字段差量写回在作业 6 完整版随
+        # FormUpdate 事件捕获补齐（对照计划作业 8 的真实增量）。
+        async for event in self._backend.stream(result_input):
+            yield event
+
+    # ---- set_info / get_info：零 LLM、零辅助对话 ----
+
+    def set_info(self, session_id: str, key: str, value: str) -> tuple[str, str]:
+        """白名单校验后写入 repository。返回 (code, msg)。"""
+        spec = FIELD_REGISTRY.get(key)
+        if spec is None or not spec.writable:
+            return "500", SET_INFO_INVALID_KEY_MSG
+        try:
+            parsed = parse_field_value(spec, value)
+        except ValueError:
+            return "500", "入参不规范：字段值无法解析"
+        self._repository.apply_update(session_id, {key: parsed})
+        return "200", ""
+
+    def get_info(self, session_id: str, key: str) -> str:
+        """按分组或单键读出，返回 JSON 字符串（兼容现有 /get_info 契约）。"""
+        import json
+
+        business = self._repository.load(session_id)
+        if key == "all":
+            return json.dumps(
+                {
+                    group.value: {
+                        k: to_external_string(business.get(k))
+                        for k in GROUP_KEYS[group]
+                    }
+                    for group in FieldGroup
+                },
+                ensure_ascii=False,
+            )
+        try:
+            group = FieldGroup(key)
+        except ValueError:
+            group = None
+        if group is not None:
+            return json.dumps(
+                {k: to_external_string(business.get(k)) for k in GROUP_KEYS[group]},
+                ensure_ascii=False,
+            )
+        # 单键：未知 key 返回 JSON 空串（兼容现状）；已知 key 返回其外部表示
+        if key not in FIELD_REGISTRY:
+            return json.dumps("", ensure_ascii=False)
+        return json.dumps(to_external_string(business.get(key)), ensure_ascii=False)
+
+    # ---- delete_session：业务状态 + 图 checkpoint 一起清理 ----
+
+    async def delete_session(self, session_id: str, drop_checkpoint) -> None:
+        """删除会话全部数据。
+
+        drop_checkpoint：由 core 层注入的 checkpoint 清理函数
+        （图生命周期归 lifespan 管理，本层不持有图引用）。
+        repository 删除幂等；checkpoint 缺失时静默（幂等语义一致）。
+        """
+        self._repository.delete(session_id)
+        if drop_checkpoint is not None:
+            try:
+                await drop_checkpoint(session_id)
+            except Exception:
+                # checkpoint 已不存在视为删除成功的一部分（幂等）
+                pass
+
+
+__all__ = ["AccidentAppService", "SET_INFO_INVALID_KEY_MSG"]
diff --git a/src/services/fields_registry.py b/src/services/fields_registry.py
new file mode 100644
index 0000000..e51b956
--- /dev/null
+++ b/src/services/fields_registry.py
@@ -0,0 +1,144 @@
+"""字段注册表（作业 6-3/6-4）：外部业务键 -> 类型/分组/可写性/敏感级别。
+
+完整覆盖当前 get_info 代码中的三个分组：
+- 事故字段（acdinfo）：ywrysw/ywfjdc/ywmtc/bjrjs/sgfssj/sfsgxc/jdcsl/sgyy
+- 本方人员（acdhuman1）：xm1/hpzl1/hphm1/sfzmhm1/sfzmwh1/sjhm1/sjwh1/csbw1
+- 对方人员（acdhuman2）：同上一组，序号 2
+
+设计约束：
+- 外部传输一律字符串；内部按 value_type 存储强类型值，
+  保留区分 空值 / False / 0 / 未知 的能力（不做 bool("0") 转换）；
+- 敏感级别用于作业 9 的访问边界与日志脱敏，注册表先行记录；
+- get 的分组名（acdinfo/acdhuman1/acdhuman2）与外部接口契约保持一致。
+"""
+
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Any
+
+
+class FieldGroup(StrEnum):
+    """get_info 对外分组名（兼容现有 /get_info 契约）。"""
+
+    ACD = "acdinfo"
+    HUMAN1 = "acdhuman1"
+    HUMAN2 = "acdhuman2"
+
+
+class SensitiveLevel(StrEnum):
+    """敏感级别（作业 9 用于访问边界/脱敏）。"""
+
+    PUBLIC = "public"        # 无个人信息
+    PERSONAL = "personal"    # 姓名等一般个人信息
+    SENSITIVE = "sensitive"  # 证件号/手机号，日志与回显需保护
+
+
+@dataclass(frozen=True)
+class FieldSpec:
+    """一个外部业务字段的注册信息。"""
+
+    key: str
+    group: FieldGroup
+    value_type: type  # str / bool / int
+    writable: bool
+    sensitive: SensitiveLevel
+    description: str = ""
+
+
+def _acd_fields() -> list[FieldSpec]:
+    return [
+        FieldSpec("ywrysw", FieldGroup.ACD, bool, True, SensitiveLevel.PUBLIC, "是否有人受伤"),
+        FieldSpec("ywfjdc", FieldGroup.ACD, bool, True, SensitiveLevel.PUBLIC, "是否撞非机动车"),
+        FieldSpec("ywmtc", FieldGroup.ACD, bool, True, SensitiveLevel.PUBLIC, "是否撞摩托车"),
+        FieldSpec("bjrjs", FieldGroup.ACD, bool, True, SensitiveLevel.PUBLIC, "报警人是否驾驶员"),
+        FieldSpec("sgfssj", FieldGroup.ACD, str, True, SensitiveLevel.PUBLIC, "事故发生时间"),
+        FieldSpec("sfsgxc", FieldGroup.ACD, bool, True, SensitiveLevel.PUBLIC, "是否在现场"),
+        FieldSpec("jdcsl", FieldGroup.ACD, int, True, SensitiveLevel.PUBLIC, "机动车数量"),
+        FieldSpec("sgyy", FieldGroup.ACD, str, True, SensitiveLevel.PUBLIC, "事故原因"),
+    ]
+
+
+def _human_fields(index: int) -> list[FieldSpec]:
+    group = FieldGroup.HUMAN1 if index == 1 else FieldGroup.HUMAN2
+    side = "本方" if index == 1 else "对方"
+    return [
+        FieldSpec(f"xm{index}", group, str, True, SensitiveLevel.PERSONAL, f"{side}姓名"),
+        FieldSpec(f"hpzl{index}", group, str, True, SensitiveLevel.PUBLIC, f"{side}号牌种类"),
+        FieldSpec(f"hphm{index}", group, str, True, SensitiveLevel.PUBLIC, f"{side}号牌号码"),
+        FieldSpec(f"sfzmhm{index}", group, str, True, SensitiveLevel.SENSITIVE, f"{side}身份证明号码"),
+        FieldSpec(f"sfzmwh{index}", group, str, True, SensitiveLevel.SENSITIVE, f"{side}身份证明文号"),
+        FieldSpec(f"sjhm{index}", group, str, True, SensitiveLevel.SENSITIVE, f"{side}手机号码"),
+        FieldSpec(f"sjwh{index}", group, str, True, SensitiveLevel.PUBLIC, f"{side}手机文号"),
+        FieldSpec(f"csbw{index}", group, str, True, SensitiveLevel.PUBLIC, f"{side}常熟部位/身份标识"),
+    ]
+
+
+# 注册表：key -> FieldSpec（构建一次，只读共享）
+FIELD_REGISTRY: dict[str, FieldSpec] = {
+    spec.key: spec for spec in [*_acd_fields(), *_human_fields(1), *_human_fields(2)]
+}
+
+# 分组 -> 组内有序字段键（保持当前 get_info 的键顺序，输出稳定）
+GROUP_KEYS: dict[FieldGroup, tuple[str, ...]] = {
+    group: tuple(spec.key for spec in FIELD_REGISTRY.values() if spec.group == group)
+    for group in FieldGroup
+}
+
+# 内部受保护字段：注册表之外、图内部使用的键，set_info 不得写入
+# （兼容策略见 docs/assignment6-compat.md）
+INTERNAL_FIELD_KEYS = frozenset(
+    {
+        "stage_code", "raw_status_code", "turn_count", "client_event",
+        "clarify_count", "no_reply_count", "state_version",
+    }
+)
+
+
+def parse_field_value(spec: FieldSpec, raw: str) -> Any:
+    """外部字符串 -> 内部强类型值；无法解析时抛 ValueError。
+
+    bool 只接受明确的 1/0/true/false/是/否；拒绝空串与模糊值，
+    避免 bool("0") == True 这类陷阱。int 必须是纯数字字面量。
+    """
+    if raw is None:
+        raise ValueError(f"{spec.key}: 值为空")
+    text = raw.strip()
+    if spec.value_type is bool:
+        lowered = text.lower()
+        if lowered in ("1", "true", "是"):
+            return True
+        if lowered in ("0", "false", "否"):
+            return False
+        raise ValueError(f"{spec.key}: 无法解析为布尔值: {raw!r}")
+    if spec.value_type is int:
+        if text.isdigit():
+            return int(text)
+        raise ValueError(f"{spec.key}: 无法解析为整数: {raw!r}")
+    return raw  # str 原样存储（空串视为有效的外部空值）
+
+
+def to_external_string(value: Any) -> str:
+    """内部值 -> 外部字符串表示（兼容现有 get_info 的 bool_to_str）。
+
+    True -> "1"，False -> "0"，None/未知 -> ""，其余 str() 原样。
+    内部仍保留 False 与 0 的区分，仅在对外序列化时收敛。
+    """
+    if value is True:
+        return "1"
+    if value is False:
+        return "0"
+    if value is None:
+        return ""
+    return str(value)
+
+
+__all__ = [
+    "FIELD_REGISTRY",
+    "GROUP_KEYS",
+    "INTERNAL_FIELD_KEYS",
+    "FieldGroup",
+    "FieldSpec",
+    "SensitiveLevel",
+    "parse_field_value",
+    "to_external_string",
+]
diff --git a/src/services/repository.py b/src/services/repository.py
new file mode 100644
index 0000000..5de875a
--- /dev/null
+++ b/src/services/repository.py
@@ -0,0 +1,52 @@
+"""会话业务状态 repository（作业 6-1/6-2）。
+
+抽象出 repository 接口，使应用服务不依赖具体存储；
+作业 6 用内存实现跑通四接口，作业 7 再替换为 PostgreSQL 实现
+（对话 checkpoint 已由 LangGraph checkpointer 承担，本层只管
+业务字段——单一事实来源，图每轮读取最新值）。
+
+语义约定：
+- 业务状态按外部业务键（hphm1、ywrysw...）扁平存储，值为内部强类型；
+- 不存在的会话视为空状态（读取返回 {}），与首轮 checkpoint 无字段一致；
+- delete 幂等：重复删除固定成功，不抛异常（行为进测试）。
+"""
+
+from typing import Protocol
+
+
+class SessionRepository(Protocol):
+    """业务字段存储接口（作业 7 由 PostgreSQL 实现替换）。"""
+
+    def load(self, session_id: str) -> dict:
+        """读取会话业务字段；无记录时返回空 dict。"""
+        ...
+
+    def apply_update(self, session_id: str, patch: dict) -> dict:
+        """合并差量并返回合并后的完整业务字段。"""
+        ...
+
+    def delete(self, session_id: str) -> None:
+        """删除会话业务状态；幂等，重复删除不抛异常。"""
+        ...
+
+
+class InMemorySessionRepository:
+    """内存实现：字典存储，仅用于单进程开发与测试。"""
+
+    def __init__(self) -> None:
+        self._store: dict[str, dict] = {}
+
+    def load(self, session_id: str) -> dict:
+        return dict(self._store.get(session_id) or {})
+
+    def apply_update(self, session_id: str, patch: dict) -> dict:
+        state = self._store.setdefault(session_id, {})
+        state.update(patch)
+        return dict(state)
+
+    def delete(self, session_id: str) -> None:
+        # 幂等：pop 的缺省值吞掉不存在的情况，重复删除行为固定为成功
+        self._store.pop(session_id, None)
+
+
+__all__ = ["SessionRepository", "InMemorySessionRepository"]
diff --git a/test/api/test_assignment6_endpoints.py b/test/api/test_assignment6_endpoints.py
new file mode 100644
index 0000000..133622e
--- /dev/null
+++ b/test/api/test_assignment6_endpoints.py
@@ -0,0 +1,106 @@
+"""作业 6 验收演示的 HTTP 层组合测试。
+
+链路（对应计划作业 6）：创建会话 -> /set_info 写车牌 -> /get_info 读回
+-> /chat 继续核实 -> 分组读值 -> /delete_session -> 再读与再聊不泄漏。
+
+fake LLM（LLM_PROVIDER=fake）保证本测试离线可跑、零真实模型调用。
+"""
+
+import json
+
+import pytest
+from fastapi.testclient import TestClient
+
+from src.agent.llm import workflow_fake_llm
+from src.core.config import Config
+from src.main import app
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    """langgraph 后端 + fake LLM 的测试客户端（lifespan 完整初始化）。"""
+    monkeypatch.setattr(Config, "AGENT_BACKEND", "langgraph")
+    monkeypatch.setattr(Config, "LLM_PROVIDER", "fake")
+
+    import src.agent.llm as llm_module
+
+    monkeypatch.setattr(
+        llm_module, "build_llm_service", lambda *a, **k: workflow_fake_llm()
+    )
+    with TestClient(app) as test_client:
+        yield test_client
+
+
+TS = "20261022120000"
+
+
+def test_assignment6_acceptance_flow(client):
+    session = "a6-acceptance-001"
+
+    # set_info 写入车牌（零 LLM：fake 未注入任何 responder，调用即报错）
+    response = client.post("/set_info", json={
+        "sessionId": session, "timeStamp": TS,
+        "key": "hphm1", "value": "苏B12345",
+    })
+    assert response.status_code == 200
+    assert response.json()["code"] == "200"
+
+    # get_info 单键读回
+    response = client.post("/get_info", json={
+        "sessionId": session, "timeStamp": TS, "key": "hphm1",
+    })
+    assert response.json()["value"] == '"苏B12345"'
+
+    # chat 使用业务状态继续（非流式，协议层解析 state 前缀）
+    response = client.post("/chat", json={
+        "sessionId": session, "timeStamp": TS,
+        "text": "我是刚才报警的驾驶员", "clientMode": "direct",
+    })
+    body = response.json()
+    assert body["code"] == "200"
+    assert body["nextStageCode"] in ("1001", "1002")
+
+    # 分组读值：三个分组齐全，未填字段为空串
+    response = client.post("/get_info", json={
+        "sessionId": session, "timeStamp": TS, "key": "all",
+    })
+    groups = json.loads(response.json()["value"])
+    assert set(groups) == {"acdinfo", "acdhuman1", "acdhuman2"}
+    assert groups["acdhuman1"]["hphm1"] == "苏B12345"
+    assert groups["acdhuman1"]["xm1"] == ""
+
+    # 未知 key 兼容：JSON 空串
+    response = client.post("/get_info", json={
+        "sessionId": session, "timeStamp": TS, "key": "no_such_key",
+    })
+    assert response.json()["value"] == '""'
+
+    # set_info 未知 key：白名单拒绝
+    response = client.post("/set_info", json={
+        "sessionId": session, "timeStamp": TS,
+        "key": "no_such_key", "value": "x",
+    })
+    assert response.json()["code"] == "500"
+
+    # 删除 -> 再读为空
+    response = client.request("DELETE", "/delete_session", json={
+        "sessionId": session, "timeStamp": TS,
+    })
+    assert response.json()["code"] == "200"
+    response = client.post("/get_info", json={
+        "sessionId": session, "timeStamp": TS, "key": "hphm1",
+    })
+    assert response.json()["value"] == '""'
+
+    # 重复删除幂等成功
+    response = client.request("DELETE", "/delete_session", json={
+        "sessionId": session, "timeStamp": TS,
+    })
+    assert response.json()["code"] == "200"
+
+    # 删除后再聊天：不注入旧业务数据，服务正常应答
+    response = client.post("/chat", json={
+        "sessionId": session, "timeStamp": TS,
+        "text": "新对话", "clientMode": "direct",
+    })
+    assert response.json()["code"] == "200"
diff --git a/test/services/test_app_service.py b/test/services/test_app_service.py
new file mode 100644
index 0000000..77dcf0b
--- /dev/null
+++ b/test/services/test_app_service.py
@@ -0,0 +1,192 @@
+"""作业 6：应用服务四接口组合测试。
+
+验收演示对应（计划作业 6）：
+创建会话 -> set_info 写车牌 -> get_info 读回 -> chat 使用车牌继续核实
+-> 分组读值 -> 删除 -> 再读与再次聊天均不泄漏旧数据。
+
+spy 断言：set_info / get_info / delete_session 全程零模型调用。
+"""
+
+import json
+
+import pytest
+
+from src.backends.chat import ChatResult, TextDelta, FormUpdate
+from src.services.app_service import AccidentAppService, SET_INFO_INVALID_KEY_MSG
+from src.services.repository import InMemorySessionRepository
+
+
+class SpyBackend:
+    """记录调用的 ChatBackend 替身；form_update 模拟图产出的字段差量。"""
+
+    def __init__(self, form_update=None):
+        self.calls = []           # complete 收到的 ChatInput
+        self.stream_calls = []
+        self._form_update = form_update or {}
+        # 用调用次数当“模型调用”spy：真实 backend 每次 complete 至少一次 LLM
+        self.llm_calls = 0
+
+    async def complete(self, chat_input):
+        self.calls.append(chat_input)
+        self.llm_calls += 1
+        return ChatResult(
+            content="<state>1002</state>收到",
+            status_code=None,
+            form_update=dict(self._form_update),
+        )
+
+    async def stream(self, chat_input):
+        self.stream_calls.append(chat_input)
+        self.llm_calls += 1
+        yield TextDelta("<state>1002</state>")
+        if self._form_update:
+            yield FormUpdate(dict(self._form_update))
+        yield TextDelta("收到")
+
+
+def make_service(form_update=None):
+    backend = SpyBackend(form_update=form_update)
+    service = AccidentAppService(backend, InMemorySessionRepository())
+    return service, backend
+
+
+# ---- set/get：白名单、分组、外部表示 ----
+
+def test_set_then_get_single_key_roundtrip():
+    service, backend = make_service()
+    code, msg = service.set_info("s1", "hphm1", "苏B12345")
+    assert (code, msg) == ("200", "")
+    assert json.loads(service.get_info("s1", "hphm1")) == "苏B12345"
+    # set/get 零模型调用
+    assert backend.llm_calls == 0
+
+
+def test_get_groups_match_existing_contract():
+    service, _ = make_service()
+    service.set_info("s1", "ywrysw", "0")       # False
+    service.set_info("s1", "jdcsl", "0")        # int 0
+    service.set_info("s1", "xm2", "张三")
+
+    acd = json.loads(service.get_info("s1", "acdinfo"))
+    assert acd["ywrysw"] == "0"      # False -> "0"
+    assert acd["jdcsl"] == "0"       # int 0 -> "0"（内部类型不同）
+    assert acd["sgyy"] == ""         # 未知 -> 空串
+
+    human2 = json.loads(service.get_info("s1", "acdhuman2"))
+    assert human2["xm2"] == "张三"
+    assert human2["sfzmwh2"] == ""   # 全字段齐全，未填为空
+
+    all_groups = json.loads(service.get_info("s1", "all"))
+    assert set(all_groups) == {"acdinfo", "acdhuman1", "acdhuman2"}
+
+
+def test_get_unknown_key_returns_json_empty_string():
+    """兼容现状：未知 key 返回 JSON 空串 '""'。"""
+    service, _ = make_service()
+    service.set_info("s1", "hphm1", "苏B12345")
+    assert service.get_info("s1", "no_such_key") == '""'
+    assert service.get_info("ghost-session", "hphm1") == '""'
+
+
+def test_set_unknown_or_internal_key_rejected():
+    service, backend = make_service()
+    code, msg = service.set_info("s1", "no_such_key", "x")
+    assert code == "500" and msg == SET_INFO_INVALID_KEY_MSG
+    code, _ = service.set_info("s1", "stage_code", "1002")
+    assert code == "500"          # 内部字段保护
+    assert service.get_info("s1", "hphm1") == '""'   # 未被污染
+    assert backend.llm_calls == 0
+
+
+def test_set_rejects_unparsable_value():
+    service, _ = make_service()
+    code, msg = service.set_info("s1", "ywrysw", "maybe")
+    assert code == "500" and "无法解析" in msg
+
+
+# ---- chat：注入 + 写回 ----
+
+@pytest.mark.asyncio
+async def test_chat_sees_set_info_values_and_writes_back_patch():
+    """set_info 写入 -> chat 输入注入 -> 图差量写回 -> get_info 读到。"""
+    service, backend = make_service(form_update={"ywrysw": False})
+    service.set_info("s1", "hphm1", "苏B12345")
+    service.set_info("s1", "xm2", "李四")
+
+    result = await service.chat(
+        _input("s1", "我是刚才报警的驾驶员")
+    )
+    assert result.content == "<state>1002</state>收到"
+
+    # 图输入必须携带 repository 最新业务字段（车牌供核实使用）
+    injected = backend.calls[0].injected_fields
+    assert injected["person_fields"]["hphm1"] == "苏B12345"
+    assert injected["person_fields"]["xm2"] == "李四"
+
+    # 图产出的差量写回 repository，get_info 立即可读
+    assert json.loads(service.get_info("s1", "acdinfo"))["ywrysw"] == "0"
+    assert json.loads(service.get_info("s1", "hphm1")) == "苏B12345"
+
+
+@pytest.mark.asyncio
+async def test_chat_without_business_state_passes_no_injection():
+    service, backend = make_service()
+    await service.chat(_input("fresh-session", "你好"))
+    assert backend.calls[0].injected_fields is None
+
+
+@pytest.mark.asyncio
+async def test_stream_injects_fields_and_passthrough_events():
+    service, backend = make_service(form_update={"ywrysw": False})
+    service.set_info("s1", "xm1", "王五")
+    events = [
+        event async for event in service.stream(_input("s1", "继续办理"))
+    ]
+    kinds = [type(event).__name__ for event in events]
+    assert kinds == ["TextDelta", "FormUpdate", "TextDelta"]
+    assert backend.stream_calls[0].injected_fields["person_fields"]["xm1"] == "王五"
+
+
+# ---- delete：清理 + 幂等 ----
+
+@pytest.mark.asyncio
+async def test_delete_clears_business_state_and_is_idempotent():
+    drops = []
+
+    async def drop_checkpoint(session_id):
+        drops.append(session_id)
+
+    service, backend = make_service()
+    service.set_info("s1", "hphm1", "苏B12345")
+    await service.chat(_input("s1", "你好"))
+
+    await service.delete_session("s1", drop_checkpoint=drop_checkpoint)
+    assert drops == ["s1"]
+    # 再读为空、再聊天不注入旧数据
+    assert service.get_info("s1", "hphm1") == '""'
+    await service.chat(_input("s1", "还在吗"))
+    assert backend.calls[-1].injected_fields is None
+
+    # 重复删除固定行为：成功返回；checkpoint 清理会再次触发，
+    # 但缺失时静默容忍（幂等语义由 drop_checkpoint 内部保证）
+    await service.delete_session("s1", drop_checkpoint=drop_checkpoint)
+    assert drops == ["s1", "s1"]
+    assert backend.llm_calls == 2   # 两次 chat 各一次；set/get/delete 零调用
+
+
+@pytest.mark.asyncio
+async def test_delete_tolerates_missing_checkpoint():
+    async def missing(_session_id):
+        raise RuntimeError("no such thread")
+
+    service, _ = make_service()
+    service.set_info("s1", "xm1", "王五")
+    # checkpoint 不存在视为删除成功的一部分，不向上抛
+    await service.delete_session("s1", drop_checkpoint=missing)
+    assert service.get_info("s1", "xm1") == '""'
+
+
+def _input(session_id, text):
+    from src.backends.chat import ChatInput
+
+    return ChatInput(session_id=session_id, text=text, client_mode="direct")
diff --git a/test/services/test_fields_registry.py b/test/services/test_fields_registry.py
new file mode 100644
index 0000000..01e0d10
--- /dev/null
+++ b/test/services/test_fields_registry.py
@@ -0,0 +1,89 @@
+"""作业 6：字段注册表测试（分组完整性、类型/敏感级别、外部表示）。"""
+
+import pytest
+
+from src.services.fields_registry import (
+    FIELD_REGISTRY,
+    GROUP_KEYS,
+    INTERNAL_FIELD_KEYS,
+    FieldGroup,
+    SensitiveLevel,
+    parse_field_value,
+    to_external_string,
+)
+
+
+def test_registry_covers_all_three_groups():
+    """注册表完整覆盖 acd/human1/human2 三个分组（作业 6-3）。"""
+    assert set(GROUP_KEYS) == set(FieldGroup)
+    assert len(FIELD_REGISTRY) == 24
+    assert len(GROUP_KEYS[FieldGroup.ACD]) == 8
+    assert len(GROUP_KEYS[FieldGroup.HUMAN1]) == 8
+    assert len(GROUP_KEYS[FieldGroup.HUMAN2]) == 8
+
+
+def test_person_fields_include_full_set():
+    """人员字段必须包含 sfzmwh/sjwh/csbw，不能只迁姓名和号码。"""
+    for index in (1, 2):
+        group_keys = set(GROUP_KEYS[FieldGroup.HUMAN1 if index == 1 else FieldGroup.HUMAN2])
+        for suffix in ("xm", "hpzl", "hphm", "sfzmhm", "sfzmwh", "sjhm", "sjwh", "csbw"):
+            assert f"{suffix}{index}" in group_keys
+
+
+def test_registry_metadata_recorded():
+    """每个字段都记录类型、可写性、敏感级别（作业 6-4）。"""
+    for spec in FIELD_REGISTRY.values():
+        assert spec.value_type in (str, bool, int)
+        assert spec.writable is True
+        assert isinstance(spec.sensitive, SensitiveLevel)
+        assert spec.description
+
+
+def test_sensitive_levels_assigned():
+    """证件号/手机号 sensitive，姓名 personal，其余 public。"""
+    assert FIELD_REGISTRY["sfzmhm1"].sensitive is SensitiveLevel.SENSITIVE
+    assert FIELD_REGISTRY["sjhm2"].sensitive is SensitiveLevel.SENSITIVE
+    assert FIELD_REGISTRY["xm1"].sensitive is SensitiveLevel.PERSONAL
+    assert FIELD_REGISTRY["hphm1"].sensitive is SensitiveLevel.PUBLIC
+
+
+@pytest.mark.parametrize("raw,expected", [
+    ("1", True), ("true", True), ("是", True),
+    ("0", False), ("false", False), ("否", False),
+])
+def test_parse_bool_explicit(raw, expected):
+    assert parse_field_value(FIELD_REGISTRY["ywrysw"], raw) is expected
+
+
+@pytest.mark.parametrize("bad", ["", "maybe", "2", "有吧"])
+def test_parse_bool_rejects_ambiguous(bad):
+    """模糊值必须拒绝——保留 空/False/0/未知 的区分能力。"""
+    with pytest.raises(ValueError):
+        parse_field_value(FIELD_REGISTRY["ywrysw"], bad)
+
+
+def test_parse_int_and_str():
+    assert parse_field_value(FIELD_REGISTRY["jdcsl"], "2") == 2
+    assert parse_field_value(FIELD_REGISTRY["jdcsl"], "0") == 0
+    with pytest.raises(ValueError):
+        parse_field_value(FIELD_REGISTRY["jdcsl"], "两辆")
+    assert parse_field_value(FIELD_REGISTRY["hphm1"], "苏B12345") == "苏B12345"
+
+
+def test_external_string_representation():
+    """对外序列化：True->1，False->0，None->空串；内部仍区分 False 与 0。"""
+    assert to_external_string(True) == "1"
+    assert to_external_string(False) == "0"
+    assert to_external_string(0) == "0"
+    assert to_external_string(None) == ""
+    assert to_external_string("苏B12345") == "苏B12345"
+    # 内部存储不混淆：False 是 bool、0 是 int
+    assert parse_field_value(FIELD_REGISTRY["ywrysw"], "0") is False
+    assert parse_field_value(FIELD_REGISTRY["jdcsl"], "0") == 0
+    assert isinstance(parse_field_value(FIELD_REGISTRY["jdcsl"], "0"), int)
+
+
+def test_internal_field_protection_listed():
+    """图内部键登记在保护清单中，set_info 白名单不会收录它们。"""
+    for key in INTERNAL_FIELD_KEYS:
+        assert key not in FIELD_REGISTRY