Compare commits

...
2 Commits
Author SHA1 Message Date
Elliot Slusky b3c57468ae fix(cli): honor enabled tools when serving (#737)
* fix(cli): honor enabled tools when serving

* fix(server): preserve tools for streaming agents
2026-08-13 17:03:51 -07:00
Elliot Slusky ff69797135 fix(web): normalize tool call arguments (#738)
* fix(web): normalize tool call arguments

* fix(web): preserve chats when repair writeback fails
2026-08-13 17:03:10 -07:00
14 changed files with 527 additions and 51 deletions
+3 -2
View File
@@ -5,6 +5,7 @@ import { useAppStore, generateId } from '../../lib/store';
import { streamChat, streamResearch } from '../../lib/sse';
import { fetchSavings, getBase } from '../../lib/api';
import { listConnectors, getSyncStatus } from '../../lib/connectors-api';
import { serializeToolCallArguments } from '../../lib/tool-call';
import { MicButton } from './MicButton';
import { useSpeech } from '../../hooks/useSpeech';
import type {
@@ -389,7 +390,7 @@ export function InputArea() {
const tc: ToolCallInfo = {
id: generateId(),
tool: data.tool,
arguments: data.arguments || '',
arguments: serializeToolCallArguments(data.arguments),
status: 'running',
};
toolCalls.push(tc);
@@ -400,7 +401,7 @@ export function InputArea() {
updateLastAssistant(convId, accumulatedContent, [...toolCalls]);
useAppStore.getState().addLogEntry({
timestamp: Date.now(), level: 'info', category: 'tool',
message: `Calling ${data.tool}(${data.arguments || ''})`,
message: `Calling ${data.tool}(${serializeToolCallArguments(data.arguments)})`,
});
} catch {}
} else if (eventName === 'tool_call_end') {
@@ -1,6 +1,7 @@
import { useState } from 'react';
import { ChevronDown, ChevronRight, Loader2, CheckCircle2, XCircle } from 'lucide-react';
import type { ToolCallInfo } from '../../types';
import { serializeToolCallArguments } from '../../lib/tool-call';
interface Props {
toolCall: ToolCallInfo;
@@ -35,7 +36,10 @@ export function ToolCallCard({ toolCall }: Props) {
const [expanded, setExpanded] = useState(false);
const config = statusConfig[toolCall.status];
const StatusIcon = config.icon;
const preview = previewArgs(toolCall.arguments);
// Persisted conversations may contain the pre-fix object payload despite
// the TypeScript contract, so normalize again at the final render boundary.
const argumentsText = serializeToolCallArguments(toolCall.arguments);
const preview = previewArgs(argumentsText);
return (
<div
@@ -95,7 +99,7 @@ export function ToolCallCard({ toolCall }: Props) {
className="px-2.5 pb-2 pt-0.5"
style={{ borderTop: '1px solid var(--color-border-subtle, var(--color-border))' }}
>
{toolCall.arguments && (
{argumentsText && (
<div className="mt-1.5">
<div
style={{
@@ -120,7 +124,7 @@ export function ToolCallCard({ toolCall }: Props) {
wordBreak: 'break-all',
}}
>
{formatJson(toolCall.arguments)}
{formatJson(argumentsText)}
</pre>
</div>
)}
+2 -1
View File
@@ -1,5 +1,6 @@
import type { ModelInfo, SavingsData, ServerInfo } from '../types';
import { SUPABASE_ANON_KEY, SUPABASE_URL } from './supabase';
import { serializeToolCallArguments } from './tool-call';
// ---------------------------------------------------------------------------
// Supabase config
@@ -741,7 +742,7 @@ export async function sendAgentMessage(
const parsed = JSON.parse(data);
callbacks?.onToolCallStart?.({
tool: parsed.tool,
arguments: parsed.arguments ?? '',
arguments: serializeToolCallArguments(parsed.arguments),
});
} catch {
/* skip */
@@ -0,0 +1,122 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
const CONVERSATIONS_KEY = 'openjarvis-conversations';
class MemoryStorage {
private store = new Map<string, string>();
getItem(key: string): string | null {
return this.store.get(key) ?? null;
}
setItem(key: string, value: string): void {
this.store.set(key, String(value));
}
removeItem(key: string): void {
this.store.delete(key);
}
}
beforeEach(() => {
vi.resetModules();
(globalThis as unknown as { localStorage: MemoryStorage }).localStorage =
new MemoryStorage();
});
afterEach(() => {
(globalThis as unknown as { localStorage?: MemoryStorage }).localStorage =
undefined;
});
describe('persisted tool calls', () => {
it('repairs parsed argument objects while loading conversations', async () => {
localStorage.setItem(
CONVERSATIONS_KEY,
JSON.stringify({
version: 1,
activeId: 'conversation-1',
conversations: {
'conversation-1': {
id: 'conversation-1',
title: 'Broken chat',
createdAt: 1,
updatedAt: 1,
model: 'test-model',
messages: [
{
id: 'assistant-1',
role: 'assistant',
content: '',
timestamp: 1,
toolCalls: [
{
id: 'call-1',
tool: 'web_search',
arguments: { query: 'python' },
status: 'success',
},
],
},
],
},
},
}),
);
const { useAppStore } = await import('./store');
expect(useAppStore.getState().messages[0].toolCalls?.[0].arguments).toBe(
'{"query":"python"}',
);
const repaired = JSON.parse(localStorage.getItem(CONVERSATIONS_KEY) ?? '{}');
expect(
repaired.conversations['conversation-1'].messages[0].toolCalls[0].arguments,
).toBe('{"query":"python"}');
});
it('keeps repaired conversations in memory when writeback fails', async () => {
localStorage.setItem(
CONVERSATIONS_KEY,
JSON.stringify({
version: 1,
activeId: 'conversation-1',
conversations: {
'conversation-1': {
id: 'conversation-1',
title: 'Readable chat',
createdAt: 1,
updatedAt: 1,
model: 'test-model',
messages: [
{
id: 'assistant-1',
role: 'assistant',
content: '',
timestamp: 1,
toolCalls: [
{
id: 'call-1',
tool: 'web_search',
arguments: { query: 'python' },
status: 'success',
},
],
},
],
},
},
}),
);
vi.spyOn(localStorage, 'setItem').mockImplementation(() => {
throw new DOMException('Storage quota exceeded', 'QuotaExceededError');
});
const { useAppStore } = await import('./store');
expect(useAppStore.getState().messages).toHaveLength(1);
expect(useAppStore.getState().messages[0].toolCalls?.[0].arguments).toBe(
'{"query":"python"}',
);
});
});
+25 -1
View File
@@ -16,6 +16,7 @@ import type {
} from '../types';
import type { ManagedAgent } from './api';
import { isEmbedOnlyModel } from './model-capabilities';
import { serializeToolCallArguments } from './tool-call';
export interface CachedConnector {
connector_id: string;
@@ -55,7 +56,30 @@ function loadConversations(): ConversationStore {
const raw = localStorage.getItem(CONVERSATIONS_KEY);
if (!raw) return { version: 1, conversations: {}, activeId: null };
const parsed = JSON.parse(raw);
if (parsed.version === 1) return parsed;
if (parsed.version === 1) {
let repaired = false;
for (const conversation of Object.values(parsed.conversations ?? {}) as Conversation[]) {
for (const message of conversation.messages ?? []) {
for (const toolCall of message.toolCalls ?? []) {
const argumentsText = serializeToolCallArguments(toolCall.arguments);
if (argumentsText !== toolCall.arguments) {
toolCall.arguments = argumentsText;
repaired = true;
}
}
}
}
if (repaired) {
try {
localStorage.setItem(CONVERSATIONS_KEY, JSON.stringify(parsed));
} catch {
// Keep the repaired conversations usable in memory when storage is
// read-only or full. A failed best-effort writeback must not make
// otherwise readable conversation history disappear from the UI.
}
}
return parsed;
}
return { version: 1, conversations: {}, activeId: null };
} catch {
return { version: 1, conversations: {}, activeId: null };
+22
View File
@@ -0,0 +1,22 @@
import { describe, expect, it } from 'vitest';
import { serializeToolCallArguments } from './tool-call';
describe('serializeToolCallArguments', () => {
it('preserves JSON strings', () => {
expect(serializeToolCallArguments('{"query":"python"}')).toBe(
'{"query":"python"}',
);
});
it('serializes parsed argument objects', () => {
expect(serializeToolCallArguments({ query: 'python' })).toBe(
'{"query":"python"}',
);
});
it('uses an empty string for missing arguments', () => {
expect(serializeToolCallArguments(null)).toBe('');
expect(serializeToolCallArguments(undefined)).toBe('');
});
});
+11
View File
@@ -0,0 +1,11 @@
/** Convert tool-call arguments from API or persisted data into display-safe text. */
export function serializeToolCallArguments(value: unknown): string {
if (typeof value === 'string') return value;
if (value == null) return '';
try {
return JSON.stringify(value) ?? String(value);
} catch {
return String(value);
}
}
+29 -35
View File
@@ -25,6 +25,30 @@ from openjarvis.intelligence import (
logger = logging.getLogger(__name__)
_DEFAULT_TOOLS = frozenset({"think", "calculator", "web_search"})
def _resolve_allowed_tools(config: object) -> tuple[set[str], bool]:
"""Return configured tool names and whether the selection was explicit.
``tools.enabled`` is the canonical setting used by ``SystemBuilder`` and
the interactive CLI. ``agent.tools`` remains as a backward-compatible
fallback, followed by the server's default tool set when neither is set.
"""
configured = config.tools.enabled or config.agent.tools
if not configured:
return set(_DEFAULT_TOOLS), False
if isinstance(configured, list):
allowed = {
tool.strip()
for tool in configured
if isinstance(tool, str) and tool.strip()
}
else:
allowed = {tool.strip() for tool in configured.split(",") if tool.strip()}
return allowed, True
def _unique_model_ids(model_ids: list[str]) -> list[str]:
"""Return model ids in first-seen order without duplicates."""
@@ -96,7 +120,7 @@ def _resolve_server_model(
"--agent",
"agent_name",
default=None,
help="Agent for non-streaming requests (simple, orchestrator, react, openhands).",
help="Agent for chat requests (simple, orchestrator, react, openhands).",
)
@click.pass_context
def serve(
@@ -305,21 +329,7 @@ def serve(
from openjarvis.core.registry import ToolRegistry
from openjarvis.tools._stubs import BaseTool
_DEFAULT_TOOLS = {"think", "calculator", "web_search"}
configured = config.agent.tools
if configured:
if isinstance(configured, list):
allowed = {
t.strip()
for t in configured
if isinstance(t, str) and t.strip()
}
else:
allowed = {
t.strip() for t in configured.split(",") if t.strip()
}
else:
allowed = _DEFAULT_TOOLS
allowed, tools_configured = _resolve_allowed_tools(config)
tools = []
for name in ToolRegistry.keys():
@@ -336,7 +346,7 @@ def serve(
# MCP server tools from config.tools.mcp.servers
# (#461 — these were silently dropped).
mcp_tools = managed_mcp_tools
if configured:
if tools_configured:
mcp_tools = [
tool
for tool in managed_mcp_tools
@@ -406,23 +416,7 @@ def serve(
from openjarvis.core.registry import ToolRegistry
from openjarvis.tools._stubs import BaseTool
_DEFAULT_TOOLS = {"think", "calculator", "web_search"}
configured = config.agent.tools
if configured:
if isinstance(configured, list):
_allowed = {
t.strip()
for t in configured
if isinstance(t, str) and t.strip()
}
else:
_allowed = {
t.strip()
for t in configured.split(",")
if t.strip()
}
else:
_allowed = _DEFAULT_TOOLS
_allowed, _tools_configured = _resolve_allowed_tools(config)
for _tname in ToolRegistry.keys():
if _tname not in _allowed:
@@ -436,7 +430,7 @@ def serve(
# Reuse the process-owned MCP pool so channels do not
# open a second transport to every configured server.
_ch_mcp_tools = managed_mcp_tools
if configured:
if _tools_configured:
_ch_mcp_tools = [
tool
for tool in managed_mcp_tools
+128 -9
View File
@@ -200,12 +200,14 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
# When the client passes `tools`, stream the model's raw
# OpenAI-compat function-calling decision directly from the engine
# (bypassing the agent) — the streaming mirror of the non-streaming
# #454 fix. Routing tools through the agent stream bridge ignored
# `request_body.tools`, ran the agent's own tool loop, and
# word-split generic filler content into fake token deltas, so the
# caller's tool_calls were dropped entirely (the streaming analog of
# #414). For plain chat (no tools), stream token-by-token directly
# from the engine for true real-time output.
# #454 fix. Routing client-supplied tools through a server-side agent
# would execute the agent's different tool set and drop the raw tool
# call the caller expects (#414).
#
# Without client-supplied tools, keep streaming requests on the
# configured server agent so its server-side tool loop is available
# to the desktop UI and other stream:true clients (#735). Fall back to
# direct token streaming when no tool-bearing agent is configured.
if request_body.tools:
return await _handle_stream_tools(
engine,
@@ -216,6 +218,16 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
bus=getattr(request.app.state, "bus", None),
memory_service=getattr(request.app.state, "memory_service", None),
)
if agent is not None and getattr(agent, "_tools", None):
return await _handle_agent_stream(
agent,
model,
request_body,
complexity_info,
trace_store=getattr(request.app.state, "trace_store", None),
bus=getattr(request.app.state, "bus", None),
memory_service=getattr(request.app.state, "memory_service", None),
)
return await _handle_stream(
engine,
model,
@@ -547,6 +559,114 @@ def _handle_agent(
)
async def _handle_agent_stream(
agent,
model: str,
req: ChatCompletionRequest,
complexity_info=None,
*,
trace_store=None,
bus=None,
memory_service=None,
):
"""Run the configured agent and return its result as an SSE response.
Agents own the tool-execution loop, which is synchronous today. Run that
loop in a worker thread and stream its final answer once complete. This
keeps ``stream:true`` clients (including the desktop UI) on the same agent
and configured toolkit as non-streaming requests instead of bypassing the
agent and silently dropping server-side tools.
Requests that explicitly supply OpenAI ``tools`` continue to use
``_handle_stream_tools`` so their raw tool-call deltas are preserved.
"""
chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
query_text = ""
for message in reversed(req.messages):
if message.role == "user" and message.content:
query_text = message.content
break
async def generate():
first_chunk = ChatCompletionChunk(
id=chunk_id,
model=model,
choices=[StreamChoice(delta=DeltaMessage(role="assistant"))],
)
yield f"data: {first_chunk.model_dump_json()}\n\n"
try:
response = await asyncio.to_thread(
_handle_agent,
agent,
model,
req,
complexity_info,
trace_store=trace_store,
bus=bus,
)
except Exception as exc:
logging.getLogger("openjarvis.server").error(
"Agent stream error: %s",
exc,
exc_info=True,
)
error_chunk = ChatCompletionChunk(
id=chunk_id,
model=model,
choices=[
StreamChoice(
delta=DeltaMessage(
content=f"Sorry, an error occurred: {exc}",
),
finish_reason="stop",
)
],
)
yield f"data: {error_chunk.model_dump_json()}\n\n"
yield "data: [DONE]\n\n"
return
content = _response_content(response)
if content:
content_chunk = ChatCompletionChunk(
id=chunk_id,
model=model,
choices=[StreamChoice(delta=DeltaMessage(content=content))],
)
yield f"data: {content_chunk.model_dump_json()}\n\n"
import json as _json
finish_chunk = ChatCompletionChunk(
id=chunk_id,
model=model,
choices=[
StreamChoice(delta=DeltaMessage(), finish_reason="stop"),
],
)
finish_data = _json.loads(finish_chunk.model_dump_json())
finish_data["usage"] = response.usage.model_dump()
if complexity_info is not None:
finish_data["complexity"] = complexity_info.model_dump()
yield f"data: {_json.dumps(finish_data)}\n\n"
_record_completed_exchange(
memory_service,
query_text,
content,
bus=bus,
source="server.chat.stream",
)
yield "data: [DONE]\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
)
async def _handle_stream_tools(
engine,
model: str,
@@ -690,11 +810,10 @@ async def _handle_stream(
):
"""Stream response using SSE format.
This path streams straight from the engine, bypassing the agent /
This no-agent fallback streams straight from the engine, bypassing the
``TraceCollector``. When *trace_store* is set we accumulate the streamed
tokens and record a minimal ``Trace`` once the stream completes
successfully — otherwise streamed chats (the desktop GUI's main path)
would never populate ``traces.db``.
successfully.
"""
import time
+6
View File
@@ -110,6 +110,12 @@ class AgentStreamBridge:
def _format_named_event(self, name: str, data: dict) -> str:
"""Format an SSE event with an explicit ``event:`` field."""
if name == "tool_call_start" and not isinstance(data.get("arguments"), str):
# The in-process event bus uses parsed arguments for trace/eval
# consumers, while the web SSE contract expects their JSON text.
# Copy before normalizing so other subscribers keep the object.
data = dict(data)
data["arguments"] = json.dumps(data.get("arguments"))
return f"event: {name}\ndata: {json.dumps(data)}\n\n"
def _run_agent(self) -> object:
+53
View File
@@ -0,0 +1,53 @@
"""Regression tests for tool selection during ``jarvis serve`` startup."""
from __future__ import annotations
import pytest
from openjarvis.cli.serve import _resolve_allowed_tools
from openjarvis.core.config import JarvisConfig
@pytest.mark.parametrize(
"configured",
[
"code_interpreter,file_read",
["code_interpreter", "file_read"],
],
)
def test_tools_enabled_is_used_by_serve(configured):
config = JarvisConfig()
config.tools.enabled = configured
allowed, explicit = _resolve_allowed_tools(config)
assert allowed == {"code_interpreter", "file_read"}
assert explicit is True
def test_tools_enabled_takes_precedence_over_legacy_agent_tools():
config = JarvisConfig()
config.tools.enabled = "file_read"
config.agent.tools = "calculator"
allowed, explicit = _resolve_allowed_tools(config)
assert allowed == {"file_read"}
assert explicit is True
def test_agent_tools_remains_a_backward_compatible_fallback():
config = JarvisConfig()
config.agent.tools = "file_read"
allowed, explicit = _resolve_allowed_tools(config)
assert allowed == {"file_read"}
assert explicit is True
def test_serve_defaults_tools_when_no_selection_is_configured():
allowed, explicit = _resolve_allowed_tools(JarvisConfig())
assert allowed == {"think", "calculator", "web_search"}
assert explicit is False
+1
View File
@@ -213,6 +213,7 @@ class TestStreamingResilience:
engine = _make_engine()
agent = MagicMock()
agent.agent_id = "simple"
agent._tools = []
agent.run.return_value = AgentResult(
content="agent response",
turns=1,
+88
View File
@@ -534,6 +534,94 @@ class TestChatCompletions:
content += delta_content
assert content == "Hello world"
def test_streaming_without_client_tools_uses_configured_agent(self):
"""Server-side tools remain available to streaming web clients (#735)."""
from openjarvis.agents.orchestrator import OrchestratorAgent
from openjarvis.core.types import ToolResult
from openjarvis.tools._stubs import BaseTool, ToolSpec
executions: list[str] = []
class _FileReadTool(BaseTool):
@property
def spec(self):
return ToolSpec(
name="file_read",
description="Read a file",
parameters={
"type": "object",
"properties": {"path": {"type": "string"}},
},
)
def execute(self, **params):
executions.append(params["path"])
return ToolResult(
tool_name="file_read",
content="README fixture contents",
success=True,
)
engine = _make_engine(content="ENGINE BYPASS")
engine.generate.side_effect = [
{
"content": "",
"tool_calls": [
{
"id": "call_1",
"name": "file_read",
"arguments": '{"path": "README.md"}',
}
],
"usage": {},
},
{
"content": "README fixture contents",
"finish_reason": "stop",
"usage": {},
},
]
agent = OrchestratorAgent(
engine,
"test-model",
tools=[_FileReadTool()],
bus=EventBus(),
max_turns=3,
temperature=0.7,
max_tokens=128,
system_prompt="Use the configured tools.",
)
app = create_app(
engine,
"test-model",
agent=agent,
bus=EventBus(),
config=_test_config(),
)
client = TestClient(app)
resp = client.post(
"/v1/chat/completions",
json={
"model": "test-model",
"messages": [{"role": "user", "content": "Read README.md"}],
"stream": True,
},
)
assert resp.status_code == 200
content = ""
for line in resp.text.strip().split("\n"):
if not line.startswith("data:") or "[DONE]" in line:
continue
data = json.loads(line[5:].strip())
delta = data.get("choices", [{}])[0].get("delta", {})
content += delta.get("content") or ""
assert content == "README fixture contents"
assert executions == ["README.md"]
assert engine.generate.call_count == 2
def test_streaming_with_tools_emits_tool_calls_and_bypasses_agent(self):
"""Regression for the streaming analog of #414.
+30
View File
@@ -0,0 +1,30 @@
import json
from openjarvis.server.stream_bridge import AgentStreamBridge
def test_tool_call_start_serializes_arguments_for_sse_without_mutating_event():
bridge = object.__new__(AgentStreamBridge)
event_data = {
"tool": "web_search",
"arguments": {"query": "python"},
"agent": "agent-1",
}
event = bridge._format_named_event("tool_call_start", event_data)
payload = json.loads(event.split("data: ", 1)[1])
assert payload["arguments"] == '{"query": "python"}'
assert event_data["arguments"] == {"query": "python"}
def test_tool_call_start_preserves_already_serialized_arguments():
bridge = object.__new__(AgentStreamBridge)
event = bridge._format_named_event(
"tool_call_start",
{"tool": "web_search", "arguments": '{"query":"python"}'},
)
payload = json.loads(event.split("data: ", 1)[1])
assert payload["arguments"] == '{"query":"python"}'