from __future__ import annotations import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, patch from uuid import uuid4 from core.loop import ( AgentLoop, ReasoningGuardExhausted, ReasoningPhaseTimeout, resolve_reasoning_effort, tool_result_succeeded_for_reasoning, ) from core.probe import probe_thinking def _text_response(text: str = "ok"): return SimpleNamespace( choices=[SimpleNamespace(message=SimpleNamespace( content=text, tool_calls=None, ))], usage=None, ) def _reasoning_chunk(text: str = "thinking"): return SimpleNamespace(choices=[SimpleNamespace(delta=SimpleNamespace( reasoning_content=text, content=None, tool_calls=None, ))]) def _stream_loop(*, configured: str = "auto", cancelled=False) -> AgentLoop: loop = object.__new__(AgentLoop) loop.caps = SimpleNamespace( default_reasoning_effort=configured, reliable_context=64_000, family="deepseek_v4", variant="flash", reasoning_replay="tool_turn", native_image_input=False, ) loop.session = SimpleNamespace( messages=[{"role": "user", "content": "hello"}], task_id="task", ) loop.user_id = "user" loop.user_root = None loop.working_dir = Path(".") loop.executor = SimpleNamespace(schemas=lambda: []) loop.cancel_check = (lambda: cancelled) loop.events = [] loop._emit = loop.events.append loop._ctx_chars_per_token = 2.5 loop._last_sent_chars = 0 loop._last_had_native_images = False loop._llm_call_count = 0 loop._previous_tool_step_succeeded = None loop._reasoning_usage = {} return loop class ReasoningDecisionTests(unittest.TestCase): def test_auto_first_call_is_high(self) -> None: self.assertEqual( resolve_reasoning_effort( "auto", first_call=True, previous_tools_succeeded=None ), ("high", "first_call"), ) def test_auto_success_is_low_and_failure_is_high(self) -> None: self.assertEqual( resolve_reasoning_effort( "auto", first_call=False, previous_tools_succeeded=True ), ("low", "previous_tools_succeeded"), ) self.assertEqual( resolve_reasoning_effort( "auto", first_call=False, previous_tools_succeeded=False ), ("high", "previous_tools_failed"), ) def test_fixed_effort_is_unchanged(self) -> None: self.assertEqual( resolve_reasoning_effort( "max", first_call=False, previous_tools_succeeded=True ), ("max", "configured"), ) def test_tool_success_error_and_quality_gate_signals(self) -> None: self.assertTrue(tool_result_succeeded_for_reasoning( "read", '{}', "file content", productive=True, )) self.assertFalse(tool_result_succeeded_for_reasoning( "read", '{}', "[Error] missing", productive=False, )) self.assertFalse(tool_result_succeeded_for_reasoning( "shell", '{"command":"build"}', "created\n[产物机检 ERROR] 发现整页贴图", productive=True, )) def test_probe_resolves_auto_before_provider_call(self) -> None: efforts = [] def chat(**kwargs): efforts.append(kwargs["reasoning_effort"]) return SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace( reasoning_content="brief reasoning", content="391", ))]) caps = SimpleNamespace( thinking_enabled=True, default_reasoning_effort="auto", reasoning_effort_levels=["low", "high", "max"], ) result = probe_thinking(SimpleNamespace(chat=chat), caps) self.assertEqual(efforts, ["high"]) self.assertEqual(result.status, "ok") class ReasoningGuardTests(unittest.TestCase): def test_pure_reasoning_timeout_resets_stream(self) -> None: loop = _stream_loop() def chat_stream(**kwargs): yield _reasoning_chunk() self.assertTrue(kwargs["cancel_check"]()) loop.llm = SimpleNamespace(chat_stream=chat_stream) loop._active_reasoning_effort = "high" loop._REASONING_PHASE_TIMEOUT_S = 0 with self.assertRaises(ReasoningPhaseTimeout): loop._collect_stream_once(loop.session.messages) self.assertEqual(loop.events, [ {"type": "reasoning", "delta": "thinking"}, {"type": "reasoning_reset"}, ]) @patch("core.loop.record_agent_guard") def test_timeout_retries_once_with_low_and_ephemeral_instruction(self, guard) -> None: loop = _stream_loop() calls = [] def run_robust(**kwargs): calls.append((loop._active_reasoning_effort, kwargs["llm_messages"])) if len(calls) == 1: raise ReasoningPhaseTimeout return _text_response(), False loop._run_robust_stream = run_robust response, cancelled = loop._stream_llm() self.assertFalse(cancelled) self.assertEqual(response.choices[0].message.content, "ok") self.assertEqual([call[0] for call in calls], ["high", "low"]) self.assertEqual(len(calls[0][1]), 1) self.assertEqual(len(calls[1][1]), 2) self.assertEqual(calls[1][1][0]["role"], "system") self.assertEqual(loop.session.messages, [{"role": "user", "content": "hello"}]) self.assertEqual(loop._reasoning_usage, { "reasoning_config": "auto", "reasoning_effort": "low", "reasoning_reason": "reasoning_guard_retry", "reasoning_guard_retry": True, }) guard.assert_called_once() @patch("core.loop.record_agent_guard") def test_second_timeout_stops(self, guard) -> None: loop = _stream_loop() loop._run_robust_stream = MagicMock(side_effect=[ ReasoningPhaseTimeout(), ReasoningPhaseTimeout(), ]) with self.assertRaises(ReasoningGuardExhausted): loop._stream_llm() self.assertEqual(guard.call_count, 2) @patch("core.loop.record_agent_guard") def test_user_cancel_wins_over_guard_retry(self, guard) -> None: loop = _stream_loop(cancelled=True) loop._run_robust_stream = MagicMock(side_effect=ReasoningPhaseTimeout()) response, cancelled = loop._stream_llm() self.assertIsNone(response) self.assertTrue(cancelled) guard.assert_not_called() def test_nonstream_fallback_uses_resolved_effort(self) -> None: loop = _stream_loop() efforts = [] def chat(**kwargs): efforts.append(kwargs["reasoning_effort"]) return _text_response() loop.llm = SimpleNamespace(chat=chat) loop._active_reasoning_effort = "low" response = loop._nonstream_once(loop.session.messages) self.assertEqual(response.choices[0].message.content, "ok") self.assertEqual(efforts, ["low"]) class _Session: def __init__(self): self.task_id = uuid4() self.messages = [{"role": "user", "content": "hello"}] self.appended = [] def append(self, message, **_kwargs): self.messages.append(message) self.appended.append(message) return uuid4() class ReasoningPersistenceTests(unittest.TestCase): def test_exhausted_guard_does_not_persist_partial_assistant(self) -> None: session = _Session() loop = AgentLoop( llm=MagicMock(), executor=MagicMock(), session=session, capabilities=SimpleNamespace(max_iterations=1), user_id=uuid4(), working_dir=Path("."), ) loop._maybe_fold_context = MagicMock() loop._stream_llm = MagicMock(side_effect=ReasoningGuardExhausted()) result = loop.run_persisted_turn() self.assertEqual(result, "[stopped: reasoning timeout]") self.assertEqual(session.appended, []) self.assertEqual(loop.events if hasattr(loop, "events") else [], []) @patch("core.loop.record_chat_usage") def test_successful_chat_records_reasoning_metadata(self, record_usage) -> None: session = _Session() caps = SimpleNamespace( max_iterations=1, family="deepseek_v4", variant="flash", input_cny_per_mtoken=0, output_cny_per_mtoken=0, cache_hit_cny_per_mtoken=0, pricing={}, ) loop = AgentLoop( llm=MagicMock(), executor=MagicMock(), session=session, capabilities=caps, user_id=uuid4(), working_dir=Path("."), ) loop._maybe_fold_context = MagicMock() def stream(): loop._reasoning_usage = { "reasoning_config": "auto", "reasoning_effort": "high", "reasoning_reason": "first_call", "reasoning_guard_retry": False, } return _text_response("done"), False loop._stream_llm = MagicMock(side_effect=stream) self.assertEqual(loop.run_persisted_turn(), "done") units = record_usage.call_args.kwargs["extra_units"] self.assertEqual(units["reasoning_config"], "auto") self.assertEqual(units["reasoning_effort"], "high") self.assertEqual(units["reasoning_reason"], "first_call") self.assertFalse(units["reasoning_guard_retry"]) if __name__ == "__main__": unittest.main()