From 5acc86d7cc519f1dbf1f3a6200908625805da568 Mon Sep 17 00:00:00 2001 From: krypticmouse Date: Tue, 26 May 2026 03:22:14 +0000 Subject: [PATCH] =?UTF-8?q?fix(server):=20managed-agent=20streaming=20pari?= =?UTF-8?q?ty=20=E2=80=94=20tool=5Fcalls=20replay,=20sampler=20params,=20t?= =?UTF-8?q?ool=20DI=20(#382,=20#386,=20#395)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_stream_managed_agent` had diverged from the canonical cli/ask.py path and lost three behaviours. All three are fixed via small extracted, unit-tested helpers: - #382: cross-request history replay dropped stored `tool_calls`, so the model never saw its own prior tool use and fabricated tool output on turn 2+. `_replay_history_messages` now reconstructs the assistant tool-use message plus matching tool-result messages (synthesised, consistent tool_call_ids). - #386: only temperature/max_tokens reached the engine. `_sampler_kwargs` forwards repetition_penalty / top_p / top_k / min_p / frequency_penalty / presence_penalty when set in the agent config (opt-in; default agents send nothing extra). Fixes degenerate repetition loops on local models with no repetition_penalty. - #395: tools were built with a bare `tool_cls()`, so memory_* / channel_* / llm tools loaded with no backend and failed on every call. `_instantiate_managed_tool` injects backend / channel / engine the same way cli/ask.py::_build_tools does. Verified empirically: replay emits user → assistant(tool_calls) → tool(result) with matching ids; sampler extraction forwards only set keys; DI gives memory tools a backend and llm the engine/model. New tests in tests/server/test_managed_agent_streaming.py (helpers are pure, so verifiable without a live engine). 38 passed locally incl. existing route tests. Closes #382 Closes #386 Closes #395 Co-Authored-By: Claude Opus 4.7 (1M context) --- src/openjarvis/server/agent_manager_routes.py | 161 +++++++++++++++++- tests/server/test_managed_agent_streaming.py | 131 ++++++++++++++ 2 files changed, 283 insertions(+), 9 deletions(-) create mode 100644 tests/server/test_managed_agent_streaming.py diff --git a/src/openjarvis/server/agent_manager_routes.py b/src/openjarvis/server/agent_manager_routes.py index 5aaea179..64c6ec83 100644 --- a/src/openjarvis/server/agent_manager_routes.py +++ b/src/openjarvis/server/agent_manager_routes.py @@ -398,6 +398,139 @@ def _resolve_tool_specs( return resolved +# Per-agent sampler params forwarded to the engine when present in config. +# The OpenAI-compat engine passes **kwargs straight through to the upstream +# payload, so these reach local servers (vLLM / mlx_lm / etc.) that support +# them. Only forwarded when explicitly set, so default agents send nothing +# extra and engines that don't support a key never receive it. (#386) +_SAMPLER_PARAM_KEYS = ( + "top_p", + "top_k", + "min_p", + "repetition_penalty", + "frequency_penalty", + "presence_penalty", +) + + +def _sampler_kwargs(config: Dict[str, Any]) -> Dict[str, Any]: + """Extract per-agent sampler params from a managed agent's config (#386).""" + out: Dict[str, Any] = {} + for key in _SAMPLER_PARAM_KEYS: + val = config.get(key) + if val is not None: + out[key] = val + return out + + +def _replay_history_messages( + history: List[Dict[str, Any]], + exclude_id: str, +) -> List[Any]: + """Rebuild prior-turn LLM messages from stored managed-agent history. + + When a stored assistant turn recorded ``tool_calls``, replay them as an + assistant tool-use message followed by the corresponding tool-result + messages — so the model sees its own prior tool pattern instead of + regressing to fabricated tool output on later turns. Without this, only + the assistant's text is replayed and the tool-use signal is lost (#382). + """ + from openjarvis.core.types import Message, Role, ToolCall + + messages: List[Any] = [] + for m in reversed(history): + if m.get("id") == exclude_id: + continue + direction = m.get("direction") + if direction == "user_to_agent": + messages.append(Message(role=Role.USER, content=m.get("content") or "")) + elif direction == "agent_to_user": + stored = m.get("tool_calls") + if stored: + calls = [] + results = [] + for i, tc in enumerate(stored): + call_id = f"hist-{m.get('id', '')}-{i}" + calls.append( + ToolCall( + id=call_id, + name=tc.get("tool", ""), + arguments=tc.get("arguments") or "", + ) + ) + results.append( + Message( + role=Role.TOOL, + content=str(tc.get("result", "")), + tool_call_id=call_id, + name=tc.get("tool", ""), + ) + ) + messages.append( + Message( + role=Role.ASSISTANT, + content=m.get("content") or None, + tool_calls=calls, + ) + ) + messages.extend(results) + else: + messages.append( + Message(role=Role.ASSISTANT, content=m.get("content") or "") + ) + return messages + + +def _instantiate_managed_tool( + tool_cls: Any, + name: str, + *, + engine: Any, + model: str, + app_state: Any, +) -> Any: + """Instantiate a tool with the same dependency injection as the canonical + ``cli/ask.py`` path, so ``memory_*`` / ``channel_*`` / ``llm`` tools work + instead of silently failing with "No backend configured" (#395). + """ + try: + from openjarvis.cli.ask import ( + _CHANNEL_TOOLS, + _MEMORY_TOOLS, + _get_memory_backend, + ) + except Exception: # pragma: no cover - cli import should always succeed + _MEMORY_TOOLS = frozenset() + _CHANNEL_TOOLS = frozenset() + _get_memory_backend = None + + if name in _MEMORY_TOOLS: + backend = getattr(app_state, "memory_backend", None) if app_state else None + if backend is None and app_state is not None and _get_memory_backend: + cfg = getattr(app_state, "config", None) + if cfg is not None: + backend = _get_memory_backend(cfg) + if backend is None: + logger.warning( + "Memory tool %r instantiated without a backend — calls will " + "return no results.", + name, + ) + return tool_cls(backend=backend) + if name in _CHANNEL_TOOLS: + channel = getattr(app_state, "channel_bridge", None) if app_state else None + if channel is None: + logger.warning( + "Channel tool %r instantiated without a channel — calls will " + "fail with 'No channel backend configured'.", + name, + ) + return tool_cls(channel=channel) + if name == "llm": + return tool_cls(engine=engine, model=model) + return tool_cls() + + def _build_deep_research_tools( engine: Any, model: str, @@ -657,15 +790,11 @@ async def _stream_managed_agent( if app_state is not None and dr_tools: app_state._dr_tools = dr_tools - # Load prior conversation context (DESC order, reverse for chronological) + # Load prior conversation context (DESC order, reverse for chronological). + # Replaying recorded tool_calls (assistant tool-use + tool results) keeps + # multi-turn tool behaviour from regressing to fabricated output (#382). history = manager.list_messages(agent_id, limit=50) - for m in reversed(history): - if m["id"] == message_id: - continue - if m["direction"] == "user_to_agent": - llm_messages.append(Message(role=Role.USER, content=m["content"])) - elif m["direction"] == "agent_to_user": - llm_messages.append(Message(role=Role.ASSISTANT, content=m["content"])) + llm_messages.extend(_replay_history_messages(history, message_id)) # Append the current user message llm_messages.append(Message(role=Role.USER, content=user_content)) @@ -960,6 +1089,10 @@ async def _stream_managed_agent( if resolved_tools: stream_kwargs["tools"] = resolved_tools + # Forward any per-agent sampler params (repetition_penalty, top_p, …) so + # locally-hosted models can be tuned per agent (#386). + stream_kwargs.update(_sampler_kwargs(config)) + # Discover MCP tools and merge into stream_kwargs mcp_adapters: Dict[str, Any] = {} if app_state is not None: @@ -1175,7 +1308,17 @@ async def _stream_managed_agent( tool_cls = ToolRegistry.get(tool_name) if tool_cls is not None: - tool_instance = tool_cls() + # Inject backend / channel / engine the same + # way cli/ask.py does, else memory_* / channel_* + # / llm tools fail with "No backend configured" + # on every call (#395). + tool_instance = _instantiate_managed_tool( + tool_cls, + tool_name, + engine=engine, + model=model, + app_state=app_state, + ) # Tools the user explicitly added to this # agent's toolkit are considered pre-approved — # selecting them in the wizard is the diff --git a/tests/server/test_managed_agent_streaming.py b/tests/server/test_managed_agent_streaming.py new file mode 100644 index 00000000..cf57047b --- /dev/null +++ b/tests/server/test_managed_agent_streaming.py @@ -0,0 +1,131 @@ +"""Managed-agent streaming parity tests (#382, #386, #395). + +These exercise the pure helpers extracted from ``_stream_managed_agent`` so the +streaming path's history replay, sampler forwarding, and tool dependency +injection can be verified without a live engine. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +pytest.importorskip("fastapi") # agent_manager_routes imports FastAPI at module load + +from openjarvis.core.types import Role # noqa: E402 +from openjarvis.server.agent_manager_routes import ( # noqa: E402 + _instantiate_managed_tool, + _replay_history_messages, + _sampler_kwargs, +) + + +class TestReplayHistoryToolCalls: + """#382 — stored tool_calls must be replayed, not dropped.""" + + def test_assistant_tool_calls_are_replayed_with_results(self): + history = [ # DESC order (newest first), as list_messages returns + {"id": "m2", "direction": "agent_to_user", "content": "", "tool_calls": [ + {"tool": "shell_exec", "arguments": '{"command":"pwd"}', + "result": "/home/u", "success": True, "latency": 1.0}, + ]}, + {"id": "m1", "direction": "user_to_agent", "content": "run pwd", + "tool_calls": None}, + ] + msgs = _replay_history_messages(history, exclude_id="current") + # Chronological: user, assistant(tool_calls), tool(result) + assert [m.role for m in msgs] == [Role.USER, Role.ASSISTANT, Role.TOOL] + assistant = msgs[1] + assert assistant.tool_calls is not None + assert assistant.tool_calls[0].name == "shell_exec" + tool_msg = msgs[2] + assert tool_msg.content == "/home/u" + # The tool result must reference the assistant's tool_call id. + assert tool_msg.tool_call_id == assistant.tool_calls[0].id + + def test_plain_assistant_without_tool_calls(self): + history = [ + {"id": "m1", "direction": "agent_to_user", "content": "hi", + "tool_calls": None}, + ] + msgs = _replay_history_messages(history, exclude_id="current") + assert len(msgs) == 1 + assert msgs[0].role == Role.ASSISTANT + assert msgs[0].tool_calls is None + + def test_excludes_current_message(self): + history = [ + {"id": "cur", "direction": "user_to_agent", "content": "now", + "tool_calls": None}, + {"id": "old", "direction": "user_to_agent", "content": "before", + "tool_calls": None}, + ] + msgs = _replay_history_messages(history, exclude_id="cur") + assert [m.content for m in msgs] == ["before"] + + +class TestSamplerKwargs: + """#386 — sampler params forwarded only when set.""" + + def test_reads_present_keys(self): + cfg = {"temperature": 0.7, "repetition_penalty": 1.1, "top_p": 0.9, + "top_k": 40, "min_p": 0.05, "frequency_penalty": 0.2, + "presence_penalty": 0.1} + out = _sampler_kwargs(cfg) + assert out == {"repetition_penalty": 1.1, "top_p": 0.9, "top_k": 40, + "min_p": 0.05, "frequency_penalty": 0.2, + "presence_penalty": 0.1} + # temperature/max_tokens are handled separately, not here. + assert "temperature" not in out + + def test_omits_absent_keys(self): + assert _sampler_kwargs({"temperature": 0.7}) == {} + + def test_none_values_skipped(self): + assert _sampler_kwargs({"top_p": None, "repetition_penalty": 1.05}) == { + "repetition_penalty": 1.05 + } + + +class _FakeTool: + def __init__(self, **kwargs): + self.kwargs = kwargs + + +class TestInstantiateManagedTool: + """#395 — tools get their dependencies injected.""" + + def test_memory_tool_gets_backend(self): + backend = object() + app_state = SimpleNamespace(memory_backend=backend, config=None, + channel_bridge=None) + tool = _instantiate_managed_tool(_FakeTool, "memory_store", + engine=object(), model="m", + app_state=app_state) + assert tool.kwargs == {"backend": backend} + + def test_llm_tool_gets_engine_and_model(self): + engine = object() + app_state = SimpleNamespace(memory_backend=None, config=None, + channel_bridge=None) + tool = _instantiate_managed_tool(_FakeTool, "llm", engine=engine, + model="qwen", app_state=app_state) + assert tool.kwargs == {"engine": engine, "model": "qwen"} + + def test_channel_tool_gets_channel(self): + bridge = object() + app_state = SimpleNamespace(memory_backend=None, config=None, + channel_bridge=bridge) + tool = _instantiate_managed_tool(_FakeTool, "channel_send", + engine=object(), model="m", + app_state=app_state) + assert tool.kwargs == {"channel": bridge} + + def test_plain_tool_gets_no_injection(self): + app_state = SimpleNamespace(memory_backend=None, config=None, + channel_bridge=None) + tool = _instantiate_managed_tool(_FakeTool, "calculator", + engine=object(), model="m", + app_state=app_state) + assert tool.kwargs == {}