Index

engine

Inference Engine primitive — LLM runtime management.

Classes

EngineConnectionError

Bases: Exception

Raised when an engine is unreachable.

EngineContextLengthError

Bases: 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.

InferenceEngine

Bases: ABC

Base class for all inference engine backends.

Subclasses must be registered via @EngineRegistry.register("name") to become discoverable.

Functions

generate abstractmethod

generate(messages: Sequence[Message], *, model: str, temperature: float = 0.7, max_tokens: int = 1024, **kwargs: Any) -> Dict[str, Any]

Synchronous completion — returns a dict with content and usage.

Source code in src/openjarvis/engine/_stubs.py

@abstractmethod
def generate(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> Dict[str, Any]:
    """Synchronous completion — returns a dict with ``content`` and ``usage``."""

stream abstractmethod async

stream(messages: Sequence[Message], *, model: str, temperature: float = 0.7, max_tokens: int = 1024, **kwargs: Any) -> AsyncIterator[str]

Yield token strings as they are generated.

Source code in src/openjarvis/engine/_stubs.py

@abstractmethod
async def stream(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[str]:
    """Yield token strings as they are generated."""
    # NOTE: must contain a yield to satisfy the type checker
    yield ""  # pragma: no cover

stream_full async

stream_full(messages: Sequence[Message], *, model: str, temperature: float = 0.7, max_tokens: int = 1024, **kwargs: Any) -> AsyncIterator['StreamChunk']

Yield full StreamChunks including tool_calls and finish_reason.

Default implementation wraps stream() for backward compatibility. Engines with native tool-call streaming should override this.

Source code in src/openjarvis/engine/_stubs.py

async def stream_full(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator["StreamChunk"]:
    """Yield full StreamChunks including tool_calls and finish_reason.

    Default implementation wraps ``stream()`` for backward compatibility.
    Engines with native tool-call streaming should override this.
    """
    async for token in self.stream(
        messages,
        model=model,
        temperature=temperature,
        max_tokens=max_tokens,
        **kwargs,
    ):
        yield StreamChunk(content=token)
    yield StreamChunk(finish_reason="stop")

list_models abstractmethod

list_models() -> List[str]

Return identifiers of models available on this engine.

Source code in src/openjarvis/engine/_stubs.py

@abstractmethod
def list_models(self) -> List[str]:
    """Return identifiers of models available on this engine."""

health abstractmethod

health() -> bool

Return True when the engine is reachable and healthy.

Source code in src/openjarvis/engine/_stubs.py

@abstractmethod
def health(self) -> bool:
    """Return ``True`` when the engine is reachable and healthy."""

can_serve

can_serve(model: str) -> bool

Return True if this engine can serve model.

Defaults to True: local engines accept any model id (whether a specific model is installed is a separate concern from engine selection). Engines that multiplex provider-specific clients (e.g. the cloud engine) override this so selection can skip an engine whose client for the model's provider isn't configured (see #532).

Source code in src/openjarvis/engine/_stubs.py

def can_serve(self, model: str) -> bool:
    """Return ``True`` if this engine can serve *model*.

    Defaults to ``True``: local engines accept any model id (whether a
    specific model is *installed* is a separate concern from engine
    selection). Engines that multiplex provider-specific clients (e.g.
    the cloud engine) override this so selection can skip an engine whose
    client for the model's provider isn't configured (see #532).
    """
    return True

close

close() -> None

Release resources (HTTP clients, connections, threads, etc.).

Source code in src/openjarvis/engine/_stubs.py

def close(self) -> None:
    """Release resources (HTTP clients, connections, threads, etc.)."""

prepare

prepare(model: str) -> None

Optional warm-up hook called before the first request.

Source code in src/openjarvis/engine/_stubs.py

def prepare(self, model: str) -> None:
    """Optional warm-up hook called before the first request."""

Functions

looks_like_context_length_error

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.

Source code in src/openjarvis/engine/_base.py

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)

messages_to_dicts

messages_to_dicts(messages: Sequence[Message]) -> List[Dict[str, Any]]

Convert Message objects to OpenAI-format dicts.

Source code in src/openjarvis/engine/_base.py

def messages_to_dicts(messages: Sequence[Message]) -> List[Dict[str, Any]]:
    """Convert ``Message`` objects to OpenAI-format dicts."""
    out: List[Dict[str, Any]] = []
    for m in messages:
        d: Dict[str, Any] = {"role": m.role.value, "content": m.content}
        if m.name:
            d["name"] = m.name
        if m.tool_calls:
            d["tool_calls"] = [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {
                        "name": tc.name,
                        "arguments": tc.arguments,
                    },
                }
                for tc in m.tool_calls
            ]
        if m.tool_call_id:
            d["tool_call_id"] = m.tool_call_id
        # Vision: forward base64 images to the engine. Ollama's /api/chat
        # accepts an "images" array on a message; text messages skip this.
        if getattr(m, "images", None):
            d["images"] = list(m.images)
        out.append(d)
    return out

discover_engines

discover_engines(config: JarvisConfig) -> List[Tuple[str, InferenceEngine]]

Probe registered engines and return [(key, instance)] for healthy ones.

Results are sorted with the config default engine first.

Source code in src/openjarvis/engine/_discovery.py

def discover_engines(config: JarvisConfig) -> List[Tuple[str, InferenceEngine]]:
    """Probe registered engines and return ``[(key, instance)]`` for healthy ones.

    Results are sorted with the config default engine first.
    """
    _maybe_register_mining_sidecar_engine()

    # Probe engines concurrently: each health() does a blocking network
    # check with its own timeout, so a serial loop costs the SUM of all
    # probe timeouts (dead localhost ports especially). Running them in
    # threads collapses that to roughly the slowest single probe. The
    # healthy.sort() below normalizes order, so completion order is
    # irrelevant and the result is identical to the serial version (#263).
    keys = list(EngineRegistry.keys())

    def _probe(key: str) -> Tuple[str, InferenceEngine] | None:
        try:
            engine = _make_engine(key, config)
            if engine.health():
                return (key, engine)
        except Exception as exc:
            logger.debug("Engine %r failed during discovery: %s", key, exc)
        return None

    healthy: List[Tuple[str, InferenceEngine]] = []
    if keys:
        with ThreadPoolExecutor(max_workers=len(keys)) as pool:
            for result in pool.map(_probe, keys):
                if result is not None:
                    healthy.append(result)

    default_key = config.engine.default

    def sort_key(item: Tuple[str, Any]) -> Tuple[int, str]:
        return (0 if item[0] == default_key else 1, item[0])

    healthy.sort(key=sort_key)
    return healthy

discover_models

discover_models(engines: List[Tuple[str, InferenceEngine]]) -> Dict[str, List[str]]

Call list_models() on each engine and return a dict.

Source code in src/openjarvis/engine/_discovery.py

def discover_models(
    engines: List[Tuple[str, InferenceEngine]],
) -> Dict[str, List[str]]:
    """Call ``list_models()`` on each engine and return a dict."""
    result: Dict[str, List[str]] = {}
    for key, engine in engines:
        try:
            result[key] = engine.list_models()
        except Exception as exc:
            logger.debug("Failed to list models for engine %r: %s", key, exc)
            result[key] = []
    return result

get_engine

get_engine(config: JarvisConfig, engine_key: str | None = None, model: str | None = None) -> Tuple[str, InferenceEngine] | None

Get a specific engine by key, or the default with a named fallback.

An explicit engine_key is authoritative and is never silently substituted. Default-engine fallback prefers the same local/cloud class and logs the selected replacement.

When model is given, an engine is selected only if it can actually serve that model (engine.can_serve(model)). This stops the cloud fallback from being chosen — when the local engine is down — for a model whose provider client is missing, which otherwise surfaces as a confusing "OpenAI client not available" instead of a helpful "start your local engine" message (see #532). When model is None selection stays model-agnostic (unchanged behaviour).

Returns (key, engine_instance) or None if no engine is available.

Source code in src/openjarvis/engine/_discovery.py

def get_engine(
    config: JarvisConfig,
    engine_key: str | None = None,
    model: str | None = None,
) -> Tuple[str, InferenceEngine] | None:
    """Get a specific engine by key, or the default with a named fallback.

    An explicit ``engine_key`` is authoritative and is never silently
    substituted. Default-engine fallback prefers the same local/cloud class
    and logs the selected replacement.

    When *model* is given, an engine is selected only if it can actually
    serve that model (``engine.can_serve(model)``). This stops the cloud
    fallback from being chosen — when the local engine is down — for a model
    whose provider client is missing, which otherwise surfaces as a confusing
    "OpenAI client not available" instead of a helpful "start your local
    engine" message (see #532). When *model* is ``None`` selection stays
    model-agnostic (unchanged behaviour).

    Returns ``(key, engine_instance)`` or ``None`` if no engine is available.
    """

    def _usable(engine: InferenceEngine) -> bool:
        return engine.health() and (model is None or engine.can_serve(model))

    if engine_key:
        if not EngineRegistry.contains(engine_key):
            logger.warning("Requested engine %r is not registered", engine_key)
            return None
        try:
            engine = _make_engine(engine_key, config)
            if _usable(engine):
                return (engine_key, engine)
        except Exception as exc:
            logger.debug("Engine %r health check failed: %s", engine_key, exc)
        logger.warning(
            "Requested engine %r is unavailable%s; no substitute was selected",
            engine_key,
            f" for model {model!r}" if model else "",
        )
        return None

    default_key = config.engine.default
    default_is_cloud: bool | None = None
    if default_key and EngineRegistry.contains(default_key):
        default_cls = EngineRegistry.get(default_key)
        default_is_cloud = bool(getattr(default_cls, "is_cloud", False))
        try:
            engine = _make_engine(default_key, config)
            default_is_cloud = bool(getattr(engine, "is_cloud", default_is_cloud))
            if _usable(engine):
                return (default_key, engine)
        except Exception as exc:
            logger.debug("Engine %r health check failed: %s", default_key, exc)

    candidates = [
        (key, engine)
        for key, engine in discover_engines(config)
        if key != default_key and (model is None or engine.can_serve(model))
    ]
    if not candidates:
        return None

    chosen: Tuple[str, InferenceEngine] | None = None
    if default_is_cloud is not None:
        chosen = next(
            (
                candidate
                for candidate in candidates
                if bool(getattr(candidate[1], "is_cloud", False)) == default_is_cloud
            ),
            None,
        )
    chosen = chosen or candidates[0]
    chosen_is_cloud = bool(getattr(chosen[1], "is_cloud", False))
    boundary = (
        " across the local/cloud boundary"
        if (default_is_cloud is not None and chosen_is_cloud != default_is_cloud)
        else ""
    )
    logger.warning(
        "Default engine %r is unavailable; using %r%s",
        default_key,
        chosen[0],
        boundary,
    )
    return chosen