fix(backends): make streamed replies and session state consistent

- Validate stage prefixes before forwarding model tokens. Invalid prefixes
  may use the existing repair budget; once output starts, validation failure
  aborts the turn without appending a regenerated reply. Bound the token
  queue and close inner streams when consumption stops.
- Replace delete-time lock eviction with reference-counted session locks
  covering holders and waiters. Serialize set/get with chat/stream/delete
  and release cancelled waiters without replacing a live lock.
- Store person synonyms under workflow-native driver keys and expose
  consistent alias views to single/group reads and graph prompt injection.
- Separate committed field patches from public formUpdate visibility.
  Commit callbacks keep repository fields synchronized even when updates
  are hidden or the consumer closes after graph completion.
- Disable OpenAI SDK automatic retries. Own bounded retry attempts in the
  application, use 1/2/4 second backoff, retry streams only before text is
  emitted, enforce attempt timeouts, and close upstream/client resources.
- Translate FastGPT SDK max_retries (total attempts) from the configured
  extra-retry count, and verify the real SDK loop uses four attempts.
- Document the per-LLM budget and streaming failure semantics without
  claiming the whole multi-call HTTP turn has the same timeout.

Validation: 267 offline tests passed, including actual SDK requests through
MockTransport, stream repair/abort, timeout/close, lock races, aliases and
formUpdate visibility; existing four dependency deprecation warnings.

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"

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

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

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)