Compare commits

...
6 Commits
Author SHA1 Message Date
github-actions[bot] a97c64c67b chore: update clone traffic data [skip ci] 2026-08-14 07:19:51 +00:00
Jon Saad-Falcon 64333651d1 Delete tools/pearl-reference-oracle directory 2026-08-13 19:27:37 -07:00
Elliot Slusky 9bee016c82 fix(server): preserve grounded agent stream content (#736)
* fix(server): preserve grounded agent stream content

* test(server): cover active grounded stream path

* fix(server): retain agent stream bridge
2026-08-13 18:15:36 -07:00
Elliot Slusky c9942961ad fix(evals): make TauBench dependency explicit (#739)
* fix(evals): make TauBench dependency explicit

* fix(evals): verify TauBench install provenance
2026-08-13 18:14:55 -07:00
Elliot Slusky c3a7ffebff fix(memory): recall auto-captured facts across sessions (#740)
* fix(memory): recall auto-captured facts

* fix(memory): harden recalled context injection

* fix(memory): preserve mixed context history

* fix(agents): preserve caller system context
2026-08-13 17:07:10 -07:00
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
26 changed files with 1183 additions and 444 deletions
+1 -1
View File
@@ -1,7 +1,7 @@
{
"schemaVersion": 1,
"label": "Git Clones",
"message": "189,482",
"message": "190,252",
"color": "green",
"namedLogo": "git"
}
+4 -3
View File
@@ -1,6 +1,6 @@
{
"total_clones": 189482,
"last_updated": "2026-08-13T07:22:13Z",
"total_clones": 190252,
"last_updated": "2026-08-14T07:19:51Z",
"daily": {
"2026-03-27": 2189,
"2026-03-28": 1874,
@@ -140,6 +140,7 @@
"2026-08-09": 1076,
"2026-08-10": 1060,
"2026-08-11": 2182,
"2026-08-12": 641
"2026-08-12": 641,
"2026-08-13": 770
}
}
+10
View File
@@ -31,6 +31,16 @@ uv sync --extra dev --extra eval-wandb # Weights & Biases run tracking
uv sync --extra dev --extra eval-sheets # Google Sheets results export
```
TauBench additionally requires Python 3.12 or newer and the upstream `tau2`
package. Install the pinned revision explicitly before running that benchmark:
```bash
uv pip install "tau2 @ git+https://github.com/sierra-research/tau2-bench.git@fc0055dc4e0a316c3f83133267fbd6faaa770992"
```
OpenJarvis does not install third-party packages automatically when an
evaluation is imported or run.
!!! note "Python version requirement"
Python 3.10 requires the `tomli` package for TOML config parsing. `openjarvis` declares it as a conditional dependency, so it is installed automatically.
+24 -2
View File
@@ -155,6 +155,9 @@ class BaseAgent(ABC):
conversation messages, and finally the user input.
"""
messages: list[Message] = []
context_messages = (
list(context.conversation.messages) if context is not None else []
)
# Check if the context already supplies a system message
_context_has_system = (
context
@@ -176,9 +179,28 @@ class BaseAgent(ABC):
except Exception:
effective_system_prompt = None
if effective_system_prompt:
context_system_text = "\n\n".join(
message.text
for message in context_messages
if message.role == Role.SYSTEM
and message.metadata.get("memory_context")
and message.text
)
if context_system_text:
effective_system_prompt = (
f"{effective_system_prompt}\n\n{context_system_text}"
)
context_messages = [
message
for message in context_messages
if not (
message.role == Role.SYSTEM
and message.metadata.get("memory_context")
)
]
messages.append(Message(role=Role.SYSTEM, content=effective_system_prompt))
if context and context.conversation.messages:
messages.extend(context.conversation.messages)
if context_messages:
messages.extend(context_messages)
messages.append(Message(role=Role.USER, content=input))
return messages
+17 -2
View File
@@ -248,6 +248,17 @@ def _get_memory_backend(config):
return None
def _get_memory_facts(config):
"""Load facts captured by the automatic memory service."""
try:
from openjarvis.memory import load_configured_facts
return load_configured_facts(config)
except Exception as exc:
logger.debug("Automatic memory facts unavailable (optional): %s", exc)
return []
_MEMORY_TOOLS = frozenset(
{"retrieval", "memory_store", "memory_search", "memory_index", "memory_retrieve"}
)
@@ -416,7 +427,8 @@ def _run_agent(
from openjarvis.tools.storage.context import ContextConfig, inject_context
backend = _get_memory_backend(config)
if backend is not None:
facts = _get_memory_facts(config)
if backend is not None or facts:
ctx_cfg = ContextConfig(
top_k=config.memory.context_top_k,
min_score=config.memory.context_min_score,
@@ -427,6 +439,7 @@ def _run_agent(
[],
backend,
config=ctx_cfg,
facts=facts,
)
for msg in context_messages:
ctx.conversation.add(msg)
@@ -963,7 +976,8 @@ def ask(
)
backend = _get_memory_backend(config)
if backend is not None:
facts = _get_memory_facts(config)
if backend is not None or facts:
ctx_cfg = ContextConfig(
top_k=config.memory.context_top_k,
min_score=config.memory.context_min_score,
@@ -974,6 +988,7 @@ def ask(
messages,
backend,
config=ctx_cfg,
facts=facts,
)
except Exception as exc:
logger.debug("Failed to inject memory context: %s", exc)
+55 -3
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import logging
import sys
from typing import List, Optional
@@ -15,6 +16,8 @@ from openjarvis.core.events import EventBus
from openjarvis.core.types import Message, Role
from openjarvis.memory import publish_completed_exchange
logger = logging.getLogger(__name__)
def _read_input(prompt: str = "You> ") -> Optional[str]:
"""Read user input with graceful EOF handling."""
@@ -194,6 +197,15 @@ def chat(
console.print(f"[yellow]Memory service unavailable: {exc}[/yellow]")
memory_service = None
# The document backend and automatic fact store are separate persistence
# mechanisms. Context injection combines both at read time so facts from
# previous sessions are immediately available without a manual index step.
memory_backend = None
if config.agent.context_from_memory:
from openjarvis.cli.ask import _get_memory_backend
memory_backend = _get_memory_backend(config)
# Conversation state
if not system_prompt:
from openjarvis.prompt.builder import SystemPromptBuilder
@@ -262,15 +274,55 @@ def chat(
# Add user message
history.append(Message(role=Role.USER, content=user_input))
# Generate response
generation_history = history
agent_context_message = None
if config.agent.context_from_memory:
try:
from openjarvis.memory import load_configured_facts
from openjarvis.tools.storage.context import (
ContextConfig,
inject_context,
)
if memory_service is not None and hasattr(memory_service, "list_facts"):
facts = memory_service.list_facts()
else:
facts = load_configured_facts(config)
ctx_cfg = ContextConfig(
top_k=config.memory.context_top_k,
min_score=config.memory.context_min_score,
max_context_tokens=config.memory.context_max_tokens,
)
context_messages = inject_context(
user_input,
[] if agent is not None else history,
memory_backend,
config=ctx_cfg,
facts=facts,
)
if agent is not None:
if context_messages:
agent_context_message = context_messages[0]
else:
generation_history = context_messages
except Exception:
logger.debug("Failed to inject memory context", exc_info=True)
# Generate response even when optional memory context is unavailable.
try:
if agent is not None:
response = agent.run(user_input)
agent_context = None
if agent_context_message is not None:
from openjarvis.agents._stubs import AgentContext
agent_context = AgentContext()
agent_context.conversation.add(agent_context_message)
response = agent.run(user_input, context=agent_context)
content = (
response.content if hasattr(response, "content") else str(response)
)
else:
result = engine.generate(history, model=model)
result = engine.generate(generation_history, model=model)
content = (
result.get("content", "")
if isinstance(result, dict)
+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
+39 -38
View File
@@ -8,13 +8,12 @@ Reference: https://github.com/sierra-research/tau2-bench
from __future__ import annotations
import json
import logging
import os
import subprocess
import sys
from importlib import metadata
from typing import Iterable, List, Optional
from openjarvis.core.paths import get_cache_dir
from openjarvis.evals.core.dataset import DatasetProvider
from openjarvis.evals.core.splits import apply_split
from openjarvis.evals.core.types import EvalRecord
@@ -22,48 +21,50 @@ from openjarvis.evals.core.types import EvalRecord
LOGGER = logging.getLogger(__name__)
TAU2_REPO = "https://github.com/sierra-research/tau2-bench.git"
CACHE_DIR = get_cache_dir() / "tau2-bench"
# v1.0.1. Keep the full commit SHA here (rather than a movable tag) so every
# TauBench setup uses the same third-party code.
TAU2_REVISION = "fc0055dc4e0a316c3f83133267fbd6faaa770992"
TAU2_INSTALL_SPEC = f"tau2 @ git+{TAU2_REPO}@{TAU2_REVISION}"
DOMAINS = ("airline", "retail", "telecom")
def _ensure_tau2() -> None:
"""Ensure tau2 package is importable; install from cache if needed."""
"""Ensure the explicitly installed, pinned tau2 package is importable."""
try:
distribution = metadata.distribution("tau2")
except metadata.PackageNotFoundError as exc:
raise ImportError(
"TauBench requires tau2, which OpenJarvis does not install at "
"runtime. Install the pinned dependency explicitly (Python >=3.12): "
f'uv pip install "{TAU2_INSTALL_SPEC}"'
) from exc
try:
direct_url_text = distribution.read_text("direct_url.json")
direct_url = json.loads(direct_url_text or "")
vcs_info = direct_url.get("vcs_info", {})
installed_repo = direct_url.get("url")
installed_revision = vcs_info.get("commit_id")
except (json.JSONDecodeError, AttributeError):
installed_repo = None
installed_revision = None
if installed_repo != TAU2_REPO or installed_revision != TAU2_REVISION:
raise ImportError(
"The installed tau2 package does not match OpenJarvis's pinned "
"source revision. Reinstall it explicitly (Python >=3.12): "
f'uv pip install --force-reinstall "{TAU2_INSTALL_SPEC}"'
)
try:
import tau2 # noqa: F401
except ImportError:
# Clone and install from source
if not CACHE_DIR.exists():
LOGGER.info("Cloning tau2-bench from %s ...", TAU2_REPO)
CACHE_DIR.parent.mkdir(parents=True, exist_ok=True)
subprocess.run(
["git", "clone", "--depth", "1", TAU2_REPO, str(CACHE_DIR)],
check=True,
capture_output=True,
)
LOGGER.info("Installing tau2-bench ...")
# Try `python -m pip` first; fall back to `uv pip` for uv-managed venvs
# which don't ship pip by default.
try:
subprocess.run(
[sys.executable, "-m", "pip", "install", "-e", str(CACHE_DIR)],
check=True,
capture_output=True,
)
except (subprocess.CalledProcessError, FileNotFoundError):
subprocess.run(
[
"uv",
"pip",
"install",
"--python",
sys.executable,
"-e",
str(CACHE_DIR),
],
check=True,
capture_output=True,
)
except ImportError as exc:
raise ImportError(
"The pinned tau2 package is installed but cannot be imported. "
"Reinstall it explicitly (Python >=3.12): "
f'uv pip install --force-reinstall "{TAU2_INSTALL_SPEC}"'
) from exc
class TauBenchDataset(DatasetProvider):
+2
View File
@@ -19,6 +19,7 @@ from openjarvis.memory.store import (
FactStore,
LocalFactStore,
create_fact_store,
load_configured_facts,
)
__all__ = [
@@ -29,5 +30,6 @@ __all__ = [
"MemoryService",
"build_memory_service",
"create_fact_store",
"load_configured_facts",
"publish_completed_exchange",
]
+28 -2
View File
@@ -16,7 +16,7 @@ import time
from abc import ABC, abstractmethod
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Iterable, List
from typing import Any, Iterable, List
from openjarvis.core.paths import get_config_dir
from openjarvis.core.registry import FactStoreRegistry
@@ -205,4 +205,30 @@ def create_fact_store(
return FactStoreRegistry.create(key, path, max_facts=max_facts)
__all__ = ["Fact", "FactStore", "LocalFactStore", "create_fact_store"]
def load_configured_facts(config: Any) -> List[Fact]:
"""Load automatic-memory facts from *config* when the service is enabled.
Context injection is also used by short-lived commands such as
``jarvis ask``, where no :class:`MemoryService` instance exists. This
helper gives those callers the same configured fact-store view without
coupling them to the service lifecycle.
"""
memory = getattr(config, "memory", None)
if memory is None or not getattr(memory, "enabled", False):
return []
store = create_fact_store(
getattr(memory, "backend", "local"),
path=getattr(memory, "facts_path", None),
max_facts=getattr(memory, "max_facts", 1000),
)
return store.list()
__all__ = [
"Fact",
"FactStore",
"LocalFactStore",
"create_fact_store",
"load_configured_facts",
]
+14 -5
View File
@@ -522,14 +522,15 @@ class Jarvis:
# Context injection
if context and self._config.agent.context_from_memory:
try:
from openjarvis.cli.ask import _get_memory_backend
from openjarvis.cli.ask import _get_memory_backend, _get_memory_facts
from openjarvis.tools.storage.context import (
ContextConfig,
inject_context,
)
backend = _get_memory_backend(self._config)
if backend is not None:
facts = _get_memory_facts(self._config)
if backend is not None or facts:
ctx_cfg = ContextConfig(
top_k=self._config.memory.context_top_k,
min_score=self._config.memory.context_min_score,
@@ -540,6 +541,7 @@ class Jarvis:
[],
backend,
config=ctx_cfg,
facts=facts,
)
for msg in context_messages:
ctx.conversation.add(msg)
@@ -570,17 +572,24 @@ class Jarvis:
) -> List[Message]:
"""Inject memory context into messages."""
try:
from openjarvis.cli.ask import _get_memory_backend
from openjarvis.cli.ask import _get_memory_backend, _get_memory_facts
from openjarvis.tools.storage.context import ContextConfig, inject_context
backend = _get_memory_backend(self._config)
if backend is not None:
facts = _get_memory_facts(self._config)
if backend is not None or facts:
ctx_cfg = ContextConfig(
top_k=self._config.memory.context_top_k,
min_score=self._config.memory.context_min_score,
max_context_tokens=self._config.memory.context_max_tokens,
)
return inject_context(query, messages, backend, config=ctx_cfg)
return inject_context(
query,
messages,
backend,
config=ctx_cfg,
facts=facts,
)
except Exception as exc:
logger.warning("Failed to inject memory context: %s", exc)
return messages
+168 -24
View File
@@ -11,7 +11,7 @@ from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import StreamingResponse
from openjarvis.core.paths import get_config_dir
from openjarvis.core.types import Message, Role
from openjarvis.core.types import Message, Role, ToolCall
from openjarvis.server.model_capabilities import is_embed_only_model
from openjarvis.server.models import (
ChatCompletionChunk,
@@ -40,6 +40,15 @@ def _to_messages(chat_messages) -> list[Message]:
role=role,
content=m.content or "",
name=m.name,
tool_calls=[
ToolCall(
id=tool_call.get("id", ""),
name=tool_call.get("function", {}).get("name", ""),
arguments=tool_call.get("function", {}).get("arguments", "{}"),
)
for tool_call in (m.tool_calls or [])
]
or None,
tool_call_id=m.tool_call_id,
)
)
@@ -114,13 +123,15 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
memory_backend = getattr(request.app.state, "memory_backend", None)
if (
config is not None
and memory_backend is not None
and config.agent.context_from_memory
and request_body.messages
):
try:
from openjarvis.tools.storage.context import ContextConfig, inject_context
memory_service = getattr(request.app.state, "memory_service", None)
facts = memory_service.list_facts() if memory_service is not None else []
# Extract query from the last user message
query_text = ""
for m in reversed(request_body.messages):
@@ -130,6 +141,7 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
if query_text:
messages = _to_messages(request_body.messages)
messages = _ensure_identity_prompt(messages, config)
ctx_cfg = ContextConfig(
top_k=config.memory.context_top_k,
min_score=config.memory.context_min_score,
@@ -140,22 +152,35 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
messages,
memory_backend,
config=ctx_cfg,
facts=facts,
)
# Rebuild request messages from enriched Message objects
if len(enriched) > len(messages):
from openjarvis.server.models import ChatMessage
# Rebuild after identity/context merging so downstream engine
# adapters always receive exactly one system message.
from openjarvis.server.models import ChatMessage
new_msgs = []
for msg in enriched:
new_msgs.append(
ChatMessage(
role=msg.role.value,
content=msg.content,
name=msg.name,
tool_call_id=getattr(msg, "tool_call_id", None),
)
new_msgs = []
for msg in enriched:
new_msgs.append(
ChatMessage(
role=msg.role.value,
content=msg.content,
name=msg.name,
tool_calls=[
{
"id": tool_call.id,
"type": "function",
"function": {
"name": tool_call.name,
"arguments": tool_call.arguments,
},
}
for tool_call in (msg.tool_calls or [])
]
or None,
tool_call_id=getattr(msg, "tool_call_id", None),
)
request_body.messages = new_msgs
)
request_body.messages = new_msgs
except Exception:
logging.getLogger("openjarvis.server").debug(
"Memory context injection failed",
@@ -200,12 +225,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 +243,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 +584,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 +835,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
+8 -55
View File
@@ -246,62 +246,15 @@ class AgentStreamBridge:
{"results": tool_results_data},
)
# Stream content using real LLM token streaming via
# engine.stream_full() when the engine is available.
# ``agent.run()`` already produced the authoritative, grounded
# response. Do not call the engine again here: a second inference
# would not have the agent's system prompt, tool transcript, or
# other internal context and could therefore contradict the
# result reported by the agent events. Replay the final content
# in chunks so the OpenAI-compatible streaming response stays
# consistent with the completed agent run.
content = agent_result.content or ""
engine = getattr(self._agent, "_engine", None)
used_real_streaming = False
if engine is not None and hasattr(engine, "stream_full") and content:
# Re-stream using the engine for real token delivery.
# Build the same messages the agent used for its final turn.
try:
from openjarvis.core.types import Message as MsgType
from openjarvis.core.types import Role as RoleType
replay_messages = []
for m in self._request.messages:
role = (
RoleType(m.role)
if m.role in {r.value for r in RoleType}
else RoleType.USER
)
replay_messages.append(
MsgType(
role=role,
content=m.content or "",
name=m.name,
tool_call_id=m.tool_call_id,
)
)
async for sc in engine.stream_full(
replay_messages,
model=self._model,
):
if sc.content:
chunk = ChatCompletionChunk(
id=self._chunk_id,
model=self._model,
choices=[
StreamChoice(
delta=DeltaMessage(content=sc.content),
)
],
)
yield f"data: {chunk.model_dump_json()}\n\n"
used_real_streaming = True
except Exception as stream_exc:
import logging as _logging
_logger = _logging.getLogger("openjarvis.server")
_logger.warning(
"Real streaming failed, falling back to word replay: %s",
stream_exc,
)
# Fallback: word-by-word replay if real streaming was not used
if not used_real_streaming and content:
if content:
words = content.split(" ")
for i, word in enumerate(words):
token = word if i == 0 else " " + word
+4 -1
View File
@@ -40,8 +40,9 @@ class QueryOrchestrator:
messages = [Message(role=Role.USER, content=query)]
if context and s.memory_backend and s.config.agent.context_from_memory:
if context and s.config.agent.context_from_memory:
try:
from openjarvis.memory import load_configured_facts
from openjarvis.tools.storage.context import (
ContextConfig,
inject_context,
@@ -52,11 +53,13 @@ class QueryOrchestrator:
min_score=s.config.memory.context_min_score,
max_context_tokens=s.config.memory.context_max_tokens,
)
facts = load_configured_facts(s.config)
messages = inject_context(
query,
messages,
s.memory_backend,
config=ctx_cfg,
facts=facts,
)
except Exception as exc:
logger.warning("Failed to inject memory context: %s", exc)
+93 -20
View File
@@ -2,13 +2,16 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import List, Optional
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING, List, Optional, Sequence
from openjarvis.core.events import EventType, get_event_bus
from openjarvis.core.types import Message, Role
from openjarvis.tools.storage._stubs import MemoryBackend, RetrievalResult
if TYPE_CHECKING:
from openjarvis.memory.store import Fact
@dataclass(slots=True)
class ContextConfig:
@@ -46,28 +49,75 @@ def format_context(results: List[RetrievalResult]) -> str:
def build_context_message(
results: List[RetrievalResult],
facts: Sequence[Fact] = (),
) -> Message:
"""Create a system message with formatted context."""
context_text = format_context(results)
content = (
"The following context was retrieved from the knowledge"
" base. Use it to inform your response, citing sources"
" where applicable:\n\n" + context_text
sections = []
if facts:
fact_text = "\n".join(f"- {fact.text}" for fact in facts)
sections.append(
"The following durable facts were remembered from prior "
"conversations. Use them when relevant to the user's request:\n\n"
+ fact_text
)
if results:
sections.append(
"The following context was retrieved from the knowledge"
" base. Use it to inform your response, citing sources"
" where applicable:\n\n" + format_context(results)
)
content = "\n\n".join(sections)
return Message(
role=Role.SYSTEM,
content=content,
metadata={"memory_context": True},
)
return Message(role=Role.SYSTEM, content=content)
def _merge_context_message(
messages: List[Message],
context_message: Message,
) -> List[Message]:
"""Return a copy with context folded into the existing system prompt."""
system_messages = [message for message in messages if message.role == Role.SYSTEM]
if not system_messages:
return [context_message, *messages]
content = "\n\n".join(
part
for part in (
*(message.text for message in system_messages),
context_message.text,
)
if part
)
combined = replace(system_messages[0], content=content)
merged: List[Message] = []
inserted = False
for message in messages:
if message.role == Role.SYSTEM:
if not inserted:
merged.append(combined)
inserted = True
continue
merged.append(message)
return merged
def inject_context(
query: str,
messages: List[Message],
backend: MemoryBackend,
backend: Optional[MemoryBackend],
*,
config: Optional[ContextConfig] = None,
facts: Sequence[Fact] = (),
) -> List[Message]:
"""Retrieve relevant context and prepend it to *messages*.
Returns a **new** list the original list is not mutated.
If no results pass the score threshold, returns the original
Automatic-memory facts are included independently of the retrieval
backend, so persisted facts remain recallable even when the document
store is empty. If no facts or results are available, returns the original
messages unchanged.
Parameters
@@ -77,33 +127,55 @@ def inject_context(
messages:
The existing message list.
backend:
The memory backend to search.
The memory backend to search, or ``None`` when only facts are available.
config:
Context injection settings (uses defaults if ``None``).
facts:
Durable facts captured by the automatic memory service.
"""
cfg = config or ContextConfig()
if not cfg.enabled:
return messages
results = backend.retrieve(query, top_k=cfg.top_k)
results = backend.retrieve(query, top_k=cfg.top_k) if backend is not None else []
# Filter by minimum score
results = [r for r in results if r.score >= cfg.min_score]
if not results:
return messages
# Truncate to max_context_tokens
truncated: List[RetrievalResult] = []
# When both sources have data, cap facts at half the total budget so they
# cannot starve query-specific document retrieval. Unused fact budget is
# still available to documents. Newest facts win within the fact budget.
fact_budget = cfg.max_context_tokens
if results:
fact_budget //= 2
selected_facts: List[Fact] = []
total_tokens = 0
for fact in reversed(facts):
tokens = _count_tokens(fact.text)
if total_tokens + tokens > fact_budget:
continue
selected_facts.append(fact)
total_tokens += tokens
# Fill the remaining context budget with retrieved documents.
truncated: List[RetrievalResult] = []
for r in results:
tokens = _count_tokens(r.content)
if total_tokens + tokens > cfg.max_context_tokens:
# A large top result should not disappear solely because facts
# consumed their reserved share. Prefer that result when it fits
# the total budget on its own.
if not truncated and selected_facts and tokens <= cfg.max_context_tokens:
selected_facts = []
total_tokens = 0
else:
break
if total_tokens + tokens > cfg.max_context_tokens:
break
truncated.append(r)
total_tokens += tokens
if not truncated:
if not selected_facts and not truncated:
return messages
# Publish event
@@ -114,13 +186,14 @@ def inject_context(
"context_injection": True,
"query": query,
"num_results": len(truncated),
"num_facts": len(selected_facts),
"total_tokens": total_tokens,
},
)
# Build context message and prepend
ctx_msg = build_context_message(truncated)
return [ctx_msg] + list(messages)
ctx_msg = build_context_message(truncated, selected_facts)
return _merge_context_message(messages, ctx_msg)
__all__ = [
+34
View File
@@ -205,6 +205,40 @@ class TestBuildMessages:
assert messages[1].content == "prev"
assert messages[2].content == "new"
def test_prompt_builder_merges_context_system_message(self):
engine = MagicMock()
prompt_builder = MagicMock()
prompt_builder.build.return_value = "You are OpenJarvis."
agent = _ConcreteAgent(engine, "m", prompt_builder=prompt_builder)
conv = Conversation()
conv.add(
Message(
role=Role.SYSTEM,
content="Remember: user likes jazz.",
metadata={"memory_context": True},
)
)
ctx = AgentContext(conversation=conv)
messages = agent._build_messages("new", ctx)
system_messages = [m for m in messages if m.role == Role.SYSTEM]
assert len(system_messages) == 1
assert "You are OpenJarvis." in system_messages[0].content
assert "user likes jazz" in system_messages[0].content
def test_prompt_builder_preserves_caller_system_context(self):
engine = MagicMock()
prompt_builder = MagicMock()
prompt_builder.build.return_value = "Agent instructions."
agent = _ConcreteAgent(engine, "m", prompt_builder=prompt_builder)
conv = Conversation()
conv.add(Message(role=Role.SYSTEM, content="You are helpful."))
messages = agent._build_messages("new", AgentContext(conversation=conv))
assert any(message.content == "You are helpful." for message in messages)
class TestGenerate:
def test_delegates_to_engine(self):
+74
View File
@@ -18,6 +18,7 @@ from openjarvis.core.config import JarvisConfig
from openjarvis.core.events import Event, EventBus, EventType
from openjarvis.core.registry import AgentRegistry, ToolRegistry
from openjarvis.core.types import ToolCall, ToolResult
from openjarvis.memory.store import LocalFactStore
from openjarvis.tools._stubs import BaseTool, ToolSpec
@@ -97,6 +98,79 @@ class TestReadInput:
class TestChatAgents:
def test_direct_chat_injects_auto_memory_facts(self, tmp_path) -> None:
facts_path = tmp_path / "facts.jsonl"
LocalFactStore(facts_path).add(
"The user's favorite color is blue",
source="auto",
)
engine = MagicMock()
engine.engine_id = "mock"
engine.generate.return_value = {"content": "Blue."}
config = JarvisConfig()
config.intelligence.default_model = "test-model"
config.memory.enabled = True
config.memory.facts_path = str(facts_path)
config.agent.context_from_memory = True
with (
patch("openjarvis.cli.chat_cmd.load_config", return_value=config),
patch("openjarvis.engine.get_engine", return_value=("mock", engine)),
patch("openjarvis.intelligence.register_builtin_models"),
patch("openjarvis.memory.build_memory_service", return_value=None),
patch("openjarvis.cli.ask._get_memory_backend", return_value=None),
):
result = CliRunner().invoke(
chat,
["--model", "test-model"],
input="What is my favorite color?\n/quit\n",
)
assert result.exit_code == 0
messages = engine.generate.call_args.args[0]
assert messages[0].role.value == "system"
assert "favorite color is blue" in messages[0].content
def test_chat_generation_survives_fact_store_failure(self) -> None:
class _FailingMemoryService:
def start(self) -> None:
pass
def stop(self, timeout: float = 2.0) -> None:
pass
def list_facts(self):
raise OSError("fact store unavailable")
engine = MagicMock()
engine.engine_id = "mock"
engine.generate.return_value = {"content": "Still working."}
config = JarvisConfig()
config.intelligence.default_model = "test-model"
config.memory.enabled = True
config.agent.context_from_memory = True
with (
patch("openjarvis.cli.chat_cmd.load_config", return_value=config),
patch("openjarvis.engine.get_engine", return_value=("mock", engine)),
patch("openjarvis.intelligence.register_builtin_models"),
patch(
"openjarvis.memory.build_memory_service",
return_value=_FailingMemoryService(),
),
patch("openjarvis.cli.ask._get_memory_backend", return_value=None),
):
result = CliRunner().invoke(
chat,
["--model", "test-model"],
input="hello\n/quit\n",
)
assert result.exit_code == 0
assert "Still working." in result.output
engine.generate.assert_called_once()
def test_simple_agent_does_not_receive_tool_only_kwargs(self) -> None:
engine = MagicMock()
engine.engine_id = "mock"
+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
+98
View File
@@ -0,0 +1,98 @@
"""Tests for the TauBench optional dependency boundary."""
from __future__ import annotations
import builtins
import sys
from types import ModuleType
from unittest.mock import Mock
import pytest
from openjarvis.evals.datasets import taubench
def _mock_direct_url(monkeypatch, direct_url):
distribution = Mock()
distribution.read_text.return_value = direct_url
monkeypatch.setattr(
taubench.metadata, "distribution", Mock(return_value=distribution)
)
def test_ensure_tau2_accepts_the_pinned_source_revision(monkeypatch):
monkeypatch.setitem(sys.modules, "tau2", ModuleType("tau2"))
_mock_direct_url(
monkeypatch,
(
'{"url": "https://github.com/sierra-research/tau2-bench.git", '
'"vcs_info": {"vcs": "git", '
f'"commit_id": "{taubench.TAU2_REVISION}"}}}}'
),
)
taubench._ensure_tau2()
def test_ensure_tau2_requires_explicit_pinned_install(monkeypatch):
monkeypatch.setitem(sys.modules, "tau2", None)
monkeypatch.setattr(
taubench.metadata,
"distribution",
Mock(side_effect=taubench.metadata.PackageNotFoundError),
)
with pytest.raises(ImportError) as exc_info:
taubench._ensure_tau2()
message = str(exc_info.value)
assert "does not install at runtime" in message
assert taubench.TAU2_REVISION in message
assert "uv pip install" in message
@pytest.mark.parametrize(
"direct_url",
[
# Editable install left behind by the previous runtime installer.
'{"url": "file:///home/user/.openjarvis/cache/tau2-bench", '
'"dir_info": {"editable": true}}',
# A git install from an arbitrary upstream revision.
'{"url": "https://github.com/sierra-research/tau2-bench.git", '
'"vcs_info": {"vcs": "git", "commit_id": "deadbeef"}}',
# Registry installs do not carry PEP 610 direct-origin metadata.
None,
],
)
def test_ensure_tau2_rejects_unpinned_install(monkeypatch, direct_url):
_mock_direct_url(monkeypatch, direct_url)
original_import = builtins.__import__
def guarded_import(name, *args, **kwargs):
if name == "tau2":
raise AssertionError("unverified tau2 package was imported")
return original_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", guarded_import)
with pytest.raises(ImportError) as exc_info:
taubench._ensure_tau2()
message = str(exc_info.value)
assert "does not match" in message
assert taubench.TAU2_REVISION in message
assert "--force-reinstall" in message
def test_verify_requirements_reports_install_instruction(monkeypatch):
monkeypatch.setitem(sys.modules, "tau2", None)
monkeypatch.setattr(
taubench.metadata,
"distribution",
Mock(side_effect=taubench.metadata.PackageNotFoundError),
)
issues = taubench.TauBenchDataset().verify_requirements()
assert len(issues) == 1
assert taubench.TAU2_REVISION in issues[0]
+108
View File
@@ -7,6 +7,7 @@ from typing import Any, Dict, List, Optional
from openjarvis.core.events import EventBus, EventType
from openjarvis.core.types import Message, Role
from openjarvis.memory.store import Fact
from openjarvis.tools.storage._stubs import MemoryBackend, RetrievalResult
from openjarvis.tools.storage.context import (
ContextConfig,
@@ -167,6 +168,113 @@ def test_inject_context_no_results_returns_original():
assert augmented is messages
def test_inject_context_adds_auto_memory_facts_without_backend():
messages = [Message(role=Role.USER, content="What is my favorite color?")]
facts = [Fact(text="The user's favorite color is blue", source="auto")]
augmented = inject_context("favorite color", messages, None, facts=facts)
assert len(augmented) == 2
assert augmented[0].role == Role.SYSTEM
assert "remembered from prior conversations" in augmented[0].content
assert "favorite color is blue" in augmented[0].content
def test_inject_context_prioritizes_newest_facts_within_token_budget():
messages = [Message(role=Role.USER, content="What do you remember?")]
facts = [
Fact(text="old fact uses four tokens"),
Fact(text="new fact uses four tokens"),
]
augmented = inject_context(
"remember",
messages,
None,
config=ContextConfig(max_context_tokens=5),
facts=facts,
)
assert "new fact uses four tokens" in augmented[0].content
assert "old fact uses four tokens" not in augmented[0].content
def test_inject_context_merges_with_existing_system_message():
messages = [
Message(role=Role.SYSTEM, content="You are OpenJarvis."),
Message(role=Role.USER, content="What is my favorite color?"),
]
facts = [Fact(text="The user's favorite color is blue")]
augmented = inject_context("favorite color", messages, None, facts=facts)
system_messages = [m for m in augmented if m.role == Role.SYSTEM]
assert len(system_messages) == 1
assert "You are OpenJarvis." in system_messages[0].content
assert "favorite color is blue" in system_messages[0].content
assert messages[0].content == "You are OpenJarvis."
def test_inject_context_collapses_multiple_system_messages():
messages = [
Message(role=Role.SYSTEM, content="Identity."),
Message(role=Role.SYSTEM, content="Persona."),
Message(role=Role.USER, content="What do you remember?"),
]
augmented = inject_context(
"remember",
messages,
None,
facts=[Fact(text="User likes jazz")],
)
system_messages = [m for m in augmented if m.role == Role.SYSTEM]
assert len(system_messages) == 1
assert "Identity." in system_messages[0].content
assert "Persona." in system_messages[0].content
assert "User likes jazz" in system_messages[0].content
def test_inject_context_reserves_budget_for_retrieved_documents():
backend = _FakeMemory(
[RetrievalResult(content="d1 d2 d3 d4 d5", score=1.0, source="doc")]
)
facts = [
Fact(text="old1 old2 old3 old4 old5"),
Fact(text="new1 new2 new3 new4 new5"),
]
augmented = inject_context(
"query",
[Message(role=Role.USER, content="query")],
backend,
config=ContextConfig(max_context_tokens=10),
facts=facts,
)
assert "new1 new2 new3 new4 new5" in augmented[0].content
assert "d1 d2 d3 d4 d5" in augmented[0].content
assert "old1 old2 old3 old4 old5" not in augmented[0].content
def test_inject_context_prefers_large_document_that_fits_total_budget():
backend = _FakeMemory(
[RetrievalResult(content="d1 d2 d3 d4 d5 d6 d7 d8", score=1.0)]
)
augmented = inject_context(
"query",
[Message(role=Role.USER, content="query")],
backend,
config=ContextConfig(max_context_tokens=10),
facts=[Fact(text="f1 f2 f3 f4 f5")],
)
assert "d1 d2 d3 d4 d5 d6 d7 d8" in augmented[0].content
assert "f1 f2 f3 f4 f5" not in augmented[0].content
def test_inject_context_publishes_event():
bus = EventBus(record_history=True)
results = [
+30 -1
View File
@@ -7,7 +7,11 @@ import json
import pytest
from openjarvis.core.registry import FactStoreRegistry
from openjarvis.memory.store import LocalFactStore, create_fact_store
from openjarvis.memory.store import (
LocalFactStore,
create_fact_store,
load_configured_facts,
)
def test_add_and_list(tmp_path):
@@ -145,3 +149,28 @@ def test_create_fact_store_default_path_uses_openjarvis_home(tmp_path, monkeypat
def test_create_fact_store_unknown_backend(tmp_path):
with pytest.raises(ValueError):
create_fact_store("cloud", path=tmp_path / "f.jsonl")
def test_load_configured_facts_reads_enabled_store(tmp_path):
from types import SimpleNamespace
path = tmp_path / "facts.jsonl"
LocalFactStore(path).add("User likes jazz", source="auto")
config = SimpleNamespace(
memory=SimpleNamespace(
enabled=True,
backend="local",
facts_path=str(path),
max_facts=1000,
)
)
assert [fact.text for fact in load_configured_facts(config)] == ["User likes jazz"]
def test_load_configured_facts_skips_disabled_memory():
from types import SimpleNamespace
config = SimpleNamespace(memory=SimpleNamespace(enabled=False))
assert load_configured_facts(config) == []
+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,
+225
View File
@@ -11,6 +11,7 @@ fastapi = pytest.importorskip("fastapi")
from fastapi.testclient import TestClient # noqa: E402
from openjarvis.core.events import EventBus, EventType # noqa: E402
from openjarvis.core.types import Role # noqa: E402
from openjarvis.server.app import create_app # noqa: E402
# ---------------------------------------------------------------------------
@@ -534,6 +535,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.
@@ -758,6 +847,47 @@ class TestIdentityPromptInjection:
assert len(system_msgs) == 1
assert system_msgs[0].content == "Be terse."
def test_stream_uses_grounded_agent_result_without_replay(self):
"""Regression for #734: web streaming emits the agent's final answer."""
from openjarvis.core.events import EventBus
captured: list = []
engine = _make_capturing_engine(captured)
agent = _make_agent(content="My name is Jarvis Prime.")
agent._tools = [object()]
agent._engine = engine
client = TestClient(
create_app(
engine,
"test-model",
agent=agent,
bus=EventBus(),
config=_identity_config(),
)
)
resp = client.post(
"/v1/chat/completions",
json={
"model": "test-model",
"messages": [{"role": "user", "content": "who are you?"}],
"stream": True,
},
)
assert resp.status_code == 200
streamed_content = ""
for line in resp.text.splitlines():
if not line.startswith("data: {"):
continue
payload = json.loads(line.removeprefix("data: "))
choices = payload.get("choices", [])
if choices and choices[0]["delta"].get("content"):
streamed_content += choices[0]["delta"]["content"]
assert streamed_content == "My name is Jarvis Prime."
assert captured == []
agent.run.assert_called_once()
def test_direct_injects_identity_when_absent(self):
captured: list = []
engine = _make_capturing_engine(captured)
@@ -798,6 +928,101 @@ class TestIdentityPromptInjection:
assert len(system_msgs) == 1
assert system_msgs[0].content == "Be terse."
def test_direct_merges_identity_and_auto_memory_into_one_system_message(self):
from openjarvis.memory.store import Fact
class _MemoryService:
def list_facts(self):
return [Fact(text="The user's favorite color is blue")]
captured: list = []
engine = _make_capturing_engine(captured)
cfg = _identity_config()
cfg.agent.context_from_memory = True
client = TestClient(
create_app(
engine,
"test-model",
config=cfg,
memory_service=_MemoryService(),
)
)
resp = client.post(
"/v1/chat/completions",
json={
"model": "test-model",
"messages": [{"role": "user", "content": "What is my favorite color?"}],
},
)
assert resp.status_code == 200
messages = engine.generate.call_args.args[0]
system_messages = [m for m in messages if m.role == Role.SYSTEM]
assert len(system_messages) == 1
assert "OpenJarvis" in system_messages[0].content
assert "favorite color is blue" in system_messages[0].content
def test_memory_context_preserves_assistant_tool_calls(self):
from openjarvis.memory.store import Fact
class _MemoryService:
def list_facts(self):
return [Fact(text="User likes jazz")]
captured: list = []
engine = _make_capturing_engine(captured)
cfg = _identity_config()
cfg.agent.context_from_memory = True
client = TestClient(
create_app(
engine,
"test-model",
config=cfg,
memory_service=_MemoryService(),
)
)
resp = client.post(
"/v1/chat/completions",
json={
"model": "test-model",
"messages": [
{"role": "user", "content": "Run the lookup"},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "lookup",
"arguments": '{"query":"jazz"}',
},
}
],
},
{
"role": "tool",
"content": "result",
"tool_call_id": "call_1",
},
{"role": "user", "content": "What did it find?"},
],
},
)
assert resp.status_code == 200
messages = engine.generate.call_args.args[0]
assistant = next(
message for message in messages if message.role == Role.ASSISTANT
)
assert assistant.tool_calls is not None
assert assistant.tool_calls[0].id == "call_1"
assert assistant.tool_calls[0].name == "lookup"
assert assistant.tool_calls[0].arguments == '{"query":"jazz"}'
def test_direct_injects_soul_persona_when_present(self, tmp_path):
"""Regression: /v1/chat/completions previously injected only the bare
``default_system_prompt`` blurb via a hand-rolled lookup, bypassing
+64 -2
View File
@@ -1,6 +1,68 @@
import json
"""Regression tests for streaming completed agent responses."""
from openjarvis.server.stream_bridge import AgentStreamBridge
from __future__ import annotations
import asyncio
import json
from unittest.mock import MagicMock
import pytest
pytest.importorskip("fastapi")
from openjarvis.agents._stubs import AgentResult # noqa: E402
from openjarvis.core.events import EventBus # noqa: E402
from openjarvis.core.types import ToolResult # noqa: E402
from openjarvis.server.models import ChatCompletionRequest # noqa: E402
from openjarvis.server.stream_bridge import AgentStreamBridge # noqa: E402
def _streamed_content(events: list[str]) -> str:
"""Join assistant content from OpenAI-compatible data chunks."""
content = []
for event in events:
if not event.startswith("data: {"):
continue
payload = json.loads(event.removeprefix("data: ").strip())
choices = payload.get("choices")
if choices and choices[0]["delta"].get("content"):
content.append(choices[0]["delta"]["content"])
return "".join(content)
def test_stream_replays_grounded_agent_result_without_second_inference():
grounded_content = "My name is Jarvis. The tool reports 72 degrees."
agent = MagicMock()
agent._model = "configured-model"
agent.run.return_value = AgentResult(
content=grounded_content,
tool_results=[
ToolResult(tool_name="weather", content="72 degrees", success=True)
],
metadata={"prompt_tokens": 10, "completion_tokens": 12, "total_tokens": 22},
)
async def ungrounded_replay(*args, **kwargs):
raise AssertionError("stream_full must not run after agent.run")
yield # pragma: no cover
agent._engine.stream_full = ungrounded_replay
request = ChatCompletionRequest(
model="requested-model",
messages=[{"role": "user", "content": "Who are you, and what's outside?"}],
stream=True,
)
bridge = AgentStreamBridge(agent, EventBus(), request.model, request)
async def collect_events() -> list[str]:
return [event async for event in bridge.stream()]
events = asyncio.run(collect_events())
assert _streamed_content(events) == grounded_content
assert any(event.startswith("event: tool_results\n") for event in events)
agent.run.assert_called_once()
assert agent._model == "configured-model"
def test_tool_call_start_serializes_arguments_for_sse_without_mutating_event():
-111
View File
@@ -1,111 +0,0 @@
# Pearl reference oracle (OpenJarvis Phase 0 deliverable)
Phase 0-B of [Spec B](../../docs/design/2026-05-05-apple-silicon-pearl-mining-design.md)
called for "build a Python reference oracle for NoisyGEMM, validate against the
Pearl CUDA reference."
**Phase 0 found the oracle already exists upstream**, in two complementary forms:
| Layer | Upstream location | What it covers |
|---|---|---|
| Pure-Rust mining algorithm exposed to Python | `pearl/py-pearl-mining` | The complete `mine()` + `verify_plain_proof()` cycle. CPU-only. Hardware-portable. |
| PyTorch reference of production NoisyGEMM | `pearl/miner/miner-base/src/miner_base/noisy_gemm.py` | The same NoisyGEMM that vllm-miner accelerates with H100 CUDA. Bit-exact denoising verified by upstream test (`tests/test_noisy_gemm.py:92`). |
So this directory contains:
1. `smoke_test.py` — a runnable script that **actually mines a block on this machine** using the upstream Rust path, demonstrating the v1 architecture works on Apple Silicon (or any platform where `py-pearl-mining` builds).
2. This README documenting where the reference math lives.
## What this is *not*
This is **not a reimplementation** of NoisyGEMM. The original Spec B planned for that;
Phase 0 made it unnecessary. If you're tempted to write `noisy_gemm.py` here, stop —
read `pearl/miner/miner-base/src/miner_base/noisy_gemm.py` instead.
## Setup
You need:
- macOS arm64 (M1/M2/M3/M4) **or** Linux x86_64 / aarch64
- Python 3.12 (`uv venv --python 3.12 .venv` is the easiest)
- Rust 1.78+ (any recent toolchain — verified with 1.94 on macOS arm64)
- The Pearl source tree somewhere on disk
Build the wheel and install it (one-time, ~60 s on a fast Mac, ~5 min on first build):
```bash
# from the Pearl repo root
cd py-pearl-mining
uv pip install maturin
maturin build --release --interpreter "$(which python)"
# install the resulting wheel
uv pip install target/wheels/py_pearl_mining-*.whl
```
Or if Pearl publishes to PyPI in the future:
```bash
uv pip install py-pearl-mining
```
## Run the smoke test
```bash
python smoke_test.py
```
Actual output on Apple Silicon M2 Max (numbers will vary by hardware and run):
```
host: macOS-26.4.1-arm64-arm-64bit (arm64)
python: 3.12.1
[ok] pearl_mining loaded from <site-packages>/pearl_mining/__init__.py
[ok] PUBLICDATA_SIZE=164 MERKLE_LEAF_SIZE=1024
[ok] mine(m=256, n=128, k=1024, rank=32) returned a proof in 0.119 s
proof.m=256 proof.n=128 proof.k=1024 noise_rank=32
a.row_indices=[177, 185, 241, 249] bt.row_indices=[80, 81, 88, 89, 112, 113, 120, 121]
[ok] verify_plain_proof: ok=True ('Mining solution verified successfully', 0.2 ms)
[ok] all checks passed — Pearl mining works on this host
```
The `a.row_indices` and `bt.row_indices` values above are not constants — they're
`(offset + ROWS_PATTERN)` and `(offset + COLS_PATTERN)` for whichever offset the
miner happened to find a jackpot at. The smoke test verifies the *deltas* match
the configured `PeriodicPattern`, not the absolute values.
If it succeeds, this host can mine Pearl using the OpenJarvis `cpu-pearl` provider
(see Spec B §13). If it fails, the `[fail]` line tells you which step broke.
## What this proves (and what it doesn't)
**Proves:**
- The Pearl mining algorithm executes correctly on this host's CPU.
- Generated proofs verify under `verify_plain_proof`. (This is the same check
validators run on the inputs to the ZK proof.)
- The whole stack — `pearl-blake3`, `zk-pow`, `py-pearl-mining` — builds and
loads as a native CPython extension.
**Does NOT prove:**
- Network-difficulty hashrate. The smoke test uses
`nbits=0x1D2FFFFF` (test difficulty), much easier than mainnet. Real mining
expected hashrate on Apple Silicon CPU is several orders of magnitude lower
per share — see Spec B §1.5.6.
- ZK proof generation throughput. The smoke test calls `verify_plain_proof`,
not `generate_proof`. Plonky2 STARK proving takes seconds-to-minutes of CPU
per block (Spec B Open Q10).
- That this host can keep up with the network's block production rate.
## When to update this
- When Pearl bumps `py-pearl-mining` API: re-run the smoke test against the
new ref pinned in `OpenJarvis/src/openjarvis/mining/_constants.py`.
- When Pearl publishes a Mac wheel to PyPI: simplify the install instructions
above, drop the local `maturin build` step.
- When Spec B v2 adds the PyTorch-MPS reference path: extend `smoke_test.py`
with an MPS path comparison. The `miner-base` reference is already in
PyTorch, so the v2 smoke test would be a different test invoking
`miner_base.NoisyGemm` and comparing CPU vs MPS outputs for parity.
-139
View File
@@ -1,139 +0,0 @@
"""Pearl mining smoke test — runs an end-to-end mine + verify cycle.
Verifies that this host can run Pearl's pure-Rust mining algorithm via the
`pearl_mining` Python package. Used as Phase 0-B of the OpenJarvis Apple Silicon
mining spec ([Spec B]).
Exit codes:
0 all checks passed
1 pearl_mining import failed
2 mine() failed
3 verify_plain_proof rejected the proof
4 timing or sanity check failed
[Spec B]: ../../docs/design/2026-05-05-apple-silicon-pearl-mining-design.md
"""
from __future__ import annotations
import platform
import sys
import time
# Test fixture values — match upstream Pearl's tests/test_python_api.py so we are
# testing the same code path that Pearl's own CI exercises. Do not change
# without re-syncing with upstream.
DEFAULT_NBITS = 0x1D2FFFFF
DEFAULT_M = 256
DEFAULT_N = 128
DEFAULT_K = 1024
DEFAULT_RANK = 32
ROWS_PATTERN = [0, 8, 64, 72]
COLS_PATTERN = [0, 1, 8, 9, 32, 33, 40, 41]
def _ok(msg: str) -> None:
print(f"[ok] {msg}")
def _fail(msg: str, code: int) -> None:
print(f"[fail] {msg}")
sys.exit(code)
def main() -> None:
print(f"host: {platform.platform()} ({platform.machine()})")
print(f"python: {sys.version.split()[0]}")
try:
import pearl_mining
except ImportError as e:
_fail(f"could not import pearl_mining — install with `uv pip install py-pearl-mining` or build from source: {e}", 1)
_ok(f"pearl_mining loaded from {pearl_mining.__file__}")
_ok(
f"PUBLICDATA_SIZE={pearl_mining.PUBLICDATA_SIZE} "
f"MERKLE_LEAF_SIZE={pearl_mining.MERKLE_LEAF_SIZE}"
)
block_header = pearl_mining.IncompleteBlockHeader(
version=0,
prev_block=b"\x00" * 32,
merkle_root=b"0123456789abcdef" * 2,
timestamp=0x66666666,
nbits=DEFAULT_NBITS,
)
mining_config = pearl_mining.MiningConfiguration(
common_dim=DEFAULT_K,
rank=DEFAULT_RANK,
mma_type=pearl_mining.MMAType.Int7xInt7ToInt32,
rows_pattern=pearl_mining.PeriodicPattern.from_list(ROWS_PATTERN),
cols_pattern=pearl_mining.PeriodicPattern.from_list(COLS_PATTERN),
reserved=pearl_mining.MiningConfiguration.RESERVED,
)
t0 = time.perf_counter()
try:
plain_proof = pearl_mining.mine(
DEFAULT_M,
DEFAULT_N,
DEFAULT_K,
block_header,
mining_config,
signal_range=None,
wrong_jackpot_hash=False,
)
except Exception as e:
_fail(f"mine() raised: {e!r}", 2)
t_mine = time.perf_counter() - t0
_ok(
f"mine(m={DEFAULT_M}, n={DEFAULT_N}, k={DEFAULT_K}, rank={DEFAULT_RANK}) "
f"returned a proof in {t_mine:.3f} s"
)
print(
f" proof.m={plain_proof.m} proof.n={plain_proof.n} proof.k={plain_proof.k} "
f"noise_rank={plain_proof.noise_rank}"
)
print(
f" a.row_indices={plain_proof.a.row_indices} "
f"bt.row_indices={plain_proof.bt.row_indices}"
)
t0 = time.perf_counter()
ok, msg = pearl_mining.verify_plain_proof(block_header, plain_proof)
t_verify_ms = (time.perf_counter() - t0) * 1000
if not ok:
_fail(f"verify_plain_proof rejected our proof: {msg}", 3)
_ok(f"verify_plain_proof: ok=True ({msg!r}, {t_verify_ms:.1f} ms)")
if plain_proof.m != DEFAULT_M or plain_proof.n != DEFAULT_N or plain_proof.k != DEFAULT_K:
_fail("plain_proof dimensions do not match request", 4)
if plain_proof.noise_rank != DEFAULT_RANK:
_fail("plain_proof noise_rank does not match request", 4)
# Row indices are (offset + base_index) for some valid offset within the
# matrix dimension — see threads_partition() in zk-pow/src/ffi/mine.rs.
# We can't assert an absolute value (different offsets are valid every run),
# but we can assert the deltas match the pattern shape.
a_idxs = list(plain_proof.a.row_indices)
bt_idxs = list(plain_proof.bt.row_indices)
a_deltas = [v - a_idxs[0] for v in a_idxs]
bt_deltas = [v - bt_idxs[0] for v in bt_idxs]
if a_deltas != ROWS_PATTERN:
_fail(f"a.row_indices deltas ({a_deltas}) != ROWS_PATTERN ({ROWS_PATTERN})", 4)
if bt_deltas != COLS_PATTERN:
_fail(f"bt.row_indices deltas ({bt_deltas}) != COLS_PATTERN ({COLS_PATTERN})", 4)
print()
print("[ok] all checks passed — Pearl mining works on this host")
print()
print("Note: this used test difficulty (nbits=0x1D2FFFFF), not mainnet.")
print("Real-network shares per second will be many orders of magnitude lower.")
print("See docs/design/2026-05-05-apple-silicon-pearl-mining-design.md §1.5.6")
if __name__ == "__main__":
main()