Recover expired proxy tokens using existing DMS authentication

CHANGELOG.md

@@ -13,9 +13,11 @@
 - 2026-10-27：Redis核验当前检查点后，可复用本请求刚写成功的大正文，省去重复回传；待处理写入、最终状态恢复、旧数据格式和TTL规则保持。
 - 2026-10-27：原文分析使用单个分块执行池，预计尾部负载至少降低5%时才长块优先，按原资料和块序合并；并发上限与失败/重试规则保持。
 - 2026-10-27：复用JSON、精确UTF-8字节量、分块计量和同来源固定提示；编号引用只在单次绑定中懒建索引，保留旧引用兼容及错误拒绝规则。
+- 2026-10-27：企查查/DeepSeek代理明确返回Token失效业务码212时，复用现有DMS认证配置获取短Token，按代理地址在进程内共享、并发合并刷新、最多重试一次，失败后30秒冷却；不改写`.env`。流式正文已开始后不重放，正常JSON不重复解析，错误仅报告安全固定码。配置和边界见[代理认证恢复](docs/proxy-auth.md)。
 
 ### 验证范围
 
 - master合并后的已跟踪合成回归847项通过：知识准备42、入库26、问答395、API115、后台128、部署脚本125、公司分类16。分类HTTP测试替身显式声明可选资源，避免Mock伪造进程接纳对象；恢复本地独立过程提示后API123项亦通过。
+- 代理恢复完成后的本地工作区回归893项通过：42准备、26入库、395问答、123 API、128后台、163脚本、16分类，含22项新增认证恢复测试。该计数包含仍独立保留的过程提示测试及本地Harness测试；它们不随本次代理恢复提交发布。
 - 用户提供的单次运行日志显示：最终Redis读取2153→19毫秒、完整状态加载2160→24毫秒，仍解码完整正文。该观测不构成固定输入A/B或稳定性能保证。
 - 综合输入新表示的真实模型质量、分析调度及本地CPU优化的实际延迟收益仍待验证；本次集成不调用真实业务服务、不部署、不推送远程。

docs/README.md

@@ -8,6 +8,7 @@
 | 人才客服三类知识、原文关联、独立前缀及离线导入 | [人才客服知识](talent-knowledge.md) |
 | 指名文件标题召回、向量候选合并与12条预算 | [标题召回](title-recall.md) |
 | DMS五栏目ID、名称及字段映射调整 | [DMS配置](dms-schema.md) |
+| 企查查/DeepSeek代理212失效后自动刷新、内存复用和有限重试 | [代理认证恢复](proxy-auth.md) |
 | 内存接纳拒绝、前后台共享名额、Linux硬内存隔离 | [内存保护](memory-safety.md) |
 | 最终会话回读慢、Redis取回量与选中解码量、反序列化拆分 | [Redis读取诊断](redis-read-timing.md) |
 | 综合引文及字段无损复用、独立输出预留和减少中间整理 | [综合预算](synthesis-budget.md) |

docs/deployment.md

@@ -1,5 +1,7 @@
 # 部署操作指南
 
+1.0.1（2026-10-27）新增企查查/DeepSeek代理Token自动恢复：沿用现有URL中的Token，明确收到代理业务码212后复用现有DMS登录取得短Token，仅在内存中替换并重试一次；不改写`.env`，不新增必填变量。需要已有可用的DMS账号密码和认证地址。失败冷却、流式边界及本地验证范围见[代理认证恢复](proxy-auth.md)。本次尚未打包部署，待用户核对完整优化清单后再交付。
+
 F160（2026-10-22）交付包为 `output/talent-update-20261022.zip`，包含当前运行代码所需更新、人才137条完整知识和操作手册；精简步骤见 `output/人才客服部署操作手册-20261022.txt`，包内同为 `操作手册.txt`。依赖文件与既有离线项目包一致，不需要重新安装依赖，沿用服务器配置和模型。
 
 `sh deploy-offline.sh --env-file-only check-milvus` 现在按已发布清单检查营商库，并在人才准备目录存在时额外检查人才三库。输出 `agents.business`、`agents.talent` 各自的数据库名、索引记录及内容分段计数、差异和快照；新增总计为 `expected_total_all_agents` / `indexed_total_all_agents`。旧顶层 `libraries` / `expected_total` / `indexed_total` 仍表示营商结果。人才部分缺失或不匹配时整体失败，不隐藏营商结果；未导入人才且没有该目录的旧安装仍只检查营商库。各库计数以本机清单为准，不能用一组库的成功代替另一组库验证。

docs/proxy-auth.md

@@ -0,0 +1,25 @@
+# 企查查与DeepSeek代理Token自动恢复
+
+版本1.0.1，优化时间2026-10-27（Asia/Shanghai）。正式企查查三接口、DeepSeek普通JSON/流式调用及复用它们的部署检查均支持此流程。
+
+## 处理流程
+
+1. 先使用现有URL中的`proxyToken`。有效时直接调用，不登录DMS，也不要求新增配置。
+2. 仅在代理返回外层JSON业务码`212`时确认Token失效；支持HTTP成功响应及401/403响应中的同一业务码。权限码201、企查查Status 115、无明确212的HTTP错误、网络错误和生成中断均不触发刷新。
+3. 复用项目现有DMS认证配置和登录实现，获取短期Token。配置为主机加`AUTH_PORT`和`DMS_PORT`时，使用认证端口的`/oauth/api/user/login`；显式`/proxy_dms`及`DMS_LOGIN_URL`沿用现有规则，不猜测新的登录地址。登录表单仍为项目已使用的`userName/password/clientId/serviceId`，不执行栏目、模型或业务数据操作。
+4. 新Token仅保存在进程内，同一配置路径、同一代理协议/主机/端口的企查查和模型请求可复用。并发失效请求共用一次刷新；不同代理地址不共享缓存Token。只替换URL中的Token值，其余query原始编码/顺序、QCC签名和时间戳、业务参数、模型payload及Bearer头保持。
+5. 当前请求最多重放一次。登录失败或新Token仍被拒绝时，30秒内不再重复登录；后续请求可重新尝试。Token没有已验证的过期时间契约，因此不猜测TTL，后续明确失效再刷新。重启进程会清空缓存，恢复从配置Token开始。
+
+## 配置及故障边界
+
+无需改写现有`.env`，也没有新的必填环境变量。刷新复用已有`DMS_BASE_URL`、`AUTH_PORT`、`DMS_PORT`、`DMS_USERNAME`、`DMS_PASSWORD`及可选`DMS_LOGIN_URL`。实际生效值仍按现有配置加载优先级确定；仅有静态`DMS_TOKEN`而没有可登录账号/密码时，不能自动生成新Token。现有Token有效时，缺少DMS登录信息不会影响正常代理请求。
+
+刷新失败报告`proxy_token_refresh_failed`，刷新后再次被拒绝报告`proxy_token_invalid`；模型包装层可能带`model_`前缀。日志和部署检查仅保留固定错误码，不输出Token、账号密码、完整URL或响应正文。DMS登录拒绝重定向并沿用有限超时，失败不会伪造成空政策或正常回答。
+
+DeepSeek流式请求只在输出模型内容之前识别代理拒绝并重试。正常SSE仍按原字节增量读取；一旦已收到正文，后续断流、格式错误或取消不会触发认证重放，避免重复输出或拼接两次回答。普通模型JSON严格解析结果在本次传输内复用，避免刷新检测增加第二次解析。
+
+## 已验证范围
+
+本地回环HTTP和合成凭据覆盖有效初始Token、明确失效、短Token再次失效、两类服务共享、并发单次登录、失败冷却、权限/供应商错误、401/403、重定向拒绝、参数/签名保持、分片JSON、流式首段及时送达、断流和取消不重放。测试逐次核对临时`.env`字节不变。
+
+本轮未调用线上代理或DMS，未改实际`.env`，没有验证生产账号权限、短Token对各代理服务的授权或实际有效期。生产联调、打包和部署须在用户核对全部优化清单后另行进行。配置连接契约见[DMS配置](dms-schema.md)，已有检查命令见[部署操作指南](deployment.md)。

scripts/deploy_check.py

@@ -206,7 +206,8 @@ def safe_code(exc):
                'dms_login_failed', 'dms_direct_api_requires_token', 'dms_column_mismatch',
                'dms_parent_mismatch', 'dms_schema_mismatch', 'dms_write_scope_denied',
                'dms_operation_denied', 'invalid_deepseek_base_url', 'missing_deepseek_configuration',
-               'request_timeout', 'response_incomplete', 'invalid_response_json'}
+               'request_timeout', 'response_incomplete', 'invalid_response_json',
+               'proxy_token_invalid', 'proxy_token_refresh_failed'}
     return code if code in allowed or (code.startswith('http_') and code[5:].isdigit()) else 'request_failed'
 
 

scripts/tests/test_proxy_auth.py

@@ -0,0 +1,404 @@
+"""Proxy recovery against loopback HTTP only; all credentials are synthetic."""
+
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
+import hashlib
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import tempfile
+import threading
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+from urllib.parse import parse_qs, urlsplit
+
+from src.proxy_auth import ProxyAuthError, ProxyTokens, proxy_identity, replace_token, token_rejected
+from src.step1_knowledge_prepare import summarize
+from src.step3_question_answer import qcc
+from src.step3_question_answer.streaming import stream_transport
+from scripts import deploy_check
+
+
+class TokenStateTests(unittest.TestCase):
+    url = 'http://example.invalid:10011/proxy/?servertype=deepseek&proxyToken=synthetic-old'
+
+    def client(self, login=None):
+        client = SimpleNamespace(config=SimpleNamespace(username='synthetic-user', password='synthetic-password'),
+                                 login_url='http://example.invalid/login', token='synthetic-short')
+        client._login = login or (lambda: None)
+        return client
+
+    def test_only_definite_outer_rejections(self):
+        for value in ({'code': 212}, {'code': '212'}):
+            self.assertTrue(token_rejected(value))
+        for value in ({'code': 201}, {'Status': '115'}, {'code': 212, 'choices': []},
+                      {'code': 212, 'Status': '200'}, {'message': 'token invalid'}, [], None):
+            self.assertFalse(token_rejected(value))
+
+    def test_diagnostics_expose_only_fixed_auth_codes(self):
+        from src.step4_api.logging import safe_code
+        for code in ('proxy_token_invalid', 'proxy_token_refresh_failed'):
+            self.assertEqual(deploy_check.safe_code(RuntimeError(code)), code)
+            self.assertEqual(safe_code(code), code)
+            self.assertEqual(safe_code('model_' + code), 'model_' + code)
+        self.assertEqual(deploy_check.safe_code(RuntimeError('synthetic-token-in-url')), 'request_failed')
+
+    def test_replace_preserves_signed_query_and_encodes_token(self):
+        url = self.url + '&raw=%2f+%20&x=1&x=2&blank=&flag'
+        updated = replace_token(url, 'synthetic+/&=?')
+        self.assertEqual(updated, url.replace('synthetic-old', 'synthetic%2B%2F%26%3D%3F'))
+        self.assertEqual(proxy_identity(updated)[1], 'synthetic+/&=?')
+        self.assertIsNone(proxy_identity(url + '&proxyToken=duplicate'))
+
+    def test_singleflight_and_origin_isolation(self):
+        calls = []
+        auth = ProxyTokens(lambda: self.client(lambda: calls.append('login')))
+        barrier = threading.Barrier(8)
+        def refresh(_):
+            barrier.wait()
+            auth.refresh(self.url)
+            return auth.url(self.url)
+        with ThreadPoolExecutor(max_workers=8) as pool:
+            urls = list(pool.map(refresh, range(8)))
+        self.assertEqual(calls, ['login'])
+        self.assertEqual(set(urls), {self.url.replace('synthetic-old', 'synthetic-short')})
+        for other in (self.url.replace('http:', 'https:'), self.url.replace(':10011', ':10012'),
+                      self.url.replace('example.invalid', 'other.invalid')):
+            self.assertEqual(auth.url(other), other)
+        self.assertEqual(repr(auth), '<ProxyTokens>')
+
+    def test_failure_cooldown_and_later_recovery(self):
+        now, calls = [10], []
+        def factory():
+            calls.append(1)
+            if len(calls) == 1:
+                raise RuntimeError('synthetic-password')
+            return self.client()
+        auth = ProxyTokens(factory, lambda: now[0])
+        for _ in range(2):
+            with self.assertRaisesRegex(ProxyAuthError, '^proxy_token_refresh_failed$'):
+                auth.refresh(self.url)
+        self.assertEqual(calls, [1])
+        now[0] += 31
+        auth.refresh(self.url)
+        self.assertIn('synthetic-short', auth.url(self.url))
+
+    def test_missing_login_same_or_malformed_token_fails_closed(self):
+        for changes in ({'login_url': None}, {'token': 'synthetic-old'}, {'token': 'bad\nvalue'},
+                        {'token': ''}, {'token': None}, {'config': SimpleNamespace(username='', password='')}):
+            with self.subTest(changes=changes):
+                client = self.client()
+                for name, value in changes.items():
+                    setattr(client, name, value)
+                auth = ProxyTokens(lambda: client)
+                with self.assertRaisesRegex(ProxyAuthError, '^proxy_token_refresh_failed$'):
+                    auth.refresh(self.url)
+                self.assertEqual(auth.url(self.url), self.url)
+
+    def test_fresh_rejection_cools_down_without_relogin(self):
+        calls, now = [], [10]
+        auth = ProxyTokens(lambda: self.client(lambda: calls.append(1)), lambda: now[0])
+        auth.refresh(self.url)
+        new = auth.url(self.url)
+        auth.reject(new)
+        with self.assertRaisesRegex(ProxyAuthError, '^proxy_token_refresh_failed$'):
+            auth.refresh(new)
+        self.assertEqual(calls, [1])
+
+
+class RecoveryHttpTests(unittest.TestCase):
+    def setUp(self):
+        self.calls, self.logins = [], []
+        self.valid = 'synthetic-short'
+        self.mode, self.login_mode, self.rejection_status = 'normal', 'normal', 200
+        self.first_received = threading.Event()
+        test = self
+
+        class Handler(BaseHTTPRequestHandler):
+            protocol_version = 'HTTP/1.1'
+
+            def log_message(self, *_):
+                pass
+
+            def handle(self):
+                try:
+                    super().handle()
+                except (ConnectionAbortedError, ConnectionResetError, BrokenPipeError):
+                    pass
+
+            def envelope(self, value, status=200):
+                body = json.dumps(value).encode()
+                self.send_response(status)
+                self.send_header('Content-Length', str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+
+            def do_POST(self):
+                body = self.rfile.read(int(self.headers['Content-Length']))
+                if self.path == '/oauth/api/user/login':
+                    test.logins.append((self.path, parse_qs(body.decode()), dict(self.headers)))
+                    if test.login_mode == 'redirect':
+                        self.send_response(302)
+                        self.send_header('Location', '/do-not-follow')
+                        self.send_header('Content-Length', '0')
+                        self.end_headers()
+                    else:
+                        self.envelope({'code': 201 if test.login_mode == 'denied' else 200,
+                                       'message': test.valid, 'content': None})
+                    return
+                self.proxy(json.loads(body))
+
+            def do_GET(self):
+                self.proxy(None)
+
+            def proxy(self, payload):
+                query = parse_qs(urlsplit(self.path).query)
+                test.calls.append((self.path, dict(self.headers), payload))
+                if test.mode == 'permission':
+                    self.envelope({'code': 201})
+                elif test.mode == 'upstream_auth':
+                    self.envelope({'Status': '115'})
+                elif test.mode == 'http401':
+                    self.envelope({'message': 'unauthorized'}, 401)
+                elif test.mode == 'http500':
+                    self.envelope({'code': 212}, 500)
+                elif test.mode == 'truncated_rejection':
+                    body = b'{"code":212}'
+                    self.send_response(401)
+                    self.send_header('Content-Length', str(len(body) + 10))
+                    self.end_headers()
+                    self.wfile.write(body)
+                    self.close_connection = True
+                elif test.mode == 'invalid_json':
+                    self.send_response(200)
+                    self.send_header('Content-Length', '1')
+                    self.end_headers()
+                    self.wfile.write(b'{')
+                elif test.mode == 'always_invalid' or query['proxyToken'][0] != test.valid:
+                    if test.mode == 'whitespace' and payload is not None:
+                        self.send_response(200)
+                        self.send_header('Transfer-Encoding', 'chunked')
+                        self.end_headers()
+                        self.chunk(b' \n ')
+                        self.chunk(b'{"code":212}')
+                        self.wfile.write(b'0\r\n\r\n')
+                    else:
+                        self.envelope({'code': '212', 'message': 'synthetic rejection'}, test.rejection_status)
+                elif payload is None:
+                    service = query['servertype'][0]
+                    company = {'KeyNo': 'synthetic-id', 'Name': '合成企业'}
+                    result = ([company] if service == 'fuzzySearch' else company
+                              if service == 'basicDetails' else {'VerifyResult': 0, 'Data': []})
+                    self.envelope({'Status': '200', 'Result': result})
+                elif payload.get('stream'):
+                    self.send_response(200)
+                    self.send_header('Content-Type', 'text/event-stream')
+                    self.send_header('Transfer-Encoding', 'chunked')
+                    self.end_headers()
+                    self.chunk(b'data: {"choices":[{"delta":{"content":"first"}}]}\n\n')
+                    if test.mode == 'partial':
+                        self.close_connection = True
+                        return
+                    if test.mode == 'incremental' and not test.first_received.wait(2):
+                        self.close_connection = True
+                        return
+                    self.chunk(b'data: {"choices":[{"delta":{"content":"OK"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n')
+                    self.wfile.write(b'0\r\n\r\n')
+                else:
+                    self.envelope({'choices': [{'message': {'content': 'OK'}}]})
+
+            def chunk(self, body):
+                self.wfile.write(f'{len(body):x}\r\n'.encode() + body + b'\r\n')
+                self.wfile.flush()
+
+        self.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+        self.thread = threading.Thread(target=lambda: self.server.serve_forever(poll_interval=.02), daemon=True)
+        self.thread.start()
+        self.temp = tempfile.TemporaryDirectory()
+        self.path = Path(self.temp.name) / '.env'
+        base = f'http://127.0.0.1:{self.server.server_port}'
+        query = '?servertype={}&proxyToken=synthetic-old&raw=%2f+%20&x=1&x=2'
+        lines = ['QCC_KEY=synthetic-key', 'QCC_SECRET_KEY=synthetic-secret', 'QCC_TIMEOUT_SECONDS=2',
+                 'DEEPSEEK_API_KEY=synthetic-api-key', 'DEEPSEEK_MODEL=synthetic-model', 'SUMMARY_TIMEOUT_SECONDS=2',
+                 'DMS_BASE_URL=http://127.0.0.1', f'AUTH_PORT={self.server.server_port}',
+                 f'DMS_PORT={self.server.server_port}', 'DMS_USERNAME=synthetic-user', 'DMS_PASSWORD=synthetic-password',
+                 'DEEPSEEK_BASE_URL=' + base + '/proxy/chat/completions' + query.format('deepseek')]
+        for name, service in (('SEARCH', 'fuzzySearch'), ('BASIC', 'basicDetails'), ('HONOR', 'honor')):
+            lines.append('QCC_' + name + '_URL=' + base + '/proxy/' + query.format(service))
+        self.path.write_text('\n'.join(lines), encoding='utf-8')
+        self.original = self.path.read_bytes()
+        self.env = patch.dict(os.environ, {'NO_PROXY': '*'}, clear=True)
+        self.env.start()
+        self.qconfig, self.mconfig = qcc.load_config(self.path), summarize.load_config(self.path)
+
+    def tearDown(self):
+        self.assertEqual(self.path.read_bytes(), self.original, '.env must stay byte-identical')
+        self.env.stop()
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join()
+        self.temp.cleanup()
+
+    def model(self):
+        return summarize.transport({'model': self.mconfig.model, 'messages': [{'role': 'user', 'content': 'synthetic'}]}, self.mconfig)
+
+    def test_qcc_login_shared_with_other_interfaces_and_model(self):
+        self.assertIs(self.qconfig.proxy_auth, self.mconfig.proxy_auth)
+        client = qcc.QccClient(self.qconfig)
+        company = client.search('合成企业')['candidates'][0]
+        client.basic(company)
+        self.assertTrue(client.honors(company)['complete'])
+        self.assertEqual(self.model()['choices'][0]['message']['content'], 'OK')
+        self.assertEqual(len(self.logins), 1)
+        self.assertEqual(len(self.calls), 5)
+        first, replay = self.calls[:2]
+        self.assertEqual(first[0].replace('synthetic-old', 'synthetic-short'), replay[0])
+        for name in ('Token', 'Timespan'):
+            self.assertEqual(first[1][name], replay[1][name])
+        expected = hashlib.md5(('synthetic-key' + replay[1]['Timespan'] + 'synthetic-secret').encode()).hexdigest().upper()
+        self.assertEqual(replay[1]['Token'], expected)
+        self.assertEqual(self.logins[0][1], {'userName': ['synthetic-user'], 'password': ['synthetic-password'],
+                                         'clientId': ['0'], 'serviceId': ['0']})
+        self.assertNotIn('Token', self.logins[0][2])
+        self.assertEqual(self.calls[-1][1]['Authorization'], 'Bearer synthetic-api-key')
+        for path, headers, body in self.calls:
+            self.assertIn('raw=%2f+%20&x=1&x=2', path)
+            self.assertNotIn('synthetic-password', str(headers) + str(body) + path)
+
+    def test_valid_long_token_needs_no_dms(self):
+        self.valid = 'synthetic-old'
+        with patch('src.step5_session_analysis.dms.load_dms_config', side_effect=AssertionError('no login expected')):
+            self.model()
+            qcc.QccClient(self.qconfig).search('合成企业')
+        self.assertEqual(self.logins, [])
+
+    def test_healthy_json_is_strictly_decoded_only_once(self):
+        self.valid = 'synthetic-old'
+        with patch.object(summarize, 'strict_json', wraps=summarize.strict_json) as decode:
+            self.model()
+        self.assertEqual(decode.call_count, 1)
+
+    def test_model_json_replays_exact_payload_once(self):
+        self.model()
+        self.assertEqual(len(self.calls), 2)
+        self.assertEqual(self.calls[0][2], self.calls[1][2])
+        self.assertEqual(len(self.logins), 1)
+        self.valid = 'synthetic-next'
+        self.model()
+        self.assertEqual(len(self.calls), 4)
+        self.assertEqual(len(self.logins), 2)
+        self.assertIn('proxyToken=synthetic-next', self.calls[-1][0])
+
+    def test_stream_recovery_keeps_first_delta_incremental(self):
+        self.mode = 'incremental'
+        parts = []
+        def consume(text):
+            parts.append(text)
+            self.first_received.set()
+        result = stream_transport({'model': 'synthetic-model'}, self.mconfig, consume)
+        self.assertEqual(parts, ['first', 'OK'])
+        self.assertEqual(result['choices'][0]['message']['content'], 'firstOK')
+        self.assertEqual(len(self.calls), 2)
+        self.assertEqual(len(self.logins), 1)
+        self.assertEqual(self.calls[0][2], self.calls[1][2])
+
+    def test_whitespace_fragmented_rejection(self):
+        self.mode = 'whitespace'
+        self.assertEqual(self.model()['choices'][0]['message']['content'], 'OK')
+        self.assertEqual(len(self.logins), 1)
+
+    def test_http_auth_requires_explicit_212(self):
+        for status in (401, 403):
+            self.rejection_status = status
+            self.valid = 'synthetic-' + str(status)
+            self.model()
+            self.assertIn('proxyToken=' + self.valid, self.calls[-1][0])
+        self.assertEqual(len(self.logins), 2)
+        # QCC uses the same outer rejection contract for HTTP auth failures.
+        self.valid = 'synthetic-qcc'
+        qcc.QccClient(self.qconfig).search('合成企业')
+        self.assertEqual(len(self.logins), 3)
+
+    def test_unrelated_failures_do_not_log_in(self):
+        for mode, model_code, qcc_code in (
+            ('permission', None, 'provider_failure'), ('upstream_auth', None, 'qcc_auth_failed'),
+            ('http401', 'http_401', 'http_401'), ('http500', 'http_500', 'http_500'),
+            ('invalid_json', 'invalid_response_json', 'invalid_json')):
+            self.mode = mode
+            if model_code is None:
+                self.model()  # Original provider schema validation belongs to the caller.
+            else:
+                with self.assertRaisesRegex(summarize.RequestFailure, '^' + model_code + '$'):
+                    self.model()
+            with self.assertRaisesRegex(qcc.QccError, '^' + qcc_code + '$'):
+                qcc.QccClient(self.qconfig).search('合成企业')
+        self.assertEqual(self.logins, [])
+        self.assertEqual(len(self.calls), 10)
+
+    def test_second_rejection_stops_and_cools_down(self):
+        self.mode = 'always_invalid'
+        with self.assertRaisesRegex(summarize.RequestFailure, '^proxy_token_invalid$') as failure:
+            self.model()
+        self.assertFalse(failure.exception.retryable)
+        self.assertEqual(len(self.calls), 2)
+        with self.assertRaisesRegex(qcc.QccError, '^proxy_token_refresh_failed$'):
+            qcc.QccClient(self.qconfig).search('合成企业')
+        self.assertEqual(len(self.logins), 1)
+        self.assertEqual(len(self.calls), 3)
+
+    def test_truncated_auth_envelope_is_not_a_definite_rejection(self):
+        self.mode = 'truncated_rejection'
+        with self.assertRaisesRegex(summarize.RequestFailure, '^response_incomplete$'):
+            self.model()
+        with self.assertRaisesRegex(qcc.QccError, '^response_incomplete$'):
+            qcc.QccClient(self.qconfig).search('合成企业')
+        self.assertEqual(self.logins, [])
+        self.assertEqual(len(self.calls), 2)
+
+    def test_qcc_second_rejection_is_bounded(self):
+        self.mode = 'always_invalid'
+        with self.assertRaisesRegex(qcc.QccError, '^proxy_token_invalid$'):
+            qcc.QccClient(self.qconfig).search('合成企业')
+        self.assertEqual(len(self.calls), 2)
+        self.assertEqual(len(self.logins), 1)
+
+    def test_login_failure_and_redirect_are_safe_and_not_replayed(self):
+        for mode in ('denied', 'redirect'):
+            self.login_mode = mode
+            # Isolate the two scenarios' shared caches using an explicit test env.
+            config = summarize.load_config(self.path, {})
+            with self.assertRaisesRegex(summarize.RequestFailure, '^proxy_token_refresh_failed$') as failure:
+                summarize.transport({}, config)
+            self.assertFalse(failure.exception.retryable)
+        self.assertEqual(len(self.logins), 2)
+        self.assertEqual(len(self.calls), 2)
+
+    def test_partial_stream_is_never_replayed(self):
+        self.valid, self.mode = 'synthetic-old', 'partial'
+        parts = []
+        with self.assertRaisesRegex(summarize.RequestFailure, '^response_incomplete$'):
+            stream_transport({}, self.mconfig, parts.append)
+        self.assertEqual(parts, ['first'])
+        self.assertEqual(len(self.calls), 1)
+        self.assertEqual(self.logins, [])
+
+    def test_cancellation_after_login_prevents_replay(self):
+        def check():
+            if self.logins:
+                raise summarize.RequestFailure('stream_cancelled', False)
+        with self.assertRaisesRegex(summarize.RequestFailure, '^stream_cancelled$'):
+            stream_transport({}, self.mconfig, lambda _: None, check)
+        self.assertEqual(len(self.calls), 1)
+        self.assertEqual(len(self.logins), 1)
+
+    def test_direct_qcc_envelope_does_not_use_proxy_auth(self):
+        config = replace(self.qconfig, search_url=qcc.DEFAULT_SEARCH_URL)
+        with self.assertRaisesRegex(qcc.QccError, '^provider_failure$'):
+            qcc.QccClient(config, send=lambda *_: {'code': 212}).search('合成企业')
+        self.assertEqual(self.logins, [])
+
+
+if __name__ == '__main__':
+    unittest.main()

src/proxy_auth.py

@@ -0,0 +1,113 @@
+"""In-memory proxy authentication recovery; configuration files stay read-only."""
+
+from pathlib import Path
+from threading import RLock
+from time import monotonic
+from urllib.parse import parse_qsl, quote, unquote_plus, urlsplit, urlunsplit
+from weakref import WeakValueDictionary
+
+
+class ProxyAuthError(RuntimeError):
+    """Fixed error code, never a URL, response body or credential."""
+
+
+def token_rejected(value):
+    return (isinstance(value, dict) and str(value.get('code')) == '212'
+            and 'choices' not in value and 'Status' not in value)
+
+
+def proxy_identity(url):
+    parts = urlsplit(url)
+    query = parse_qsl(parts.query, keep_blank_values=True)
+    tokens = [v for k, v in query if k == 'proxyToken']
+    services = [v for k, v in query if k == 'servertype']
+    if len(tokens) != 1 or not tokens[0] or len(services) != 1 or not services[0]:
+        return None
+    return (parts.scheme, parts.hostname, parts.port), tokens[0]
+
+
+def replace_token(url, token):
+    """Only replace the token value; preserve every other signed query byte."""
+    parts = urlsplit(url)
+    query = []
+    for item in parts.query.split('&'):
+        key, _, _ = item.partition('=')
+        query.append(key + '=' + quote(token, safe='') if unquote_plus(key) == 'proxyToken' else item)
+    return urlunsplit(parts._replace(query='&'.join(query)))
+
+
+class ProxyTokens:
+    def __init__(self, client_factory, clock=monotonic):
+        self._factory, self._clock = client_factory, clock
+        self._lock = RLock()
+        self._client = None
+        self._tokens = {}
+        self._retry_at = 0
+
+    def __repr__(self):
+        return '<ProxyTokens>'
+
+    def url(self, configured):
+        identity = proxy_identity(configured)
+        if identity is None:
+            return configured
+        with self._lock:
+            token = self._tokens.get(identity[0])
+        return replace_token(configured, token) if token else configured
+
+    def refresh(self, rejected_url):
+        identity = proxy_identity(rejected_url)
+        if identity is None:
+            raise ProxyAuthError('proxy_token_refresh_failed')
+        origin, rejected = identity
+        with self._lock:
+            current = self._tokens.get(origin)
+            if current and current != rejected:
+                return
+            if self._clock() < self._retry_at:
+                raise ProxyAuthError('proxy_token_refresh_failed')
+            try:
+                if self._client is None:
+                    self._client = self._factory()
+                client = self._client
+                if not (client.config.username and client.config.password and client.login_url):
+                    raise ProxyAuthError('proxy_token_refresh_failed')
+                client._login()  # Existing AUTH_PORT/login form contract.
+                token = client.token
+                if (not isinstance(token, str) or not token.strip() or token == rejected
+                        or any(ord(c) < 32 or ord(c) == 127 for c in token)):
+                    raise ProxyAuthError('proxy_token_refresh_failed')
+            except Exception:
+                self._retry_at = self._clock() + 30
+                raise ProxyAuthError('proxy_token_refresh_failed') from None
+            self._tokens[origin] = token
+            self._retry_at = 0
+
+    def reject(self, url):
+        """A freshly acquired token also failed: cool down repeated logins."""
+        identity = proxy_identity(url)
+        if identity is not None:
+            with self._lock:
+                if self._tokens.get(identity[0]) == identity[1]:
+                    self._retry_at = self._clock() + 30
+
+
+_registry = WeakValueDictionary()
+_registry_lock = RLock()
+
+
+def proxy_tokens(path, environ=None):
+    """Share within this process/config path; explicit test envs stay local."""
+    source = Path(path).resolve()
+    def client():
+        # A valid existing token does not log in or require DMS at startup.
+        from src.step5_session_analysis.dms import DmsClient, load_dms_config
+        return DmsClient(load_dms_config(source, environ, scope_names=()))
+    if environ is not None:
+        return ProxyTokens(client)
+    with _registry_lock:
+        result = _registry.get(str(source))
+        if result is None:
+            result = ProxyTokens(client)
+            _registry[str(source)] = result
+        return result

src/step1_knowledge_prepare/summarize.py

@@ -3,6 +3,7 @@
 import argparse
 import hashlib
 from http.client import HTTPConnection, HTTPSConnection, HTTPException
+from io import BytesIO
 import json
 import os
 import re
@@ -21,6 +22,7 @@ from pathlib import Path
 
 from .prepare import ROOT, dump, strict_json
 from .summary_input import PROMPT_VERSION, chat_text, facts, messages, scenario
+from src.proxy_auth import ProxyAuthError, proxy_tokens, token_rejected
 
 KINDS = ("policy", "finance", "general", "qa")
 DEFAULTS = {"SUMMARY_CONTEXT_TOKENS": "32768", "SUMMARY_MAX_OUTPUT_TOKENS": "4096",
@@ -28,6 +30,7 @@ DEFAULTS = {"SUMMARY_CONTEXT_TOKENS": "32768", "SUMMARY_MAX_OUTPUT_TOKENS": "409
             "SUMMARY_TIMEOUT_SECONDS": "120", "SUMMARY_MAX_ATTEMPTS": "3",
             "SUMMARY_MAX_OUTPUT_LIMIT": "32768", "SUMMARY_MAX_CONTEXT_LIMIT": "131072"}
 CONFIG_KEYS = set(DEFAULTS) | {"DEEPSEEK_API_KEY", "DEEPSEEK_BASE_URL", "DEEPSEEK_MODEL"}
+_JSON_UNREAD = object()
 
 
 def fingerprint(value):
@@ -51,6 +54,7 @@ class Config:
     max_attempts: int = 3
     max_output_limit: int = 32768
     max_context_limit: int = 131072
+    proxy_auth: object = field(default=None, repr=False, compare=False)
 
     @property
     def endpoint(self):
@@ -75,10 +79,44 @@ def chat_endpoint(base_url):
     return urllib.parse.urlunsplit(parts._replace(path=path))
 
 
+class _BufferedResponse(BytesIO):
+    def __init__(self, body, decoded):
+        super().__init__(body)
+        self.decoded, self.length = decoded, 0
+
+
+class _PrefixedResponse:
+    """Replay a bounded preflight read without changing JSON or SSE bytes."""
+    def __init__(self, prefix, response):
+        self.prefix, self.offset, self.response = prefix, 0, response
+
+    def __getattr__(self, name):
+        return getattr(self.response, name)
+
+    @property
+    def length(self):
+        remaining = getattr(self.response, 'length', None)
+        return None if remaining is None else remaining + len(self.prefix) - self.offset
+
+    def read1(self, count=-1):
+        if self.offset < len(self.prefix):
+            end = len(self.prefix) if count < 0 else min(len(self.prefix), self.offset + count)
+            result = self.prefix[self.offset:end]
+            self.offset = end
+            return result
+        return self.response.read1(count)
+
+    def read(self, count=-1):
+        first = self.read1(count) if self.offset < len(self.prefix) else b''
+        return first + self.response.read(-1 if count < 0 else count - len(first))
+
+
 @contextmanager
-def deepseek_response(payload, config, *, stream=False):
+def deepseek_response(payload, config, *, stream=False, check=lambda: None):
     """One request, with a private connection per gateway call; always close it."""
-    endpoint = config.endpoint
+    configured = config.endpoint
+    auth = getattr(config, 'proxy_auth', None)
+    endpoint = auth.url(configured) if auth is not None else configured
     parts = urllib.parse.urlsplit(endpoint)
     query = dict(urllib.parse.parse_qsl(parts.query))
     gateway = query.get('servertype') == 'deepseek' and bool(query.get('proxyToken'))
@@ -91,18 +129,70 @@ def deepseek_response(payload, config, *, stream=False):
             yield response
         return
     headers['Connection'] = 'keep-alive'
-    connection = (HTTPSConnection if parts.scheme == 'https' else HTTPConnection)(
-        parts.hostname, parts.port, timeout=config.timeout)
-    try:
-        target = (parts.path or '/') + ('?' + parts.query if parts.query else '')
-        connection.request('POST', target, body=body, headers=headers)
-        with connection.getresponse() as response:
-            if not 200 <= response.status < 300:
-                code = response.status
-                raise RequestFailure(f'http_{code}', not stream and (code in {408, 429} or code >= 500))
-            yield response
-    finally:
-        connection.close()
+    for attempt in range(2):
+        check()
+        parts = urllib.parse.urlsplit(endpoint)
+        connection = (HTTPSConnection if parts.scheme == 'https' else HTTPConnection)(
+            parts.hostname, parts.port, timeout=config.timeout)
+        rejected = False
+        try:
+            target = (parts.path or '/') + ('?' + parts.query if parts.query else '')
+            connection.request('POST', target, body=body, headers=headers)
+            with connection.getresponse() as response:
+                buffered, decoded = None, _JSON_UNREAD
+                if not 200 <= response.status < 300 and response.status not in (401, 403):
+                    code = response.status
+                    raise RequestFailure(f'http_{code}', not stream and (code in {408, 429} or code >= 500))
+                # read1 validates HTTP chunk framing (peek can swallow broken
+                # chunks). Replay all inspected bytes for normal SSE streams.
+                check()
+                prefix = bytearray(response.read1(4096))
+                deadline = time.monotonic() + config.timeout
+                while prefix and not prefix.lstrip() and len(prefix) <= 4_000_000:
+                    check()
+                    if time.monotonic() >= deadline:
+                        raise RequestFailure('request_timeout', False)
+                    more = response.read1(4096)
+                    if not more:
+                        break
+                    prefix.extend(more)
+                if len(prefix) > 4_000_000:
+                    raise RequestFailure('response_too_large', False)
+                response = _PrefixedResponse(bytes(prefix), response)
+                if prefix.lstrip().startswith(b'{'):
+                    buffered = response.read(4_000_001)
+                    if len(buffered) > 4_000_000:
+                        raise RequestFailure('response_too_large', False)
+                    if getattr(response, 'length', None) not in (None, 0):
+                        raise RequestFailure('response_incomplete', False)
+                    try:
+                        decoded = strict_json(buffered.decode('utf-8'))
+                        rejected = token_rejected(decoded)
+                    except (ValueError, UnicodeError):
+                        pass
+                if not rejected:
+                    if not 200 <= response.status < 300:
+                        code = response.status
+                        raise RequestFailure(f'http_{code}', not stream and (code in {408, 429} or code >= 500))
+                    if buffered is None:
+                        yield response
+                    else:
+                        with _BufferedResponse(buffered, decoded) as cached:
+                            yield cached
+                    return
+        finally:
+            connection.close()
+        if rejected:
+            if attempt or auth is None:
+                if auth is not None:
+                    auth.reject(endpoint)
+                raise RequestFailure('proxy_token_invalid', False)
+            check()
+            try:
+                auth.refresh(endpoint)
+            except ProxyAuthError:
+                raise RequestFailure('proxy_token_refresh_failed', False) from None
+            endpoint = auth.url(configured)
 
 
 def load_config(path, environ=None):
@@ -141,7 +231,8 @@ def load_config(path, environ=None):
         raise ValueError("output_budget_leaves_no_input_space")
     if numbers["SUMMARY_MAX_OUTPUT_TOKENS"] > numbers["SUMMARY_MAX_OUTPUT_LIMIT"] or numbers["SUMMARY_CONTEXT_TOKENS"] > numbers["SUMMARY_MAX_CONTEXT_LIMIT"]:
         raise ValueError("initial_budget_exceeds_retry_limit")
-    return Config(values["DEEPSEEK_API_KEY"], values["DEEPSEEK_BASE_URL"], values["DEEPSEEK_MODEL"], *numbers.values())
+    return Config(values["DEEPSEEK_API_KEY"], values["DEEPSEEK_BASE_URL"], values["DEEPSEEK_MODEL"],
+                  *numbers.values(), proxy_auth=proxy_tokens(path, environ))
 
 
 def make_job(record, config, count, *, workbook_llm=False):
@@ -183,11 +274,12 @@ def transport(payload, config):
     try:
         with deepseek_response(payload, config) as response:
             content = response.read(4_000_001)
+            decoded = response.decoded if isinstance(response, _BufferedResponse) else _JSON_UNREAD
             if getattr(response, 'length', None) not in (None, 0) and len(content) <= 4_000_000:
                 raise RequestFailure('response_incomplete')
         if len(content) > 4_000_000:
             raise RequestFailure("response_too_large", False)
-        return strict_json(content.decode("utf-8"))
+        return strict_json(content.decode("utf-8")) if decoded is _JSON_UNREAD else decoded
     except urllib.error.HTTPError as exc:
         code = exc.code
         exc.close()

src/step3_question_answer/qcc.py

@@ -9,6 +9,7 @@ import urllib.error
 import urllib.parse
 import urllib.request
 from dataclasses import dataclass, field
+from src.proxy_auth import ProxyAuthError, proxy_identity, proxy_tokens, token_rejected
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -49,6 +50,7 @@ class Config:
     search_url: str = field(default=DEFAULT_SEARCH_URL, repr=False)
     basic_url: str = field(default=DEFAULT_BASIC_URL, repr=False)
     honor_url: str = field(default=DEFAULT_HONOR_URL, repr=False)
+    proxy_auth: object = field(default=None, repr=False, compare=False)
 
     def __post_init__(self):
         if not self.key or not self.secret:
@@ -76,7 +78,7 @@ def load_config(path=Path('.env'), environ=None):
                       int(values.get('QCC_HONOR_MAX_PAGES', '3')),
                       values.get('QCC_SEARCH_URL', DEFAULT_SEARCH_URL),
                       values.get('QCC_BASIC_URL', DEFAULT_BASIC_URL),
-                      values.get('QCC_HONOR_URL', DEFAULT_HONOR_URL))
+                      values.get('QCC_HONOR_URL', DEFAULT_HONOR_URL), proxy_auth=proxy_tokens(path, environ))
     except (TypeError, ValueError):
         raise ValueError('invalid_qcc_configuration') from None
 
@@ -98,6 +100,16 @@ def transport(url, headers, timeout):
                                headers={**headers, 'Accept': '*/*', 'Connection': 'keep-alive'})
             with connection.getresponse() as response:
                 if not 200 <= response.status < 300:
+                    if response.status in (401, 403):
+                        rejected = response.read(65537)
+                        if response.length not in (None, 0) and len(rejected) <= 65536:
+                            raise QccError('response_incomplete')
+                        try:
+                            envelope = json.loads(rejected) if len(rejected) <= 65536 else None
+                        except (ValueError, UnicodeError):
+                            envelope = None
+                        if token_rejected(envelope):
+                            return envelope
                     raise QccError(f'http_{response.status}')
                 body = response.read(4_000_001)
                 if response.length not in (None, 0) and len(body) <= 4_000_000:
@@ -136,7 +148,21 @@ class QccClient:
         timestamp = str(int(self.clock()))
         token = hashlib.md5((self.config.key + timestamp + self.config.secret).encode()).hexdigest().upper()
         url = endpoint + ('&' if urllib.parse.urlsplit(endpoint).query else '?') + urllib.parse.urlencode({'key': self.config.key, **params})
-        payload = self.send(url, {'Token': token, 'Timespan': timestamp}, self.config.timeout)
+        auth = self.config.proxy_auth
+        effective = auth.url(url) if auth is not None else url
+        for attempt in range(2):
+            payload = self.send(effective, {'Token': token, 'Timespan': timestamp}, self.config.timeout)
+            if proxy_identity(effective) is None or not token_rejected(payload):
+                break
+            if attempt or auth is None:
+                if auth is not None:
+                    auth.reject(effective)
+                raise QccError('proxy_token_invalid')
+            try:
+                auth.refresh(effective)
+            except ProxyAuthError:
+                raise QccError('proxy_token_refresh_failed') from None
+            effective = auth.url(url)
         status = str(payload.get('Status')) if isinstance(payload, dict) else None
         if status == '115':
             raise QccError('qcc_auth_failed')

src/step3_question_answer/recommendation.py

@@ -209,7 +209,7 @@ class JsonModel:
             result = model_response(payload, self.config, transport)
         except RequestFailure as exc:
             allowed = {'request_timeout', 'network_failure', 'invalid_response_json', 'response_too_large',
-                       'stream_incomplete', 'stream_cancelled'}
+                       'stream_incomplete', 'stream_cancelled', 'proxy_token_invalid', 'proxy_token_refresh_failed'}
             code = exc.code if exc.code in allowed or re.fullmatch(r'http_[1-5][0-9]{2}', str(exc.code)) else 'request_failed'
             raise AnalysisError('model_' + code) from None
         except Exception:

src/step3_question_answer/streaming.py

@@ -225,7 +225,7 @@ def stream_transport(payload, config, on_content, check=lambda: None):
     started = time.monotonic()
     try:
         check()
-        with deepseek_response({**payload, 'stream': True}, config, stream=True) as response:
+        with deepseek_response({**payload, 'stream': True}, config, stream=True, check=check) as response:
             for line in response_lines(response, check, started + config.timeout):
                 check()
                 if time.monotonic() - started > config.timeout:

src/step4_api/logging.py

@@ -46,6 +46,7 @@ def safe_code(value):
         'model_response_too_large', 'model_request_failed', 'model_unexpected_error',
         'model_invalid_response_schema', 'model_output_token_limit', 'model_content_filtered',
         'model_incomplete_generation', 'model_invalid_content_json', 'model_invalid_content_schema',
+        'proxy_token_invalid', 'proxy_token_refresh_failed', 'model_proxy_token_invalid', 'model_proxy_token_refresh_failed',
         'network_failure', 'provider_failure', 'response_too_large', 'invalid_json',
         'invalid_result', 'invalid_search_result', 'invalid_candidate', 'duplicate_candidate',
         'invalid_paging', 'company_identity_mismatch', 'invalid_honor_result',