diff --git a/src/agent_engine/engine/langgraph/nodes.py b/src/agent_engine/engine/langgraph/nodes.py index bfc39ddf..4ffd7dc8 100644 --- a/src/agent_engine/engine/langgraph/nodes.py +++ b/src/agent_engine/engine/langgraph/nodes.py @@ -17,7 +17,7 @@ from langchain_core.language_models import BaseChatModel from langchain_core.tools import BaseTool, StructuredTool from langgraph.errors import GraphInterrupt -from pydantic import BaseModel +from pydantic import BaseModel, Field, ValidationError from agent_engine.approvals.coordinator import ApprovalCoordinator from agent_engine.approvals.invocation import ToolInvocation @@ -33,7 +33,7 @@ run_tool_loop, ) from agent_engine.loaders.resolver_loader import ResolverLoader -from agent_engine.logging_config import log, safe_error +from agent_engine.logging_config import log, safe_error, validation_problems from agent_engine.runtime.execution import ( ExecutionLimitExceeded, blocked_message, @@ -69,7 +69,16 @@ class _AgentCall(BaseModel): """Input schema for a child-agent tool.""" - message: str + # Every child agent takes exactly this one argument, so describing it once + # here is what keeps an orchestrator from inventing a shape (a dict of + # fields, a nested object) that can only fail validation. + message: str = Field( + description=( + "The complete request for this agent, as a single plain-text string. " + "Not an object or a list: everything the agent needs — including any " + "context it cannot see — goes into the text itself." + ) + ) @dataclass(frozen=True) @@ -130,6 +139,30 @@ def _elapsed_ms(start: float) -> int: return int((time.perf_counter() - start) * 1000) +def _invalid_args_message(subject: str, exc: ValidationError) -> str: + """Tell the model which fields it got wrong, so it can correct the call. + + Built from the structured errors rather than ``str(exc)``: pydantic echoes + the rejected ``input_value`` back in its own message, which is the caller's + arguments and may hold credentials or personal data. + """ + problems = validation_problems(exc) + return f"Invalid arguments for '{subject}': {problems}. Correct them and call it again." + + +def _child_failure_message(subject: str, exc: Exception) -> str: + """Report a child-agent failure back to the calling model, cause included. + + A model can only correct a call whose failure it can read, so the reason is + passed through rather than replaced by a generic notice — the same contract + a tool failure already has with its caller. ``safe_error`` bounds it and + keeps a schema rejection from echoing the arguments back into the context. + """ + if isinstance(exc, ValidationError): + return _invalid_args_message(subject, exc) + return f"Agent '{subject}' failed to complete this request: {safe_error(exc)}" + + class AgentNode: """Callable that implements one agent turn inside a LangGraph node. @@ -345,6 +378,11 @@ async def _record_error( """Record a failed call, fire ``on_tool_error``, and return the error text. The failure is returned (not raised) so the model can read it and recover. + A tool's own exception text is the developer's message to the model and + is passed through; a schema rejection is reworded so the model learns + which field it got wrong without its arguments being echoed back — the + same rewording is used for the trace/hook copy of the error, which a + pydantic ``str(exc)`` would otherwise leak its rejected input through. """ error = safe_error(exc) used_tools.append(self._usage(call, "failed", error=error)) @@ -354,10 +392,15 @@ async def _record_error( current_run_context.get(), self._call_context(call, "failed", latency_ms, error=error), ) + model_error = ( + _invalid_args_message(call.name, exc) + if isinstance(exc, ValidationError) + else f"Tool error: {exc}" + ) await self._execution_manager.finish_execution( - call.exec_id, status="failed", result=f"Tool error: {exc}" + call.exec_id, status="failed", result=model_error ) - return f"Tool error: {exc}" + return model_error async def _record_success( self, @@ -597,10 +640,7 @@ async def invoke(message: str) -> str: error=type(exc).__name__, message=exc, ) - return ( - f"Agent '{entry.id}' failed to complete this request. " - "This is an internal error, not a problem with the arguments." - ) + return _child_failure_message(entry.id, exc) finally: current_streams.reset(sink_token) @@ -680,10 +720,7 @@ async def invoke_tool(tc: dict[str, Any]) -> str: error=type(exc).__name__, message=exc, ) - return ( - f"Agent '{tc['name']}' failed to complete this request. " - "This is an internal error, not a problem with the arguments." - ) + return _child_failure_message(tc["name"], exc) response = await run_tool_loop(bound_model, messages, state, self._node_path, invoke_tool) answer = as_text(response.content) diff --git a/tests/engine/test_engine_flow.py b/tests/engine/test_engine_flow.py index 891cb979..c8c75cf7 100644 --- a/tests/engine/test_engine_flow.py +++ b/tests/engine/test_engine_flow.py @@ -157,6 +157,32 @@ def _user_text(messages: list[Any]) -> str: return "" +def _logged_fields(caplog: pytest.LogCaptureFixture) -> str: + """Every structured field value emitted during the capture, as one string.""" + return "\n".join(str(getattr(r, "fields", "")) for r in caplog.records) + + +class BadArgsChatModel(FakeChatModel): + """Calls its first tool with args that fail the tool's own schema.""" + + BAD_ARG_VALUE = "secret-arg-value" + + def _respond(self, messages: list[Any]) -> AIMessage: + _record_tool_results(messages) + already_called = any(isinstance(m, ToolMessage) for m in messages) + if self._tool_names and not already_called: + call = ToolCall( + name=self._select(messages), + args={"message": {"leaked": self.BAD_ARG_VALUE}}, + id="call_1", + ) + return AIMessage(content="", tool_calls=[call]) + return AIMessage(content=self._answer) + + def bind_tools(self, tools: list[Any]) -> BadArgsChatModel: + return BadArgsChatModel(self._answer, [t.name for t in tools]) + + @pytest.fixture def model_factory() -> Callable[[str, str, float | None], BaseChatModel]: def factory(provider: str, name: str, temperature: float | None) -> BaseChatModel: @@ -164,6 +190,8 @@ def factory(provider: str, name: str, temperature: float | None) -> BaseChatMode return cast(BaseChatModel, FailingChatModel()) if name == "successful-fallback": return cast(BaseChatModel, FakeChatModel(answer="recovered ok")) + if name == "bad-args": + return cast(BaseChatModel, BadArgsChatModel()) return cast(BaseChatModel, FakeChatModel()) return factory @@ -286,12 +314,42 @@ async def test_orchestrator_routes_to_matching_child(tmp_path: Path, model_facto assert result.visited == ["root", "root/super"] +async def test_orchestrator_recovers_from_invalid_child_tool_args( + tmp_path: Path, model_factory: Any, caplog: pytest.LogCaptureFixture +) -> None: + # Regression: a malformed child-agent tool call (args failing _AgentCall's + # schema) must not crash the run. The model is told which field it got + # wrong so it can correct the call, but never has its own arguments echoed + # back at it; the untrimmed detail goes to the server log instead. + SEEN_TOOL_RESULTS.clear() + bad_args_model = ModelConfig(provider="fake", name="bad-args", temperature=None) + spec = system(orchestrator("root", [agent("flights")], model=bad_args_model)) + + with caplog.at_level(logging.WARNING): + result = await run_message(spec, tmp_path, model_factory, "please handle flights") + + assert result.answer == "ok" + assert result.visited == ["root"] + + returned = "\n".join(SEEN_TOOL_RESULTS) + assert "Invalid arguments for 'flights'" in returned + assert "message: Input should be a valid string" in returned + assert BadArgsChatModel.BAD_ARG_VALUE not in returned + + record = next(r for r in caplog.records if r.getMessage() == "child agent call failed") + assert getattr(record, "fields", {})["error"] == "ValidationError" + # The rejected arguments must not reach the log either — not through this + # record, and not through the trace callback's own tool-error line. + assert BadArgsChatModel.BAD_ARG_VALUE not in str(getattr(record, "fields", {})) + assert BadArgsChatModel.BAD_ARG_VALUE not in _logged_fields(caplog) + + async def test_orchestrator_recovers_from_child_agent_crash( tmp_path: Path, model_factory: Any, caplog: pytest.LogCaptureFixture ) -> None: - # Regression: a child agent's own run raising (its model is down) must not - # crash the whole orchestrator run — the orchestrator still answers, and - # the model gets a generic notice instead of the run dying with a 500. + # The child's own run raises (its model is down) rather than its arguments + # being rejected: the orchestrator still answers, and the cause reaches the + # model so it can decide whether the call is worth correcting. SEEN_TOOL_RESULTS.clear() down = ModelConfig(provider="fake", name="failing-primary", temperature=None) spec = system(orchestrator("root", [agent("flights", model=down)])) @@ -301,8 +359,8 @@ async def test_orchestrator_recovers_from_child_agent_crash( assert result.answer == "ok" returned = "\n".join(SEEN_TOOL_RESULTS) - assert "Agent 'flights' failed to complete this request." in returned - assert "not a problem with the arguments" in returned + assert "Agent 'flights' failed to complete this request" in returned + assert "Model execution failed" in returned # the cause, not a generic notice record = next(r for r in caplog.records if r.getMessage() == "child agent call failed") assert getattr(record, "fields", {})["error"] == "RuntimeError"