Babak Alizadeh
0c4079d765
Replace regex/LLM shims for continue/stop decisions with structural evidence (API tool_calls, Harmony parser, ObservationStore). Add DB regression fixtures and remove unused aux.agent_continuation prompt. Co-authored-by: Cursor <cursoragent@cursor.com>
91 lines
3 KiB
Python
91 lines
3 KiB
Python
"""Tests for Harmony token sanitization and trace finalization."""
|
|
from app.services.ai.ai_content_sanitize import (
|
|
extract_leaked_function_calls,
|
|
prepare_assistant_content_for_persist,
|
|
resolve_round_function_calls,
|
|
sanitize_assistant_content,
|
|
)
|
|
from app.services.ai.ai_trace import finalize_trace_steps_for_persist
|
|
|
|
|
|
_HARMONY_SAMPLE = (
|
|
"در ابتدا بازهٔ زمانی «۳ ماه گذشته» را مشخص میکنم."
|
|
"<|start|>assistant<|channel|>commentary to=functions.resolve_date_range "
|
|
"<|constrain|>json<|message|>{"
|
|
' "relative": "last_3_months",'
|
|
' "calendar_type": "jalali"'
|
|
"}<|call|>"
|
|
)
|
|
|
|
|
|
def test_sanitize_removes_harmony_tool_leak():
|
|
cleaned = sanitize_assistant_content(_HARMONY_SAMPLE)
|
|
assert "<|start|>" not in cleaned
|
|
assert "<|channel|>" not in cleaned
|
|
assert "resolve_date_range" not in cleaned
|
|
assert "۳ ماه گذشته" in cleaned
|
|
|
|
|
|
def test_extract_leaked_function_calls():
|
|
calls = extract_leaked_function_calls(_HARMONY_SAMPLE)
|
|
assert len(calls) == 1
|
|
assert calls[0]["name"] == "resolve_date_range"
|
|
assert calls[0]["arguments"]["relative"] == "last_3_months"
|
|
assert calls[0]["arguments"]["calendar_type"] == "jalali"
|
|
|
|
|
|
def test_resolve_round_does_not_infer_from_natural_language():
|
|
text = "We will call resolve_date_range then get_financial_summary."
|
|
resolved = resolve_round_function_calls(
|
|
api_function_calls=None,
|
|
round_text=text,
|
|
round_reasoning="",
|
|
)
|
|
assert resolved is None
|
|
|
|
|
|
def test_resolve_round_prefers_api_calls():
|
|
api = [{"id": "x", "name": "get_report", "arguments": {}}]
|
|
resolved = resolve_round_function_calls(
|
|
api_function_calls=api,
|
|
round_text="will call search_persons",
|
|
round_reasoning="",
|
|
)
|
|
assert resolved == api
|
|
|
|
|
|
def test_finalize_trace_closes_active_steps():
|
|
trace = [
|
|
{
|
|
"step_id": "ctx_thinking",
|
|
"kind": "context",
|
|
"state": "active",
|
|
"body_markdown": _HARMONY_SAMPLE,
|
|
},
|
|
{"step_id": "answer_1", "kind": "answer", "state": "done", "body_markdown": "ok"},
|
|
]
|
|
finalized = finalize_trace_steps_for_persist(trace)
|
|
assert finalized[0]["state"] == "done"
|
|
assert "<|start|>" not in finalized[0]["body_markdown"]
|
|
assert finalized[1]["state"] == "done"
|
|
|
|
|
|
def test_prepare_assistant_content_for_persist_uses_sanitized_trace():
|
|
trace = [
|
|
{
|
|
"kind": "answer",
|
|
"body_markdown": _HARMONY_SAMPLE,
|
|
}
|
|
]
|
|
content = prepare_assistant_content_for_persist("", trace)
|
|
assert "<|start|>" not in content
|
|
assert "۳ ماه گذشته" in content
|
|
|
|
|
|
def test_merge_trace_into_function_results_finalizes_active():
|
|
from app.services.ai.ai_trace import merge_trace_into_function_results
|
|
|
|
trace = [{"step_id": "ctx_thinking", "kind": "context", "state": "active"}]
|
|
merged = merge_trace_into_function_results({}, trace)
|
|
stored = merged["_agent_trace"][0]
|
|
assert stored["state"] == "done"
|