diff --git a/.env.example b/.env.example
index 3539ca8..96dc4e1 100644
--- a/.env.example
+++ b/.env.example
@@ -17,8 +17,9 @@ VOICE_CONFIG=config/voice.json
 # 每组 LLM_<名>_BASE_URL / _API_KEY / _MODEL 必填；
 # 可选 _TIMEOUT_SECONDS / _MAX_RETRIES / _TEMPERATURE / _EXTRA_BODY（JSON 字符串，
 # 如思考开关 KIMI 网关为 {"reasoning_effort":"none"}）。
-# 超时/重试缺省 60 秒 / 3 次，与 FastGPT 客户端 SLA 对齐
-# （契约测试 test/backends/test_sla_alignment.py）。
+# 超时/重试缺省：每次尝试 60 秒，最多额外重试 3 次，退避 1、2、4 秒。
+# SDK 自动重试关闭；流式发出首个文本 token 后不重试。
+# 单次 LLM 调用预算最多 247 秒（另加响应关闭清理）；不是整轮聊天总超时。
 # AGENT_HISTORY_MESSAGES=50：注入模型的对话历史上限（最近 N 条，
 # 双向计数）；调大支持长对话；0 = 不截断（对齐 FastGPT 全量历史）。
 # AGENT_HISTORY_MESSAGES=0
diff --git a/docs/assignment6-compat.md b/docs/assignment6-compat.md
index bda81d7..65c2e7c 100644
--- a/docs/assignment6-compat.md
+++ b/docs/assignment6-compat.md
@@ -71,22 +71,42 @@
   这修正了旧实现"workflow 真实人员数据无法分组读到"的问题；
 - `driver{i}License` 非 state 键，对其 set_info 按未知键拒绝
   （workflow 不消费它）；
-- 同义键（driver1ID vs sfzmhm1）暂不合并：合并需调用方确认
-  实际使用侧后作为后续优化，已列入审计项。
+- 同义键统一存储：`sfzmhm{i}` 与 `driver{i}ID` 共用后者，
+  `sjhm{i}` 与 `driver{i}Phonenumber` 共用后者。单键和分组读取
+  继续提供两套键名；图输入同时注入一致的别名视图，避免旧 driver 值
+  遮盖 API 键的新值。同一批差量包含相互冲突的别名值时拒绝提交。
 
 ## 待审计项（作业 6-7 要求的后续工作）
 
 - [ ] 审计现有调用方是否存在对 `set_info` 任意 key 写入的依赖
       （如动态自定义键）。存在则在注册表补登记或提供逃生口。
-      **对账新增**：workflow 20261022 state 人员键
-      （`driver1ID`/`driver1License`/`driver1Phonenumber` ×2）
-      与 API 分组键不同名——确认线上调用方实际使用哪套键名，
-      必要时注册表补别名。
+      workflow 人员键 `driver{i}ID` / `driver{i}Phonenumber` 及对应
+      API 别名均已登记并统一存储；`driver{i}License` 只是车牌的
+      输出名称，外部写车牌继续使用 `hphm{i}`。
 - [x] ~~流式 `chat` 的 `formUpdate` 差量写回 repository~~
-      （完整版已实现：FormUpdate 事件捕获后经映射写回，见 `app_service.stream`）。
+      （图提交时即同步业务字段；FormUpdate 是否返回不影响写回）。
 - [ ] `get_info` 返回值中的敏感字段（证件号/手机号）回显边界：
       注册表已记录 `SensitiveLevel`，访问控制随作业 9 落地。
 
+## 流式、并发与字段一致性约定
+
+- 阶段前缀先缓冲并按当前节点的转换规则校验，非法前缀不推送；
+  未输出时仍可使用一次修复预算。已开始输出后，全文格式或时间依据
+  校验失败会以错误终止，不在同一流中拼接修复答案，不提交本轮业务字段。
+  已经到达客户端的正文不可撤回；客户端只在收到 done 后将一轮视为成功。
+- 会话锁引用包括持有者与等待者，只有最后一个使用者退出才回收。
+  chat、stream、set、get、delete 全部按同一会话串行；不同会话独立。
+  取消流时先关闭内部生成器、取消图任务，再释放会话锁。
+- needFormUpdate 只决定对外是否返回差量。图提交时通过内部回调同步
+  repository，关闭差量显示或消费者在提交后断开不会丢失业务字段。
+- SDK 自动重试关闭，由应用层统一控制；默认首次尝试加 3 次重试，
+  每次尝试总超时 60 秒，退避 1、2、4 秒。单次 LLM 调用理论预算
+  4×60+1+2+4=247 秒（另有响应关闭的清理时间）。FastGPT SDK 的
+  max_retries 实际代表总尝试数，适配层传入 4 以表达额外重试 3 次；
+  其 timeout 是 HTTP SDK 超时，不将它等同于整个多节点请求总期限。
+  流式首个文本 token 前可重试，发出文本后失败则终止，不重放。
+  该预算并非整个多次 LLM 调用的聊天回合总超时。
+
 ## 验证
 
 - `test/services/test_fields_registry.py`：注册表完整性、类型/敏感级别、
diff --git a/src/agent/collect.py b/src/agent/collect.py
index 64890b5..572f282 100644
--- a/src/agent/collect.py
+++ b/src/agent/collect.py
@@ -100,11 +100,20 @@ async def validated_reply(
                 + f"\n当前阶段为 {current}。请重新检查该阶段允许的转换"
                 "和当前 clientMode，不得跳步或切换到另一模式的出口。"
             )
-        raw = await llm.complete(prompt, user_message, history=history)
+        streaming_complete = getattr(llm, "complete_validated", None)
+        if streaming_complete is None:
+            raw = await llm.complete(prompt, user_message, history=history)
+        else:
+            raw = await streaming_complete(
+                prompt, user_message, history=history,
+                stage_validator=lambda target: validator(current, target),
+            )
         parsed = _try_parse(raw)
         if (parsed is not None and validator(current, parsed[0])
                 and (reply_validator is None or reply_validator(parsed[0], parsed[1]))):
             return raw, parsed[0]
+        if getattr(llm, "response_started", False):
+            raise CollectReplyError("流式回复已开始后校验失败，已终止本轮，未提交业务状态")
     raise CollectReplyError("模型回复格式、阶段转换或时间依据不合规（一次修复后仍失败）")
 
 
diff --git a/src/agent/llm.py b/src/agent/llm.py
index 376aff9..8b2c2a7 100644
--- a/src/agent/llm.py
+++ b/src/agent/llm.py
@@ -87,176 +87,153 @@ class FakeLLMService:
 
 
 class OpenAILLMService:
-    """真实 LLM provider（OpenAI 兼容接口，如 OpenAI / DeepSeek 等）。
+    """单层重试：首次 + max_retries；每次尝试超时覆盖整个响应读取。
 
-    参考实现：async 客户端 + asyncio.wait_for 超时（取消会向外传播，
-    客户端断连时图任务随之取消）；仅对可重试错误（429 / 5xx / 连接错）
-    按 LLM_MAX_RETRIES 指数退避重试；错误统一转换为
-    LLMTimeoutError / LLMAPIError，不向上层泄漏原始响应体。
+    SDK 自动重试关闭。流式仅在首个文本 token 前重试，输出后失败终止。
+    默认 60 秒、3 次重试、1/2/4 秒退避：单次 LLM 调用预算上界 247 秒
+    （不含响应关闭的清理时间），不代表多次 LLM 调用的整轮 HTTP 超时。
     """
 
-    def __init__(
-        self,
-        model: str,
-        api_key: str,
-        base_url: str = "",
-        timeout_seconds: float = 60.0,
-        max_retries: int = 3,
-        temperature: float | None = None,
-        extra_body: dict | None = None,
-    ):
+    RETRY_DELAY_SECONDS = 1.0
+
+    def __init__(self, model: str, api_key: str, base_url: str = "",
+                 timeout_seconds: float = 60.0, max_retries: int = 3,
+                 temperature: float | None = None, extra_body: dict | None = None):
+        if timeout_seconds <= 0 or max_retries < 0:
+            raise ValueError("timeout_seconds must be positive; max_retries must be nonnegative")
         self._model = model
         self._api_key = api_key
         self._base_url = base_url
         self._timeout_seconds = timeout_seconds
         self._max_retries = max_retries
         self._temperature = temperature
-        # 附加请求体（如思考开关），原样合并进 chat.completions.create 参数。
         self._extra_body = dict(extra_body or {})
+        if {"model", "messages", "stream"} & self._extra_body.keys():
+            raise ValueError("extra_body cannot override model, messages or stream")
         self._client = None
 
     def _get_client(self):
-        # 延迟构造：fake provider 使用时无需安装/初始化真实客户端。
-        # AsyncOpenAI 的创建不绑定事件循环，可在首次调用前安全缓存。
-        # trust_env=False：不走系统代理，本地网关直连（代理曾导致 503/500）。
         if self._client is None:
             import httpx
             from openai import AsyncOpenAI
-
             from ..core.config import Config
 
-            kwargs = {"api_key": self._api_key, "timeout": self._timeout_seconds}
+            options = {
+                "api_key": self._api_key, "timeout": self._timeout_seconds,
+                "max_retries": 0,
+                "http_client": httpx.AsyncClient(trust_env=Config.LLM_HTTP_TRUST_ENV),
+            }
             if self._base_url:
-                kwargs["base_url"] = self._base_url
-            kwargs["http_client"] = httpx.AsyncClient(trust_env=Config.LLM_HTTP_TRUST_ENV)
-            self._client = AsyncOpenAI(**kwargs)
+                options["base_url"] = self._base_url
+            self._client = AsyncOpenAI(**options)
         return self._client
 
-    async def complete(
-        self, system_prompt: str, user_message: str, *,
-        history: list[dict[str, str]] | None = None,
-    ) -> str:
-        import asyncio
+    async def aclose(self):
+        if self._client is not None:
+            await self._client.close()
+            self._client = None
 
-        from openai import (
-            APIConnectionError,
-            APITimeoutError,
-            AuthenticationError,
-            RateLimitError,
+    def _request_args(self, system_prompt, user_message, history, *, stream=False):
+        options = {
+            "model": self._model,
+            "messages": [
+                {"role": "system", "content": system_prompt},
+                *[dict(message) for message in (history or [])],
+                {"role": "user", "content": user_message},
+            ],
+        }
+        if stream:
+            options["stream"] = True
+        if self._temperature is not None:
+            options["temperature"] = self._temperature
+        if self._extra_body:
+            options["extra_body"] = dict(self._extra_body)
+        return options
+
+    @staticmethod
+    def _retryable(exc):
+        from openai import APIConnectionError, APIStatusError
+
+        return (
+            isinstance(exc, (TimeoutError, APIConnectionError))
+            or isinstance(exc, APIStatusError)
+            and (exc.status_code in (408, 409, 429) or exc.status_code >= 500)
         )
 
+    def _error(self, exc):
+        from openai import APITimeoutError, AuthenticationError, RateLimitError
+
+        if isinstance(exc, AuthenticationError):
+            return LLMAuthError("LLM provider 认证失败")
+        if isinstance(exc, RateLimitError):
+            return LLMRateLimitError("LLM provider 限流（重试预算耗尽或已开始输出）")
+        if isinstance(exc, (TimeoutError, APITimeoutError)):
+            return LLMTimeoutError(f"LLM 调用超时（每次尝试 {self._timeout_seconds}s）")
+        if isinstance(exc, LLMError):
+            return exc
+        return LLMAPIError(f"LLM 调用失败：{type(exc).__name__}")
+
+    async def _retry_delay(self, attempt):
+        import asyncio
+        await asyncio.sleep(self.RETRY_DELAY_SECONDS * (2 ** attempt))
+
+    async def complete(self, system_prompt: str, user_message: str, *,
+                       history: list[dict[str, str]] | None = None) -> str:
+        import asyncio
+
         client = self._get_client()
-        messages = [
-            {"role": "system", "content": system_prompt},
-            *[dict(message) for message in (history or [])],
-            {"role": "user", "content": user_message},
-        ]
-        kwargs: dict = {"model": self._model, "messages": messages}
-        if self._temperature is not None:
-            kwargs["temperature"] = self._temperature
-        kwargs.update(self._extra_body)
-        retries = 0
-        while True:
+        options = self._request_args(system_prompt, user_message, history)
+        for attempt in range(self._max_retries + 1):
             try:
-                response = await asyncio.wait_for(
-                    client.chat.completions.create(**kwargs),
-                    timeout=self._timeout_seconds,
-                )
-            except (APITimeoutError, asyncio.TimeoutError) as exc:
-                # 超时：可重试，重试耗尽后转换为 LLMTimeoutError
-                if retries < self._max_retries:
-                    retries += 1
-                    await asyncio.sleep(min(2 ** retries, 10))
-                    continue
-                raise LLMTimeoutError(
-                    f"LLM 调用超时（{self._timeout_seconds}s，"
-                    f"重试 {self._max_retries} 次后仍失败）"
-                ) from exc
-            except (RateLimitError, APIConnectionError) as exc:
-                # 限流 / 连接错误：可重试
-                if retries < self._max_retries:
-                    retries += 1
-                    await asyncio.sleep(min(2 ** retries, 10))
-                    continue
-                if isinstance(exc, RateLimitError):
-                    # 限流耗尽：区分上报，API 层映射为 429（两后端错误码等价）
-                    raise LLMRateLimitError(
-                        f"LLM provider 限流（重试 {self._max_retries} 次后仍失败）"
-                    ) from exc
-                raise LLMAPIError(f"LLM provider 不可用：{type(exc).__name__}") from exc
-            except AuthenticationError as exc:
-                # 401 鉴权失败：不可重试，区分上报，API 层映射为 401
-                raise LLMAuthError("LLM provider 认证失败") from exc
+                async with asyncio.timeout(self._timeout_seconds):
+                    response = await client.chat.completions.create(**options)
+                content = response.choices[0].message.content if response.choices else None
+                if not content:
+                    raise LLMAPIError("LLM 返回了空内容")
+                return content
+            except LLMError:
+                raise
             except Exception as exc:
-                # 其他错误（如 4xx 参数错）不可重试
-                raise LLMAPIError(f"LLM 调用失败：{type(exc).__name__}") from exc
+                if attempt < self._max_retries and self._retryable(exc):
+                    await self._retry_delay(attempt)
+                    continue
+                raise self._error(exc) from exc
 
-            content = response.choices[0].message.content if response.choices else None
-            if not content:
-                raise LLMAPIError("LLM 返回了空内容")
-            return content
-
-    async def stream_tokens(
-        self, system_prompt: str, user_message: str, *,
-        history: list[dict[str, str]] | None = None,
-    ) -> AsyncIterator[str]:
-        """逐 token 产出对话文本（作业 8 真实增量的模型侧）。
-
-        语义约束（与 complete 对齐的部分）：
-        - 同样的消息构造（system + role-preserving history + user）、
-          temperature 与 extra_body 透传；
-        - 同样的错误分类：401 -> LLMAuthError、429 -> LLMRateLimitError、
-          超时 -> LLMTimeoutError、其余 -> LLMAPIError。
-
-        有意差异：
-        - **不做重试**——首个 token 发出后重试没有意义（已流出的内容
-          无法收回）；首包前的连接/鉴权错误照样按上述分类抛出，
-          调用方（透传包装器）决定如何呈现；
-        - 超时用 asyncio.timeout 包整个迭代（末 token 迟到同样算超时）。
-        """
+    async def stream_tokens(self, system_prompt: str, user_message: str, *,
+                            history: list[dict[str, str]] | None = None) -> AsyncIterator[str]:
         import asyncio
-
-        from openai import (
-            APIConnectionError,
-            AuthenticationError,
-            RateLimitError,
-        )
+        from contextlib import suppress
 
         client = self._get_client()
-        messages = [
-            {"role": "system", "content": system_prompt},
-            *[dict(message) for message in (history or [])],
-            {"role": "user", "content": user_message},
-        ]
-        kwargs: dict = {"model": self._model, "messages": messages, "stream": True}
-        if self._temperature is not None:
-            kwargs["temperature"] = self._temperature
-        kwargs.update(self._extra_body)
-        try:
-            async with asyncio.timeout(self._timeout_seconds):
-                response = await client.chat.completions.create(**kwargs)
-                async for event in response:
-                    choices = getattr(event, "choices", None)
-                    if not choices:
-                        continue
-                    delta = getattr(choices[0], "delta", None)
-                    token = getattr(delta, "content", None)
-                    if token:
-                        yield token
-        except AuthenticationError as exc:
-            raise LLMAuthError("LLM provider 认证失败") from exc
-        except RateLimitError as exc:
-            raise LLMRateLimitError("LLM provider 限流") from exc
-        except TimeoutError as exc:
-            raise LLMTimeoutError(
-                f"LLM 流式调用超时（{self._timeout_seconds}s）"
-            ) from exc
-        except APIConnectionError as exc:
-            raise LLMAPIError(f"LLM provider 连接失败:{type(exc).__name__}") from exc
-        except LLMError:
-            raise
-        except Exception as exc:
-            raise LLMAPIError(f"LLM 流式调用失败:{type(exc).__name__}") from exc
+        options = self._request_args(system_prompt, user_message, history, stream=True)
+        emitted = False
+        for attempt in range(self._max_retries + 1):
+            response = None
+            try:
+                async with asyncio.timeout(self._timeout_seconds):
+                    response = await client.chat.completions.create(**options)
+                    async for event in response:
+                        choices = getattr(event, "choices", None)
+                        if not choices:
+                            continue
+                        token = getattr(getattr(choices[0], "delta", None), "content", None)
+                        if token:
+                            emitted = True
+                            yield token
+                    if not emitted:
+                        raise LLMAPIError("LLM 返回了空内容")
+                return
+            except LLMError:
+                raise
+            except Exception as exc:
+                if emitted or attempt == self._max_retries or not self._retryable(exc):
+                    raise self._error(exc) from exc
+            finally:
+                close = getattr(response, "close", None)
+                if close is not None:
+                    with suppress(Exception):
+                        await asyncio.wait_for(close(), timeout=5)
+            await self._retry_delay(attempt)
 
 
 def _default_fake_responder(system_prompt: str, user_message: str) -> str:
diff --git a/src/api/endpoints.py b/src/api/endpoints.py
index 13907b4..2e4e129 100644
--- a/src/api/endpoints.py
+++ b/src/api/endpoints.py
@@ -556,7 +556,7 @@ async def set_info(
     json_data = request.model_dump()
 
     if Config.AGENT_BACKEND == "langgraph":
-        code, msg = get_app_service().set_info(
+        code, msg = await get_app_service().set_info(
             json_data['sessionId'], json_data['key'], json_data['value']
         )
         return ProcessResponse_set(
@@ -668,7 +668,7 @@ async def get_info(
         return ProcessResponse_get(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
-            value=get_app_service().get_info(json_data['sessionId'], json_data['key']),
+            value=await get_app_service().get_info(json_data['sessionId'], json_data['key']),
             code="200",
             msg="",
         )
diff --git a/src/backends/chat.py b/src/backends/chat.py
index 4c43c06..3a3f916 100644
--- a/src/backends/chat.py
+++ b/src/backends/chat.py
@@ -1,6 +1,6 @@
 """Backend-neutral chat contract."""
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Callable
 from dataclasses import dataclass, field
 from typing import Any, Protocol
 
@@ -17,6 +17,8 @@ class ChatInput:
     # 人员进 person_fields）。LangGraph 后端合并进图输入；
     # FastGPT 后端忽略该字段。
     injected_fields: dict | None = None
+    # 图提交后、公开结果之前同步业务字段；不进入图状态或 HTTP schema。
+    on_commit: Callable[[dict], None] | None = field(default=None, repr=False, compare=False)
 
 
 @dataclass(frozen=True)
@@ -47,6 +49,8 @@ class ChatResult:
     content: str
     status_code: str | None = None
     form_update: Any = field(default_factory=dict)
+    # 内部提交差量，独立于 needFormUpdate 对外显隐。
+    committed_patch: dict | None = field(default=None, repr=False, compare=False)
 
 
 class ChatBackend(Protocol):
diff --git a/src/backends/langgraph.py b/src/backends/langgraph.py
index 447e932..2d8c785 100644
--- a/src/backends/langgraph.py
+++ b/src/backends/langgraph.py
@@ -15,9 +15,10 @@ test/backends/test_langgraph_streaming.py
 """
 
 import asyncio
-from collections.abc import AsyncIterator, Callable
+from contextlib import suppress
+from collections.abc import AsyncIterator, Awaitable, Callable
 
-from ..agent.collect import _EXTRACTION_MARKER
+from ..agent.collect import STATE_PREFIX_RE, parse_state_reply
 from ..agent.graph import GraphDeps, build_graph, workflow_fake_llm
 from ..agent.llm import LLMAuthError, LLMRateLimitError, LLMService
 from .chat import (
@@ -36,42 +37,58 @@ _DONE = object()
 
 
 class _TokenPassThroughLLM:
-    """LLMService 包装器：对话调用逐 token 旁路给 sink，提取调用静默。
+    """先验证阶段前缀；已经输出的回复不能重生成后拼接。"""
 
-    图节点仍按批处理接口 ``complete()`` 使用 LLM（校验/解析依赖完整
-    文本，语义不变）；本包装器在收集 token 的同时推给 SSE 桥：
-    - 系统提示含【字段提取】标记的辅助调用走 inner.complete，
-      不推流（提取 JSON 不应出现在客户端流里）；
-    - 其余调用若 inner 提供 stream_tokens 则逐 token 推送；
-      否则退化为整段推送（旧行为，fake/测试 LLM 兼容）。
-    """
-
-    def __init__(self, inner: LLMService, sink: Callable[[str], None]):
+    def __init__(self, inner: LLMService, sink: Callable[[str], Awaitable[None]]):
         self._inner = inner
         self._sink = sink
+        self.response_started = False
 
-    async def complete(
-        self, system_prompt: str, user_message: str, *,
-        history: list[dict[str, str]] | None = None,
+    async def complete(self, system_prompt, user_message, *, history=None):
+        # 提取等未经过 validated_reply 的辅助调用始终静默。
+        return await self._inner.complete(system_prompt, user_message, history=history)
+
+    async def complete_validated(
+        self, system_prompt, user_message, *, history=None, stage_validator,
     ) -> str:
-        if _EXTRACTION_MARKER in system_prompt:
-            return await self._inner.complete(
-                system_prompt, user_message, history=history
-            )
         stream_tokens = getattr(self._inner, "stream_tokens", None)
         if stream_tokens is None:
-            text = await self._inner.complete(
-                system_prompt, user_message, history=history
-            )
-            self._sink(text)
-            return text
+            raw = await self.complete(system_prompt, user_message, history=history)
+            try:
+                code, _ = parse_state_reply(raw)
+            except ValueError:
+                return raw
+            if stage_validator(code):
+                self.response_started = True
+                await self._sink(raw)
+            return raw
+
         chunks: list[str] = []
-        async for token in stream_tokens(
-            system_prompt, user_message, history=history
-        ):
-            chunks.append(token)
-            self._sink(token)
-        return "".join(chunks)
+        prefix = ""
+        accepted = False
+        source = stream_tokens(system_prompt, user_message, history=history)
+        try:
+            async for token in source:
+                chunks.append(token)
+                if not accepted:
+                    prefix += token
+                    match = STATE_PREFIX_RE.match(prefix)
+                    if match is None:
+                        if len(prefix) > 1024 or "</state>" in prefix:
+                            return "".join(chunks)
+                        continue
+                    if not stage_validator(match.group(1)):
+                        return "".join(chunks)
+                    accepted = True
+                    self.response_started = True
+                    await self._sink(prefix)
+                else:
+                    await self._sink(token)
+            return "".join(chunks)
+        finally:
+            close = getattr(source, "aclose", None)
+            if close is not None:
+                await close()
 
 
 class LangGraphBackend(ChatBackend):
@@ -99,10 +116,11 @@ class LangGraphBackend(ChatBackend):
           （作业 6 起实现）。
         """
         result = await self._complete_turn(chat_input)
+        patch = result.get("turn_patch") or {}
         return ChatResult(
-          content = result["response_text"],
-          status_code = None,
-          form_update = result.get("turn_patch") or {}
+            content=result["response_text"],
+            form_update=patch if chat_input.need_form_update else {},
+            committed_patch=patch,
         )
 
     async def _complete_turn(
@@ -127,11 +145,15 @@ class LangGraphBackend(ChatBackend):
         if chat_input.injected_fields:
             graph_input.update(chat_input.injected_fields)
         try:
-            return await self._graph.ainvoke(
+            result = await self._graph.ainvoke(
                 graph_input,
                 config={"configurable": {"thread_id": chat_input.session_id}},
                 context=GraphDeps(llm=llm or self._llm),
             )
+            # 即使消费者在图完成后断开，业务 repository 仍与 checkpoint 一致。
+            if chat_input.on_commit is not None:
+                chat_input.on_commit(result.get("turn_patch") or {})
+            return result
         except LLMAuthError as exc:
             raise ChatBackendAuthenticationError(str(exc)) from exc
         except LLMRateLimitError as exc:
@@ -148,23 +170,25 @@ class LangGraphBackend(ChatBackend):
         - 生成器被关闭（客户端断开）时取消图任务，停止 LLM 调用；
         - 图任务异常在消费端重新抛出（错误映射沿用 _complete_turn）。
         """
-        queue: asyncio.Queue = asyncio.Queue()
-        token_count = {"n": 0}   # sink 同步回调不可直接闭包递增，用盒
+        queue: asyncio.Queue = asyncio.Queue(maxsize=32)
+        token_count = 0
 
-        def sink(token: str) -> None:
-            # LLM 在图任务的事件循环线程内同步回调
-            token_count["n"] += 1
-            queue.put_nowait(token)
+        async def sink(token: str) -> None:
+            nonlocal token_count
+            token_count += 1
+            await queue.put(token)
 
         async def run_turn() -> dict:
             try:
                 result = await self._complete_turn(
                     chat_input, llm=_TokenPassThroughLLM(self._llm, sink)
                 )
-            except BaseException as exc:
-                queue.put_nowait(exc)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                await queue.put(exc)
                 return {}
-            queue.put_nowait(_DONE)
+            await queue.put(_DONE)
             return result
 
         task = asyncio.ensure_future(run_turn())
@@ -177,19 +201,17 @@ class LangGraphBackend(ChatBackend):
                     raise item
                 yield TextDelta(item)
             result = await task
-            if token_count["n"] == 0:
+            if token_count == 0:
                 # 本回合无 LLM 调用（welcome 等固定应答轮）：
                 # sink 无 token 可推，回退为完整文本切片（旧行为）
                 text = result.get("response_text") or ""
                 if text:
                     yield TextDelta(text)
             form_update = result.get("turn_patch") or {}
-            if form_update:
+            if form_update and chat_input.need_form_update:
                 yield FormUpdate(form_update)
         finally:
             if not task.done():
                 task.cancel()
-                try:
-                    await task
-                except (asyncio.CancelledError, Exception):
-                    pass
+            with suppress(asyncio.CancelledError):
+                await task
diff --git a/src/core/fastgpt_client.py b/src/core/fastgpt_client.py
index 9e2f7c7..4fd9e92 100644
--- a/src/core/fastgpt_client.py
+++ b/src/core/fastgpt_client.py
@@ -16,11 +16,11 @@ _chat_backend: ChatBackend | None = None
 _langgraph_graph = None
 _app_service: AccidentAppService | None = None
 
-# 后端超时/重试 SLA（两后端对齐；契约测试防止再漂移）。
-# FastGPT 侧：客户端每次请求总超时 + 重试次数；
-# LangGraph 侧：LLM 每次调用总超时 + 重试次数（退避为指数式，
-# 与 FastGPT 固定 1s 不同——总尝试次数与单次超时可比，见
-# test/backends/test_sla_alignment.py）。
+# 两后端缺省 60 秒、最多额外重试 3 次，1/2/4 秒退避。
+# FastGPT SDK 的 max_retries 实为总尝试数，注入时转换为 1 + 重试次数。
+# LLM 关闭 SDK 自动重试；流式发出文本后不再重试。
+# LLM 预算按单次模型调用计算，FastGPT timeout 为 HTTP SDK 超时，
+# 均不承诺多次模型调用的整个聊天回合具有相同总期限。
 FASTGPT_TIMEOUT_SECONDS = 60.0
 FASTGPT_MAX_RETRIES = 3
 
@@ -30,6 +30,7 @@ async def lifespan(app: FastAPI):
     """Manage chat backend resource lifecycle."""
     global _chat_backend, _fastgpt_client
     Config.validate()
+    llm = None
 
     if Config.AGENT_BACKEND == "langgraph":
         # 作业 2：LangGraph fake 模式不要求 FastGPT 密钥。
@@ -38,7 +39,8 @@ async def lifespan(app: FastAPI):
 
         global _langgraph_graph, _app_service
         _langgraph_graph = build_graph()
-        _chat_backend = LangGraphBackend(_langgraph_graph, llm=build_llm_service())
+        llm = build_llm_service()
+        _chat_backend = LangGraphBackend(_langgraph_graph, llm=llm)
         # 作业 6：应用服务 + 内存 repository（作业 7 换 PostgreSQL）。
         _app_service = AccidentAppService(
             _chat_backend, InMemorySessionRepository()
@@ -48,7 +50,7 @@ async def lifespan(app: FastAPI):
             api_key=Config.FASTGPT_API_KEY,
             base_url=Config.FASTGPT_BASE_URL,
             timeout=FASTGPT_TIMEOUT_SECONDS,
-            max_retries=FASTGPT_MAX_RETRIES,
+            max_retries=FASTGPT_MAX_RETRIES + 1,
             retry_delay=1.0,
             enable_logging=Config.DEBUG,
         )
@@ -61,6 +63,8 @@ async def lifespan(app: FastAPI):
         _chat_backend = None
         _langgraph_graph = None
         _app_service = None
+        if llm is not None and hasattr(llm, "aclose"):
+            await llm.aclose()
         if _fastgpt_client:
             await _fastgpt_client.__aexit__(None, None, None)
         _fastgpt_client = None
diff --git a/src/services/app_service.py b/src/services/app_service.py
index 7ccfd33..478ad95 100644
--- a/src/services/app_service.py
+++ b/src/services/app_service.py
@@ -5,11 +5,9 @@
   （对照旧 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 原子操作，不存在混合中间态。
+- chat / stream / set_info / get_info / delete_session 都在同会话锁内。
+  锁的引用计数包含等待者，取消和异常都会归还引用；没有使用者时回收。
+  不同会话独立执行，同会话 set_info 不会被在途聊天的旧结果覆盖。
 - 未知 key 兼容策略：get 对未知 key 返回 JSON 空串 '""'（现状兼容）；
   set 对未知 key 拒绝写入（白名单保护，差异见兼容清单）；
 - includeInputInfo 继续被 endpoint 接受，但本层不再使用
@@ -18,7 +16,8 @@
 endpoint 只负责校验、协议（SSE / 响应模型）与依赖注入。
 """
 
-import asyncio
+from contextlib import aclosing
+from dataclasses import replace
 
 from ..backends.chat import ChatBackend, ChatInput, ChatResult, FormUpdate
 from .fields_registry import (
@@ -27,10 +26,13 @@ from .fields_registry import (
     BUSINESS_TO_FORM_KEY,
     FieldGroup,
     business_patch_from_form,
+    business_value,
+    canonical_key,
     parse_field_value,
     to_external_string,
 )
 from .repository import SessionRepository
+from .session_locks import SessionLocks
 
 # set_info 对未知/受保护 key 的拒绝码（与现有“入参不规范”文案对齐）
 SET_INFO_INVALID_KEY_MSG = "入参不规范：未知或受保护的字段 key"
@@ -42,17 +44,7 @@ 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
+        self._locks = SessionLocks()
 
     # ---- chat：图每轮读取最新业务状态，产出差量写回 ----
 
@@ -67,9 +59,10 @@ class AccidentAppService:
         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:
+        # 注入同义键的一致视图，覆盖 checkpoint 里的旧值。
+        for key, spec in FIELD_REGISTRY.items():
+            value = business_value(business, key)
+            if value is None:
                 continue
             if spec.group is FieldGroup.ACD:
                 form_key = BUSINESS_TO_FORM_KEY.get(key)
@@ -87,49 +80,64 @@ class AccidentAppService:
 
     async def chat(self, chat_input: ChatInput) -> ChatResult:
         """非流式一轮对话：注入最新业务字段 -> 执行图 -> 写回字段差量。"""
-        async with self._lock_for(chat_input.session_id):
+        async with self._locks.hold(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)
+        committed = False
+
+        def commit(patch):
+            nonlocal committed
+            self._writeback_patch(chat_input.session_id, patch)
+            committed = True
+
         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,
-            )
+            replace(chat_input, injected_fields=injected or None, on_commit=commit)
+        )
+        if not committed:
+            patch = result.committed_patch
+            if patch is None:
+                patch = result.form_update
+            if isinstance(patch, dict):
+                commit(patch)
+        return replace(
+            result, form_update=result.form_update if chat_input.need_form_update else {},
         )
-        # 图产出的字段差量写回 repository，
-        # 使 set/get_info 与后续轮次看到同一事实来源
-        if isinstance(result.form_update, dict):
-            self._writeback_patch(chat_input.session_id, result.form_update)
-        return result
 
     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 with self._locks.hold(chat_input.session_id):
+            async with aclosing(self._stream_locked(chat_input)) as events:
+                async for event in events:
+                    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,
-            text=chat_input.text,
-            need_form_update=chat_input.need_form_update,
-            client_mode=chat_input.client_mode,
-            injected_fields=injected or None,
-        )
-        async for event in self._backend.stream(result_input):
-            if isinstance(event, FormUpdate) and isinstance(event.data, dict):
-                self._writeback_patch(chat_input.session_id, event.data)
-            yield event
+        committed = False
+
+        def commit(patch):
+            nonlocal committed
+            self._writeback_patch(chat_input.session_id, patch)
+            committed = True
+
+        result_input = replace(chat_input, injected_fields=injected or None, on_commit=commit)
+        async with aclosing(self._backend.stream(result_input)) as events:
+            async for event in events:
+                if isinstance(event, FormUpdate):
+                    if not committed and isinstance(event.data, dict):
+                        commit(event.data)
+                    if not chat_input.need_form_update:
+                        continue
+                yield event
 
     # ---- set_info / get_info：零 LLM、零辅助对话 ----
 
-    def set_info(self, session_id: str, key: str, value: str) -> tuple[str, str]:
+    async def set_info(self, session_id: str, key: str, value: str) -> tuple[str, str]:
+        async with self._locks.hold(session_id):
+            return self._set_info(session_id, key, value)
+
+    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:
@@ -138,10 +146,14 @@ class AccidentAppService:
             parsed = parse_field_value(spec, value)
         except ValueError:
             return "500", "入参不规范：字段值无法解析"
-        self._repository.apply_update(session_id, {key: parsed})
+        self._repository.apply_update(session_id, {canonical_key(key): parsed})
         return "200", ""
 
-    def get_info(self, session_id: str, key: str) -> str:
+    async def get_info(self, session_id: str, key: str) -> str:
+        async with self._locks.hold(session_id):
+            return self._get_info(session_id, key)
+
+    def _get_info(self, session_id: str, key: str) -> str:
         """按分组或单键读出，返回 JSON 字符串（兼容现有 /get_info 契约）。"""
         import json
 
@@ -150,7 +162,7 @@ class AccidentAppService:
             return json.dumps(
                 {
                     group.value: {
-                        k: to_external_string(business.get(k))
+                        k: to_external_string(business_value(business, k))
                         for k in GROUP_KEYS[group]
                     }
                     for group in FieldGroup
@@ -163,13 +175,13 @@ class AccidentAppService:
             group = None
         if group is not None:
             return json.dumps(
-                {k: to_external_string(business.get(k)) for k in GROUP_KEYS[group]},
+                {k: to_external_string(business_value(business, 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)
+        return json.dumps(to_external_string(business_value(business, key)), ensure_ascii=False)
 
     # ---- delete_session：业务状态 + 图 checkpoint 一起清理 ----
 
@@ -178,10 +190,11 @@ class AccidentAppService:
 
         drop_checkpoint：由 core 层注入的 checkpoint 清理函数
         （图生命周期归 lifespan 管理，本层不持有图引用）。
-        与会话锁互斥：删除不与在途 chat 交错。删除后清理锁项。
+        与会话锁互斥：删除不与在途 chat/set/get 交错。
+        释放锁引用时保留等待者使用的锁，最后一个使用者退出后回收。
         repository 删除幂等；checkpoint 缺失时静默（幂等语义一致）。
         """
-        async with self._lock_for(session_id):
+        async with self._locks.hold(session_id):
             self._repository.delete(session_id)
             if drop_checkpoint is not None:
                 try:
@@ -189,8 +202,6 @@ class AccidentAppService:
                 except Exception:
                     # checkpoint 已不存在视为删除成功的一部分（幂等）
                     pass
-            # 清理锁项：引用随等待者自然释放，新会话重建新锁
-            self._locks.pop(session_id, None)
 
 
 __all__ = ["AccidentAppService", "SET_INFO_INVALID_KEY_MSG"]
diff --git a/src/services/fields_registry.py b/src/services/fields_registry.py
index 121929c..b3209fa 100644
--- a/src/services/fields_registry.py
+++ b/src/services/fields_registry.py
@@ -125,6 +125,22 @@ BUSINESS_TO_FORM_KEY: dict[str, str] = {
 }
 
 
+# 同义人员键只存 workflow 原生键；API 键保留为读写别名。
+FIELD_ALIASES = {
+    f"{external}{index}": f"driver{index}{native}"
+    for index in (1, 2)
+    for external, native in (("sfzmhm", "ID"), ("sjhm", "Phonenumber"))
+}
+
+
+def canonical_key(key: str) -> str:
+    return FIELD_ALIASES.get(key, key)
+
+
+def business_value(fields: dict, key: str):
+    return fields.get(canonical_key(key))
+
+
 def business_patch_from_form(form_patch: dict) -> dict:
     """图产出的表单差量 -> repository 业务键差量（白名单内）。
 
@@ -137,8 +153,10 @@ def business_patch_from_form(form_patch: dict) -> dict:
         spec = FIELD_REGISTRY.get(business_key)
         if spec is None or not spec.writable:
             continue
-        # 类型以注册表为准：图表单值已是内部类型，直接采用
-        patch[business_key] = value
+        canonical = canonical_key(business_key)
+        if canonical in patch and patch[canonical] != value:
+            raise ValueError("同一字段的别名包含冲突值")
+        patch[canonical] = value
     return patch
 
 
diff --git a/src/services/session_locks.py b/src/services/session_locks.py
new file mode 100644
index 0000000..5920e42
--- /dev/null
+++ b/src/services/session_locks.py
@@ -0,0 +1,28 @@
+"""单事件循环内的会话互斥：引用计数包含持有者和等待者。"""
+
+import asyncio
+from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
+
+
+@dataclass
+class _Entry:
+    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
+    users: int = 0
+
+
+class SessionLocks:
+    def __init__(self):
+        self._entries: dict[str, _Entry] = {}
+
+    @asynccontextmanager
+    async def hold(self, session_id: str):
+        entry = self._entries.setdefault(session_id, _Entry())
+        entry.users += 1
+        try:
+            async with entry.lock:
+                yield
+        finally:
+            entry.users -= 1
+            if entry.users == 0:
+                del self._entries[session_id]
diff --git a/test/agent/test_llm_stream_tokens.py b/test/agent/test_llm_stream_tokens.py
index d672d95..a94c3fc 100644
--- a/test/agent/test_llm_stream_tokens.py
+++ b/test/agent/test_llm_stream_tokens.py
@@ -64,7 +64,7 @@ async def test_stream_tokens_yields_incrementally_and_forwards_kwargs():
     assert kwargs["model"] == "test-model"
     assert kwargs["stream"] is True
     assert kwargs["temperature"] == 0
-    assert kwargs["reasoning_effort"] == "none"    # extra_body 透传
+    assert kwargs["extra_body"]["reasoning_effort"] == "none"    # extra_body 透传
     assert kwargs["messages"][0] == {"role": "system", "content": "系统"}
     assert kwargs["messages"][-1] == {"role": "user", "content": "输入"}
 
diff --git a/test/backends/test_langgraph_streaming.py b/test/backends/test_langgraph_streaming.py
index e03dbd1..ae8131e 100644
--- a/test/backends/test_langgraph_streaming.py
+++ b/test/backends/test_langgraph_streaming.py
@@ -58,7 +58,7 @@ class StreamingFakeLLM:
 
 
 def make_input(session="stream-s1", text="继续办理"):
-    return ChatInput(session_id=session, text=text, client_mode="direct")
+    return ChatInput(session_id=session, text=text, client_mode="direct", need_form_update=True)
 
 
 async def _second_turn_events(backend, session="stream-s1"):
diff --git a/test/backends/test_sla_alignment.py b/test/backends/test_sla_alignment.py
index 686b4f8..9e96ca6 100644
--- a/test/backends/test_sla_alignment.py
+++ b/test/backends/test_sla_alignment.py
@@ -1,11 +1,8 @@
-"""后端超时/重试 SLA 对齐契约测试。
+"""超时/重试配置与 SDK 适配契约。
 
-背景（可替代性清单）：FastGPT 客户端为 60s/3 次，LangGraph LLM 侧
-曾是 30s/2 次，客户端感知到的超时行为不等价。本测试把两端常量
-钉在一起，防止再次漂移。
-
-注：退避策略有意不同——FastGPT 固定 1s，LLM 指数退避（限流场景
-更友好）；对齐的是"单次调用总超时"与"最大重试次数"两个 SLA 参数。
+LLM max_retries 表示额外重试次数；FastGPT SDK 同名参数实际表示
+总尝试次数，需要 +1 转换。退避统一为 1、2、4 秒。
+超时分别约束一次 LLM 尝试与 FastGPT HTTP SDK，不代表整轮总期限。
 """
 
 import inspect
@@ -40,3 +37,40 @@ def test_named_profile_defaults_match_fastgpt_client(monkeypatch):
     profile = Config.llm_profile("probe")
     assert profile["timeout_seconds"] == FASTGPT_TIMEOUT_SECONDS
     assert profile["max_retries"] == FASTGPT_MAX_RETRIES
+
+
+import asyncio
+from unittest.mock import AsyncMock
+import httpx
+import pytest
+from fastapi import FastAPI
+from fastgpt_client.exceptions import APIError
+from src.core import fastgpt_client as wiring
+from src.core.config import Config
+
+
+@pytest.mark.asyncio
+async def test_fastgpt_sdk_uses_same_attempt_count_and_backoff(monkeypatch):
+    monkeypatch.setattr(Config, "validate", lambda: None)
+    monkeypatch.setattr(Config, "AGENT_BACKEND", "fastgpt")
+    monkeypatch.setattr(Config, "FASTGPT_BASE_URL", "http://audit")
+    monkeypatch.setattr(Config, "FASTGPT_API_KEY", "audit")
+    sleep = AsyncMock()
+    monkeypatch.setattr(asyncio, "sleep", sleep)
+    async with wiring.lifespan(FastAPI()):
+        client = wiring.get_fastgpt_client()
+        request = AsyncMock(return_value=httpx.Response(503))
+        with pytest.raises(APIError):
+            await client._retry_request_async(request, "audit")
+        assert request.await_count == 4
+        assert [call.args[0] for call in sleep.await_args_list] == [1, 2, 4]
+
+
+@pytest.mark.asyncio
+async def test_llm_backoff_uses_same_delays(monkeypatch):
+    sleep = AsyncMock()
+    monkeypatch.setattr(asyncio, "sleep", sleep)
+    llm = OpenAILLMService(model="audit", api_key="audit")
+    for attempt in range(3):
+        await llm._retry_delay(attempt)
+    assert [call.args[0] for call in sleep.await_args_list] == [1, 2, 4]
diff --git a/test/services/test_app_service.py b/test/services/test_app_service.py
index ae17de5..a58cbab 100644
--- a/test/services/test_app_service.py
+++ b/test/services/test_app_service.py
@@ -52,55 +52,60 @@ def make_service(form_update=None):
 
 # ---- set/get：白名单、分组、外部表示 ----
 
-def test_set_then_get_single_key_roundtrip():
+@pytest.mark.asyncio
+async def test_set_then_get_single_key_roundtrip():
     service, backend = make_service()
-    code, msg = service.set_info("s1", "hphm1", "苏B12345")
+    code, msg = await service.set_info("s1", "hphm1", "苏B12345")
     assert (code, msg) == ("200", "")
-    assert json.loads(service.get_info("s1", "hphm1")) == "苏B12345"
+    assert json.loads(await service.get_info("s1", "hphm1")) == "苏B12345"
     # set/get 零模型调用
     assert backend.llm_calls == 0
 
 
-def test_get_groups_match_existing_contract():
+@pytest.mark.asyncio
+async 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", "张三")
+    await service.set_info("s1", "ywrysw", "0")       # False
+    await service.set_info("s1", "jdcsl", "0")        # int 0
+    await service.set_info("s1", "xm2", "张三")
 
-    acd = json.loads(service.get_info("s1", "acdinfo"))
+    acd = json.loads(await 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"))
+    human2 = json.loads(await service.get_info("s1", "acdhuman2"))
     assert human2["xm2"] == "张三"
     assert human2["sfzmwh2"] == ""   # 全字段齐全，未填为空
 
-    all_groups = json.loads(service.get_info("s1", "all"))
+    all_groups = json.loads(await service.get_info("s1", "all"))
     assert set(all_groups) == {"acdinfo", "acdhuman1", "acdhuman2"}
 
 
-def test_get_unknown_key_returns_json_empty_string():
+@pytest.mark.asyncio
+async 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") == '""'
+    await service.set_info("s1", "hphm1", "苏B12345")
+    assert await service.get_info("s1", "no_such_key") == '""'
+    assert await service.get_info("ghost-session", "hphm1") == '""'
 
 
-def test_set_unknown_or_internal_key_rejected():
+@pytest.mark.asyncio
+async def test_set_unknown_or_internal_key_rejected():
     service, backend = make_service()
-    code, msg = service.set_info("s1", "no_such_key", "x")
+    code, msg = await 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")
+    code, _ = await service.set_info("s1", "stage_code", "1002")
     assert code == "500"          # 内部字段保护
-    assert service.get_info("s1", "hphm1") == '""'   # 未被污染
+    assert await service.get_info("s1", "hphm1") == '""'   # 未被污染
     assert backend.llm_calls == 0
 
 
-def test_set_rejects_unparsable_value():
+@pytest.mark.asyncio
+async def test_set_rejects_unparsable_value():
     service, _ = make_service()
-    code, msg = service.set_info("s1", "ywrysw", "maybe")
+    code, msg = await service.set_info("s1", "ywrysw", "maybe")
     assert code == "500" and "无法解析" in msg
 
 
@@ -110,8 +115,8 @@ def test_set_rejects_unparsable_value():
 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", "李四")
+    await service.set_info("s1", "hphm1", "苏B12345")
+    await service.set_info("s1", "xm2", "李四")
 
     result = await service.chat(
         _input("s1", "我是刚才报警的驾驶员")
@@ -124,8 +129,8 @@ async def test_chat_sees_set_info_values_and_writes_back_patch():
     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"
+    assert json.loads(await service.get_info("s1", "acdinfo"))["ywrysw"] == "0"
+    assert json.loads(await service.get_info("s1", "hphm1")) == "苏B12345"
 
 
 @pytest.mark.asyncio
@@ -135,21 +140,21 @@ async def test_chat_form_patch_mapped_to_business_keys():
         "has_injury": False, "vehicle_count": 2, "accident_cause": "追尾",
     })
     await service.chat(_input("s1", "两辆汽车追尾，没有人受伤"))
-    acd = json.loads(service.get_info("s1", "acdinfo"))
+    acd = json.loads(await service.get_info("s1", "acdinfo"))
     assert acd["ywrysw"] == "0"
     assert acd["jdcsl"] == "2"
     assert acd["sgyy"] == "追尾"
     # 表单键本身不落 repository（未知业务键 -> JSON 空串）
-    assert json.loads(service.get_info("s1", "has_injury")) == ""
+    assert json.loads(await service.get_info("s1", "has_injury")) == ""
 
 
 @pytest.mark.asyncio
 async def test_set_info_acd_value_mapped_into_form_injection():
     """set_info 写业务键 -> chat 注入为图表单键（反向映射）。"""
     service, backend = make_service()
-    service.set_info("s1", "ywrysw", "0")
-    service.set_info("s1", "jdcsl", "2")
-    service.set_info("s1", "bjrjs", "1")   # 无表单来源，不注入
+    await service.set_info("s1", "ywrysw", "0")
+    await service.set_info("s1", "jdcsl", "2")
+    await service.set_info("s1", "bjrjs", "1")   # 无表单来源，不注入
     await service.chat(_input("s1", "继续办理"))
     injected = backend.calls[0].injected_fields
     assert injected["accident_fields"] == {"has_injury": False, "vehicle_count": 2}
@@ -166,7 +171,7 @@ async def test_chat_without_business_state_passes_no_injection():
 async def test_stream_injects_fields_writes_back_form_update():
     """流式：注入同 chat；FormUpdate 事件差量捕获写回后透传。"""
     service, backend = make_service(form_update={"has_injury": False})
-    service.set_info("s1", "xm1", "王五")
+    await service.set_info("s1", "xm1", "王五")
     events = [
         event async for event in service.stream(_input("s1", "继续办理"))
     ]
@@ -174,7 +179,7 @@ async def test_stream_injects_fields_writes_back_form_update():
     assert kinds == ["TextDelta", "FormUpdate", "TextDelta"]
     assert backend.stream_calls[0].injected_fields["person_fields"]["xm1"] == "王五"
     # FormUpdate 差量已写回 repository（表单键 -> 业务键）
-    assert json.loads(service.get_info("s1", "acdinfo"))["ywrysw"] == "0"
+    assert json.loads(await service.get_info("s1", "acdinfo"))["ywrysw"] == "0"
 
 
 # ---- delete：清理 + 幂等 ----
@@ -187,13 +192,13 @@ async def test_delete_clears_business_state_and_is_idempotent():
         drops.append(session_id)
 
     service, backend = make_service()
-    service.set_info("s1", "hphm1", "苏B12345")
+    await 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") == '""'
+    assert await service.get_info("s1", "hphm1") == '""'
     await service.chat(_input("s1", "还在吗"))
     assert backend.calls[-1].injected_fields is None
 
@@ -210,16 +215,16 @@ async def test_delete_tolerates_missing_checkpoint():
         raise RuntimeError("no such thread")
 
     service, _ = make_service()
-    service.set_info("s1", "xm1", "王五")
+    await service.set_info("s1", "xm1", "王五")
     # checkpoint 不存在视为删除成功的一部分，不向上抛
     await service.delete_session("s1", drop_checkpoint=missing)
-    assert service.get_info("s1", "xm1") == '""'
+    assert await 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")
+    return ChatInput(session_id=session_id, text=text, client_mode="direct", need_form_update=True)
 
 
 # ---- 并发隔离（A2）：同会话串行、不同会话并行、delete 与 chat 互斥 ----
@@ -311,13 +316,13 @@ async def test_concurrent_delete_serialization():
 
 @pytest.mark.asyncio
 async def test_concurrent_chat_and_set_info_no_corruption():
-    """set_info（同步无锁）与并发 chat 交错：repository 键值保持完整。"""
+    """set_info 与 chat 按同一会话串行，最终保留写入的完整车牌。"""
     backend = SerialProbeBackend()
     service = AccidentAppService(backend, InMemorySessionRepository())
     results = await asyncio.gather(
         service.chat(_input("s1", "a")),
-        asyncio.to_thread(service.set_info, "s1", "hphm1", "苏B12345"),
+        service.set_info("s1", "hphm1", "苏B12345"),
     )
     assert results[1] == ("200", "")
-    # 无论时序如何，最终两个写者之一的车牌值完整存在（无撕裂）
-    assert json.loads(service.get_info("s1", "hphm1")) in ("苏B12345", "")
+    # set 写入在 chat 结束后执行，不能丢失。
+    assert json.loads(await service.get_info("s1", "hphm1")) == "苏B12345"
diff --git a/test/services/test_field_commit_and_retry.py b/test/services/test_field_commit_and_retry.py
new file mode 100644
index 0000000..9d9fdff
--- /dev/null
+++ b/test/services/test_field_commit_and_retry.py
@@ -0,0 +1,158 @@
+import asyncio
+import json
+from contextlib import aclosing
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+from openai import AsyncOpenAI
+
+from src.agent.llm import LLMAPIError, LLMRateLimitError, OpenAILLMService
+from src.agent.prompts import SINGLE_CONFIRM, render_prompt
+from src.backends.chat import ChatInput, FormUpdate, TextDelta
+from src.backends.langgraph import LangGraphBackend
+from src.services.app_service import AccidentAppService
+from src.services.repository import InMemorySessionRepository
+
+
+class PatchGraph:
+    async def ainvoke(self, *args, **kwargs):
+        return {'response_text': '<state>1002</state>ok', 'turn_patch': {'has_injury': False}}
+
+
+@pytest.mark.asyncio
+async def test_person_synonyms_share_storage_and_prompt_values():
+    repo = InMemorySessionRepository()
+    service = AccidentAppService(LangGraphBackend(graph=PatchGraph()), repo)
+    for index in (1, 2):
+        for api, native in ((f'sfzmhm{index}', f'driver{index}ID'), (f'sjhm{index}', f'driver{index}Phonenumber')):
+            await service.set_info('s', native, 'old')
+            await service.set_info('s', api, 'new')
+            assert json.loads(await service.get_info('s', api)) == 'new'
+            assert json.loads(await service.get_info('s', native)) == 'new'
+            assert api not in repo.load('s')
+    fields = service._injected_fields('s')['person_fields']
+    prompt = render_prompt(SINGLE_CONFIRM, {'person_fields': fields})
+    assert 'old' not in prompt
+    assert 'new' in prompt
+    await service.set_info('s', 'driver1ID', 'latest')
+    group = json.loads(await service.get_info('s', 'acdhuman1'))
+    assert group['driver1ID'] == group['sfzmhm1'] == 'latest'
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('stream', [False, True])
+@pytest.mark.parametrize('visible', [False, True])
+async def test_form_visibility_does_not_control_internal_commit(stream, visible):
+    backend = LangGraphBackend(graph=PatchGraph())
+    service = AccidentAppService(backend, InMemorySessionRepository())
+    request = ChatInput('s', 'hello', need_form_update=visible)
+    if stream:
+        events = [e async for e in service.stream(request)]
+        assert any(isinstance(e, FormUpdate) for e in events) is visible
+    else:
+        result = await service.chat(request)
+        assert bool(result.form_update) is visible
+    assert json.loads(await service.get_info('s', 'ywrysw')) == '0'
+
+
+@pytest.mark.asyncio
+async def test_commit_is_not_lost_when_consumer_closes_after_graph_completes():
+    backend = LangGraphBackend(graph=PatchGraph())
+    repo = InMemorySessionRepository()
+    service = AccidentAppService(backend, repo)
+    async with aclosing(service.stream(ChatInput('s', 'hello'))) as events:
+        assert isinstance(await anext(events), TextDelta)
+    assert json.loads(await service.get_info('s', 'ywrysw')) == '0'
+    assert not service._locks._entries
+
+
+@pytest.mark.asyncio
+async def test_factory_disables_sdk_retries():
+    llm = OpenAILLMService(model='m', api_key='audit')
+    try:
+        assert llm._get_client().max_retries == 0
+    finally:
+        await llm.aclose()
+
+
+def completion(stream):
+    if not stream:
+        return httpx.Response(200, json={
+            'id': 'r', 'object': 'chat.completion', 'created': 0, 'model': 'm',
+            'choices': [{'index': 0, 'message': {'role': 'assistant', 'content': 'ok'}, 'finish_reason': 'stop'}],
+        })
+    data = {'id': 'r', 'object': 'chat.completion.chunk', 'created': 0, 'model': 'm',
+            'choices': [{'index': 0, 'delta': {'content': 'ok'}, 'finish_reason': None}]}
+    return httpx.Response(200, headers={'content-type': 'text/event-stream'},
+                          content='data: '+json.dumps(data)+'\n\ndata: [DONE]\n\n')
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('stream', [False, True])
+@pytest.mark.parametrize('status', [429, 503])
+async def test_actual_http_attempt_count_is_not_multiplied_by_sdk(stream, status):
+    calls = []
+    def handler(request):
+        calls.append(json.loads(request.content))
+        if len(calls) <= 3:
+            return httpx.Response(status, json={'error': {'message': 'retry'}})
+        return completion(stream)
+    llm = OpenAILLMService(model='m', api_key='audit', max_retries=3,
+                          extra_body={'reasoning_effort': 'none'})
+    llm._client = AsyncOpenAI(api_key='audit', base_url='http://audit/v1', max_retries=0,
+                             http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
+    llm._retry_delay = AsyncMock()
+    try:
+        result = ''.join([t async for t in llm.stream_tokens('system', 'user')]) if stream else await llm.complete('system', 'user')
+        assert result == 'ok'
+        assert len(calls) == 4
+        assert llm._retry_delay.await_count == 3
+        assert all(c['reasoning_effort'] == 'none' for c in calls)
+    finally:
+        await llm.aclose()
+
+
+@pytest.mark.asyncio
+async def test_stream_failure_after_first_token_closes_response_without_retry():
+    class Response:
+        def __init__(self): self.closed = False
+        def __aiter__(self): return self.generate()
+        async def generate(self):
+            yield SimpleNamespace(choices=[SimpleNamespace(delta=SimpleNamespace(content='partial'))])
+            raise OSError('broken socket')
+        async def close(self): self.closed = True
+    response = Response()
+    create = AsyncMock(return_value=response)
+    llm = OpenAILLMService(model='m', api_key='audit')
+    llm._client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=create)))
+    tokens = []
+    with pytest.raises(LLMAPIError):
+        async for token in llm.stream_tokens('system', 'user'):
+            tokens.append(token)
+    assert tokens == ['partial']
+    assert create.await_count == 1
+    assert response.closed
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('stream', [False, True])
+async def test_retry_budget_exhaustion_is_exact(stream):
+    calls = []
+    def handler(request):
+        calls.append(request)
+        return httpx.Response(429, json={'error': {'message': 'retry'}})
+    llm = OpenAILLMService(model='m', api_key='audit', max_retries=2)
+    llm._retry_delay = AsyncMock()
+    llm._client = AsyncOpenAI(api_key='audit', base_url='http://audit/v1', max_retries=0,
+                             http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
+    try:
+        with pytest.raises(LLMRateLimitError):
+            if stream:
+                async for _ in llm.stream_tokens('system', 'user'): pass
+            else:
+                await llm.complete('system', 'user')
+        assert len(calls) == 3
+    finally:
+        await llm.aclose()
diff --git a/test/services/test_stream_and_session_safety.py b/test/services/test_stream_and_session_safety.py
new file mode 100644
index 0000000..871f578
--- /dev/null
+++ b/test/services/test_stream_and_session_safety.py
@@ -0,0 +1,151 @@
+import asyncio
+
+import pytest
+
+from src.agent.collect import CollectReplyError, validated_reply
+from src.agent.llm import FakeLLMService
+from src.backends.chat import ChatInput, ChatResult, TextDelta
+from src.backends.langgraph import LangGraphBackend
+from src.services.app_service import AccidentAppService
+from src.services.repository import InMemorySessionRepository
+
+
+class ValidatingGraph:
+    def __init__(self):
+        self.committed = None
+
+    async def ainvoke(self, data, *, config, context):
+        raw, stage = await validated_reply(context.llm, 'system', 'user', '1002')
+        self.committed = stage
+        return {'response_text': raw, 'turn_patch': {}}
+
+
+class ScriptedStream:
+    def __init__(self, replies):
+        self.replies = iter(replies)
+        self.calls = 0
+
+    async def complete(self, *args, **kwargs):
+        return '{}'
+
+    async def stream_tokens(self, *args, **kwargs):
+        self.calls += 1
+        for chunk in next(self.replies):
+            yield chunk
+
+
+@pytest.mark.asyncio
+async def test_invalid_prefix_is_not_sent_and_repair_is_the_only_reply():
+    graph = ValidatingGraph()
+    llm = ScriptedStream([
+        ['<state>9999</state>invalid'],
+        ['<sta', 'te>1002</state>', 'repaired'],
+    ])
+    backend = LangGraphBackend(graph=graph, llm=llm)
+    events = [e async for e in backend.stream(ChatInput('s', 'hello'))]
+    assert ''.join(e.text for e in events if isinstance(e, TextDelta)) == '<state>1002</state>repaired'
+    assert graph.committed == '1002'
+    assert llm.calls == 2
+
+
+@pytest.mark.asyncio
+async def test_failure_after_output_aborts_without_appending_a_repair():
+    graph = ValidatingGraph()
+    llm = ScriptedStream([
+        ['<state>1002</state>', 'partial', '<state>9999</state>'],
+        ['<state>1002</state>must not appear'],
+    ])
+    backend = LangGraphBackend(graph=graph, llm=llm)
+    events = []
+    with pytest.raises(CollectReplyError):
+        async for event in backend.stream(ChatInput('s', 'hello')):
+            events.append(event)
+    assert llm.calls == 1
+    assert graph.committed is None
+    assert 'must not appear' not in ''.join(e.text for e in events)
+
+
+@pytest.mark.asyncio
+async def test_repair_exhaustion_emits_no_invalid_text():
+    llm = ScriptedStream([['<state>9999</state>x'], ['malformed']])
+    backend = LangGraphBackend(graph=ValidatingGraph(), llm=llm)
+    events = []
+    with pytest.raises(CollectReplyError):
+        async for event in backend.stream(ChatInput('s', 'hello')):
+            events.append(event)
+    assert events == []
+
+
+class BlockingBackend:
+    def __init__(self):
+        self.entered = asyncio.Queue()
+        self.release = asyncio.Event()
+        self.active = 0
+        self.peak = 0
+
+    async def complete(self, request):
+        self.active += 1
+        self.peak = max(self.peak, self.active)
+        await self.entered.put(request.text)
+        try:
+            await self.release.wait()
+            return ChatResult('<state>1002</state>ok', form_update={'vehicle_count': 1})
+        finally:
+            self.active -= 1
+
+
+@pytest.mark.asyncio
+async def test_delete_waiter_and_new_chat_keep_one_lock():
+    backend = BlockingBackend()
+    service = AccidentAppService(backend, InMemorySessionRepository())
+    dropping, drop_release = asyncio.Event(), asyncio.Event()
+
+    async def drop(_):
+        dropping.set()
+        await drop_release.wait()
+
+    deletion = asyncio.create_task(service.delete_session('s', drop))
+    await dropping.wait()
+    first = asyncio.create_task(service.chat(ChatInput('s', 'a')))
+    await asyncio.sleep(0)
+    drop_release.set()
+    await deletion
+    assert await asyncio.wait_for(backend.entered.get(), 1) == 'a'
+    second = asyncio.create_task(service.chat(ChatInput('s', 'b')))
+    await asyncio.sleep(0)
+    assert backend.entered.empty()
+    backend.release.set()
+    await asyncio.wait_for(asyncio.gather(first, second), 2)
+    assert backend.peak == 1
+    assert not service._locks._entries
+
+
+@pytest.mark.asyncio
+async def test_set_waits_for_chat_and_cannot_be_overwritten_by_old_result():
+    backend = BlockingBackend()
+    service = AccidentAppService(backend, InMemorySessionRepository())
+    chat = asyncio.create_task(service.chat(ChatInput('s', 'a')))
+    await backend.entered.get()
+    setting = asyncio.create_task(service.set_info('s', 'jdcsl', '2'))
+    await asyncio.sleep(0)
+    assert not setting.done()
+    backend.release.set()
+    await asyncio.gather(chat, setting)
+    assert await service.get_info('s', 'jdcsl') == '"2"'
+
+
+@pytest.mark.asyncio
+async def test_cancelled_waiter_does_not_leak_or_replace_lock():
+    backend = BlockingBackend()
+    service = AccidentAppService(backend, InMemorySessionRepository())
+    chat = asyncio.create_task(service.chat(ChatInput('s', 'a')))
+    await backend.entered.get()
+    waiting = asyncio.create_task(service.get_info('s', 'jdcsl'))
+    await asyncio.sleep(0)
+    waiting.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await waiting
+    assert len(service._locks._entries) == 1
+    backend.release.set()
+    await chat
+    assert not service._locks._entries
diff --git a/test/services/test_stream_retry_cleanup.py b/test/services/test_stream_retry_cleanup.py
new file mode 100644
index 0000000..1f4cd22
--- /dev/null
+++ b/test/services/test_stream_retry_cleanup.py
@@ -0,0 +1,94 @@
+import asyncio
+from contextlib import aclosing
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from src.agent.collect import validated_reply
+from src.agent.llm import LLMTimeoutError, OpenAILLMService
+from src.backends.chat import ChatInput, TextDelta
+from src.backends.langgraph import LangGraphBackend
+from src.services.app_service import AccidentAppService
+from src.services.repository import InMemorySessionRepository
+
+
+class Graph:
+    async def ainvoke(self, data, *, config, context):
+        raw, _ = await validated_reply(context.llm, 'system', 'user', '1002')
+        return {'response_text': raw, 'turn_patch': {}}
+
+
+@pytest.mark.asyncio
+async def test_valid_prefix_reaches_client_before_model_finishes():
+    release = asyncio.Event()
+    finished = asyncio.Event()
+    class LLM:
+        async def stream_tokens(self, *args, **kwargs):
+            yield '<sta'
+            yield 'te>1002</state>'
+            await release.wait()
+            yield 'tail'
+            finished.set()
+    backend = LangGraphBackend(graph=Graph(), llm=LLM())
+    async with aclosing(backend.stream(ChatInput('s', 'hello'))) as events:
+        first = await asyncio.wait_for(anext(events), 1)
+        assert first == TextDelta('<state>1002</state>')
+        assert not finished.is_set()
+        release.set()
+        remaining = [e async for e in events]
+    assert ''.join(e.text for e in remaining) == 'tail'
+    assert finished.is_set()
+
+
+@pytest.mark.asyncio
+async def test_closing_application_stream_cancels_model_before_unlocking():
+    cancelled = asyncio.Event()
+    class LLM:
+        async def stream_tokens(self, *args, **kwargs):
+            yield '<state>1002</state>'
+            try:
+                await asyncio.Future()
+            finally:
+                cancelled.set()
+    service = AccidentAppService(LangGraphBackend(graph=Graph(), llm=LLM()), InMemorySessionRepository())
+    async with aclosing(service.stream(ChatInput('s', 'hello'))) as events:
+        await asyncio.wait_for(anext(events), 1)
+    assert cancelled.is_set()
+    assert not service._locks._entries
+    assert await service.set_info('s', 'hphm1', 'test') == ('200', '')
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('stream', [False, True])
+async def test_timeout_covers_whole_attempt_and_retries_are_bounded(stream):
+    responses = []
+    class HangingResponse:
+        closed = False
+        def __aiter__(self): return self.generate()
+        async def generate(self):
+            await asyncio.Future()
+            yield None
+        async def close(self): self.closed = True
+    async def create(**kwargs):
+        if stream:
+            response = HangingResponse()
+            responses.append(response)
+            return response
+        await asyncio.Future()
+    llm = OpenAILLMService(model='m', api_key='audit', timeout_seconds=0.01, max_retries=2)
+    call = AsyncMock(side_effect=create)
+    llm._client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=call)))
+    llm._retry_delay = AsyncMock()
+    # 先加载 SDK 错误类；外层看门狗不测模块惰性导入耗时。
+    assert llm._retryable(TimeoutError())
+    assert isinstance(llm._error(TimeoutError()), LLMTimeoutError)
+    async with asyncio.timeout(10):
+        with pytest.raises(LLMTimeoutError):
+            if stream:
+                async for _ in llm.stream_tokens('system', 'user'): pass
+            else:
+                await llm.complete('system', 'user')
+    assert call.await_count == 3
+    assert llm._retry_delay.await_count == 2
+    assert all(response.closed for response in responses)