270 lines
9.4 KiB
Python
270 lines
9.4 KiB
Python
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()
|