mirror of
https://github.com/open-jarvis/OpenJarvis.git
synced 2026-08-15 17:31:58 +00:00
Compare commits
18
Commits
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"schemaVersion": 1,
|
||||
"label": "Git Clones",
|
||||
"message": "140,724",
|
||||
"message": "159,322",
|
||||
"color": "green",
|
||||
"namedLogo": "git"
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"total_clones": 140724,
|
||||
"last_updated": "2026-07-02T07:10:21Z",
|
||||
"total_clones": 159322,
|
||||
"last_updated": "2026-07-16T08:08:14Z",
|
||||
"daily": {
|
||||
"2026-03-27": 2189,
|
||||
"2026-03-28": 1874,
|
||||
@@ -98,6 +98,20 @@
|
||||
"2026-06-28": 1028,
|
||||
"2026-06-29": 765,
|
||||
"2026-06-30": 951,
|
||||
"2026-07-01": 1134
|
||||
"2026-07-01": 1134,
|
||||
"2026-07-02": 593,
|
||||
"2026-07-03": 537,
|
||||
"2026-07-04": 411,
|
||||
"2026-07-05": 485,
|
||||
"2026-07-06": 555,
|
||||
"2026-07-07": 905,
|
||||
"2026-07-08": 1171,
|
||||
"2026-07-09": 1857,
|
||||
"2026-07-10": 1181,
|
||||
"2026-07-11": 2185,
|
||||
"2026-07-12": 1917,
|
||||
"2026-07-13": 2102,
|
||||
"2026-07-14": 2337,
|
||||
"2026-07-15": 2362
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
<p><i>Personal AI, On Personal Devices.</i></p>
|
||||
|
||||
<p>
|
||||
<a href="https://scalingintelligence.stanford.edu/blogs/openjarvis/"><img src="https://img.shields.io/badge/project-OpenJarvis-blue" alt="Project"></a>
|
||||
<a href="https://openjarvis.stanford.edu/"><img src="https://img.shields.io/badge/project-OpenJarvis-blue" alt="Project"></a>
|
||||
<a href="https://open-jarvis.github.io/OpenJarvis/"><img src="https://img.shields.io/badge/docs-mkdocs-blue" alt="Docs"></a>
|
||||
<img src="https://img.shields.io/badge/python-%3E%3D3.10-blue" alt="Python">
|
||||
<img src="https://img.shields.io/badge/license-Apache%202.0-green" alt="License">
|
||||
@@ -23,7 +23,7 @@
|
||||
|
||||
> **[Documentation](https://open-jarvis.github.io/OpenJarvis/)**
|
||||
>
|
||||
> **[Project Site](https://scalingintelligence.stanford.edu/blogs/openjarvis/)**
|
||||
> **[Project Site](https://openjarvis.stanford.edu/)**
|
||||
>
|
||||
> **[Paper](https://arxiv.org/abs/2605.17172)**
|
||||
>
|
||||
|
||||
+1
-1
@@ -215,7 +215,7 @@ OpenJarvis is built around five composable layers. Each has a clean interface an
|
||||
|
||||
OpenJarvis is part of [Intelligence Per Watt](https://www.intelligence-per-watt.ai/), a research initiative studying the efficiency of on-device AI systems. Developed at [Hazy Research](https://hazyresearch.stanford.edu/) and the [Scaling Intelligence Lab](https://scalingintelligence.stanford.edu/) at [Stanford SAIL](https://ai.stanford.edu/).
|
||||
|
||||
Read the [blog post](https://scalingintelligence.stanford.edu/blogs/openjarvis/) for the full research motivation, architecture details, and experimental results.
|
||||
Read the [blog post](https://openjarvis.stanford.edu/) for the full research motivation, architecture details, and experimental results.
|
||||
|
||||
## Citation
|
||||
|
||||
|
||||
+1
-1
@@ -55,5 +55,5 @@ See how the OpenJarvis community saves money, energy, and compute by running AI
|
||||
<div id="leaderboard-pagination" class="lb-pagination"></div>
|
||||
|
||||
<p style="font-size:12px;opacity:0.6;margin-top:12px">
|
||||
*Dollar savings estimated vs. Claude Opus 4.6 API pricing ($5/1M input, $25/1M output tokens). Assumes local open-source models produce roughly the same number of tokens per request as cloud models.
|
||||
*Dollar savings estimated vs. Claude Fable 5 API pricing ($10/1M input, $50/1M output tokens). Assumes local open-source models produce roughly the same number of tokens per request as cloud models.
|
||||
</p>
|
||||
|
||||
@@ -29,8 +29,8 @@ interface TelemetryStats {
|
||||
}
|
||||
|
||||
const CLOUD_PRICING = [
|
||||
{ name: 'GPT-5.3', input: 2.00, output: 10.00, primary: true },
|
||||
{ name: 'Claude Opus 4.6', input: 5.00, output: 25.00, primary: false },
|
||||
{ name: 'GPT-5.6 Sol', input: 5.00, output: 30.00, primary: true },
|
||||
{ name: 'Claude Fable 5', input: 10.00, output: 50.00, primary: false },
|
||||
{ name: 'Gemini 3.1 Pro', input: 2.00, output: 12.00, primary: false },
|
||||
];
|
||||
|
||||
|
||||
@@ -2,8 +2,8 @@ import { DollarSign, TrendingDown, Cloud, HardDrive } from 'lucide-react';
|
||||
import { useAppStore } from '../../lib/store';
|
||||
|
||||
const CLOUD_PRICING = [
|
||||
{ name: 'GPT-5.3', input: 2.00, output: 10.00 },
|
||||
{ name: 'Claude Opus 4.6', input: 5.00, output: 25.00 },
|
||||
{ name: 'GPT-5.6 Sol', input: 5.00, output: 30.00 },
|
||||
{ name: 'Claude Fable 5', input: 10.00, output: 50.00 },
|
||||
{ name: 'Gemini 3.1 Pro', input: 2.00, output: 12.00 },
|
||||
];
|
||||
|
||||
|
||||
@@ -222,8 +222,8 @@ const styles: Record<string, React.CSSProperties> = {
|
||||
};
|
||||
|
||||
const PROVIDER_COLORS: Record<string, string> = {
|
||||
'gpt-5.3': colors.green,
|
||||
'claude-opus-4.6': colors.yellow,
|
||||
'gpt-5.6-sol': colors.green,
|
||||
'claude-fable-5': colors.yellow,
|
||||
'gemini-3.1-pro': colors.accent,
|
||||
};
|
||||
|
||||
|
||||
@@ -3740,8 +3740,8 @@ export function AgentsPage() {
|
||||
const paramsB = paramMatch ? parseFloat(paramMatch[1]) : 9;
|
||||
const flops = 2 * paramsB * 1e9 * (inTok + outTok);
|
||||
const providers = [
|
||||
{ label: 'GPT-5.3', inPer1M: 2.0, outPer1M: 10.0 },
|
||||
{ label: 'Claude Opus 4.6', inPer1M: 5.0, outPer1M: 25.0 },
|
||||
{ label: 'GPT-5.6 Sol', inPer1M: 5.0, outPer1M: 30.0 },
|
||||
{ label: 'Claude Fable 5', inPer1M: 10.0, outPer1M: 50.0 },
|
||||
{ label: 'Gemini 3.1 Pro', inPer1M: 2.0, outPer1M: 12.0 },
|
||||
];
|
||||
const energyWh = (inTok + outTok) / 1000 * 0.4;
|
||||
|
||||
@@ -805,7 +805,7 @@ export function SettingsPage() {
|
||||
</p>
|
||||
<div className="flex gap-3 mt-3 text-xs">
|
||||
<a
|
||||
href="https://scalingintelligence.stanford.edu/blogs/openjarvis/"
|
||||
href="https://openjarvis.stanford.edu/"
|
||||
target="_blank"
|
||||
rel="noopener noreferrer"
|
||||
style={{ color: 'var(--color-accent)' }}
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from openjarvis.engine._base import looks_like_context_length_error
|
||||
|
||||
|
||||
class AgentTickError(Exception):
|
||||
"""Base class for agent tick errors."""
|
||||
@@ -64,6 +66,14 @@ def classify_error(exc: Exception) -> AgentTickError:
|
||||
|
||||
msg = str(exc).lower()
|
||||
|
||||
# A context-window overflow is deterministic: retrying the identical
|
||||
# over-length request can never succeed, so fail fast instead of burning
|
||||
# the retry budget on it.
|
||||
if getattr(exc, "is_context_length_error", False) or (
|
||||
looks_like_context_length_error(msg)
|
||||
):
|
||||
return FatalError(str(exc))
|
||||
|
||||
# Check fatal patterns first (more specific)
|
||||
if isinstance(exc, PermissionError):
|
||||
return FatalError(str(exc))
|
||||
@@ -90,6 +100,11 @@ def retry_delay(attempt: int) -> int:
|
||||
def suggest_action(error: AgentTickError) -> str:
|
||||
"""Return a human-readable suggested action for the given error."""
|
||||
msg = str(error).lower()
|
||||
if looks_like_context_length_error(msg):
|
||||
return (
|
||||
"Conversation too long for the model's context window \u2014 "
|
||||
"start a new chat or shorten the conversation"
|
||||
)
|
||||
if any(p in msg for p in ("rate limit", "rate_limit", "429", "too many requests")):
|
||||
return "Rate limited \u2014 agent will auto-retry on next tick"
|
||||
if any(p in msg for p in ("timeout", "timed out", "connection", "unavailable")):
|
||||
|
||||
@@ -1403,12 +1403,10 @@ def _loop_local(
|
||||
# server walled the call. Compact aggressively (keep_last=1)
|
||||
# and retry once. Re-raise on anything else or on a second
|
||||
# failure — the runner records the row as errored.
|
||||
from openjarvis.engine._base import looks_like_context_length_error
|
||||
|
||||
msg = str(exc)
|
||||
is_ctx = (
|
||||
"maximum context length" in msg
|
||||
or "context length" in msg.lower()
|
||||
and "exceed" in msg.lower()
|
||||
)
|
||||
is_ctx = looks_like_context_length_error(msg)
|
||||
if not is_ctx:
|
||||
raise
|
||||
_record_event(
|
||||
|
||||
@@ -20,6 +20,7 @@ from openjarvis.core.events import EventBus, EventType
|
||||
from openjarvis.core.types import Message, Role
|
||||
from openjarvis.engine import (
|
||||
EngineConnectionError,
|
||||
EngineContextLengthError,
|
||||
discover_engines,
|
||||
discover_models,
|
||||
get_engine,
|
||||
@@ -881,6 +882,11 @@ def ask(
|
||||
capability_policy=sec.capability_policy,
|
||||
memory_files_config=effective_mf,
|
||||
)
|
||||
except EngineContextLengthError as exc:
|
||||
# Not a reachability problem — pointing the user at server/host
|
||||
# config (hint_no_engine) would be misleading here.
|
||||
console.print(f"[red]{exc}[/red]")
|
||||
sys.exit(1)
|
||||
except EngineConnectionError as exc:
|
||||
console.print(f"[red]Engine error:[/red] {exc}")
|
||||
console.print(hint_no_engine())
|
||||
@@ -990,6 +996,11 @@ def ask(
|
||||
temperature=temperature,
|
||||
max_tokens=max_tokens,
|
||||
)
|
||||
except EngineContextLengthError as exc:
|
||||
# Not a reachability problem — pointing the user at server/host
|
||||
# config (hint_no_engine) would be misleading here.
|
||||
console.print(f"[red]{exc}[/red]")
|
||||
sys.exit(1)
|
||||
except EngineConnectionError as exc:
|
||||
console.print(f"[red]Engine error:[/red] {exc}")
|
||||
console.print(hint_no_engine())
|
||||
|
||||
@@ -9,7 +9,9 @@ import openjarvis.engine.ollama # noqa: F401
|
||||
import openjarvis.engine.openai_compat_engines # noqa: F401
|
||||
from openjarvis.engine._base import (
|
||||
EngineConnectionError,
|
||||
EngineContextLengthError,
|
||||
InferenceEngine,
|
||||
looks_like_context_length_error,
|
||||
messages_to_dicts,
|
||||
)
|
||||
from openjarvis.engine._discovery import discover_engines, discover_models, get_engine
|
||||
@@ -23,9 +25,11 @@ for _optional in ("cloud", "litellm", "gemma_cpp"):
|
||||
|
||||
__all__ = [
|
||||
"EngineConnectionError",
|
||||
"EngineContextLengthError",
|
||||
"InferenceEngine",
|
||||
"discover_engines",
|
||||
"discover_models",
|
||||
"get_engine",
|
||||
"looks_like_context_length_error",
|
||||
"messages_to_dicts",
|
||||
]
|
||||
|
||||
@@ -13,6 +13,46 @@ class EngineConnectionError(Exception):
|
||||
"""Raised when an engine is unreachable."""
|
||||
|
||||
|
||||
class EngineContextLengthError(EngineConnectionError):
|
||||
"""The prompt exceeds the served model's maximum context window.
|
||||
|
||||
Subclasses ``EngineConnectionError`` so existing ``except
|
||||
EngineConnectionError`` handlers keep catching it, while callers that want a
|
||||
distinct, user-facing "conversation too long" message can branch on this type
|
||||
(or the ``is_context_length_error`` marker) instead of surfacing a generic
|
||||
engine failure.
|
||||
"""
|
||||
|
||||
is_context_length_error: bool = True
|
||||
|
||||
|
||||
# Substrings that identify an error body as a context-window overflow (vLLM,
|
||||
# SGLang, and OpenAI-compatible servers phrase this a few different ways).
|
||||
# Every marker is anchored on "context" on purpose: generic phrases like
|
||||
# "please reduce" or "too many tokens" also appear in unrelated 400 bodies
|
||||
# (max_tokens validation, rate limiting, oversized images) and would
|
||||
# misclassify those as "conversation too long".
|
||||
CONTEXT_LENGTH_MARKERS = (
|
||||
"context length",
|
||||
"maximum context",
|
||||
"context window",
|
||||
"maximum_context",
|
||||
"context_length_exceeded",
|
||||
)
|
||||
|
||||
|
||||
def looks_like_context_length_error(text: str) -> bool:
|
||||
"""True when *text* reads like a context-window overflow error.
|
||||
|
||||
The single shared heuristic for recognizing vendor context-overflow
|
||||
phrasings — used by the engine layer (typing upstream 400s), agent error
|
||||
classification, and the server stream bridge, so a new vendor phrasing
|
||||
only ever needs to be added here.
|
||||
"""
|
||||
low = (text or "").lower()
|
||||
return any(marker in low for marker in CONTEXT_LENGTH_MARKERS)
|
||||
|
||||
|
||||
_REASONING_METADATA_KEYS = ("reasoning_content", "thinking")
|
||||
|
||||
|
||||
@@ -80,8 +120,11 @@ def estimate_prompt_tokens(messages: Sequence[Message]) -> int:
|
||||
|
||||
|
||||
__all__ = [
|
||||
"CONTEXT_LENGTH_MARKERS",
|
||||
"EngineConnectionError",
|
||||
"EngineContextLengthError",
|
||||
"InferenceEngine",
|
||||
"estimate_prompt_tokens",
|
||||
"looks_like_context_length_error",
|
||||
"messages_to_dicts",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
"""Shared async-HTTP plumbing for engines that stream over httpx.
|
||||
|
||||
Home of the pieces the OpenAI-compat and Ollama engines were each hand-rolling:
|
||||
the async-client factory (with the configured timeout applied), a cached
|
||||
long-lived client so consecutive streams reuse pooled connections instead of
|
||||
paying a fresh TCP/TLS handshake per turn, the transport-error set that maps to
|
||||
``EngineConnectionError``, and the non-2xx → engine-error translation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import NoReturn
|
||||
|
||||
import httpx
|
||||
|
||||
from openjarvis.engine._base import (
|
||||
EngineConnectionError,
|
||||
EngineContextLengthError,
|
||||
looks_like_context_length_error,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Transport failures that map to EngineConnectionError on the streaming paths.
|
||||
# ``RemoteProtocolError``/``ReadError`` cover a server dying MID-STREAM (peer
|
||||
# closed between tokens); a wedged read trips the configured timeout
|
||||
# (TimeoutException). Kept exactly this narrow on purpose:
|
||||
# ``asyncio.CancelledError``/``GeneratorExit`` are NOT ``httpx.TransportError``
|
||||
# subclasses and must keep propagating for correct cancellation.
|
||||
STREAM_TRANSPORT_ERRORS = (
|
||||
httpx.ConnectError,
|
||||
httpx.TimeoutException,
|
||||
httpx.RemoteProtocolError,
|
||||
httpx.ReadError,
|
||||
)
|
||||
|
||||
_CONTEXT_LENGTH_USER_MESSAGE = (
|
||||
"The conversation is too long for the model's context window. "
|
||||
"Start a new chat or shorten the conversation, then try again."
|
||||
)
|
||||
|
||||
|
||||
class AsyncHTTPEngineMixin:
|
||||
"""Async streaming plumbing shared by httpx-backed engines.
|
||||
|
||||
Expects the engine to provide ``engine_id``, ``_host``, ``_timeout``, an
|
||||
``_async_transport`` test seam (``httpx.MockTransport`` in tests, ``None``
|
||||
in production), and optionally ``_headers``.
|
||||
"""
|
||||
|
||||
engine_id: str
|
||||
_host: str
|
||||
_timeout: float
|
||||
_async_transport: httpx.AsyncBaseTransport | None
|
||||
|
||||
# Set True by engines whose upstream reports context-window overflows in
|
||||
# 400 bodies (OpenAI-compat servers). Ollama has no such signal.
|
||||
_stream_400_signals_context_length: bool = False
|
||||
|
||||
# Lazily-created shared client (and the loop it belongs to). Class-level
|
||||
# ``None`` defaults keep engine ``__init__``s free of mixin bookkeeping.
|
||||
_async_client: httpx.AsyncClient | None = None
|
||||
_async_client_loop: asyncio.AbstractEventLoop | None = None
|
||||
|
||||
def _make_async_client(self) -> httpx.AsyncClient:
|
||||
"""Build an async client that honours the configured timeout."""
|
||||
return httpx.AsyncClient(
|
||||
base_url=self._host,
|
||||
timeout=self._timeout,
|
||||
headers=getattr(self, "_headers", None),
|
||||
transport=self._async_transport,
|
||||
)
|
||||
|
||||
def _get_async_client(self) -> httpx.AsyncClient:
|
||||
"""Return the shared async client for the running event loop.
|
||||
|
||||
Reusing one client across calls preserves connection pooling — without
|
||||
it every conversation turn pays a fresh TCP (and TLS) handshake. The
|
||||
client is cached per event loop: pooled connections die with their
|
||||
loop, so CLI flows that run ``asyncio.run()`` per turn transparently
|
||||
get a fresh client while a long-lived server loop keeps one pool.
|
||||
"""
|
||||
loop = asyncio.get_running_loop()
|
||||
client = self._async_client
|
||||
if client is None or client.is_closed or self._async_client_loop is not loop:
|
||||
# Any previous client belonged to a finished loop; its pooled
|
||||
# connections are already dead, so just drop the reference.
|
||||
client = self._make_async_client()
|
||||
self._async_client = client
|
||||
self._async_client_loop = loop
|
||||
return client
|
||||
|
||||
def _close_async_client(self) -> None:
|
||||
"""Best-effort close of the shared async client (for ``close()``)."""
|
||||
client = self._async_client
|
||||
loop = self._async_client_loop
|
||||
self._async_client = None
|
||||
self._async_client_loop = None
|
||||
if client is None or client.is_closed:
|
||||
return
|
||||
try:
|
||||
if loop is not None and not loop.is_closed():
|
||||
if loop.is_running():
|
||||
loop.create_task(client.aclose())
|
||||
else:
|
||||
loop.run_until_complete(client.aclose())
|
||||
except Exception: # noqa: BLE001 — cleanup must never mask the close
|
||||
logger.debug("Async client did not close cleanly", exc_info=True)
|
||||
|
||||
def _raise_stream_http_error(self, status: int, detail: str) -> NoReturn:
|
||||
"""Map a non-success streaming HTTP response to a clean engine error."""
|
||||
detail = (detail or "").strip()
|
||||
if (
|
||||
status == 400
|
||||
and self._stream_400_signals_context_length
|
||||
and looks_like_context_length_error(detail)
|
||||
):
|
||||
raise EngineContextLengthError(_CONTEXT_LENGTH_USER_MESSAGE)
|
||||
detail_suffix = f": {detail}" if detail else ""
|
||||
raise EngineConnectionError(
|
||||
f"{self.engine_id} engine at {self._host} returned HTTP "
|
||||
f"{status}{detail_suffix}"
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["AsyncHTTPEngineMixin", "STREAM_TRANSPORT_ERRORS"]
|
||||
@@ -12,18 +12,27 @@ import httpx
|
||||
from openjarvis.core.types import Message
|
||||
from openjarvis.engine._base import (
|
||||
EngineConnectionError,
|
||||
EngineContextLengthError,
|
||||
InferenceEngine,
|
||||
estimate_prompt_tokens,
|
||||
messages_to_dicts,
|
||||
)
|
||||
from openjarvis.engine._http_async import (
|
||||
STREAM_TRANSPORT_ERRORS,
|
||||
AsyncHTTPEngineMixin,
|
||||
)
|
||||
from openjarvis.engine._stubs import StreamChunk
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class _OpenAICompatibleEngine(InferenceEngine):
|
||||
class _OpenAICompatibleEngine(AsyncHTTPEngineMixin, InferenceEngine):
|
||||
"""Base for engines that serve the OpenAI ``/v1/chat/completions`` API."""
|
||||
|
||||
# vLLM/SGLang report context-window overflows in 400 bodies; the shared
|
||||
# ``_raise_stream_http_error`` types those as ``EngineContextLengthError``.
|
||||
_stream_400_signals_context_length = True
|
||||
|
||||
engine_id: str = ""
|
||||
_default_host: str = "http://localhost:8000"
|
||||
_api_prefix: str = "/v1"
|
||||
@@ -50,6 +59,16 @@ class _OpenAICompatibleEngine(InferenceEngine):
|
||||
headers = (
|
||||
{"Authorization": f"Bearer {self._api_key}"} if self._api_key else None
|
||||
)
|
||||
# Used by the shared async streaming plumbing (AsyncHTTPEngineMixin) so
|
||||
# the bounded request timeout is applied to streaming reads, not just
|
||||
# the synchronous methods (a wedged token read fails at ``timeout``
|
||||
# rather than hanging the caller for the httpx default).
|
||||
self._timeout = timeout
|
||||
self._headers = headers
|
||||
# Injection seam for tests: an ``httpx.MockTransport`` swapped in here lets
|
||||
# the async stream path be exercised with a mocked transport and no real
|
||||
# server. ``None`` in production so httpx uses its default networking.
|
||||
self._async_transport: httpx.AsyncBaseTransport | None = None
|
||||
self._client = httpx.Client(
|
||||
base_url=self._host, timeout=timeout, headers=headers
|
||||
)
|
||||
@@ -168,11 +187,26 @@ class _OpenAICompatibleEngine(InferenceEngine):
|
||||
# Default to tool_choice=auto when tools are provided
|
||||
if "tools" in payload and "tool_choice" not in payload:
|
||||
payload["tool_choice"] = "auto"
|
||||
url = f"{self._api_prefix}/chat/completions"
|
||||
try:
|
||||
url = f"{self._api_prefix}/chat/completions"
|
||||
with self._client.stream("POST", url, json=payload) as resp:
|
||||
resp.raise_for_status()
|
||||
for line in resp.iter_lines():
|
||||
# ASYNC streaming: ``httpx.AsyncClient`` + ``aiter_lines`` never
|
||||
# blocks the event loop between tokens (the previous SYNC
|
||||
# ``httpx.Client`` + ``iter_lines`` inside this ``async def`` blocked
|
||||
# the single uvicorn worker on every inter-token wait, serializing all
|
||||
# concurrent chats and letting one wedged read freeze the whole API).
|
||||
# The shared client keeps pooled connections across turns.
|
||||
client = self._get_async_client()
|
||||
async with client.stream("POST", url, json=payload) as resp:
|
||||
# ``not is_success`` covers 3xx as well as 4xx/5xx. With
|
||||
# ``follow_redirects`` off (the default) an unexpected redirect
|
||||
# would otherwise fall through to ``aiter_lines`` and surface as
|
||||
# a silent EMPTY stream instead of a clean engine error.
|
||||
if not resp.is_success:
|
||||
# Load the (short) error body before touching ``.text``:
|
||||
# a streaming response is otherwise unread.
|
||||
await resp.aread()
|
||||
self._raise_stream_http_error(resp.status_code, resp.text)
|
||||
async for line in resp.aiter_lines():
|
||||
if not line.startswith("data:"):
|
||||
continue
|
||||
data_str = line[len("data:") :].strip()
|
||||
@@ -186,7 +220,11 @@ class _OpenAICompatibleEngine(InferenceEngine):
|
||||
content = delta.get("content")
|
||||
if content:
|
||||
yield content
|
||||
except (httpx.ConnectError, httpx.TimeoutException) as exc:
|
||||
except STREAM_TRANSPORT_ERRORS as exc:
|
||||
# A wedged upstream read trips ``timeout`` (ReadTimeout) and is mapped
|
||||
# here, so the request fails cleanly at the configured bound instead
|
||||
# of hanging indefinitely (see STREAM_TRANSPORT_ERRORS for why the
|
||||
# set is exactly this narrow).
|
||||
raise EngineConnectionError(
|
||||
f"{self.engine_id} engine not reachable at {self._host}"
|
||||
) from exc
|
||||
@@ -212,11 +250,20 @@ class _OpenAICompatibleEngine(InferenceEngine):
|
||||
}
|
||||
if "tools" in payload and "tool_choice" not in payload:
|
||||
payload["tool_choice"] = "auto"
|
||||
url = f"{self._api_prefix}/chat/completions"
|
||||
try:
|
||||
url = f"{self._api_prefix}/chat/completions"
|
||||
with self._client.stream("POST", url, json=payload) as resp:
|
||||
resp.raise_for_status()
|
||||
for line in resp.iter_lines():
|
||||
# ASYNC streaming (see ``stream``): non-blocking shared client so
|
||||
# rich streaming never stalls the event loop and honours ``timeout``.
|
||||
client = self._get_async_client()
|
||||
async with client.stream("POST", url, json=payload) as resp:
|
||||
# ``not is_success`` covers 3xx as well as 4xx/5xx. With
|
||||
# ``follow_redirects`` off (the default) an unexpected redirect
|
||||
# would otherwise fall through to ``aiter_lines`` and surface as
|
||||
# a silent EMPTY stream instead of a clean engine error.
|
||||
if not resp.is_success:
|
||||
await resp.aread()
|
||||
self._raise_stream_http_error(resp.status_code, resp.text)
|
||||
async for line in resp.aiter_lines():
|
||||
if not line.startswith("data:"):
|
||||
continue
|
||||
data_str = line[len("data:") :].strip()
|
||||
@@ -240,7 +287,10 @@ class _OpenAICompatibleEngine(InferenceEngine):
|
||||
finish_reason=finish,
|
||||
usage=usage,
|
||||
)
|
||||
except (httpx.ConnectError, httpx.TimeoutException) as exc:
|
||||
except STREAM_TRANSPORT_ERRORS as exc:
|
||||
# See ``stream``: transport failures (incl. a mid-stream server
|
||||
# disconnect) map to a clean error; the set is kept narrow so
|
||||
# cancellation still propagates.
|
||||
raise EngineConnectionError(
|
||||
f"{self.engine_id} engine not reachable at {self._host}"
|
||||
) from exc
|
||||
@@ -279,6 +329,9 @@ class _OpenAICompatibleEngine(InferenceEngine):
|
||||
|
||||
def close(self) -> None:
|
||||
self._client.close()
|
||||
self._close_async_client()
|
||||
|
||||
|
||||
__all__ = ["_OpenAICompatibleEngine"]
|
||||
# ``EngineContextLengthError`` moved to ``openjarvis.engine._base``; re-exported
|
||||
# here for callers/tests that import it from this module.
|
||||
__all__ = ["_OpenAICompatibleEngine", "EngineContextLengthError"]
|
||||
|
||||
@@ -123,8 +123,12 @@ class LiteLLMEngine(InferenceEngine):
|
||||
call_kwargs["api_base"] = self._api_base
|
||||
call_kwargs.update(kwargs)
|
||||
|
||||
resp = litellm.completion(**call_kwargs)
|
||||
for chunk in resp:
|
||||
# ``acompletion`` + ``async for``: the sync ``litellm.completion`` used
|
||||
# before made a blocking network call (and blocking per-chunk reads)
|
||||
# inside this ``async def``, stalling the whole event loop between
|
||||
# tokens — the same bug the httpx engines' streaming paths had.
|
||||
resp = await litellm.acompletion(**call_kwargs)
|
||||
async for chunk in resp:
|
||||
delta = chunk.choices[0].delta if chunk.choices else None
|
||||
if delta and delta.content:
|
||||
yield delta.content
|
||||
|
||||
@@ -18,6 +18,10 @@ from openjarvis.engine._base import (
|
||||
estimate_prompt_tokens,
|
||||
messages_to_dicts,
|
||||
)
|
||||
from openjarvis.engine._http_async import (
|
||||
STREAM_TRANSPORT_ERRORS,
|
||||
AsyncHTTPEngineMixin,
|
||||
)
|
||||
from openjarvis.engine._stubs import StreamChunk
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -80,11 +84,15 @@ def _default_num_ctx() -> int:
|
||||
|
||||
|
||||
@EngineRegistry.register("ollama")
|
||||
class OllamaEngine(InferenceEngine):
|
||||
class OllamaEngine(AsyncHTTPEngineMixin, InferenceEngine):
|
||||
"""Ollama backend via its native HTTP API."""
|
||||
|
||||
engine_id = "ollama"
|
||||
|
||||
# Ollama has no context-length overflow signal in its 400 bodies, so the
|
||||
# shared ``_raise_stream_http_error`` keeps its default (no
|
||||
# ``EngineContextLengthError`` branch, unlike the OpenAI-compat engines).
|
||||
|
||||
_DEFAULT_HOST = "http://localhost:11434"
|
||||
|
||||
def __init__(
|
||||
@@ -98,6 +106,14 @@ class OllamaEngine(InferenceEngine):
|
||||
env_host = os.environ.get("OLLAMA_HOST")
|
||||
host = env_host or self._DEFAULT_HOST
|
||||
self._host = host.rstrip("/")
|
||||
# Used by the shared async streaming plumbing (AsyncHTTPEngineMixin) so a
|
||||
# wedged token read is bounded by ``timeout`` instead of hanging the
|
||||
# single event loop for the httpx default.
|
||||
self._timeout = timeout
|
||||
# Injection seam for tests: an ``httpx.MockTransport`` swapped in here drives
|
||||
# the async stream path with no real Ollama server. ``None`` in production so
|
||||
# httpx uses its default networking.
|
||||
self._async_transport: httpx.AsyncBaseTransport | None = None
|
||||
self._client = httpx.Client(base_url=self._host, timeout=timeout)
|
||||
# Last stream usage — captured from Ollama's final chunk
|
||||
self._last_stream_usage: Dict[str, int] = {}
|
||||
@@ -263,9 +279,26 @@ class OllamaEngine(InferenceEngine):
|
||||
elif kwargs["think"] is not None:
|
||||
payload["think"] = kwargs["think"]
|
||||
try:
|
||||
with self._client.stream("POST", "/api/chat", json=payload) as resp:
|
||||
resp.raise_for_status()
|
||||
for line in resp.iter_lines():
|
||||
# ASYNC streaming: ``httpx.AsyncClient`` + ``aiter_lines`` never
|
||||
# blocks the event loop between tokens (the previous SYNC
|
||||
# ``self._client`` + ``iter_lines`` inside this ``async def`` blocked
|
||||
# the single uvicorn worker on every inter-token wait, serializing all
|
||||
# concurrent chats and letting one wedged read freeze the whole API).
|
||||
# The shared client keeps pooled connections across turns.
|
||||
client = self._get_async_client()
|
||||
async with client.stream("POST", "/api/chat", json=payload) as resp:
|
||||
# ``not is_success`` covers 3xx as well as 4xx/5xx and maps
|
||||
# to ``EngineConnectionError`` (matching the OpenAI-compat
|
||||
# path) instead of leaking a raw ``httpx.HTTPStatusError``.
|
||||
# With redirects off (the default) an unexpected 3xx would
|
||||
# otherwise fall through to ``aiter_lines`` and surface as a
|
||||
# silent EMPTY stream rather than a clean engine error.
|
||||
if not resp.is_success:
|
||||
# Read the (short) error body before touching ``.text``:
|
||||
# a streaming response is otherwise unread.
|
||||
await resp.aread()
|
||||
self._raise_stream_http_error(resp.status_code, resp.text)
|
||||
async for line in resp.aiter_lines():
|
||||
if not line.strip():
|
||||
continue
|
||||
try:
|
||||
@@ -290,7 +323,10 @@ class OllamaEngine(InferenceEngine):
|
||||
"total_tokens": full_prompt + comp,
|
||||
}
|
||||
break
|
||||
except (httpx.ConnectError, httpx.TimeoutException) as exc:
|
||||
except STREAM_TRANSPORT_ERRORS as exc:
|
||||
# Transport failures (incl. a mid-stream server disconnect) map to a
|
||||
# clean error; the set is kept narrow (see STREAM_TRANSPORT_ERRORS)
|
||||
# so cancellation still propagates.
|
||||
raise EngineConnectionError(
|
||||
f"Ollama not reachable at {self._host}"
|
||||
) from exc
|
||||
@@ -356,19 +392,34 @@ class OllamaEngine(InferenceEngine):
|
||||
) -> AsyncIterator[StreamChunk]:
|
||||
"""Execute the streaming request and yield parsed StreamChunks."""
|
||||
try:
|
||||
with self._client.stream("POST", "/api/chat", json=payload) as resp:
|
||||
# ASYNC streaming (see ``stream``): shared ``AsyncClient`` +
|
||||
# ``aiter_lines`` so rich streaming never stalls the event loop and
|
||||
# honours ``timeout``.
|
||||
client = self._get_async_client()
|
||||
async with client.stream("POST", "/api/chat", json=payload) as resp:
|
||||
if resp.status_code == 400 and retry_without_tools:
|
||||
# Model doesn't support tools — retry without them.
|
||||
# PRESERVED: this specific 400 path must still trigger the
|
||||
# tools-less retry; only OTHER non-2xx responses map to
|
||||
# EngineConnectionError below.
|
||||
payload.pop("tools", None)
|
||||
async for c in self._run_stream(
|
||||
payload, messages, retry_without_tools=False
|
||||
):
|
||||
yield c
|
||||
return
|
||||
resp.raise_for_status()
|
||||
# ``not is_success`` covers 3xx as well as 4xx/5xx and maps
|
||||
# to ``EngineConnectionError`` (matching the OpenAI-compat
|
||||
# path) instead of leaking a raw ``httpx.HTTPStatusError``.
|
||||
# With redirects off (the default) an unexpected 3xx would
|
||||
# otherwise fall through to ``aiter_lines`` and surface as a
|
||||
# silent EMPTY stream rather than a clean engine error.
|
||||
if not resp.is_success:
|
||||
await resp.aread()
|
||||
self._raise_stream_http_error(resp.status_code, resp.text)
|
||||
|
||||
finish_reason: str | None = None
|
||||
for line in resp.iter_lines():
|
||||
async for line in resp.aiter_lines():
|
||||
if not line.strip():
|
||||
continue
|
||||
try:
|
||||
@@ -441,7 +492,10 @@ class OllamaEngine(InferenceEngine):
|
||||
usage=dict(self._last_stream_usage),
|
||||
)
|
||||
break
|
||||
except (httpx.ConnectError, httpx.TimeoutException) as exc:
|
||||
except STREAM_TRANSPORT_ERRORS as exc:
|
||||
# See ``stream``: transport failures (incl. a mid-stream server
|
||||
# disconnect) map to a clean error; the set is kept narrow so
|
||||
# cancellation still propagates.
|
||||
raise EngineConnectionError(
|
||||
f"Ollama not reachable at {self._host}"
|
||||
) from exc
|
||||
@@ -474,6 +528,7 @@ class OllamaEngine(InferenceEngine):
|
||||
|
||||
def close(self) -> None:
|
||||
self._client.close()
|
||||
self._close_async_client()
|
||||
|
||||
|
||||
__all__ = ["OllamaEngine"]
|
||||
|
||||
@@ -15,7 +15,7 @@ HARD RULE: Every reply MUST be ≤280 characters. Count before sending.
|
||||
- GitHub: https://github.com/open-jarvis/OpenJarvis
|
||||
- Docs: https://open-jarvis.github.io/OpenJarvis/
|
||||
- Discord: https://discord.gg/wfXEkpPX
|
||||
- Blog: https://scalingintelligence.stanford.edu/blogs/openjarvis/
|
||||
- Blog: https://openjarvis.stanford.edu/
|
||||
- Install: `git clone https://github.com/open-jarvis/OpenJarvis.git && cd OpenJarvis && uv sync`
|
||||
- CLI commands (ONLY these exist):
|
||||
- `jarvis init` — auto-detects hardware, configures engine
|
||||
|
||||
@@ -741,8 +741,11 @@ async def websocket_chat_stream(websocket: WebSocket):
|
||||
)
|
||||
except TypeError:
|
||||
# stream() didn't return an iterable; fall back to
|
||||
# generate()
|
||||
result = engine.generate(messages, model=model)
|
||||
# generate(). It makes a blocking upstream call, so run
|
||||
# it in a worker thread to keep the event loop free.
|
||||
result = await asyncio.to_thread(
|
||||
engine.generate, messages, model=model
|
||||
)
|
||||
content = (
|
||||
result.get("content", "")
|
||||
if isinstance(
|
||||
@@ -767,8 +770,11 @@ async def websocket_chat_stream(websocket: WebSocket):
|
||||
ended_at=_time.time(),
|
||||
)
|
||||
else:
|
||||
# No stream method — single-shot generate
|
||||
result = engine.generate(messages, model=model)
|
||||
# No stream method — single-shot generate. Blocking upstream
|
||||
# call, so run in a worker thread to keep the event loop free.
|
||||
result = await asyncio.to_thread(
|
||||
engine.generate, messages, model=model
|
||||
)
|
||||
content = (
|
||||
result.get("content", "")
|
||||
if isinstance(
|
||||
|
||||
@@ -215,8 +215,8 @@ COMPARISON_HTML = """\
|
||||
<tr>
|
||||
<th></th>
|
||||
<th>OpenJarvis (Local)</th>
|
||||
<th>GPT-5.3</th>
|
||||
<th>Claude Opus 4.6</th>
|
||||
<th>GPT-5.6 Sol</th>
|
||||
<th>Claude Fable 5</th>
|
||||
<th>Gemini 3.1 Pro</th>
|
||||
</tr>
|
||||
</thead>
|
||||
@@ -261,11 +261,11 @@ COMPARISON_HTML = """\
|
||||
<div class="cc-value">$0.00/mo</div>
|
||||
</div>
|
||||
<div class="calc-card cloud">
|
||||
<div class="cc-label">GPT-5.3</div>
|
||||
<div class="cc-label">GPT-5.6 Sol</div>
|
||||
<div class="cc-value" id="calc-gpt">--</div>
|
||||
</div>
|
||||
<div class="calc-card cloud">
|
||||
<div class="cc-label">Claude Opus 4.6</div>
|
||||
<div class="cc-label">Claude Fable 5</div>
|
||||
<div class="cc-value" id="calc-claude">--</div>
|
||||
</div>
|
||||
<div class="calc-card cloud">
|
||||
@@ -292,13 +292,13 @@ COMPARISON_HTML = """\
|
||||
<script>
|
||||
// Embedded data -- avoids API calls, keeps the page static and fast.
|
||||
const CLOUD_PRICING = {
|
||||
"gpt-5.3": {
|
||||
input_per_1m: 2.00, output_per_1m: 10.00,
|
||||
label: "GPT-5.3"
|
||||
"gpt-5.6-sol": {
|
||||
input_per_1m: 5.00, output_per_1m: 30.00,
|
||||
label: "GPT-5.6 Sol"
|
||||
},
|
||||
"claude-opus-4.6": {
|
||||
input_per_1m: 5.00, output_per_1m: 25.00,
|
||||
label: "Claude Opus 4.6"
|
||||
"claude-fable-5": {
|
||||
input_per_1m: 10.00, output_per_1m: 50.00,
|
||||
label: "Claude Fable 5"
|
||||
},
|
||||
"gemini-3.1-pro": {
|
||||
input_per_1m: 2.00, output_per_1m: 12.00,
|
||||
@@ -376,8 +376,8 @@ function updateTable() {
|
||||
const sc = SCENARIOS[activeScenario];
|
||||
const i = sc.avg_input_tokens, o = sc.avg_output_tokens;
|
||||
const c = sc.calls_per_month;
|
||||
const gpt = calcMonthlyCost(c, i, o, 'gpt-5.3');
|
||||
const claude = calcMonthlyCost(c, i, o, 'claude-opus-4.6');
|
||||
const gpt = calcMonthlyCost(c, i, o, 'gpt-5.6-sol');
|
||||
const claude = calcMonthlyCost(c, i, o, 'claude-fable-5');
|
||||
const gemini = calcMonthlyCost(c, i, o, 'gemini-3.1-pro');
|
||||
|
||||
document.getElementById('t-gpt-m').textContent = fmtDollar(gpt);
|
||||
@@ -410,8 +410,8 @@ function updateCalc() {
|
||||
const avgOut = tpc - avgIn;
|
||||
const callsPerMonth = cpd * 30;
|
||||
|
||||
const gpt = calcMonthlyCost(callsPerMonth, avgIn, avgOut, 'gpt-5.3');
|
||||
const claude = calcMonthlyCost(callsPerMonth, avgIn, avgOut, 'claude-opus-4.6');
|
||||
const gpt = calcMonthlyCost(callsPerMonth, avgIn, avgOut, 'gpt-5.6-sol');
|
||||
const claude = calcMonthlyCost(callsPerMonth, avgIn, avgOut, 'claude-fable-5');
|
||||
const gemini = calcMonthlyCost(callsPerMonth, avgIn, avgOut, 'gemini-3.1-pro');
|
||||
|
||||
document.getElementById('calc-gpt').textContent = fmtDollar(gpt) + '/mo';
|
||||
|
||||
@@ -184,7 +184,7 @@ DASHBOARD_HTML = """\
|
||||
<div class="providers">
|
||||
<div class="provider-card openai">
|
||||
<div class="pname">OpenAI</div>
|
||||
<div class="pmodel">GPT-5.3 — $2.00 / $10.00 per 1M tokens</div>
|
||||
<div class="pmodel">GPT-5.6 Sol — $5.00 / $30.00 per 1M tokens</div>
|
||||
<div class="savings-amount" id="save-openai">$0.00</div>
|
||||
<div class="breakdown">
|
||||
<div class="item">
|
||||
@@ -199,7 +199,7 @@ DASHBOARD_HTML = """\
|
||||
</div>
|
||||
<div class="provider-card anthropic">
|
||||
<div class="pname">Anthropic</div>
|
||||
<div class="pmodel">Claude Opus 4.6 — $5.00 / $25.00 per 1M tokens</div>
|
||||
<div class="pmodel">Claude Fable 5 — $10.00 / $50.00 per 1M tokens</div>
|
||||
<div class="savings-amount" id="save-anthropic">$0.00</div>
|
||||
<div class="breakdown">
|
||||
<div class="item">
|
||||
@@ -281,12 +281,12 @@ DASHBOARD_HTML = """\
|
||||
<div class="providers-heading">Energy & Compute Avoided</div>
|
||||
<div class="metrics-row">
|
||||
<div class="metric-card">
|
||||
<div class="mheading">Energy Saved (vs GPT-5.3)</div>
|
||||
<div class="mheading">Energy Saved (vs GPT-5.6 Sol)</div>
|
||||
<div class="mvalue green" id="energy-joules">0 <span class="munit">J</span></div>
|
||||
<div class="msub" id="energy-kwh">0 kWh of cloud datacenter energy avoided</div>
|
||||
</div>
|
||||
<div class="metric-card">
|
||||
<div class="mheading">FLOPs Avoided (vs GPT-5.3)</div>
|
||||
<div class="mheading">FLOPs Avoided (vs GPT-5.6 Sol)</div>
|
||||
<div class="mvalue purple" id="flops-val">0 <span class="munit">FLOP</span></div>
|
||||
<div class="msub" id="flops-sub">cloud compute operations not needed</div>
|
||||
</div>
|
||||
@@ -354,8 +354,8 @@ async function refresh() {
|
||||
providerMap[p.provider] = p;
|
||||
});
|
||||
|
||||
// OpenAI / GPT-5.3
|
||||
const oa = providerMap['gpt-5.3'] || {};
|
||||
// OpenAI / GPT-5.6 Sol
|
||||
const oa = providerMap['gpt-5.6-sol'] || {};
|
||||
document.getElementById('save-openai')
|
||||
.textContent = fmtDollar(oa.total_cost || 0);
|
||||
document.getElementById('save-openai-in')
|
||||
@@ -363,8 +363,8 @@ async function refresh() {
|
||||
document.getElementById('save-openai-out')
|
||||
.textContent = fmtDollar(oa.output_cost || 0);
|
||||
|
||||
// Anthropic / Claude Opus 4.6
|
||||
const an = providerMap['claude-opus-4.6'] || {};
|
||||
// Anthropic / Claude Fable 5
|
||||
const an = providerMap['claude-fable-5'] || {};
|
||||
document.getElementById('save-anthropic')
|
||||
.textContent = fmtDollar(an.total_cost || 0);
|
||||
document.getElementById('save-anthropic-in')
|
||||
@@ -384,13 +384,13 @@ async function refresh() {
|
||||
// Monthly projections
|
||||
const proj = d.monthly_projection || {};
|
||||
document.getElementById('proj-openai')
|
||||
.textContent = fmtDollar(proj['gpt-5.3'] || 0);
|
||||
.textContent = fmtDollar(proj['gpt-5.6-sol'] || 0);
|
||||
document.getElementById('proj-anthropic')
|
||||
.textContent = fmtDollar(proj['claude-opus-4.6'] || 0);
|
||||
.textContent = fmtDollar(proj['claude-fable-5'] || 0);
|
||||
document.getElementById('proj-google')
|
||||
.textContent = fmtDollar(proj['gemini-3.1-pro'] || 0);
|
||||
|
||||
// Energy / FLOPs (use GPT-5.3 as reference)
|
||||
// Energy / FLOPs (use GPT-5.6 Sol as reference)
|
||||
const ej = oa.energy_joules || 0;
|
||||
const eWh = oa.energy_wh || 0;
|
||||
const fl = oa.flops || 0;
|
||||
|
||||
@@ -231,8 +231,13 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
|
||||
# tools (e.g. injecting MCP tools through this endpoint and wanting
|
||||
# the agent to execute them), add an explicit opt-in header rather
|
||||
# than removing this guard — silent re-routing is what produced #414.
|
||||
# ``_handle_agent`` (sync ``agent.run()``) and ``_handle_direct`` (sync
|
||||
# ``engine.generate()``) both make blocking upstream calls; run them in a
|
||||
# worker thread so a slow/wedged non-streaming request can't stall the
|
||||
# event loop and every other concurrent request with it.
|
||||
if agent is not None and not request_body.tools:
|
||||
response = _handle_agent(
|
||||
response = await asyncio.to_thread(
|
||||
_handle_agent,
|
||||
agent,
|
||||
model,
|
||||
request_body,
|
||||
@@ -242,7 +247,8 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
|
||||
)
|
||||
else:
|
||||
bus = getattr(request.app.state, "bus", None)
|
||||
response = _handle_direct(
|
||||
response = await asyncio.to_thread(
|
||||
_handle_direct,
|
||||
engine,
|
||||
model,
|
||||
request_body,
|
||||
|
||||
@@ -23,19 +23,19 @@ from openjarvis.core.types import TOKEN_COUNTING_VERSION # noqa: E402,F401
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
CLOUD_PRICING: Dict[str, Dict[str, float]] = {
|
||||
"gpt-5.3": {
|
||||
"input_per_1m": 2.00,
|
||||
"output_per_1m": 10.00,
|
||||
"label": "GPT-5.3",
|
||||
"gpt-5.6-sol": {
|
||||
"input_per_1m": 5.00,
|
||||
"output_per_1m": 30.00,
|
||||
"label": "GPT-5.6 Sol",
|
||||
"provider": "OpenAI",
|
||||
"params_b": 200.0,
|
||||
"energy_wh_per_1k_tokens": 0.4,
|
||||
"flops_per_token": 3.0e12,
|
||||
},
|
||||
"claude-opus-4.6": {
|
||||
"input_per_1m": 5.00,
|
||||
"output_per_1m": 25.00,
|
||||
"label": "Claude Opus 4.6",
|
||||
"claude-fable-5": {
|
||||
"input_per_1m": 10.00,
|
||||
"output_per_1m": 50.00,
|
||||
"label": "Claude Fable 5",
|
||||
"provider": "Anthropic",
|
||||
"params_b": 137.0,
|
||||
"energy_wh_per_1k_tokens": 0.5,
|
||||
|
||||
@@ -16,6 +16,7 @@ from fastapi.responses import StreamingResponse
|
||||
|
||||
from openjarvis.agents._stubs import AgentContext, BaseAgent
|
||||
from openjarvis.core.events import Event, EventBus, EventType
|
||||
from openjarvis.engine._base import looks_like_context_length_error
|
||||
from openjarvis.server.models import (
|
||||
ChatCompletionChunk,
|
||||
ChatCompletionRequest,
|
||||
@@ -194,8 +195,10 @@ class AgentStreamBridge:
|
||||
logger.error("Agent stream error: %s", exc, exc_info=True)
|
||||
|
||||
error_str = str(exc)
|
||||
if "context length" in error_str.lower() or (
|
||||
"400" in error_str and "too long" in error_str.lower()
|
||||
if (
|
||||
getattr(exc, "is_context_length_error", False)
|
||||
or looks_like_context_length_error(error_str)
|
||||
or ("400" in error_str and "too long" in error_str.lower())
|
||||
):
|
||||
error_content = (
|
||||
"The input is too long for the model's context window. "
|
||||
|
||||
@@ -6,6 +6,7 @@ and filtering operations that BM25 search cannot handle.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import sqlite3
|
||||
from typing import Any, Optional
|
||||
|
||||
@@ -16,10 +17,25 @@ from openjarvis.tools._stubs import BaseTool, ToolSpec
|
||||
|
||||
_MAX_ROWS = 50
|
||||
|
||||
# Write keywords are matched on word boundaries (mirroring db_query.py) so that
|
||||
# a read-only SELECT is not rejected just because a column/alias/literal happens
|
||||
# to contain one as a substring (e.g. "deleted_at", "created_at").
|
||||
_FORBIDDEN_RE = re.compile(
|
||||
r"\b(DROP|DELETE|INSERT|UPDATE|ALTER|CREATE|TRUNCATE|ATTACH)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
# String literals are stripped before the keyword scan so that data mentioning
|
||||
# a write keyword (e.g. WHERE content LIKE '%delete%') is not rejected. A write
|
||||
# "hidden" in a literal still cannot execute: the query must start with SELECT
|
||||
# and sqlite3 refuses multi-statement strings.
|
||||
_STRING_LITERAL_RE = re.compile(r"'[^']*'")
|
||||
|
||||
_SCHEMA_DESCRIPTION = (
|
||||
"Table: knowledge_chunks\n"
|
||||
"Columns: id, content, source, doc_type, doc_id, title, author, "
|
||||
"participants, timestamp, thread_id, url, metadata, chunk_index"
|
||||
"participants, timestamp, thread_id, url, metadata, chunk_index, "
|
||||
"created_at, deleted_at (NULL for active rows)"
|
||||
)
|
||||
|
||||
|
||||
@@ -84,21 +100,20 @@ class KnowledgeSQLTool(BaseTool):
|
||||
success=False,
|
||||
)
|
||||
|
||||
_FORBIDDEN = ("DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "CREATE", "ATTACH")
|
||||
for forbidden in _FORBIDDEN:
|
||||
if forbidden in normalized:
|
||||
return ToolResult(
|
||||
tool_name="knowledge_sql",
|
||||
content=(
|
||||
f"Query contains forbidden keyword: {forbidden}."
|
||||
" Only SELECT queries allowed."
|
||||
),
|
||||
success=False,
|
||||
)
|
||||
forbidden = _FORBIDDEN_RE.search(_STRING_LITERAL_RE.sub("''", query))
|
||||
if forbidden:
|
||||
return ToolResult(
|
||||
tool_name="knowledge_sql",
|
||||
content=(
|
||||
f"Query contains forbidden keyword: {forbidden.group(1).upper()}."
|
||||
" Only SELECT queries allowed."
|
||||
),
|
||||
success=False,
|
||||
)
|
||||
|
||||
try:
|
||||
rows = self._store._conn.execute(query).fetchmany(_MAX_ROWS)
|
||||
except sqlite3.OperationalError as exc:
|
||||
except sqlite3.Error as exc:
|
||||
return ToolResult(
|
||||
tool_name="knowledge_sql",
|
||||
content=f"SQL error: {exc}",
|
||||
|
||||
@@ -56,3 +56,30 @@ class TestErrorClassification:
|
||||
assert retry_delay(2) == 40
|
||||
# Capped at 300 seconds
|
||||
assert retry_delay(10) == 300
|
||||
|
||||
def test_classify_context_length_is_fatal(self):
|
||||
# A context-window overflow is deterministic — retrying the identical
|
||||
# over-length request can never succeed, so it must NOT be classified
|
||||
# retryable (which would burn ~30s of backoff on guaranteed failures).
|
||||
from openjarvis.agents.errors import classify_error
|
||||
from openjarvis.engine._base import EngineContextLengthError
|
||||
|
||||
typed = classify_error(
|
||||
EngineContextLengthError(
|
||||
"The conversation is too long for the model's context window."
|
||||
)
|
||||
)
|
||||
assert typed.retryable is False
|
||||
|
||||
# Same for untyped errors whose message reads like a context overflow
|
||||
# (e.g. raw vendor errors from engines without the typed mapping).
|
||||
untyped = classify_error(
|
||||
Exception("This model's maximum context length is 4096 tokens.")
|
||||
)
|
||||
assert untyped.retryable is False
|
||||
|
||||
def test_suggest_action_context_length(self):
|
||||
from openjarvis.agents.errors import FatalError, suggest_action
|
||||
|
||||
action = suggest_action(FatalError("prompt exceeds the model's context window"))
|
||||
assert "context window" in action or "too long" in action.lower()
|
||||
|
||||
@@ -158,6 +158,9 @@ class TestLiteLLMEngineGenerate:
|
||||
|
||||
class TestLiteLLMEngineStream:
|
||||
def test_stream(self) -> None:
|
||||
# stream() must use the ASYNC litellm entry point (acompletion): the
|
||||
# sync litellm.completion makes blocking network reads inside an
|
||||
# ``async def``, stalling the event loop between tokens.
|
||||
chunk1 = SimpleNamespace(
|
||||
choices=[SimpleNamespace(delta=SimpleNamespace(content="Hel"))]
|
||||
)
|
||||
@@ -168,8 +171,12 @@ class TestLiteLLMEngineStream:
|
||||
choices=[SimpleNamespace(delta=SimpleNamespace(content=None))]
|
||||
)
|
||||
|
||||
async def _chunks():
|
||||
for c in (chunk1, chunk2, chunk3):
|
||||
yield c
|
||||
|
||||
fake_litellm = mock.MagicMock()
|
||||
fake_litellm.completion.return_value = iter([chunk1, chunk2, chunk3])
|
||||
fake_litellm.acompletion = mock.AsyncMock(return_value=_chunks())
|
||||
|
||||
with mock.patch.dict("sys.modules", {"litellm": fake_litellm}):
|
||||
engine = LiteLLMEngine()
|
||||
@@ -186,6 +193,8 @@ class TestLiteLLMEngineStream:
|
||||
tokens = asyncio.run(collect())
|
||||
|
||||
assert tokens == ["Hel", "lo!"]
|
||||
fake_litellm.acompletion.assert_awaited_once()
|
||||
assert fake_litellm.completion.call_count == 0
|
||||
|
||||
|
||||
class TestLiteLLMEngineListModels:
|
||||
|
||||
+232
-1
@@ -6,13 +6,27 @@ import json
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
try:
|
||||
import respx
|
||||
|
||||
_HAS_RESPX = True
|
||||
except ImportError: # respx is an optional test-only dep; the async MockTransport
|
||||
respx = None # type: ignore[assignment] # pins below run without it.
|
||||
_HAS_RESPX = False
|
||||
|
||||
from openjarvis.core.registry import EngineRegistry
|
||||
from openjarvis.core.types import Message, Role
|
||||
from openjarvis.engine._base import EngineConnectionError
|
||||
from openjarvis.engine.ollama import OllamaEngine, _is_control_token_only_args
|
||||
|
||||
# respx-backed tests exercise the SYNC client paths (generate/list_models/health)
|
||||
# and the respx-driven stream tests; they skip cleanly when respx is absent. The
|
||||
# async pins in TestOllamaStreamIsAsyncAndBounded use httpx.MockTransport directly.
|
||||
requires_respx = pytest.mark.skipif(
|
||||
not _HAS_RESPX, reason="respx not installed (optional test-only dependency)"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def engine() -> OllamaEngine:
|
||||
@@ -20,6 +34,7 @@ def engine() -> OllamaEngine:
|
||||
return OllamaEngine(host="http://testhost:11434")
|
||||
|
||||
|
||||
@requires_respx
|
||||
class TestOllamaGenerate:
|
||||
def test_generate_returns_content(self, engine: OllamaEngine) -> None:
|
||||
with respx.mock:
|
||||
@@ -53,6 +68,7 @@ class TestOllamaGenerate:
|
||||
)
|
||||
|
||||
|
||||
@requires_respx
|
||||
class TestOllamaListModels:
|
||||
def test_list_models(self, engine: OllamaEngine) -> None:
|
||||
with respx.mock:
|
||||
@@ -66,6 +82,7 @@ class TestOllamaListModels:
|
||||
assert models == ["qwen3:8b", "llama3.2:3b"]
|
||||
|
||||
|
||||
@requires_respx
|
||||
class TestOllamaHealth:
|
||||
def test_health_true(self, engine: OllamaEngine) -> None:
|
||||
with respx.mock:
|
||||
@@ -120,6 +137,7 @@ class TestControlTokenFilter:
|
||||
assert _is_control_token_only_args(raw_args) is False
|
||||
|
||||
|
||||
@requires_respx
|
||||
class TestOllamaGenerateControlToken:
|
||||
def test_generate_drops_control_token_tool_call(self, engine: OllamaEngine) -> None:
|
||||
with respx.mock:
|
||||
@@ -219,6 +237,7 @@ class TestOllamaGenerateControlToken:
|
||||
assert json.loads(result["tool_calls"][0]["arguments"]) == {"command": "date"}
|
||||
|
||||
|
||||
@requires_respx
|
||||
class TestOllamaStreamFullControlToken:
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_full_drops_control_token_tool_call(
|
||||
@@ -257,6 +276,7 @@ class TestOllamaStreamFullControlToken:
|
||||
assert all(not c.tool_calls for c in chunks)
|
||||
|
||||
|
||||
@requires_respx
|
||||
class TestOllamaStream:
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_yields_content(self, engine: OllamaEngine) -> None:
|
||||
@@ -275,3 +295,214 @@ class TestOllamaStream:
|
||||
):
|
||||
tokens.append(tok)
|
||||
assert "Hello" in tokens
|
||||
|
||||
|
||||
def _ndjson_transport(lines: list[str]) -> httpx.MockTransport:
|
||||
body = "\n".join(lines) + "\n"
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, text=body)
|
||||
|
||||
return httpx.MockTransport(handler)
|
||||
|
||||
|
||||
class TestOllamaStreamIsAsyncAndBounded:
|
||||
"""Regression pins: the Ollama stream paths are async (never iterate the SYNC
|
||||
httpx client between tokens, which blocked the single uvicorn worker on every
|
||||
inter-token wait) and a mid-stream disconnect is bounded and mapped to a clean
|
||||
error. Uses httpx.MockTransport directly, so it runs without respx."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_does_not_use_blocking_sync_client(self) -> None:
|
||||
# PIN: the old code iterated ``self._client`` (a SYNC httpx.Client) via
|
||||
# ``iter_lines`` inside this ``async def``. The async path must not touch the
|
||||
# sync client at all — swap in a bomb that explodes if ``.stream`` is used.
|
||||
engine = OllamaEngine(host="http://localhost:11434")
|
||||
engine._async_transport = _ndjson_transport(
|
||||
[
|
||||
json.dumps({"message": {"content": "Hi"}, "done": False}),
|
||||
json.dumps({"message": {"content": " there"}, "done": True}),
|
||||
]
|
||||
)
|
||||
|
||||
class _Boom:
|
||||
def stream(self, *a, **k): # pragma: no cover - must never run
|
||||
raise AssertionError("streaming used the blocking sync client")
|
||||
|
||||
engine._client = _Boom() # type: ignore[assignment]
|
||||
tokens = [
|
||||
tok
|
||||
async for tok in engine.stream(
|
||||
[Message(role=Role.USER, content="Hi")], model="qwen3:8b"
|
||||
)
|
||||
]
|
||||
assert tokens == ["Hi", " there"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_full_does_not_use_blocking_sync_client(self) -> None:
|
||||
# stream_full delegates to _run_stream; prove that path is async too.
|
||||
engine = OllamaEngine(host="http://localhost:11434")
|
||||
engine._async_transport = _ndjson_transport(
|
||||
[
|
||||
json.dumps({"message": {"content": "Hi"}, "done": False}),
|
||||
json.dumps(
|
||||
{"message": {"content": "", "tool_calls": []}, "done": True}
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
class _Boom:
|
||||
def stream(self, *a, **k): # pragma: no cover - must never run
|
||||
raise AssertionError("stream_full used the blocking sync client")
|
||||
|
||||
engine._client = _Boom() # type: ignore[assignment]
|
||||
chunks = [
|
||||
c
|
||||
async for c in engine.stream_full(
|
||||
[Message(role=Role.USER, content="Hi")], model="qwen3:8b"
|
||||
)
|
||||
]
|
||||
assert any(c.content == "Hi" for c in chunks)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_timeout_is_applied_to_async_stream_client(self) -> None:
|
||||
# The configured timeout must be APPLIED to the async stream client, so a
|
||||
# wedged read is actually bounded (not just stored on the engine).
|
||||
engine = OllamaEngine(host="http://localhost:11434", timeout=0.05)
|
||||
assert engine._timeout == 0.05
|
||||
client = engine._make_async_client()
|
||||
try:
|
||||
assert client.timeout == httpx.Timeout(0.05)
|
||||
finally:
|
||||
await client.aclose()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mid_stream_disconnect_maps_to_connection_error(self) -> None:
|
||||
# PIN: a server dying MID-STREAM raises httpx.RemoteProtocolError from
|
||||
# aiter_lines; it must surface as a clean EngineConnectionError, not raw.
|
||||
class _MidStreamCrashStream(httpx.AsyncByteStream):
|
||||
def __init__(self, request: httpx.Request) -> None:
|
||||
self._request = request
|
||||
|
||||
async def __aiter__(self):
|
||||
yield b'{"message": {"content": "Hi"}, "done": false}\n'
|
||||
raise httpx.RemoteProtocolError(
|
||||
"peer closed connection mid-stream", request=self._request
|
||||
)
|
||||
|
||||
async def aclose(self) -> None: # pragma: no cover - trivial
|
||||
pass
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, stream=_MidStreamCrashStream(request))
|
||||
|
||||
engine = OllamaEngine(host="http://localhost:11434")
|
||||
engine._async_transport = httpx.MockTransport(handler)
|
||||
tokens: list[str] = []
|
||||
with pytest.raises(EngineConnectionError):
|
||||
async for tok in engine.stream(
|
||||
[Message(role=Role.USER, content="Hi")], model="qwen3:8b"
|
||||
):
|
||||
tokens.append(tok)
|
||||
# The disconnect happened AFTER the first token was delivered (mid-stream).
|
||||
assert tokens == ["Hi"]
|
||||
|
||||
|
||||
class TestOllamaStreamHttpErrorMapping:
|
||||
"""Regression pins: Ollama streaming non-2xx responses map to the same
|
||||
``EngineConnectionError`` as the OpenAI-compat engine (via
|
||||
``_raise_stream_http_error``), instead of leaking a raw
|
||||
``httpx.HTTPStatusError`` from ``raise_for_status()``. A 3xx must NOT fall
|
||||
through to a silent empty stream. Uses ``httpx.MockTransport`` directly, so
|
||||
it runs without respx."""
|
||||
|
||||
@staticmethod
|
||||
def _status_transport(
|
||||
status: int, *, text: str = "", headers: dict | None = None
|
||||
) -> httpx.MockTransport:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(status, text=text, headers=headers or {})
|
||||
|
||||
return httpx.MockTransport(handler)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_500_maps_to_connection_error(self) -> None:
|
||||
# PIN: the old code called ``resp.raise_for_status()`` and leaked a raw
|
||||
# httpx.HTTPStatusError on a streaming 500. It must now be a clean
|
||||
# EngineConnectionError carrying the status + body, like the compat path.
|
||||
engine = OllamaEngine(host="http://localhost:11434")
|
||||
engine._async_transport = self._status_transport(500, text="internal boom")
|
||||
tokens: list[str] = []
|
||||
with pytest.raises(EngineConnectionError) as excinfo:
|
||||
async for tok in engine.stream(
|
||||
[Message(role=Role.USER, content="Hi")], model="qwen3:8b"
|
||||
):
|
||||
tokens.append(tok)
|
||||
assert not isinstance(excinfo.value, httpx.HTTPStatusError)
|
||||
assert "500" in str(excinfo.value)
|
||||
assert "internal boom" in str(excinfo.value)
|
||||
assert tokens == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_full_500_maps_to_connection_error(self) -> None:
|
||||
# Same pin for the rich (_run_stream-backed) path.
|
||||
engine = OllamaEngine(host="http://localhost:11434")
|
||||
engine._async_transport = self._status_transport(500, text="internal boom")
|
||||
with pytest.raises(EngineConnectionError) as excinfo:
|
||||
async for _ in engine.stream_full(
|
||||
[Message(role=Role.USER, content="Hi")], model="qwen3:8b"
|
||||
):
|
||||
pass
|
||||
assert not isinstance(excinfo.value, httpx.HTTPStatusError)
|
||||
assert "500" in str(excinfo.value)
|
||||
assert "internal boom" in str(excinfo.value)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_3xx_maps_to_connection_error_not_silent(self) -> None:
|
||||
# PIN: with redirects off, a 3xx must map to EngineConnectionError, NOT
|
||||
# fall through to ``aiter_lines`` as a silent empty stream.
|
||||
engine = OllamaEngine(host="http://localhost:11434")
|
||||
engine._async_transport = self._status_transport(
|
||||
302, headers={"location": "http://elsewhere/api/chat"}
|
||||
)
|
||||
tokens: list[str] = []
|
||||
with pytest.raises(EngineConnectionError) as excinfo:
|
||||
async for tok in engine.stream(
|
||||
[Message(role=Role.USER, content="Hi")], model="qwen3:8b"
|
||||
):
|
||||
tokens.append(tok)
|
||||
assert "302" in str(excinfo.value)
|
||||
assert tokens == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_400_tools_retry_still_fires(self) -> None:
|
||||
# REGRESSION GUARD: the tools-retry (400 WITH tools -> retry WITHOUT
|
||||
# tools) must keep working; only OTHER non-2xx map to
|
||||
# EngineConnectionError. A 400 carrying tools must NOT be treated as a
|
||||
# generic connection error.
|
||||
calls: list[bool] = [] # whether each request carried "tools"
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
payload = json.loads(request.content)
|
||||
had_tools = "tools" in payload
|
||||
calls.append(had_tools)
|
||||
if had_tools:
|
||||
return httpx.Response(400, text="model does not support tools")
|
||||
body = (
|
||||
json.dumps({"message": {"content": "recovered"}, "done": True}) + "\n"
|
||||
)
|
||||
return httpx.Response(200, text=body)
|
||||
|
||||
engine = OllamaEngine(host="http://localhost:11434")
|
||||
engine._async_transport = httpx.MockTransport(handler)
|
||||
chunks = [
|
||||
c
|
||||
async for c in engine.stream_full(
|
||||
[Message(role=Role.USER, content="Hi")],
|
||||
model="qwen3:8b",
|
||||
tools=[{"type": "function", "function": {"name": "shell_exec"}}],
|
||||
)
|
||||
]
|
||||
# First request had tools (400), second retried without them (200).
|
||||
assert calls == [True, False]
|
||||
assert any(c.content == "recovered" for c in chunks)
|
||||
|
||||
@@ -4,13 +4,37 @@ from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
try:
|
||||
import respx
|
||||
|
||||
_HAS_RESPX = True
|
||||
except ImportError: # respx is an optional test-only dep; MockTransport tests
|
||||
respx = None # type: ignore[assignment] # still run without it.
|
||||
_HAS_RESPX = False
|
||||
|
||||
from openjarvis.core.registry import EngineRegistry
|
||||
from openjarvis.core.types import Message, Role
|
||||
from openjarvis.engine._base import EngineConnectionError
|
||||
from openjarvis.engine._openai_compat import EngineContextLengthError
|
||||
from openjarvis.engine.openai_compat_engines import VLLMEngine
|
||||
|
||||
# respx-backed tests exercise the SYNC client paths (generate/list_models/health)
|
||||
# and skip cleanly when respx is absent; the async stream/timeout/disconnect tests
|
||||
# below use httpx.MockTransport directly and never need respx.
|
||||
requires_respx = pytest.mark.skipif(
|
||||
not _HAS_RESPX, reason="respx not installed (optional test-only dependency)"
|
||||
)
|
||||
|
||||
|
||||
def _sse_transport(sse_lines: list[str]) -> httpx.MockTransport:
|
||||
body = "\n".join(sse_lines) + "\n"
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, text=body)
|
||||
|
||||
return httpx.MockTransport(handler)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def engine() -> VLLMEngine:
|
||||
@@ -18,6 +42,7 @@ def engine() -> VLLMEngine:
|
||||
return VLLMEngine(host="http://testhost:8000")
|
||||
|
||||
|
||||
@requires_respx
|
||||
class TestOpenAICompatGenerate:
|
||||
def test_generate_returns_content(self, engine: VLLMEngine) -> None:
|
||||
with respx.mock:
|
||||
@@ -83,6 +108,7 @@ class TestOpenAICompatGenerate:
|
||||
)
|
||||
|
||||
|
||||
@requires_respx
|
||||
class TestOpenAICompatListModels:
|
||||
def test_list_models(self, engine: VLLMEngine) -> None:
|
||||
with respx.mock:
|
||||
@@ -95,6 +121,7 @@ class TestOpenAICompatListModels:
|
||||
assert engine.list_models() == ["model-a", "model-b"]
|
||||
|
||||
|
||||
@requires_respx
|
||||
class TestOpenAICompatHealth:
|
||||
def test_health_true(self, engine: VLLMEngine) -> None:
|
||||
with respx.mock:
|
||||
@@ -111,6 +138,7 @@ class TestOpenAICompatHealth:
|
||||
assert engine.health() is False
|
||||
|
||||
|
||||
@requires_respx
|
||||
class TestOpenAICompatStream:
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_sse(self, engine: VLLMEngine) -> None:
|
||||
@@ -129,3 +157,199 @@ class TestOpenAICompatStream:
|
||||
):
|
||||
tokens.append(tok)
|
||||
assert tokens == ["Hi", " there"]
|
||||
|
||||
|
||||
class TestStreamIsAsyncAndBounded:
|
||||
"""Regression pins for BC2: the stream path is async (never blocks the event
|
||||
loop on the SYNC httpx client) and a wedged/oversized upstream is bounded and
|
||||
mapped to a clean error instead of hanging or surfacing raw HTTP."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_timeout_is_applied_to_async_stream_client(self) -> None:
|
||||
# HARDENED: assert the configured timeout is actually APPLIED to the async
|
||||
# stream client, not merely stored on the engine. Deleting
|
||||
# ``timeout=self._timeout`` from ``_make_async_client`` drops the client to
|
||||
# httpx's 5s default and fails this (a stored-only assertion would not).
|
||||
engine = VLLMEngine(host="http://testhost:8000", timeout=180.0)
|
||||
assert engine._timeout == 180.0
|
||||
client = engine._make_async_client()
|
||||
try:
|
||||
assert client.timeout == httpx.Timeout(180.0)
|
||||
finally:
|
||||
await client.aclose()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_does_not_use_blocking_sync_client(self) -> None:
|
||||
# PIN: the old code iterated ``self._client`` (a SYNC httpx.Client) inside
|
||||
# this ``async def``, blocking the single uvicorn worker between tokens.
|
||||
# The async path must not touch the sync client at all.
|
||||
engine = VLLMEngine(host="http://localhost:8000")
|
||||
engine._async_transport = _sse_transport(
|
||||
[
|
||||
'data: {"choices":[{"delta":{"content":"Hi"}}]}',
|
||||
'data: {"choices":[{"delta":{"content":" there"}}]}',
|
||||
"data: [DONE]",
|
||||
]
|
||||
)
|
||||
|
||||
class _Boom:
|
||||
def stream(self, *a, **k): # pragma: no cover - must never run
|
||||
raise AssertionError("streaming used the blocking sync client")
|
||||
|
||||
engine._client = _Boom() # type: ignore[assignment]
|
||||
tokens = [
|
||||
tok
|
||||
async for tok in engine.stream(
|
||||
[Message(role=Role.USER, content="Hello")], model="m"
|
||||
)
|
||||
]
|
||||
assert tokens == ["Hi", " there"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_wedged_read_is_bounded_by_timeout(self) -> None:
|
||||
# A wedged upstream read trips the (small, honoured) timeout and maps to a
|
||||
# clean EngineConnectionError rather than hanging the caller.
|
||||
seen: dict = {}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
# httpx populates request.extensions["timeout"] with the per-op timeouts
|
||||
# that were actually applied to THIS request; capturing it here proves
|
||||
# the configured 0.05s reached the wire, not just the engine attribute.
|
||||
seen["timeout"] = request.extensions["timeout"]
|
||||
raise httpx.ReadTimeout("wedged upstream", request=request)
|
||||
|
||||
engine = VLLMEngine(host="http://localhost:8000", timeout=0.05)
|
||||
engine._async_transport = httpx.MockTransport(handler)
|
||||
with pytest.raises(EngineConnectionError):
|
||||
async for _ in engine.stream(
|
||||
[Message(role=Role.USER, content="Hello")], model="m"
|
||||
):
|
||||
pass
|
||||
# HARDENED: the configured timeout was APPLIED to the request. Deleting
|
||||
# ``timeout=self._timeout`` from ``_make_async_client`` drops this to httpx's
|
||||
# 5.0s default and fails the assertion.
|
||||
assert seen["timeout"]["read"] == 0.05
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_context_length_400_maps_to_context_error(self) -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(
|
||||
400,
|
||||
text=(
|
||||
"This model's maximum context length is 4096 tokens. "
|
||||
"However, you requested 5200 tokens. Please reduce the length."
|
||||
),
|
||||
)
|
||||
|
||||
engine = VLLMEngine(host="http://localhost:8000")
|
||||
engine._async_transport = httpx.MockTransport(handler)
|
||||
with pytest.raises(EngineContextLengthError) as excinfo:
|
||||
async for _ in engine.stream(
|
||||
[Message(role=Role.USER, content="Hello")], model="m"
|
||||
):
|
||||
pass
|
||||
assert getattr(excinfo.value, "is_context_length_error", False) is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_other_upstream_error_maps_to_connection_error(self) -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(500, text="internal server error")
|
||||
|
||||
engine = VLLMEngine(host="http://localhost:8000")
|
||||
engine._async_transport = httpx.MockTransport(handler)
|
||||
with pytest.raises(EngineConnectionError) as excinfo:
|
||||
async for _ in engine.stream(
|
||||
[Message(role=Role.USER, content="Hello")], model="m"
|
||||
):
|
||||
pass
|
||||
# A generic upstream failure is NOT reported as a context-length problem.
|
||||
assert not isinstance(excinfo.value, EngineContextLengthError)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mid_stream_disconnect_maps_to_connection_error(self) -> None:
|
||||
# PIN: a server dying MID-STREAM raises httpx.RemoteProtocolError from
|
||||
# aiter_lines. Before the except tuple was widened it propagated raw; it
|
||||
# must now surface as a clean EngineConnectionError.
|
||||
class _MidStreamCrashStream(httpx.AsyncByteStream):
|
||||
def __init__(self, request: httpx.Request) -> None:
|
||||
self._request = request
|
||||
|
||||
async def __aiter__(self):
|
||||
yield b'data: {"choices":[{"delta":{"content":"Hi"}}]}\n\n'
|
||||
raise httpx.RemoteProtocolError(
|
||||
"peer closed connection mid-stream", request=self._request
|
||||
)
|
||||
|
||||
async def aclose(self) -> None: # pragma: no cover - trivial
|
||||
pass
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, stream=_MidStreamCrashStream(request))
|
||||
|
||||
engine = VLLMEngine(host="http://localhost:8000")
|
||||
engine._async_transport = httpx.MockTransport(handler)
|
||||
tokens: list[str] = []
|
||||
with pytest.raises(EngineConnectionError) as excinfo:
|
||||
async for tok in engine.stream(
|
||||
[Message(role=Role.USER, content="Hello")], model="m"
|
||||
):
|
||||
tokens.append(tok)
|
||||
# The disconnect happened AFTER the first token was delivered (mid-stream),
|
||||
# and did not masquerade as a context-length error.
|
||||
assert tokens == ["Hi"]
|
||||
assert not isinstance(excinfo.value, EngineContextLengthError)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unrelated_400_is_not_context_error(self) -> None:
|
||||
# PIN: the context-length markers are anchored on "context" so generic
|
||||
# 400 bodies containing phrases like "please reduce" (max_tokens
|
||||
# validation, rate limiting, oversized images) are NOT misclassified
|
||||
# as "conversation too long" — that message would send the user off to
|
||||
# shorten a conversation that isn't the problem.
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(
|
||||
400,
|
||||
text=(
|
||||
"Invalid max_tokens: the maximum number of tokens you can "
|
||||
"request is 4096; please reduce max_tokens and retry."
|
||||
),
|
||||
)
|
||||
|
||||
engine = VLLMEngine(host="http://localhost:8000")
|
||||
engine._async_transport = httpx.MockTransport(handler)
|
||||
with pytest.raises(EngineConnectionError) as excinfo:
|
||||
async for _ in engine.stream(
|
||||
[Message(role=Role.USER, content="Hello")], model="m"
|
||||
):
|
||||
pass
|
||||
assert not isinstance(excinfo.value, EngineContextLengthError)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_client_is_reused_across_streams(self) -> None:
|
||||
# PIN: consecutive streams on the same event loop share one AsyncClient
|
||||
# (connection pooling). A per-call client would pay a fresh TCP/TLS
|
||||
# handshake on every conversation turn.
|
||||
engine = VLLMEngine(host="http://localhost:8000")
|
||||
engine._async_transport = _sse_transport(
|
||||
[
|
||||
'data: {"choices":[{"delta":{"content":"Hi"}}]}',
|
||||
"data: [DONE]",
|
||||
]
|
||||
)
|
||||
|
||||
async def one_turn() -> list[str]:
|
||||
return [
|
||||
tok
|
||||
async for tok in engine.stream(
|
||||
[Message(role=Role.USER, content="Hello")], model="m"
|
||||
)
|
||||
]
|
||||
|
||||
assert await one_turn() == ["Hi"]
|
||||
first_client = engine._async_client
|
||||
assert first_client is not None and not first_client.is_closed
|
||||
assert await one_turn() == ["Hi"]
|
||||
assert engine._async_client is first_client
|
||||
# close() tears the shared client down with the sync one.
|
||||
engine.close()
|
||||
assert engine._async_client is None
|
||||
|
||||
@@ -5,12 +5,27 @@ from __future__ import annotations
|
||||
import json
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any, Dict, List
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from openjarvis.core.types import Message, Role
|
||||
from openjarvis.engine._stubs import InferenceEngine, StreamChunk
|
||||
from openjarvis.engine.openai_compat_engines import VLLMEngine
|
||||
|
||||
|
||||
def _sse_transport(sse_lines: list[str]) -> httpx.MockTransport:
|
||||
"""A MockTransport that replies to /v1/chat/completions with SSE ``sse_lines``.
|
||||
|
||||
Exercises the REAL async httpx streaming path (aiter_lines) with no server.
|
||||
"""
|
||||
body = "\n".join(sse_lines) + "\n"
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, text=body)
|
||||
|
||||
return httpx.MockTransport(handler)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# StreamChunk dataclass tests
|
||||
@@ -147,8 +162,6 @@ class TestOpenAICompatStreamFull:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_parses_sse_with_content_and_finish(self):
|
||||
from openjarvis.engine._openai_compat import _OpenAICompatibleEngine
|
||||
|
||||
# Build mock SSE lines
|
||||
sse_lines = []
|
||||
for token in ["Hello", " world"]:
|
||||
@@ -164,22 +177,8 @@ class TestOpenAICompatStreamFull:
|
||||
sse_lines.append(f"data: {json.dumps(final)}")
|
||||
sse_lines.append("data: [DONE]")
|
||||
|
||||
# Mock the httpx client stream context manager
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_resp.iter_lines.return_value = iter(sse_lines)
|
||||
|
||||
engine = _OpenAICompatibleEngine.__new__(_OpenAICompatibleEngine)
|
||||
engine.engine_id = "test"
|
||||
engine._host = "http://localhost:8000"
|
||||
engine._api_prefix = "/v1"
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_stream_ctx = MagicMock()
|
||||
mock_stream_ctx.__enter__ = MagicMock(return_value=mock_resp)
|
||||
mock_stream_ctx.__exit__ = MagicMock(return_value=False)
|
||||
mock_client.stream.return_value = mock_stream_ctx
|
||||
engine._client = mock_client
|
||||
engine = VLLMEngine(host="http://localhost:8000")
|
||||
engine._async_transport = _sse_transport(sse_lines)
|
||||
|
||||
chunks = []
|
||||
async for chunk in engine.stream_full(
|
||||
@@ -198,8 +197,6 @@ class TestOpenAICompatStreamFull:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_parses_tool_call_fragments(self):
|
||||
from openjarvis.engine._openai_compat import _OpenAICompatibleEngine
|
||||
|
||||
# Simulate streamed tool_call fragments
|
||||
_tc1 = (
|
||||
'{"choices": [{"delta": {"tool_calls": [{"index": 0, "id": "call_1",'
|
||||
@@ -218,21 +215,8 @@ class TestOpenAICompatStreamFull:
|
||||
"data: [DONE]",
|
||||
]
|
||||
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_resp.iter_lines.return_value = iter(sse_lines)
|
||||
|
||||
engine = _OpenAICompatibleEngine.__new__(_OpenAICompatibleEngine)
|
||||
engine.engine_id = "test"
|
||||
engine._host = "http://localhost:8000"
|
||||
engine._api_prefix = "/v1"
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_stream_ctx = MagicMock()
|
||||
mock_stream_ctx.__enter__ = MagicMock(return_value=mock_resp)
|
||||
mock_stream_ctx.__exit__ = MagicMock(return_value=False)
|
||||
mock_client.stream.return_value = mock_stream_ctx
|
||||
engine._client = mock_client
|
||||
engine = VLLMEngine(host="http://localhost:8000")
|
||||
engine._async_transport = _sse_transport(sse_lines)
|
||||
|
||||
chunks = []
|
||||
async for chunk in engine.stream_full(
|
||||
|
||||
@@ -347,7 +347,7 @@ class TestCostCalculator:
|
||||
calls_per_month=1000,
|
||||
avg_input_tokens=500,
|
||||
avg_output_tokens=200,
|
||||
provider_key="gpt-5.3",
|
||||
provider_key="gpt-5.6-sol",
|
||||
)
|
||||
assert est.monthly_cost > 0
|
||||
assert est.annual_cost == est.monthly_cost * 12
|
||||
|
||||
@@ -64,6 +64,41 @@ def test_rejects_drop(store: KnowledgeStore) -> None:
|
||||
assert not result.success
|
||||
|
||||
|
||||
def test_allows_select_with_keyword_substring(store: KnowledgeStore) -> None:
|
||||
"""A read-only SELECT must not be rejected because a column/alias merely
|
||||
contains a write keyword as a substring (e.g. 'created' -> CREATE)."""
|
||||
from openjarvis.tools.knowledge_sql import KnowledgeSQLTool
|
||||
|
||||
tool = KnowledgeSQLTool(store=store)
|
||||
result = tool.execute(
|
||||
query="SELECT author AS created_author FROM knowledge_chunks"
|
||||
)
|
||||
assert result.success, result.content
|
||||
assert "Alice" in result.content
|
||||
|
||||
|
||||
def test_allows_keyword_inside_string_literal(store: KnowledgeStore) -> None:
|
||||
"""A write keyword appearing only inside a string literal must not be
|
||||
treated as a forbidden statement."""
|
||||
from openjarvis.tools.knowledge_sql import KnowledgeSQLTool
|
||||
|
||||
tool = KnowledgeSQLTool(store=store)
|
||||
result = tool.execute(
|
||||
query="SELECT content FROM knowledge_chunks WHERE content LIKE '%delete%'"
|
||||
)
|
||||
assert result.success, result.content
|
||||
|
||||
|
||||
def test_rejects_multi_statement(store: KnowledgeStore) -> None:
|
||||
"""Multi-statement strings fail with a ToolResult, not an exception."""
|
||||
from openjarvis.tools.knowledge_sql import KnowledgeSQLTool
|
||||
|
||||
tool = KnowledgeSQLTool(store=store)
|
||||
result = tool.execute(query="SELECT 1; VACUUM")
|
||||
assert not result.success
|
||||
assert "error" in result.content.lower()
|
||||
|
||||
|
||||
def test_handles_bad_sql(store: KnowledgeStore) -> None:
|
||||
from openjarvis.tools.knowledge_sql import KnowledgeSQLTool
|
||||
|
||||
|
||||
Reference in New Issue
Block a user