docs/superpowers/specs/2026-07-14-mcp-open-number-format-guidance-design.md

@@ -0,0 +1,80 @@
+# MCP 精准订单开放单号格式引导设计
+
+## 背景
+
+`query_order_exact` 通过字段说明和示例引导 AI 选择精准查询参数。现有说明虽然要求
+单号类型不明确时先追问，但部分字段示例或说明仍可能让 AI 把 `USC`、`PC`、`SO`
+等前缀误当成固定格式，并据此猜测号码类型。
+
+实际业务中，订单号、客户参考号、快递单号、排舱单号、柜号、SO 号和
+Shipment ID 都应视为开放格式。用户明确指定业务类型后，工具按对应字段原值精准
+查询；格式、前缀、长度和字符组合不参与字段选择。
+
+## 目标
+
+- 所有单号只按用户明确指定的业务类型选择查询字段。
+- 明确所有单号格式开放，禁止根据前缀、长度、字符组合或示例猜测类型。
+- 保留含糊“单号”先追问、无结果不跨字段重试的现有规则。
+- 保持 MCP 参数、PHP 查询、数据库字段和权限逻辑不变。
+
+## 适用字段
+
+统一规则覆盖以下单值字段及其现有批量字段：
+
+- `order_number` / `order_numbers`
+- `reference_number` / `reference_numbers`
+- `tracking_number` / `tracking_numbers`
+- `outbound_number` / `outbound_numbers`
+- `container_code` / `container_codes`
+- `so_number` / `so_numbers`
+- `shipment_id`
+
+`shipment_id` 当前没有批量字段，本次不新增。
+
+## 方案
+
+### 工具级规则
+
+`query_order_exact` 总说明新增统一规则：所有单号格式均为开放格式，只能根据用户
+明确说出的业务类型选择字段，不得根据号码的前缀、长度、字符组合或示例猜测。
+
+### 字段说明
+
+- 每个单值字段说明其业务含义，并要求用户明确指定该类型时使用。
+- 批量字段继续表达“多个同类号码使用数组”，不声明固定格式。
+- `outbound_number` 保留“排舱单号、不是海外仓出库单号”的业务区分，删除
+  “以 `PC` 开头”的规则。
+- `so_number` 说明格式不固定，只有用户明确说“SO号”时使用。
+
+### 示例
+
+- 示例只展示参数类型和调用形式，不代表格式规则。
+- 排舱单号可继续使用 `PC...` 作为普通示例，但说明和测试不得把 `PC` 设为必要
+  条件。
+- SO 号单值示例使用纯数字 `97964454`；批量示例同时包含 `97964454` 和
+  `OOLU12345678`，体现格式多样性。
+- 其他现有示例可以保留，但不得用于推断或校验格式。
+
+## 查询与错误处理
+
+- 用户明确类型：直接使用对应字段精准查询。
+- 用户只说“单号”或类型含糊：先追问，不调用工具猜测。
+- 查询无结果：原样返回，不改用其他单号字段重试。
+- Gateway 不新增正则、前缀、长度或字符集校验；现有非空、字符串长度和批量数量
+  边界保持不变。
+
+## 测试
+
+元数据回归测试需要验证：
+
+1. 工具总说明包含“所有单号格式开放”和禁止按前缀、长度、字符组合、示例猜测。
+2. 排舱单号说明不再声明 `PC` 前缀规则，但仍使用正确业务名称并区分海外仓出库
+   单号。
+3. SO 号说明强调格式不固定和用户明确指定，单值/批量示例覆盖不同格式。
+4. 所有单号字段名、Schema 类型和调用转发行为保持不变。
+
+## 文档覆盖关系
+
+本文覆盖 `2026-07-14-mcp-outbound-number-label-design.md` 中关于排舱单号必须以
+`PC` 开头、示例必须全部使用 `PC` 前缀的要求。旧设计中“业务名称统一为排舱单号、
+不是海外仓出库单号、字段名和查询链不变”的结论继续有效。

docs/superpowers/specs/2026-07-14-mcp-outbound-number-label-design.md

@@ -0,0 +1,50 @@
+# MCP 排舱单号字段文案统一设计
+
+> **部分废止：** 本文关于排舱单号“以 `PC` 开头”以及示例必须使用 `PC`
+> 前缀的要求，已被
+> `2026-07-14-mcp-open-number-format-guidance-design.md` 覆盖。保留“业务名称统一为
+> 排舱单号、不是海外仓出库单号、字段名和查询链不变”的结论。
+
+## 背景
+
+`query_order_exact` 使用 `outbound_number` 和 `outbound_numbers` 查询
+`fms_outbound.outbound_number`。当前工具说明将该字段称为“出库单号”，但业务页面
+使用的准确名称是“排舱单号”，容易使 AI 和员工误解字段含义。
+
+## 目标
+
+- 对外说明统一使用“排舱单号”。
+- 说明排舱单号以 `PC` 开头，并使用真实格式的示例。
+- 明确该字段不是海外仓出库单号，减少号码类型误判。
+- 保持现有接口和查询行为不变。
+
+## 方案
+
+保留 `outbound_number` 和 `outbound_numbers` 字段名，因为它们已经与数据库字段、
+Python Gateway、PHP 校验及查询逻辑对齐。仅修改以下文案和回归断言：
+
+1. `outbound_number` 说明改为单个排舱单号精准查询，注明号码以 `PC` 开头，
+   示例使用 `PC20210331070002001`，并提示不是海外仓出库单号。
+2. `outbound_numbers` 说明改为多个排舱单号批量精准查询，示例中的每个号码均以
+   `PC` 开头。
+3. 工具总说明中的“出库单号”改为“排舱单号”。
+4. 工具元数据测试断言同步改为“排舱单号”，并验证区分海外仓出库单号的提示。
+5. 项目文档记录本次业务术语修正。
+
+## 不在范围内
+
+- 不根据 `PC` 前缀在 Gateway 增加格式校验；号码是否合法仍由精准查询结果判断，
+  避免阻断历史数据或特殊业务数据。
+- 不修改数据库字段名。
+- 不修改 MCP 入参或返回字段名。
+- 不修改 PHP 查询、校验和权限逻辑。
+- 不把 `outbound_date_start`、`outbound_date_end` 的“出库日期”改为“排舱日期”；
+  日期字段与本次单号术语修正不是同一业务概念。
+
+## 验证
+
+- 运行 `tests.test_query_order_exact_tool`，确认工具元数据文案正确，且单值、批量
+  示例都使用 `PC` 前缀。
+- 运行 Python Gateway 全量单元测试，确认字段结构和调用行为无回归。
+- 搜索 `query_order_exact` 相关说明，确认不再将 `outbound_number(s)` 标注为
+  “出库单号”。

docs/superpowers/specs/2026-07-14-mcp-python-coverage-100-design.md

@@ -0,0 +1,111 @@
+# MCP Python Gateway 100% 覆盖率设计
+
+## 背景
+
+Python Gateway 当前执行：
+
+```powershell
+python -m coverage run -m unittest discover -s tests -p "test_*.py"
+python -m coverage report -m
+```
+
+全量 `297` 个测试通过，但 `.coveragerc` 启用了语句和分支覆盖，当前 TOTAL 仅为
+`93.63%`。报告包含 1153 条语句和 418 条分支，其中缺少 52 条语句、48 条分支，
+共有 9 个生产文件未达到 100%。
+
+## 目标
+
+- 保持现有 `.coveragerc` 的 `source`、`omit`、`branch=True` 和报告口径不变。
+- 所有纳入统计的非空生产文件逐个达到 `100.00%`。
+- TOTAL 达到 `100.00%`，并通过 `coverage report --fail-under=100`。
+- 全量 unittest 保持通过，测试不依赖真实 Redis、外部 HTTP 服务或长期运行线程。
+- 不覆盖、不回退当前工作区已有的单号元数据和测试改动。
+
+## 不采用的方案
+
+### 扩大 omit 或添加 pragma
+
+不通过 `.coveragerc omit`、`exclude_lines` 或 `# pragma: no cover` 隐藏当前生产代码
+缺口。这会改变统计边界，无法证明现有代码路径经过测试。
+
+### 为覆盖率改写业务逻辑
+
+当前函数级报告未发现必须重构才能触达的代码。默认只修改测试；若实施中证明某条
+分支结构上不可达，应暂停并重新评审，而不是直接修改生产逻辑或排除该分支。
+
+## 实施结构
+
+### 第一批：叶子模块和数据边界
+
+| 生产文件 | 测试文件 | 覆盖重点 |
+|----------|----------|----------|
+| `tools/query_order_exact.py` | `tests/test_query_order_exact_tool.py` | 非法 scalar ID、逗号字符串 ID 列表、字符串列表去重回边 |
+| `services/token_store.py` | `tests/test_token_store_coverage.py` | 无目录文件、延迟读取、已删除文件清理、bytes 请求、非法 Redis 行 |
+| `services/gateway_session_store.py` | `tests/test_gateway_session_store_unit.py` | 不存在 session 的 touch |
+| `config.py` | `tests/test_config_compat.py` | timeout 环境变量优先级、无效 dotenv 行、带引号值和循环回边 |
+
+这一批不改生产代码，只补可观察的输入、返回值、异常和依赖调用断言。
+
+### 第二批：Gateway 应用和 CLI
+
+| 生产文件 | 测试文件 | 覆盖重点 |
+|----------|----------|----------|
+| `app.py` | `tests/test_app_coverage.py` | 注册工具名称、动态列表错误、request ID、缺少刷新客户端、public 启动、CLI 必填/可选参数及最终防御分支 |
+
+`serve-public` 使用 mock 替换 Redis client、session store、public app 和 server 入口；
+CLI 防御分支通过模拟参数解析结果触发，不启动真实服务。
+
+### 第三批：协议与公网服务
+
+| 生产文件 | 测试文件 | 覆盖重点 |
+|----------|----------|----------|
+| `mcp_protocol.py` | `tests/test_mcp_protocol_coverage.py` | 无 flush 输出、无 input schema、工具响应数据形态、错误 data/meta、空轨迹 tips |
+| `public_gateway.py` | `tests/test_public_gateway_unit.py` | 动态列表无效响应、不支持 touch_session 的 store |
+| `public_server.py` | `tests/test_public_server_coverage.py` | input schema 归一化、启用限流、清理线程、KeyboardInterrupt 关闭 |
+
+HTTP server、线程和无限清理循环都使用 mock 控制。清理循环测试捕获
+`threading.Thread` 的 target，并让受控 `time.sleep` 抛出测试专用异常终止循环，不做
+真实等待。
+
+## 测试质量要求
+
+- 测试名称描述具体行为，不以“覆盖第 N 行”作为唯一目的。
+- 每个异常分支同时断言异常类型和关键消息。
+- 每个成功分支断言返回值或依赖调用参数，不能只调用代码而不验证结果。
+- mock 仅用于 Redis、socket、HTTP server、线程、时钟和 CLI 解析等外部边界。
+- 不添加只为测试存在的生产 API。
+- 每批完成后运行目标测试，再重新执行全量 coverage，依据新的 Missing 列继续收敛。
+
+## 验证流程
+
+每批执行：
+
+```powershell
+python -m unittest <本批目标测试模块>
+python -m coverage run -m unittest discover -s tests -p "test_*.py"
+python -m coverage report -m
+```
+
+最终验收：
+
+```powershell
+python -m coverage run -m unittest discover -s tests -p "test_*.py"
+python -m coverage report -m --fail-under=100
+git diff --check
+```
+
+验收标准：
+
+1. 全量 unittest 返回 0。
+2. 报告中每个非空生产文件为 `100.00%`。
+3. TOTAL 为 `100.00%`，Missing 为空。
+4. `.coveragerc` 的统计范围和分支覆盖设置未被放宽。
+5. 工作区原有单号工具改动保持完整。
+
+## 风险与处理
+
+- 覆盖率 100% 不等于业务逻辑绝对正确；测试仍以可观察行为和关键错误边界为主。
+- 线程/服务器测试若泄漏真实后台线程会导致套件不稳定，因此必须 mock `Thread.start`
+  或捕获 target 后同步执行。
+- 分支覆盖可能在新增测试后暴露新的回边提示；以最新 `coverage report -m` 为准逐项
+  处理，不通过删除有效断言或放宽配置收尾。

tests/test_app_coverage.py

@@ -12,6 +12,7 @@ import json
 import os
 import tempfile
 import unittest
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 from app import GatewayApp, main
@@ -101,6 +102,43 @@ class DummyAuthClient:
         pass
 
 
+class GatewayAppBoundaryTest(unittest.TestCase):
+    def test_registered_tool_names_returns_local_registry(self):
+        app = GatewayApp(api_client=DummyApiClient())
+
+        self.assertEqual(tuple(app._tools.keys()), app.registered_tool_names())
+
+    def test_enabled_tool_names_rejects_non_dict_response(self):
+        with self.assertRaisesRegex(RuntimeError, 'invalid enabled tool response'):
+            GatewayApp()._enabled_tool_names(None)
+
+    def test_enabled_tool_names_rejects_missing_tool_code_list(self):
+        response = {'code': 'MCP_0000', 'data': {}}
+
+        with self.assertRaisesRegex(RuntimeError, 'invalid enabled tool response'):
+            GatewayApp()._enabled_tool_names(response)
+
+    def test_load_enabled_tools_requires_capable_client(self):
+        for api_client in (None, object()):
+            with self.subTest(api_client=api_client):
+                with self.assertRaisesRegex(RuntimeError, 'client unavailable'):
+                    GatewayApp(api_client=api_client)._load_enabled_tool_names()
+
+    def test_build_request_id_preserves_provided_value(self):
+        self.assertEqual(
+            'rq_given',
+            GatewayApp().build_request_id(' rq_given '),
+        )
+
+    def test_ensure_session_requires_auth_client_for_expiring_token(self):
+        token_store = MagicMock()
+        token_store.get.return_value = {'token': 'MT_test'}
+        token_store.is_expiring.return_value = True
+
+        with self.assertRaisesRegex(RuntimeError, 'auth client missing'):
+            GatewayApp(token_store=token_store).ensure_session()
+
+
 def _make_app_with_token():
     """创建有有效 token 的 GatewayApp，附带一个自定义 fake_tool 工具。"""
     with tempfile.TemporaryDirectory() as tmp_dir:
@@ -145,6 +183,125 @@ class RunCliCallElseBranchTest(unittest.TestCase):
             payload = json.loads(stdout.getvalue())
             self.assertEqual('MCP_0000', payload['code'])
 
+
+class RunCliBranchCoverageTest(unittest.TestCase):
+    def _run_with_mocked_call(self, argv):
+        app = GatewayApp(api_client=DummyApiClient())
+        stdout = io.StringIO()
+        with patch.object(
+            app,
+            'call_tool',
+            return_value={'code': 'MCP_0000'},
+        ) as call_tool:
+            result = app.run_cli(argv, stdout=stdout)
+        return result, call_tool
+
+    def test_query_order_requires_keyword(self):
+        with self.assertRaisesRegex(ValueError, 'keyword is required'):
+            GatewayApp().run_cli([
+                'call', '--tool', 'query_order',
+            ], stdout=io.StringIO())
+
+    def test_query_track_forwards_each_supported_identifier(self):
+        cases = (
+            (['--order-id', '5'], {'order_id': 5}),
+            (['--order-number', 'ORDER-1'], {'order_number': 'ORDER-1'}),
+            (['--tracking-number', 'TRACK-1'], {'tracking_number': 'TRACK-1'}),
+        )
+        for cli_args, expected in cases:
+            with self.subTest(cli_args=cli_args):
+                result, call_tool = self._run_with_mocked_call([
+                    'call', '--tool', 'query_track', *cli_args,
+                ])
+                self.assertEqual(0, result)
+                payload = call_tool.call_args.args[1]
+                for key, value in expected.items():
+                    self.assertEqual(value, payload[key])
+
+    def test_query_track_requires_one_identifier(self):
+        with self.assertRaisesRegex(ValueError, 'order-id'):
+            GatewayApp().run_cli([
+                'call', '--tool', 'query_track',
+            ], stdout=io.StringIO())
+
+    def test_query_order_exact_forwards_scalar_ids(self):
+        result, call_tool = self._run_with_mocked_call([
+            'call', '--tool', 'query_order_exact',
+            '--order-number', 'ORDER-1',
+            '--sales-id', '7',
+            '--department-id', '8',
+        ])
+
+        self.assertEqual(0, result)
+        payload = call_tool.call_args.args[1]
+        self.assertEqual(7, payload['sales_id'])
+        self.assertEqual(8, payload['department_id'])
+
+    def test_filter_options_requires_filter_type(self):
+        with self.assertRaisesRegex(ValueError, 'filter-type is required'):
+            GatewayApp().run_cli([
+                'call', '--tool', 'list_order_filter_options',
+            ], stdout=io.StringIO())
+
+    def test_serve_public_builds_scoped_dependencies(self):
+        config = SimpleNamespace(
+            redis_host='redis.test',
+            redis_port=6380,
+            redis_db=2,
+            redis_password='secret',
+            timeout_seconds=9,
+            redis_prefix='gateway:',
+            gateway_session_ttl_seconds=600,
+            tools_base_url='https://tools.test',
+            rate_limit_enabled=True,
+            rate_limit_max_requests=12,
+            rate_limit_window_seconds=34,
+        )
+        with patch('app.GatewayConfig.from_env', return_value=config), \
+                patch('app.RedisSocketClient') as redis_cls, \
+                patch('app.GatewaySessionStore') as store_cls, \
+                patch('app.ScopedApiClient') as api_cls, \
+                patch('app.PublicGatewayApp') as public_app_cls, \
+                patch('app.serve_public', return_value=17) as serve:
+            result = GatewayApp().run_cli([
+                'serve-public', '--host', '127.0.0.1', '--port', '9000',
+            ])
+
+        self.assertEqual(17, result)
+        redis_cls.assert_called_once_with(
+            host='redis.test',
+            port=6380,
+            db=2,
+            password='secret',
+            timeout=9,
+        )
+        store_cls.assert_called_once_with(
+            redis_cls.return_value,
+            prefix='gateway:',
+            ttl_seconds=600,
+        )
+        api_cls.assert_called_once_with('https://tools.test', timeout=9)
+        public_app_cls.assert_called_once_with(
+            session_store=store_cls.return_value,
+            api_client=api_cls.return_value,
+        )
+        serve.assert_called_once_with(
+            public_app_cls.return_value,
+            host='127.0.0.1',
+            port=9000,
+            enable_rate_limit=True,
+            rate_limit_max_requests=12,
+            rate_limit_window_seconds=34,
+        )
+
+    def test_unsupported_command_raises_runtime_error(self):
+        with patch(
+            'app.argparse.ArgumentParser.parse_args',
+            return_value=SimpleNamespace(command='unknown'),
+        ):
+            with self.assertRaisesRegex(RuntimeError, 'unsupported command'):
+                GatewayApp().run_cli([], stdout=io.StringIO())
+
     def test_call_else_branch_without_keyword_still_calls_tool(self):
         """
         --tool fake_tool（非 query_order / query_track）且 --keyword 为空 →

tests/test_config_compat.py

@@ -110,6 +110,39 @@ class GatewayConfigCompatTest(unittest.TestCase):
 
         self.assertEqual(1, config.timeout_seconds)
 
+    def test_timeout_resolution_covers_supported_env_keys(self):
+        cases = (
+            ({'MCP_TIMEOUT_SECONDS': '11'}, {}, 11),
+            ({'FMS_TIMEOUT_MS': '2500'}, {}, 2),
+            ({}, {'FMS_TIMEOUT_SECONDS': '12'}, 12),
+        )
+        for preferred, fallback, expected in cases:
+            with self.subTest(preferred=preferred, fallback=fallback):
+                self.assertEqual(
+                    expected,
+                    GatewayConfig._resolve_timeout_seconds(
+                        preferred,
+                        fallback,
+                    ),
+                )
+
+    def test_load_dotenv_skips_invalid_lines_and_unquotes_values(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = os.path.join(tmp_dir, '.env')
+            with open(path, 'w', encoding='utf-8') as file:
+                file.write('# comment\n')
+                file.write('invalid-line\n')
+                file.write('=ignored\n')
+                file.write('DOUBLE="double value"\n')
+                file.write("SINGLE='single value'\n")
+
+            values = GatewayConfig._load_dotenv(path)
+
+        self.assertEqual({
+            'DOUBLE': 'double value',
+            'SINGLE': 'single value',
+        }, values)
+
 class RateLimitConfigTest(unittest.TestCase):
     """Tests for _parse_int / _parse_bool helpers and rate limit config parsing."""
 

tests/test_gateway_session_store_unit.py

@@ -118,6 +118,14 @@ class TestGatewaySessionStore(unittest.TestCase):
 
         self.assertTrue(key.startswith('custom:prefix:session:'))
 
+    def test_touch_missing_session_returns_none(self):
+        self.mock_redis.get.return_value = None
+
+        result = self.store.touch_session('GWS_missing')
+
+        self.assertIsNone(result)
+        self.mock_redis.set.assert_not_called()
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_mcp_protocol_coverage.py

@@ -229,6 +229,89 @@ class RunStdioParseErrorTest(unittest.TestCase):
         parse_err = json.loads(lines[0])
         self.assertEqual(-32700, parse_err['error']['code'])
 
+    def test_stdout_without_flush_writes_all_responses(self):
+        class WriteOnlyOutput:
+            def __init__(self):
+                self.values = []
+
+            def write(self, value):
+                self.values.append(value)
+
+        stdout = WriteOnlyOutput()
+        handler = McpProtocolHandler(None)
+
+        result = handler.run_stdio(
+            stdin=io.StringIO('bad-json\nstill-bad\n'),
+            stdout=stdout,
+        )
+
+        self.assertEqual(0, result)
+        self.assertEqual(2, len(stdout.values))
+
+
+class ProtocolResponseBoundaryTest(unittest.TestCase):
+    def test_normalize_tool_preserves_metadata_without_input_schema(self):
+        tool = {'name': 'plain_tool', 'description': 'plain'}
+
+        self.assertEqual(tool, McpProtocolHandler(None)._normalize_tool(tool))
+
+    def test_tool_call_response_rejects_non_dict(self):
+        with self.assertRaisesRegex(RuntimeError, 'invalid tool response'):
+            McpProtocolHandler._tool_call_response('1', 'invalid')
+
+    def test_success_response_wraps_non_dict_data(self):
+        response = McpProtocolHandler._tool_call_response(
+            '1',
+            {'code': 'MCP_0000', 'data': ['A']},
+        )
+
+        self.assertEqual(
+            {'data': ['A']},
+            response['result']['structuredContent'],
+        )
+
+    def test_success_response_uses_empty_structured_content_without_data(self):
+        response = McpProtocolHandler._tool_call_response(
+            '1',
+            {'code': '0', 'data': None},
+        )
+
+        self.assertEqual({}, response['result']['structuredContent'])
+        self.assertEqual('ok', response['result']['content'][0]['text'])
+
+    def test_error_response_includes_data_without_meta(self):
+        response = McpProtocolHandler._tool_call_response(
+            '1',
+            {'code': 'MCP_9001', 'msg': 'failed', 'data': ['detail']},
+        )
+
+        structured = response['result']['structuredContent']
+        self.assertEqual(['detail'], structured['data'])
+        self.assertNotIn('meta', structured)
+        self.assertTrue(response['result']['isError'])
+
+    def test_empty_table_without_tips_returns_headers(self):
+        result = McpProtocolHandler._render_text({
+            'columns': [{'key': 'number'}],
+            'records': [],
+        })
+
+        self.assertIn('1. number (number)', result)
+        self.assertNotIn('提示', result)
+
+    def test_empty_track_with_tips_returns_tip(self):
+        result = McpProtocolHandler._render_track_like_text(
+            {'tips': ['暂无轨迹']},
+            [],
+        )
+
+        self.assertEqual('提示：暂无轨迹', result)
+
+    def test_empty_track_without_tips_returns_empty_text(self):
+        result = McpProtocolHandler._render_track_like_text({}, [])
+
+        self.assertEqual('', result)
+
 
 # ---------------------------------------------------------------------------
 # handle_message / handle_request 边界情况

tests/test_public_gateway_unit.py

@@ -100,6 +100,16 @@ class TestPublicGatewayApp(unittest.TestCase):
         with self.assertRaisesRegex(RuntimeError, 'registry unavailable'):
             self.app.list_tools('GWS_test')
 
+    def test_enabled_tool_names_rejects_non_dict_response(self):
+        with self.assertRaisesRegex(RuntimeError, 'invalid enabled tool response'):
+            self.app._enabled_tool_names(None)
+
+    def test_enabled_tool_names_rejects_missing_tool_code_list(self):
+        response = {'code': 'MCP_0000', 'data': {}}
+
+        with self.assertRaisesRegex(RuntimeError, 'invalid enabled tool response'):
+            self.app._enabled_tool_names(response)
+
     def test_build_request_id_generates_id_when_empty(self):
         request_id = self.app.build_request_id('')
 
@@ -174,6 +184,33 @@ class TestPublicGatewayApp(unittest.TestCase):
         self.assertEqual(call_args['request_id'], 'rq_test')
         self.assertEqual(result['code'], '0')
 
+    def test_call_tool_succeeds_when_store_has_no_touch_method(self):
+        class ReadOnlySessionStore:
+            def get(self, gateway_session_id):
+                return {
+                    'mcp_token': 'MT_read_only',
+                    'admin_id': 1,
+                    'company_id': 2,
+                }
+
+        api_client = MagicMock()
+        api_client.list_enabled_tools.return_value = {
+            'code': 'MCP_0000',
+            'data': {'tool_codes': ['query_order']},
+        }
+        api_client.call_tool.return_value = {'code': 'MCP_0000'}
+        app = PublicGatewayApp(ReadOnlySessionStore(), api_client)
+
+        result = app.call_tool(
+            'GWS_read_only',
+            'query_order',
+            {'keyword': 'ORDER-1'},
+            'rq_read_only',
+        )
+
+        self.assertEqual('MCP_0000', result['code'])
+        api_client.call_tool.assert_called_once()
+
     def test_call_tool_rejects_dynamically_disabled_tool_before_forwarding(self):
         self.mock_session_store.get.return_value = {
             'mcp_token': 'MT_token',

tests/test_public_server_coverage.py

@@ -12,8 +12,14 @@ import socket
 import threading
 import unittest
 from http.server import HTTPServer
+from unittest.mock import MagicMock, patch
 
-from public_server import PublicMcpHttpHandler, create_http_handler, extract_client_ip
+from public_server import (
+    PublicMcpHttpHandler,
+    create_http_handler,
+    extract_client_ip,
+    serve_public,
+)
 
 
 # ---------------------------------------------------------------------------
@@ -59,6 +65,57 @@ class HandleJsonRpcUnknownMethodTest(unittest.TestCase):
         self.assertIn('error', response)
         self.assertEqual(-32601, response['error']['code'])
 
+    def test_tools_list_preserves_metadata_without_input_schema(self):
+        gateway = FakeGateway()
+        gateway.list_tools = MagicMock(return_value=[{'name': 'query_order'}])
+        handler = PublicMcpHttpHandler(gateway)
+
+        response = handler.handle_json_rpc(
+            headers={'X-Gateway-Session': 'GWS_A'},
+            message={'jsonrpc': '2.0', 'id': 1, 'method': 'tools/list'},
+        )
+
+        tool = response['result']['tools'][0]
+        self.assertEqual({'name': 'query_order'}, tool)
+        self.assertNotIn('inputSchema', tool)
+
+
+class ServePublicLifecycleTest(unittest.TestCase):
+    def test_rate_limiter_cleanup_and_keyboard_interrupt_shutdown(self):
+        class StopCleanupLoop(Exception):
+            pass
+
+        server = MagicMock()
+        server.serve_forever.side_effect = KeyboardInterrupt
+        limiter = MagicMock()
+
+        with patch('public_server.SimpleRateLimiter', return_value=limiter) as limiter_class, \
+                patch('public_server.ThreadingHTTPServer', return_value=server) as server_class, \
+                patch('public_server.threading.Thread') as thread_class:
+            serve_public(
+                FakeGateway(),
+                host='127.0.0.1',
+                port='9876',
+                rate_limit_max_requests=12,
+                rate_limit_window_seconds=34,
+            )
+
+        limiter_class.assert_called_once_with(max_requests=12, window_seconds=34)
+        self.assertEqual(('127.0.0.1', 9876), server_class.call_args.args[0])
+        thread_class.assert_called_once_with(
+            target=thread_class.call_args.kwargs['target'],
+            daemon=True,
+        )
+        thread_class.return_value.start.assert_called_once_with()
+        server.serve_forever.assert_called_once_with()
+        server.shutdown.assert_called_once_with()
+
+        cleanup_target = thread_class.call_args.kwargs['target']
+        with patch('public_server.time.sleep', side_effect=[None, StopCleanupLoop]):
+            with self.assertRaises(StopCleanupLoop):
+                cleanup_target()
+        limiter.cleanup.assert_called_once_with(max_age_seconds=34)
+
 
 # ---------------------------------------------------------------------------
 # 集成测试：启动真实 HTTP 服务器，覆盖 do_GET 404 和 do_POST JSON 错误

tests/test_query_order_exact_tool.py

@@ -95,25 +95,31 @@ class QueryOrderExactToolTest(unittest.TestCase):
 
         for phrase in (
             '单号类型不明确', '先询问用户', '不得改用其他字段',
-            '同一字段使用 IN', '不同字段使用 AND',
+            '同一字段使用 IN', '不同字段使用 AND', '排舱单号',
+            '所有单号格式均为开放格式',
+            '只能根据用户明确说出的业务类型',
+            '前缀、长度、字符组合或示例',
         ):
             self.assertIn(phrase, description)
+        self.assertNotIn('出库单号、柜号', description)
 
-        single_fields = (
+        single_number_fields = (
             'order_number', 'reference_number', 'tracking_number',
-            'outbound_number', 'container_code', 'so_number',
+            'outbound_number', 'container_code', 'so_number', 'shipment_id',
         )
         batch_fields = (
             'order_numbers', 'reference_numbers', 'tracking_numbers',
             'outbound_numbers', 'container_codes', 'so_numbers',
         )
-        for field in single_fields:
+        for field in single_number_fields:
             self.assertTrue(properties[field]['description'])
             self.assertIsInstance(properties[field]['examples'][0], str)
+            self.assertIn('用户明确', properties[field]['description'])
         for field in batch_fields:
             self.assertTrue(properties[field]['description'])
             self.assertIsInstance(properties[field]['examples'][0], list)
             self.assertGreater(len(properties[field]['examples'][0]), 1)
+            self.assertIn('用户明确', properties[field]['description'])
 
         self.assertIn('系统订单号', properties['order_number']['description'])
         self.assertIn(
@@ -121,12 +127,30 @@ class QueryOrderExactToolTest(unittest.TestCase):
             properties['reference_number']['description'],
         )
         self.assertIn('承运商', properties['tracking_number']['description'])
-        self.assertIn('出库单号', properties['outbound_number']['description'])
+        outbound_description = properties['outbound_number']['description']
+        self.assertIn('排舱单号', outbound_description)
+        self.assertIn('不是海外仓出库单号', outbound_description)
+        self.assertNotIn('以 PC 开头', outbound_description)
+        self.assertNotIn(
+            '每个号码均以 PC 开头',
+            properties['outbound_numbers']['description'],
+        )
+        so_description = properties['so_number']['description']
+        self.assertIn('格式不固定', so_description)
+        self.assertIn('用户明确', so_description)
+        self.assertEqual(
+            '97964454',
+            properties['so_number']['examples'][0],
+        )
+        self.assertEqual(
+            ['97964454', 'OOLU12345678'],
+            properties['so_numbers']['examples'][0],
+        )
         self.assertIn('柜号', properties['container_code']['description'])
-        self.assertIn('Shipping Order', properties['so_number']['description'])
+        self.assertIn('Shipping Order', so_description)
 
         for field in (
-            'shipment_id', 'receiver_country', 'product_ids', 'customer_ids',
+            'receiver_country', 'product_ids', 'customer_ids',
             'sales_id', 'warehouse_ids', 'department_id',
             'inbound_date_start', 'inbound_date_end',
             'outbound_date_start', 'outbound_date_end', 'page', 'limit',
@@ -206,6 +230,29 @@ class QueryOrderExactToolTest(unittest.TestCase):
         with self.assertRaisesRegex(RuntimeError, 'api client is required'):
             QueryOrderExactTool().call(order_number='USC001')
 
+    def test_call_rejects_non_positive_scalar_ids(self):
+        tool = QueryOrderExactTool(api_client=RecordingApiClient())
+
+        for field in ('sales_id', 'department_id'):
+            with self.subTest(field=field):
+                with self.assertRaisesRegex(ValueError, 'must be greater than 0'):
+                    tool.call(order_number='O1', **{field: 0})
+
+    def test_normalize_int_list_accepts_comma_separated_values(self):
+        self.assertEqual(
+            [9, 12],
+            QueryOrderExactTool._normalize_int_list('9, 12, 9, 0'),
+        )
+
+    def test_normalize_string_list_deduplicates_values(self):
+        self.assertEqual(
+            ['A', 'B'],
+            QueryOrderExactTool._normalize_string_list(
+                ['A', 'B', 'A'],
+                'numbers',
+            ),
+        )
+
     def test_cli_forwards_exact_fields_and_id_lists(self):
         client = RecordingApiClient()
         output = StringIO()

tests/test_token_store_coverage.py

@@ -12,6 +12,8 @@ Coverage补充测试：services/token_store.py
 """
 import io
 import json
+import os
+import tempfile
 import unittest
 from datetime import datetime, timedelta, timezone
 from unittest.mock import MagicMock, patch
@@ -89,6 +91,40 @@ class InMemoryRequireTokenTest(unittest.TestCase):
         self.assertEqual('MT_abc123', store.require_token())
 
 
+class FileTokenStoreBoundaryTest(unittest.TestCase):
+    def test_save_without_parent_directory(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            old_cwd = os.getcwd()
+            try:
+                os.chdir(tmp_dir)
+                store = FileTokenStore('token.json')
+                store.save('MT_test', '2099-01-01T00:00:00')
+                self.assertTrue(os.path.exists('token.json'))
+            finally:
+                os.chdir(old_cwd)
+
+    def test_get_reloads_when_session_is_missing_in_memory(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = os.path.join(tmp_dir, 'token.json')
+            store = FileTokenStore(path)
+            with open(path, 'w', encoding='utf-8') as handle:
+                json.dump({
+                    'token': 'MT_disk',
+                    'expire_time': '2099-01-01T00:00:00',
+                }, handle)
+
+            self.assertEqual('MT_disk', store.get()['token'])
+
+    def test_clear_when_file_is_already_missing(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = os.path.join(tmp_dir, 'missing-token.json')
+            store = FileTokenStore(path)
+
+            store.clear()
+
+            self.assertIsNone(store.get())
+
+
 # ---------------------------------------------------------------------------
 # RedisTokenStore（使用 MagicMock client 隔离真实 Redis）
 # ---------------------------------------------------------------------------
@@ -265,6 +301,19 @@ class RedisReadResponseTest(unittest.TestCase):
             self._client()._read_response(self._stream(b''))
         self.assertIn('empty redis response', str(ctx.exception))
 
+    def test_read_line_rejects_missing_crlf(self):
+        with self.assertRaisesRegex(RuntimeError, 'invalid redis line'):
+            RedisSocketClient._read_line(self._stream(b'invalid'))
+
+
+class RedisSendTest(unittest.TestCase):
+    def test_send_accepts_bytes(self):
+        sock = MagicMock()
+
+        RedisSocketClient._send(sock, b'PING')
+
+        self.assertIn(b'PING', sock.sendall.call_args.args[0])
+
 
 # ---------------------------------------------------------------------------
 # RedisSocketClient._execute（用 patch mock socket，覆盖 AUTH/SELECT 分支）

tools/query_order_exact.py

@@ -42,7 +42,7 @@ class QueryOrderExactTool:
             ['USC26070917207'],
         ),
         'order_numbers': (
-            '多个系统订单号的批量精准查询；用户给出多个订单号时使用。',
+            '多个系统订单号的批量精准查询；用户明确给出多个订单号时使用。',
             [['USC26070917207', 'USC26070917208']],
         ),
         'reference_number': (
@@ -51,43 +51,49 @@ class QueryOrderExactTool:
             ['REF-20260713-001'],
         ),
         'reference_numbers': (
-            '多个客户参考号的批量精准查询。',
+            '多个客户参考号的批量精准查询；用户明确给出多个客户参考号时使用。',
             [['REF-001', 'REF-002']],
         ),
         'tracking_number': (
-            '快递单号、物流跟踪号或承运商跟踪号码；不是系统订单号。',
+            '快递单号、物流跟踪号或承运商跟踪号码；用户明确说“快递单号”、'
+            '“物流跟踪号”或“承运商跟踪号”且只有一个值时使用；'
+            '不是系统订单号。',
             ['1Z999AA10123456784'],
         ),
         'tracking_numbers': (
-            '多个快递单号或承运商跟踪号码的批量精准查询。',
+            '多个快递单号或承运商跟踪号码的批量精准查询；'
+            '用户明确给出多个此类号码时使用。',
             [['TRACK-001', 'TRACK-002']],
         ),
         'outbound_number': (
-            '出库单号，用户明确说“出库单号”且只有一个值时使用。',
-            ['OUT-20260713-001'],
+            '排舱单号，用户明确说“排舱单号”且只有一个值时使用；'
+            '不是海外仓出库单号。',
+            ['PC20210331070002001'],
         ),
         'outbound_numbers': (
-            '多个出库单号的批量精准查询。',
-            [['OUT-001', 'OUT-002']],
+            '多个排舱单号的批量精准查询；用户明确给出多个排舱单号时使用。',
+            [['PC20210331070002001', 'PC20210331070002002']],
         ),
         'container_code': (
-            '柜号或集装箱号，只有一个值时使用。',
+            '柜号或集装箱号，用户明确说“柜号”或“集装箱号”且只有一个值时使用。',
             ['MSCU1234567'],
         ),
         'container_codes': (
-            '多个柜号或集装箱号的批量精准查询。',
+            '多个柜号或集装箱号的批量精准查询；用户明确给出多个此类号码时使用。',
             [['MSCU1234567', 'TGHU7654321']],
         ),
         'so_number': (
-            'SO号，即 Shipping Order 编号，只有一个值时使用。',
-            ['SO-20260713-001'],
+            'SO号，即 Shipping Order 编号，格式不固定；'
+            '用户明确说“SO号”且只有一个值时使用。',
+            ['97964454'],
         ),
         'so_numbers': (
-            '多个 Shipping Order 编号的批量精准查询。',
-            [['SO-001', 'SO-002']],
+            '多个格式不固定的 SO 号或 Shipping Order 编号的批量精准查询；'
+            '用户明确给出多个 SO 号时使用。',
+            [['97964454', 'OOLU12345678']],
         ),
         'shipment_id': (
-            'Shipment ID 的单值精准匹配。',
+            'Shipment ID，用户明确说“Shipment ID”且只有一个值时使用。',
             ['FBA123456789'],
         ),
         'receiver_country': (
@@ -172,8 +178,10 @@ class QueryOrderExactTool:
             'name': self.name,
             'description': (
                 '按用户明确指定的字段精准查询订单，并应用当前员工权限。'
-                '订单号、客户参考号、快递单号、出库单号、柜号和 SO 号必须'
+                '订单号、客户参考号、快递单号、排舱单号、柜号和 SO 号必须'
                 '使用各自对应字段。'
+                '所有单号格式均为开放格式，只能根据用户明确说出的业务类型'
+                '选择字段，不得根据号码的前缀、长度、字符组合或示例猜测。'
                 '单号类型不明确时先询问用户，不要调用工具猜测。'
                 '一个值使用单数字段，多个同类值使用复数字段数组；'
                 '同一字段使用 IN，不同字段使用 AND。'