fix: quarantine invalid tool results

This commit is contained in:
hamb1y
2026-07-22 01:59:09 +08:00
committed by Xubin Ren
parent c6a4d46a2a
commit a9867a5a4e
4 changed files with 113 additions and 5 deletions
+39
View File
@@ -0,0 +1,39 @@
from nanobot.agent.context_governance import ContextGovernor
def _assistant_tool_call(call_id: str) -> dict:
return {
"role": "assistant",
"content": "",
"tool_calls": [{
"id": call_id,
"type": "function",
"function": {"name": "exec", "arguments": "{}"},
}],
}
def test_drop_orphan_tool_results_drops_missing_tool_call_id() -> None:
messages = [
_assistant_tool_call("call_1"),
{"role": "tool", "name": "exec", "content": "missing id"},
{"role": "tool", "tool_call_id": "call_1", "name": "exec", "content": "ok"},
]
result = ContextGovernor.drop_orphan_tool_results(messages)
assert [m.get("tool_call_id") for m in result if m.get("role") == "tool"] == ["call_1"]
def test_drop_orphan_tool_results_drops_duplicate_tool_result() -> None:
messages = [
_assistant_tool_call("call_1"),
{"role": "tool", "tool_call_id": "call_1", "name": "exec", "content": "first"},
{"role": "tool", "tool_call_id": "call_1", "name": "exec", "content": "duplicate"},
]
result = ContextGovernor.drop_orphan_tool_results(messages)
tool_results = [m for m in result if m.get("role") == "tool"]
assert len(tool_results) == 1
assert tool_results[0]["content"] == "first"
+55
View File
@@ -1870,3 +1870,58 @@ def test_save_turn_keeps_tool_results_declared_in_prior_history() -> None:
)
assert [m["role"] for m in session.messages] == ["assistant", "tool"]
def test_save_turn_drops_tool_result_already_fulfilled_in_history() -> None:
loop = _mk_loop()
session = Session(key="test:prior-fulfilled")
session.add_message(
"assistant",
"",
tool_calls=[{
"id": "call_prior",
"type": "function",
"function": {"name": "exec", "arguments": "{}"},
}],
)
session.add_message(
"tool",
"first",
tool_call_id="call_prior",
name="exec",
)
loop._save_turn(
session,
[{"role": "tool", "tool_call_id": "call_prior", "name": "exec", "content": "duplicate"}],
skip=0,
)
assert [m["role"] for m in session.messages] == ["assistant", "tool"]
assert session.messages[1]["content"] == "first"
def test_save_turn_drops_duplicate_tool_result_ids() -> None:
loop = _mk_loop()
session = Session(key="test:duplicate-tool-result")
loop._save_turn(
session,
[
{
"role": "assistant",
"content": "",
"tool_calls": [{
"id": "call_dupe",
"type": "function",
"function": {"name": "exec", "arguments": "{}"},
}],
},
{"role": "tool", "tool_call_id": "call_dupe", "name": "exec", "content": "first"},
{"role": "tool", "tool_call_id": "call_dupe", "name": "exec", "content": "second"},
],
skip=0,
)
assert [m["role"] for m in session.messages] == ["assistant", "tool"]
assert session.messages[1]["content"] == "first"