优化订单查询提示词规则，提升单测覆盖率100%

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。'