Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 83 additions & 3 deletions openviking_cli/doctor.py
Original file line number Diff line number Diff line change
Expand Up @@ -339,8 +339,8 @@ def check_embedding() -> tuple[bool, str, Optional[str]]:
return _probe_embedding_provider(embedding, dense)


def check_vlm() -> tuple[bool, str, Optional[str]]:
"""Load VLM config and verify it's configured."""
def check_vlm() -> CheckResult:
"""Load VLM config and verify it's configured, then probe Function Calling support."""
config_path = _find_config()
if config_path is None:
return False, "Cannot check (no config file)", None
Expand Down Expand Up @@ -393,7 +393,87 @@ def check_vlm() -> tuple[bool, str, Optional[str]]:
"Set vlm.api_key in ov.conf",
)

return True, f"{provider}/{model}", None
# Non-ollama litellm (ollama already early-returned above): best-effort registry
# short-circuit before spending a live request.
# ponytail: litellm registry lags new models — treat only an explicit False as a
# signal, unknown ⇒ live-probe.
if provider == "litellm":
try:
import litellm

if litellm.supports_function_calling(model) is False:
return _fc_unsupported_warn(f"{provider}/{model}")
except Exception:
pass

return _probe_vlm_function_calling(VLMConfig(**normalized_vlm))


def _fc_unsupported_warn(label: str) -> tuple[CheckStatus, str, Optional[str]]:
"""WARN tuple for a model detected (live or via registry) to lack Function Calling."""
return (
"warn",
f"{label} does not support Function Calling (tool use); "
"Agent (--with-bot) and session-memory features will fail",
"Configure a Function-Calling-capable vlm.model "
"(see .wiki/embedding-and-vlm-providers.md), "
"or don't use --with-bot / session memory",
)


def _probe_vlm_function_calling(vlm_config: VLMConfig) -> tuple[CheckStatus, str, Optional[str]]:
"""Send one trivial tool-call request to verify the model supports Function Calling.

Mirrors ``_probe_embedding_provider``'s mechanism but reports a WARN (not a hard
fail): Function Calling is required by Agent (``--with-bot``) mode and by session
working-memory/extraction, yet pure parse/retrieval deployments never use it, so a
missing/unverifiable capability must be visible and actionable but non-blocking.
"""
_, provider = vlm_config.get_provider_config()
model = vlm_config.model or ""
label = f"{provider}/{model}"

probe_tool = {
"type": "function",
"function": {
"name": "ov_doctor_probe",
"description": "Return ok.",
"parameters": {"type": "object", "properties": {"ok": {"type": "boolean"}}},
},
}

async def _run_probe():
return await asyncio.wait_for(
vlm_config.get_completion_async(
messages=[{"role": "user", "content": "ping"}],
tools=[probe_tool],
tool_choice="auto",
),
timeout=10.0,
)

try:
asyncio.run(_run_probe())
except TimeoutError:
# str(TimeoutError()) is "" — catch explicitly (mirrors _probe_embedding_provider:239).
return (
"warn",
f"{label} (function-calling probe timed out)",
"Check the VLM provider is reachable and the model/API key are correct",
)
except Exception as exc:
text = str(exc).lower()
# ponytail: substring match on the provider 400 body (e.g. code 20037); widen
# patterns if a provider phrases FC-unsupported differently.
if "20037" in text or ("function call" in text and "not supported" in text):
return _fc_unsupported_warn(label)
return (
"warn",
f"{label} (function-calling probe could not complete: {exc})",
"Check the VLM provider is reachable and the model/API key are correct",
)

return "pass", f"{label} function-calling ok", None


def check_ollama() -> tuple[bool, str, Optional[str]]:
Expand Down
228 changes: 226 additions & 2 deletions tests/cli/test_doctor.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

from openviking_cli.doctor import (
_probe_embedding_provider,
_probe_vlm_function_calling,
check_agfs,
check_config,
check_disk,
Expand All @@ -27,6 +28,7 @@
check_vlm,
run_doctor,
)
from openviking_cli.utils.config.vlm_config import VLMConfig


class TestCheckConfig:
Expand Down Expand Up @@ -500,7 +502,11 @@ def test_pass_with_config(self, tmp_path: Path):
)
)
with patch("openviking_cli.doctor._find_config", return_value=config):
ok, detail, fix = check_vlm()
with patch(
"openviking_cli.doctor._probe_vlm_function_calling",
return_value=("pass", "openai/gpt-4o-mini function-calling ok", None),
):
ok, detail, fix = check_vlm()
assert ok

def test_fail_no_provider(self, tmp_path: Path):
Expand Down Expand Up @@ -577,6 +583,196 @@ def test_pass_with_default_provider_codex_oauth(self, tmp_path: Path):
assert "openai-codex/gpt-5.3-codex" in detail
assert "oauth via openviking" in detail

def test_folds_probe_warn(self, tmp_path: Path):
# A valid-key config reaches the probe; check_vlm returns the probe's tri-state.
config = tmp_path / "ov.conf"
config.write_text(
json.dumps(
{"vlm": {"provider": "openai", "model": "gpt-4o-mini", "api_key": "sk-test"}}
)
)
with patch("openviking_cli.doctor._find_config", return_value=config):
with patch(
"openviking_cli.doctor._probe_vlm_function_calling",
return_value=("warn", "openai/gpt-4o-mini does not support Function Calling", "fix"),
):
status, detail, fix = check_vlm()
assert status == "warn"
assert "Function Calling" in detail
assert fix == "fix"

def test_check_vlm_drives_real_probe_20037(self, tmp_path: Path):
# Integration (§8b): exercise the real check_vlm -> _probe_vlm_function_calling
# -> VLMConfig.get_completion_async chain with ONLY the LLM boundary mocked.
# The probe is NOT stubbed, so a WARN here proves the live chain wires up.
config = tmp_path / "ov.conf"
config.write_text(
json.dumps(
{"vlm": {"provider": "openai", "model": "gpt-4o-mini", "api_key": "sk-test"}}
)
)

async def raise_20037(self, *args, **kwargs):
raise Exception(
"Error code: 400 - {'code': 20037, 'message': "
"'Function call is not supported for this model.', 'data': None}"
)

with patch("openviking_cli.doctor._find_config", return_value=config):
with patch.object(VLMConfig, "get_completion_async", raise_20037):
status, detail, fix = check_vlm()
assert status == "warn"
assert "does not support Function Calling" in detail
assert "Function-Calling-capable" in fix

def test_config_fail_skips_probe(self, tmp_path: Path):
config = tmp_path / "ov.conf"
config.write_text(json.dumps({"vlm": {}}))
with patch("openviking_cli.doctor._find_config", return_value=config):
with patch("openviking_cli.doctor._probe_vlm_function_calling") as probe:
status, detail, fix = check_vlm()
assert status is False
probe.assert_not_called()

def test_codex_and_ollama_paths_skip_probe(self, tmp_path: Path):
# Early-return paths (codex-oauth :361, ollama-via-litellm :385) never probe.
codex = tmp_path / "codex.conf"
codex.write_text(
json.dumps({"vlm": {"provider": "openai-codex", "model": "gpt-5.3-codex"}})
)
with patch("openviking_cli.doctor._find_config", return_value=codex):
with patch(
"openviking.models.vlm.backends.codex_auth.resolve_codex_runtime_credentials",
return_value={"source": "openviking"},
):
with patch("openviking_cli.doctor._probe_vlm_function_calling") as probe:
ok, _, _ = check_vlm()
assert ok
probe.assert_not_called()

ollama = tmp_path / "ollama.conf"
ollama.write_text(
json.dumps({"vlm": {"provider": "litellm", "model": "ollama/llama3"}})
)
with patch("openviking_cli.doctor._find_config", return_value=ollama):
with patch("openviking_cli.doctor._probe_vlm_function_calling") as probe:
ok, _, _ = check_vlm()
assert ok
probe.assert_not_called()

@pytest.mark.parametrize(
"fc_return, fc_raises, expect_warn, expect_probe_called",
[
(False, False, True, False), # registry says False -> WARN, no live request
(True, False, False, True), # registry says True -> fall through to live probe
(None, True, False, True), # registry raises (unknown) -> fall through
],
)
def test_litellm_registry_prefilter_warns(
self, tmp_path: Path, fc_return, fc_raises, expect_warn, expect_probe_called
):
config = tmp_path / "ov.conf"
config.write_text(
json.dumps(
{"vlm": {"provider": "litellm", "model": "gpt-4o", "api_key": "sk-test"}}
)
)

def supports_fc(*args, **kwargs):
if fc_raises:
raise Exception("unknown model")
return fc_return

with patch("openviking_cli.doctor._find_config", return_value=config):
with patch("litellm.supports_function_calling", side_effect=supports_fc):
with patch(
"openviking_cli.doctor._probe_vlm_function_calling",
return_value=("pass", "litellm/gpt-4o function-calling ok", None),
) as probe:
status, detail, fix = check_vlm()

if expect_warn:
assert status == "warn"
assert "Function Calling" in detail
else:
assert status == "pass"
assert probe.called is expect_probe_called


def _fc_probe_config() -> VLMConfig:
return VLMConfig(provider="openai", model="gpt-4o-mini", api_key="sk-test")


def _patch_completion(async_fn):
"""Patch VLMConfig.get_completion_async with an async replacement."""
return patch.object(VLMConfig, "get_completion_async", async_fn)


class TestProbeVlmFunctionCalling:
"""Live Function-Calling probe (§5 rows 1-5, 11)."""

def test_passes_on_fc_model(self):
async def ok(self, *args, **kwargs):
return SimpleNamespace(content="ok")

with _patch_completion(ok):
status, detail, fix = _probe_vlm_function_calling(_fc_probe_config())
assert status == "pass"
assert "function-calling ok" in detail
assert fix is None

def test_detects_missing_fc_20037(self):
# The faithful repro: exact 400 / code 20037 body raised at tool-call time.
async def raise_20037(self, *args, **kwargs):
raise Exception(
"Error code: 400 - {'code': 20037, 'message': "
"'Function call is not supported for this model.', 'data': None}"
)

with _patch_completion(raise_20037):
status, detail, fix = _probe_vlm_function_calling(_fc_probe_config())
assert status == "warn"
assert "Function Calling" in detail
assert "Function-Calling-capable" in fix

def test_detects_missing_fc_by_phrase(self):
async def raise_phrase(self, *args, **kwargs):
raise Exception("function call is not supported")

with _patch_completion(raise_phrase):
status, detail, fix = _probe_vlm_function_calling(_fc_probe_config())
assert status == "warn"
assert "Function Calling" in detail

def test_timeout_warns_not_fails(self):
async def raise_timeout(self, *args, **kwargs):
raise TimeoutError()

with _patch_completion(raise_timeout):
status, detail, fix = _probe_vlm_function_calling(_fc_probe_config())
assert status == "warn"
assert "timed out" in detail

def test_other_error_warns(self):
async def raise_conn(self, *args, **kwargs):
raise RuntimeError("connection refused")

with _patch_completion(raise_conn):
status, detail, fix = _probe_vlm_function_calling(_fc_probe_config())
assert status == "warn"
assert "could not complete" in detail

def test_transient_error_not_mislabeled(self):
# A transient outage must never be reported as an FC verdict (row 11).
async def raise_conn(self, *args, **kwargs):
raise RuntimeError("connection refused")

with _patch_completion(raise_conn):
status, detail, fix = _probe_vlm_function_calling(_fc_probe_config())
assert status == "warn"
assert "could not complete" in detail
assert "does not support Function Calling" not in detail


class TestCheckOllama:
def test_pass_when_config_is_missing(self):
Expand Down Expand Up @@ -1018,7 +1214,11 @@ def test_returns_zero_when_all_pass(self, tmp_path: Path, capsys):
"openviking_cli.doctor._probe_embedding_provider",
return_value=(True, "openai/m probe ok", None),
):
code = run_doctor()
with patch(
"openviking_cli.doctor._probe_vlm_function_calling",
return_value=("pass", "openai/m function-calling ok", None),
):
code = run_doctor()
captured = capsys.readouterr()
assert "OpenViking Doctor" in captured.out
# May not be 0 if native engine is missing, but the function should complete
Expand All @@ -1031,6 +1231,30 @@ def test_returns_one_on_failure(self, capsys):
captured = capsys.readouterr()
assert "FAIL" in captured.out

def test_vlm_probe_warn_formats_and_exit_zero(self, tmp_path: Path, capsys):
config = tmp_path / "ov.conf"
config.write_text(
json.dumps(
{"vlm": {"provider": "openai", "model": "gpt-4o-mini", "api_key": "sk-test"}}
)
)
with patch("openviking_cli.doctor._CHECKS", [("VLM", check_vlm)]):
with patch("openviking_cli.doctor._find_config", return_value=config):
with patch(
"openviking_cli.doctor._probe_vlm_function_calling",
return_value=(
"warn",
"openai/gpt-4o-mini does not support Function Calling (tool use)",
"Configure a Function-Calling-capable vlm.model",
),
):
code = run_doctor()
assert code == 0
captured = capsys.readouterr()
assert "WARN" in captured.out
assert "Fix:" in captured.out
assert "Function Calling" in captured.out

def test_returns_zero_on_warning_only(self, capsys):
with patch(
"openviking_cli.doctor._CHECKS",
Expand Down