"""Order references in a tool turn resolve against its pending list, never an older page.""" import json import unittest from unittest.mock import patch import test_model_chat as fixtures from test_model_tool_loop import evidence, final_facts, tool_plan from test_quotes import frames from logistics_chat import db, order_data def order_arguments(question, action='followup', condition='all'): return {'question': question, 'action': action, 'condition': condition, 'stage': '', 'status': '', 'reference': ''} class ModelOrderLoopTests(unittest.TestCase): def setUp(self): fixtures.ModelChatTests.setUp(self) fixtures.ModelChatTests.login(self) def ask(self, question): events = frames(self.client.post('/stream_chat', json={'msg': question, 'conv_id': self.cid})) self.assertTrue(events[-1].get('d'), events) self.assertNotIn('SECRET', json.dumps(events, ensure_ascii=False)) return events def saved(self): with self.app.app_context(): connection = db.connect() try: rows = connection.execute('SELECT role,content FROM messages WHERE conv_id=? ORDER BY id', (self.cid,)).fetchall() state = connection.execute('SELECT order_list_state FROM chat_history WHERE id=?', (self.cid,)).fetchone()[0] return rows, json.loads(state) if state else None finally: connection.close() def add_orders(self, count): with self.app.app_context(): connection = db.connect() try: connection.execute("UPDATE order_test_data SET source_updated_at='2026-11-01 12:00:00' WHERE tenant_id='a'") order_data.import_records(connection, [ {'tenant_id': 'a', 'order_no': f'NEWNORMAL{index:08d}', 'inbound_code': f'NEWINBOUND{index:08d}', 'status': 'COMPLETED', 'source_updated_at': '2026-11-06 12:00:00'} for index in range(1, count + 1)]) connection.commit() finally: connection.close() def test_new_list_is_available_to_followup_before_any_state_is_saved(self): before = self.saved() calls = 0 def provider(body, **kwargs): nonlocal calls calls += 1 self.assertEqual(self.saved(), before) if calls == 1: return tool_plan('query_orders', order_arguments('列出我的异常订单', 'list', 'exception'), 'new_list') if calls == 2: self.assertIn('INBOUND20260001', evidence(body)[0]['result']) return tool_plan('query_orders', order_arguments('第一个订单的轨迹'), 'first_order') facts = evidence(body) self.assertIn('VISIBLE20260001', facts[1]['result']) self.assertNotIn('请先查看', facts[1]['result']) return final_facts('fact_0', 'fact_1') with patch('logistics_chat.ai_client.post_completion', side_effect=provider): events = self.ask('列出我的异常订单，再看第一票的轨迹') self.assertEqual(calls, 3) self.assertIn('VISIBLE20260001', events[0]['t']) rows, state = self.saved() self.assertEqual(len(rows), len(before[0]) + 2) self.assertTrue(state['query']['exception']) self.assertEqual(state['selected_index'], 1) self.assertEqual(state['shown'][0]['order_no'], 'VISIBLE20260001') def test_current_filter_replaces_an_older_persisted_list_for_ordinal_lookup(self): self.add_orders(2) with patch('logistics_chat.ai_client.post_completion', side_effect=[ tool_plan('query_orders', order_arguments('列出我的订单', 'list'), 'old_list'), final_facts('fact_0')]): self.ask('先列出全部订单') old_state = self.saved()[1] self.assertEqual(old_state['shown'][0]['order_no'], 'NEWNORMAL00000002') calls = 0 def provider(body, **kwargs): nonlocal calls calls += 1 if calls == 1: return tool_plan('query_orders', order_arguments('列出我的异常订单', 'list', 'exception'), 'exceptions') if calls == 2: self.assertEqual(self.saved()[1], old_state) return tool_plan('query_orders', order_arguments('第一个订单的轨迹'), 'current_first') current = evidence(body)[1]['result'] self.assertIn('VISIBLE20260001', current) self.assertNotIn('NEWNORMAL00000002', current) return final_facts('fact_0', 'fact_1') with patch('logistics_chat.ai_client.post_completion', side_effect=provider): self.ask('改看异常订单列表，再查其中第一票') self.assertEqual(self.saved()[1]['shown'][0]['order_no'], 'VISIBLE20260001') def test_repeated_next_page_calls_advance_the_pending_page_instead_of_reusing_cache(self): self.add_orders(45) before = self.saved() self.app.config['MODEL_MAX_TOOL_ROUNDS'] = 3 self.app.config['MODEL_MAX_EVIDENCE_CHARS'] = 50000 calls = 0 def provider(body, **kwargs): nonlocal calls calls += 1 self.assertEqual(self.saved(), before) if calls == 1: return tool_plan('query_orders', order_arguments('列出我的订单', 'list'), 'page_one') if calls == 2: self.assertIn('第1页', evidence(body)[-1]['result']) return tool_plan('query_orders', order_arguments('下一页'), 'page_two') if calls == 3: self.assertIn('第2页', evidence(body)[-1]['result']) return tool_plan('query_orders', order_arguments('下一页'), 'page_three') self.assertIn('第3页', evidence(body)[-1]['result']) return final_facts('fact_0', 'fact_1', 'fact_2') with patch('logistics_chat.ai_client.post_completion', side_effect=provider): self.ask('列出订单，接着看第二页和第三页') self.assertEqual(calls, 4) state = self.saved()[1] self.assertEqual(state['page'], 3) self.assertEqual(len(state['shown']), 6) def test_selected_index_advances_within_the_turn_and_remains_available_next_turn(self): self.add_orders(2) calls = 0 def provider(body, **kwargs): nonlocal calls calls += 1 if calls == 1: return tool_plan('query_orders', order_arguments('列出我的订单', 'list'), 'orders') if calls == 2: return tool_plan('query_orders', order_arguments('第一个订单的轨迹'), 'first') if calls == 3: self.assertIn('NEWNORMAL00000002', evidence(body)[-1]['result']) return tool_plan('query_orders', order_arguments('下一票的轨迹'), 'next') self.assertIn('NEWNORMAL00000001', evidence(body)[-1]['result']) return final_facts('fact_0', 'fact_1', 'fact_2') with patch('logistics_chat.ai_client.post_completion', side_effect=provider): self.ask('列出全部订单，依次查第一票和下一票的轨迹') self.assertEqual(self.saved()[1]['selected_index'], 2) with patch('logistics_chat.ai_client.post_completion', side_effect=[ tool_plan('query_orders', order_arguments('下一票的轨迹'), 'third'), final_facts('fact_0')]): events = self.ask('再看下一票') serialized = json.dumps(events, ensure_ascii=False) self.assertIn('VISIBLE20260001', serialized) self.assertEqual(self.saved()[1]['selected_index'], 3) def test_list_then_followup_in_one_batch_uses_the_new_candidate(self): calls = [tool_plan('query_orders', order_arguments('列出我的异常订单', 'list', 'exception'), 'batch_list').message['tool_calls'][0], tool_plan('query_orders', order_arguments('第一个订单的轨迹'), 'batch_followup').message['tool_calls'][0]] batch = fixtures.Completion({'role': 'assistant', 'content': '', 'tool_calls': calls}, 'tool_calls') with patch('logistics_chat.ai_client.post_completion', side_effect=[batch, final_facts('fact_0', 'fact_1')]) as model: events = self.ask('列异常订单并查询第一票的轨迹') second_fact = evidence(model.call_args_list[1].args[0])[1] self.assertIn('VISIBLE20260001', second_fact['result']) self.assertNotIn('请先查看', events[0]['t']) if __name__ == '__main__': unittest.main()