from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from langchain_core.tools import tool as tool_decorator from lilith_agent.app import _build_tool_node, _cooldown_limit_for, _route_after_model, build_react_agent from lilith_agent.config import Config @tool_decorator def echo_tool(text: str) -> str: """Echoes back the text.""" return f"echoed: {text}" def _ai_with_calls(calls): return AIMessage(content="", tool_calls=calls) def test_router_goes_to_tools_when_tool_calls_present(): state = {"messages": [_ai_with_calls([{"id": "1", "name": "echo_tool", "args": {"text": "hi"}}])]} assert _route_after_model(state) == "tools" def test_router_ends_when_no_tool_calls(): state = {"messages": [AIMessage(content="done")]} assert _route_after_model(state) == "extract_memory" def test_graph_returns_fail_safe_answer_when_hard_cap_hits_near_recursion_limit(monkeypatch, tmp_path, capsys): class FakeModel: def __init__(self): self.calls = 0 def bind_tools(self, tools, **kwargs): return self def invoke(self, messages): if "SYSTEM EMERGENCY OVERRIDE" in str(messages[0].content): return AIMessage(content="Final Answer: best effort answer") self.calls += 1 return _ai_with_calls([ { "id": f"call-{self.calls}", "name": "echo_tool", "args": {"text": str(self.calls)}, } ]) fake_model = FakeModel() cfg = Config.from_env() cfg.recursion_limit = 4 cfg.budget_hard_cap = 2 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg: fake_model) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: fake_model) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="answer this")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "hard-cap-test"}}, ) captured = capsys.readouterr().out assert "[route] recursion threshold reached" in captured assert "[fail_safe] emergency override" in captured assert result["messages"][-1].content == "Final Answer: best effort answer" def test_build_react_agent_prints_effective_recursion_limit(monkeypatch, tmp_path, capsys): class FakeModel: def bind_tools(self, tools, **kwargs): return self cfg = Config.from_env() cfg.recursion_limit = 50 cfg.budget_hard_cap = 25 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg: FakeModel()) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: FakeModel()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) build_react_agent(cfg) captured = capsys.readouterr().out assert "[graph] effective_recursion_limit=79 logical_recursion_limit=50 budget_hard_cap=25 headroom=4" in captured def test_model_prompt_includes_youtube_fallback_strategy(monkeypatch, tmp_path): class FakeModel: def __init__(self): self.system_prompt = "" def bind_tools(self, tools, **kwargs): return self def invoke(self, messages): if not self.system_prompt: self.system_prompt = str(messages[0].content) return AIMessage(content="Final Answer: inspected") fake_model = FakeModel() cfg = Config.from_env() cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg: fake_model) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: fake_model) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) graph.invoke( {"messages": [HumanMessage(content="What happens in https://www.youtube.com/watch?v=abcdefghijk?")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "youtube-fallback-prompt-test"}}, ) prompt = fake_model.system_prompt.lower() assert "youtube fallback strategy" in prompt assert "video id" in prompt assert "transcript" in prompt assert "do not repeatedly retry" in prompt def test_fail_safe_uses_unbound_model_to_prevent_more_tool_calls(monkeypatch, tmp_path): class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 return _ai_with_calls([ { "id": f"bound-call-{self.calls}", "name": "echo_tool", "args": {"text": str(self.calls)}, } ]) class FakeModel: def __init__(self): self.bound = FakeBoundModel() def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): return AIMessage(content="Final Answer: unbound best effort") fake_model = FakeModel() cfg = Config.from_env() cfg.recursion_limit = 4 cfg.budget_hard_cap = 1 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg: fake_model) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: fake_model) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="answer this")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "unbound-fail-safe-test"}}, ) assert result["messages"][-1].content == "Final Answer: unbound best effort" assert not getattr(result["messages"][-1], "tool_calls", None) def test_fail_safe_prompt_reinforces_original_question_contract(monkeypatch, tmp_path): class FakeBoundModel: def invoke(self, messages): return _ai_with_calls([ { "id": "bound-call", "name": "echo_tool", "args": {"text": "intermediate"}, } ]) class FakeModel: def __init__(self): self.bound = FakeBoundModel() self.fail_safe_prompt = "" def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): self.fail_safe_prompt = str(messages[0].content) return AIMessage(content="Final Answer: best effort") fake_model = FakeModel() cfg = Config.from_env() cfg.recursion_limit = 4 cfg.budget_hard_cap = 1 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg: fake_model) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: fake_model) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) graph.invoke( {"messages": [HumanMessage(content="What country corresponds to this capital?")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "fail-safe-contract-prompt-test"}}, ) prompt = fake_model.fail_safe_prompt.lower() assert "original question" in prompt assert "not an intermediate" in prompt assert "bare final answer" in prompt def test_supervisor_nudges_agent_to_answer_when_evidence_is_enough(monkeypatch, tmp_path): class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 if any("SUPERVISOR" in str(getattr(m, "content", "")) for m in messages): return AIMessage(content="Final Answer: backtick") return _ai_with_calls([ { "id": f"call-{self.calls}", "name": "echo_tool", "args": {"text": "Add `: 'For penguins\\n'"}, } ]) class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() self.supervisor_calls = 0 def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): self.supervisor_calls += 1 return AIMessage(content='{"status":"nudge","best_answer":"backtick","guidance":"You have verified the answer. Stop and answer backtick."}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 10 cfg.budget_hard_cap = 99 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Unlambda question")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "supervisor-nudge-test"}}, ) assert result["messages"][-1].content == "Final Answer: backtick" assert strong.supervisor_calls == 1 assert strong.bound.calls == 2 def test_supervisor_uses_extra_strong_model_not_cheap_model(monkeypatch, tmp_path): class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 if any("SUPERVISOR" in str(getattr(m, "content", "")) for m in messages): return AIMessage(content="Final Answer: backtick") return _ai_with_calls([ { "id": f"call-{self.calls}", "name": "echo_tool", "args": {"text": "evidence"}, } ]) class FakeExtraStrongModel: def __init__(self): self.bound = FakeBoundModel() self.supervisor_calls = 0 def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): self.supervisor_calls += 1 return AIMessage(content='{"status":"nudge","best_answer":"backtick","guidance":"Stop and answer backtick."}') strong = FakeExtraStrongModel() def cheap_should_not_be_used(cfg): raise AssertionError("supervisor should use extra strong model, not cheap model") cfg = Config.from_env() cfg.recursion_limit = 10 cfg.budget_hard_cap = 99 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", cheap_should_not_be_used) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Unlambda question")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "supervisor-extra-strong-test"}}, ) assert result["messages"][-1].content == "Final Answer: backtick" assert strong.supervisor_calls == 1 def test_supervisor_finalizer_prompt_reinforces_original_question_contract(monkeypatch, tmp_path): class FakeBoundModel: def invoke(self, messages): return _ai_with_calls([ { "id": "call", "name": "echo_tool", "args": {"text": "evidence"}, } ]) class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() self.finalizer_prompt = "" def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): prompt = str(messages[0].content) if "SUPERVISOR FINALIZER" in prompt: self.finalizer_prompt = prompt return AIMessage(content="Final Answer: final") return AIMessage(content='{"status":"finalize","best_answer":"","guidance":"Existing evidence is enough."}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 10 cfg.budget_hard_cap = 99 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) graph.invoke( {"messages": [HumanMessage(content="What final entity answers the original question?")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "supervisor-finalizer-contract-prompt-test"}}, ) prompt = strong.finalizer_prompt.lower() assert "original question" in prompt assert "not an intermediate" in prompt assert "bare final answer" in prompt def test_supervisor_finalizer_rejects_unknown_best_answer_and_forces_best_guess(monkeypatch, tmp_path): class FakeBoundModel: def invoke(self, messages): return _ai_with_calls([ { "id": "call", "name": "echo_tool", "args": {"text": "evidence"}, } ]) class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() self.finalizer_calls = 0 def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): prompt = str(messages[0].content) if "SUPERVISOR FINALIZER" in prompt: self.finalizer_calls += 1 return AIMessage(content="Final Answer: best guess") return AIMessage(content='{"status":"finalize","best_answer":"unknown","guidance":"Looping; make a best guess."}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 10 cfg.budget_hard_cap = 99 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Unlambda question")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "supervisor-unknown-best-answer-test"}}, ) assert result["messages"][-1].content == "Final Answer: best guess" assert strong.finalizer_calls == 1 def test_supervisor_finalizes_even_with_placeholder_if_requested(monkeypatch, tmp_path): class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 return _ai_with_calls([ { "id": f"call-{self.calls}", "name": "echo_tool", "args": {"text": "partial evidence"}, } ]) class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() self.supervisor_calls = 0 self.finalizer_calls = 0 def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): prompt = str(messages[0].content) if "SUPERVISOR FINALIZER" in prompt: self.finalizer_calls += 1 return AIMessage(content="Final Answer: finalizer output") self.supervisor_calls += 1 if self.supervisor_calls == 1: return AIMessage(content='{"status":"nudge","best_answer":"","guidance":"Use the evidence to make a best guess."}') # On second call, it asks to finalize but with a placeholder answer. The code should allow finalization. return AIMessage(content='{"status":"finalize","best_answer":"Unknown","guidance":"You were already nudged. Provide your final answer based on the best available information."}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 12 cfg.budget_hard_cap = 99 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Question requiring a concrete answer")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "supervisor-finalize-after-nudge-test"}}, ) assert result["messages"][-1].content == "Final Answer: finalizer output" assert strong.finalizer_calls == 1 assert strong.supervisor_calls == 2 def test_supervisor_finalizer_forbids_tool_calls_to_prevent_markup_leak(monkeypatch, tmp_path): # DeepSeek thinking-off still emits raw tool-call markup when invoked unbound # against a tool-laden history. The finalizer must bind tools with # tool_choice="none" so the model is forced to answer in plain text. dsml_leak = '<｜｜DSML｜｜tool_calls><｜｜DSML｜｜invoke name="web_search">' class FakeBoundModel: def invoke(self, messages): return _ai_with_calls([ {"id": "call", "name": "echo_tool", "args": {"text": "evidence"}} ]) class FakeFinalizerBound: def invoke(self, messages): return AIMessage(content="Final Answer: synthesized report") class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() self.finalizer_tool_choice = None def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": self.finalizer_tool_choice = "none" return FakeFinalizerBound() return self.bound def invoke(self, messages): prompt = str(messages[0].content) if "SUPERVISOR FINALIZER" in prompt: # Reached only if the finalizer wrongly invokes the unbound model. return AIMessage(content=dsml_leak) return AIMessage(content='{"status":"finalize","best_answer":"","guidance":"Evidence is enough."}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 10 cfg.budget_hard_cap = 99 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Research the competitor landscape.")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "supervisor-finalizer-markup-leak-test"}}, ) final = result["messages"][-1].content assert "DSML" not in final assert final == "Final Answer: synthesized report" assert strong.finalizer_tool_choice == "none" def test_supervisor_forces_finalize_after_max_nudges(monkeypatch, tmp_path): class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 return _ai_with_calls([ { "id": f"call-{self.calls}", "name": "echo_tool", "args": {"text": "partial evidence"}, } ]) class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() self.supervisor_calls = 0 self.finalizer_calls = 0 def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): prompt = str(messages[0].content) if "SUPERVISOR FINALIZER" in prompt: self.finalizer_calls += 1 return AIMessage(content="Final Answer: max nudges forced this") self.supervisor_calls += 1 return AIMessage(content='{"status":"nudge","best_answer":"concrete candidate","guidance":"Check one more constraint before final answer."}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 20 cfg.budget_hard_cap = 99 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MAX_NUDGES", 5, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Question requiring more checking")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "supervisor-max-nudges-test"}}, ) assert strong.finalizer_calls == 0 assert result["messages"][-1].content == "Final Answer: concrete candidate" assert strong.supervisor_calls == 5 assert result["supervisor_decision"] == "finalize" def test_final_answer_gets_supervisor_review_and_can_be_returned_for_revision(monkeypatch, tmp_path): class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 if any("FINAL ANSWER REVIEW FAILED" in str(getattr(m, "content", "")) for m in messages): return AIMessage(content="Final Answer: corrected answer") return AIMessage(content="Final Answer: wrong answer") class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() self.review_calls = 0 def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): prompt = str(messages[0].content) if "FINAL ANSWER REVIEW" in prompt: self.review_calls += 1 if self.review_calls == 1: return AIMessage(content='{"status":"nudge","best_answer":"","guidance":"The proposed answer violates a constraint; fix it."}') return AIMessage(content='{"status":"finalize","best_answer":"corrected answer","guidance":"Approved."}') return AIMessage(content='{"status":"continue","best_answer":"","guidance":""}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 10 cfg.budget_hard_cap = 99 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: []) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Question with constraint")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "final-answer-review-revision-test"}}, ) assert strong.review_calls == 2 assert "Final Answer: corrected answer" in result["messages"][-1].content assert all("wrong answer" != getattr(m, "content", "") for m in result["messages"][-1:]) def test_fail_safe_falls_back_to_supervisor_best_answer_when_empty(monkeypatch, tmp_path): """fail_safe model returning empty content must not propagate; supervisor_best_answer wins.""" class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 return _ai_with_calls([ { "id": f"call-{self.calls}", "name": "echo_tool", "args": {"text": str(self.calls)}, } ]) class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() self.fail_safe_calls = 0 self.supervisor_calls = 0 def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): prompt = str(messages[0].content) if "EMERGENCY OVERRIDE" in prompt: self.fail_safe_calls += 1 return AIMessage(content="") self.supervisor_calls += 1 return AIMessage(content='{"status":"nudge","best_answer":"backtick","guidance":"Stop and answer backtick."}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 4 cfg.budget_hard_cap = 2 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Question")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "fail-safe-best-answer-test"}}, ) last = result["messages"][-1] assert last.content assert "Final Answer:" in last.content assert "backtick" in last.content def test_fail_safe_never_returns_empty_answer_without_best_answer(monkeypatch, tmp_path): """No supervisor_best_answer, fail_safe model empty: still produce non-empty descriptive Final Answer.""" class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 return _ai_with_calls([ { "id": f"call-{self.calls}", "name": "echo_tool", "args": {"text": str(self.calls)}, } ]) class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): return AIMessage(content="") strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 4 cfg.budget_hard_cap = 2 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 99, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Question")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "fail-safe-default-answer-test"}}, ) last = result["messages"][-1] content = str(getattr(last, "content", "")) assert content.strip(), "fail_safe must never propagate an empty answer" assert "Final Answer:" in content def test_supervisor_review_auto_approves_after_fail_safe(monkeypatch, tmp_path): """fail_safe -> supervisor_review must auto-approve (no infinite loop back to model).""" class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 return _ai_with_calls([ { "id": f"call-{self.calls}", "name": "echo_tool", "args": {"text": str(self.calls)}, } ]) class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() self.review_calls = 0 def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): prompt = str(messages[0].content) if "EMERGENCY OVERRIDE" in prompt: return AIMessage(content="Final Answer: best effort") if "FINAL ANSWER REVIEW" in prompt: self.review_calls += 1 return AIMessage(content='{"status":"nudge","best_answer":"","guidance":"reject"}') return AIMessage(content='{"status":"continue","best_answer":"","guidance":""}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 4 cfg.budget_hard_cap = 2 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 99, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Question")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "fail-safe-review-no-loop-test"}}, ) last = result["messages"][-1] assert "Final Answer: best effort" in str(getattr(last, "content", "")) assert strong.review_calls == 0 def test_tool_node_invokes_tool_successfully(): node = _build_tool_node([echo_tool]) state = {"messages": [ HumanMessage(content="say hi"), _ai_with_calls([{"id": "1", "name": "echo_tool", "args": {"text": "hi"}}]), ]} out = node(state) assert len(out["messages"]) == 1 msg = out["messages"][0] assert isinstance(msg, ToolMessage) assert msg.tool_call_id == "1" assert "echoed: hi" in msg.content def test_supervisor_finalizes_when_agent_ignores_prior_nudge(monkeypatch, tmp_path): class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 sup_count = sum( 1 for m in messages if "SUPERVISOR:" in str(getattr(m, "content", "")) ) if sup_count >= 2: return AIMessage(content="Final Answer: backtick") return _ai_with_calls([ { "id": f"call-{self.calls}", "name": "echo_tool", "args": {"text": str(self.calls)}, } ]) class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() self.supervisor_calls = 0 def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): self.supervisor_calls += 1 return AIMessage(content='{"status":"nudge","best_answer":"backtick","guidance":"Stop. Existing evidence supports backtick."}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 12 cfg.budget_hard_cap = 99 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: object()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="Unlambda question")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "supervisor-finalize-test"}}, ) assert result["messages"][-1].content == "Final Answer: backtick" assert strong.supervisor_calls == 2 assert strong.bound.calls == 3 def test_supervisor_overhead_leaves_room_for_hard_cap_fail_safe(monkeypatch, tmp_path): class FakeBoundModel: def __init__(self): self.calls = 0 def invoke(self, messages): self.calls += 1 return _ai_with_calls([ { "id": f"call-{self.calls}", "name": "echo_tool", "args": {"text": str(self.calls)}, } ]) class FakeStrongModel: def __init__(self): self.bound = FakeBoundModel() def bind_tools(self, tools, **kwargs): if kwargs.get("tool_choice") == "none": return self return self.bound def invoke(self, messages): return AIMessage(content="Final Answer: hard cap fallback") class FakeSupervisorModel: def invoke(self, messages): return AIMessage(content='{"status":"continue"}') strong = FakeStrongModel() cfg = Config.from_env() cfg.recursion_limit = 8 cfg.budget_hard_cap = 5 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg, thinking=True: strong) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: FakeSupervisorModel()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="force hard cap")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "supervisor-hard-cap-headroom-test"}}, ) assert result["messages"][-1].content == "Final Answer: hard cap fallback" def test_supervisor_overhead_leaves_room_for_iteration_fail_safe(monkeypatch, tmp_path): class FakeBoundModel: def invoke(self, messages): return _ai_with_calls([ { "id": "call", "name": "echo_tool", "args": {"text": "loop"}, } ]) class FakeStrongModel: def bind_tools(self, tools): return FakeBoundModel() def invoke(self, messages): return AIMessage(content="Final Answer: iteration fallback") class FakeSupervisorModel: def invoke(self, messages): return AIMessage(content='{"status":"continue"}') cfg = Config.from_env() cfg.recursion_limit = 5 cfg.budget_hard_cap = 99 cfg.budget_warn_at = 99 cfg.compact_summarize = False monkeypatch.setenv("LILITH_HOME", str(tmp_path / ".lilith")) monkeypatch.setattr("lilith_agent.app._SUPERVISOR_MIN_TOOL_CALLS", 1, raising=False) monkeypatch.setattr("lilith_agent.app.get_strong_model", lambda cfg: FakeStrongModel()) monkeypatch.setattr("lilith_agent.app.get_cheap_model", lambda cfg: FakeSupervisorModel()) monkeypatch.setattr("lilith_agent.tools.build_tools", lambda cfg: [echo_tool]) monkeypatch.setattr("lilith_agent.memory.extract_and_compress_facts", lambda messages, model: None) graph = build_react_agent(cfg) result = graph.invoke( {"messages": [HumanMessage(content="force iteration cap")], "iterations": 0, "todos": []}, {"configurable": {"thread_id": "supervisor-iteration-headroom-test"}}, ) assert result["messages"][-1].content == "Final Answer: iteration fallback" def test_tool_node_invokes_tool_and_returns_tool_message(): node = _build_tool_node([echo_tool]) state = {"messages": [ HumanMessage(content="say hi"), _ai_with_calls([{"id": "1", "name": "echo_tool", "args": {"text": "hi"}}]), ]} out = node(state) assert len(out["messages"]) == 1 msg = out["messages"][0] assert isinstance(msg, ToolMessage) assert msg.tool_call_id == "1" assert "echoed: hi" in msg.content def test_tool_node_dedups_repeat_tool_call_without_invoking(): calls = 0 @tool_decorator def counting_tool(x: str) -> str: """Counting tool.""" nonlocal calls calls += 1 return f"ran {calls}" node = _build_tool_node([counting_tool]) # History: earlier AI message already called counting_tool(x="a") prior_call = {"id": "old", "name": "counting_tool", "args": {"x": "a"}} prior_ai = _ai_with_calls([prior_call]) prior_result = ToolMessage(tool_call_id="old", name="counting_tool", content="ran 0") # Now a new AI message asks for the same tool with the same args. new_call = {"id": "new", "name": "counting_tool", "args": {"x": "a"}} state = {"messages": [ HumanMessage(content="go"), prior_ai, prior_result, _ai_with_calls([new_call]), ]} out = node(state) assert calls == 0, "deduped call must not invoke the tool again" msg = out["messages"][0] assert isinstance(msg, ToolMessage) assert msg.tool_call_id == "new" assert "already called" in msg.content.lower() def test_tool_node_prints_semantic_dedup_queries(capsys): node = _build_tool_node([echo_tool], semantic_dedup_threshold=0.5) prior_call = {"id": "old", "name": "web_search", "args": {"query": "apollo moon landing transcript"}} new_call = {"id": "new", "name": "web_search", "args": {"query": "moon landing apollo transcript"}} state = {"messages": [ HumanMessage(content="go"), _ai_with_calls([prior_call]), ToolMessage(tool_call_id="old", name="web_search", content="done"), _ai_with_calls([new_call]), ]} out = node(state) printed = capsys.readouterr().out assert "[tools] semantic_dedup score=1.00 tool=web_search" in printed assert "query='moon landing apollo transcript'" in printed assert "prior_query='apollo moon landing transcript'" in printed assert "REDUNDANT SEARCH PATH" in out["messages"][0].content def test_tool_node_handles_unknown_tool_name(): node = _build_tool_node([echo_tool]) state = {"messages": [_ai_with_calls([{"id": "1", "name": "ghost", "args": {}}])]} out = node(state) msg = out["messages"][0] assert isinstance(msg, ToolMessage) assert "unknown tool" in msg.content.lower() def test_dedup_does_not_emit_warning_level_logs(caplog): """Routine dedup fires on many turns; WARNING floods stderr during normal runs. Regression guard: `[dedup]`, `[semantic_dedup]`, `[loop_breaker]` stay at INFO.""" import logging node = _build_tool_node([echo_tool]) prior_call = {"id": "old", "name": "echo_tool", "args": {"text": "x"}} state = {"messages": [ HumanMessage(content="go"), _ai_with_calls([prior_call]), ToolMessage(tool_call_id="old", name="echo_tool", content="done"), _ai_with_calls([{"id": "new", "name": "echo_tool", "args": {"text": "x"}}]), ]} with caplog.at_level(logging.DEBUG, logger="lilith_agent.app"): node(state) for rec in caplog.records: if "[dedup]" in rec.getMessage() or "[semantic_dedup]" in rec.getMessage() or "[loop_breaker]" in rec.getMessage(): assert rec.levelno < logging.WARNING, f"routine guard log at {rec.levelname}: {rec.message}" def test_cooldown_limit_for_known_tool_is_positive_int(): """Each tool must declare a positive cooldown limit. Regression guard against the `3 if name == 'web_search' else 3` no-op ternary.""" limit = _cooldown_limit_for("web_search") assert isinstance(limit, int) and limit > 0 assert _cooldown_limit_for("fetch_url") == _cooldown_limit_for("web_search") def test_tool_node_catches_tool_exceptions_and_feeds_back(): @tool_decorator def boom_tool(x: str) -> str: """Always raises.""" raise RuntimeError("kaboom") node = _build_tool_node([boom_tool]) state = {"messages": [_ai_with_calls([{"id": "1", "name": "boom_tool", "args": {"x": "y"}}])]} out = node(state) msg = out["messages"][0] assert isinstance(msg, ToolMessage) assert "kaboom" in msg.content @tool_decorator def todo_sentinel_tool(action: str) -> str: """Returns todo sentinel output.""" if action == "write": return "SET_TODOS: ['first', 'second']" return "DONE_TODO: 0" def test_tool_node_consumes_todo_sentinels_into_state(): node = _build_tool_node([todo_sentinel_tool]) write_out = node({ "messages": [_ai_with_calls([{"id": "1", "name": "todo_sentinel_tool", "args": {"action": "write"}}])], "todos": [], }) assert write_out["todos"] == ["first", "second"] done_out = node({ "messages": [_ai_with_calls([{"id": "2", "name": "todo_sentinel_tool", "args": {"action": "done"}}])], "todos": write_out["todos"], }) assert done_out["todos"] == ["second"]