m0_11_ollama_smoke.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/scripts/m0_11_ollama_smoke.py
 
# OCT-LINT: disable=tests_exist reason="Live smoke harness — exercising it against a real Ollama daemon IS the test; no unit test applies."
"""
M0.11 live-smoke harness for the Ollama provider (O1 / O4 / O5).
 
Purpose
-------
Exercises the Ollama provider against a real daemon and records
pass/fail for the three deliverables that shipped with unit tests
but were never live-verified:
 
    O1  provider construction, model listing, a streamed turn,
        model_info, and the clean-error path on an unreachable host
    O4  generation options (top_k) reaching the /api/chat request body
    O5  PerformanceStats / tok-s footer from real generation timing
 
Run
---
    OLLAMA_BASE_URL=http://192.168.1.20:11434 \\
        python -m oc_chat.scripts.m0_11_ollama_smoke [--model llama3.2:3b]
 
Exit code 0 when every smoke passes, 1 otherwise.
 
Responsibilities
----------------
- Build the provider/client exactly as OC Chat does at runtime.
- Capture the live /api/chat request body for the O4 assertion.
- Print one ``PASS``/``FAIL`` line per smoke plus an evidence block.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 - smoke lifecycle events
    L3 - per-smoke evidence
    L4 - per-call payload traces
 
Contracts
---------
- Read-only against the daemon apart from normal generation calls.
- Never mutates repo state; prints a record to stdout only.
 
Dependencies
------------
- oc_diagnostics (_dbg)
- oc_chat.core.generation_options (lazy; GenerationOptions for the O4 top_k smoke)
- providers.ollama.provider (lazy; OllamaProvider, the provider under test: client build, model_info, generation-option translation and performance parsing)
- providers.ollama.client (lazy; _post, patched to capture the live /api/chat request body)
"""
from __future__ import annotations
 
import argparse
import os
import sys
from typing import Any
 
from oc_diagnostics import _dbg
 
DEAD_URL = "http://localhost:11500"  # nothing listens here -> instant refuse
 
 
def _smoke_o1(base_url: str, model: str) -> tuple[bool, list[str]]:
    """O1 — provider build, list_models, streamed turn, model_info, clean error."""
    func = "_smoke_o1"
    ev: list[str] = []
    ok = True
    from providers.ollama.provider import OllamaProvider
 
    provider = OllamaProvider()
    client = provider.build_client(api_key="", base_url=base_url)
    ev.append(f"O1a provider/client built — base_url={client.base_url!r}")
 
    models = client.list_models()
    has_model = model in models
    ok &= has_model
    ev.append(f"O1b list_models -> {len(models)} tags; {model!r} present: {has_model}")
 
    chunks: list[str] = []
    usage: dict[str, Any] = {}
    for kind, data in client.stream_chat(
        [{"role": "user", "content": "Reply with exactly: smoke ok"}],
        model,
        max_tokens=32,
    ):
        if kind == "content":
            chunks.append(data)
        elif kind == "usage":
            usage = data
    text = "".join(chunks).strip()
    streamed_ok = len(text) > 0
    ok &= streamed_ok
    ev.append(f"O1c streamed turn -> {len(text)} chars: {text[:60]!r}")
 
    info = provider.model_info(model, client)
    info_ok = info is not None
    ok &= info_ok
    if info is not None:
        ev.append(f"O1d model_info -> family={info.family!r} params={info.parameter_count!r}")
    else:
        ev.append("O1d model_info -> None (FAIL)")
 
    dead = provider.build_client(api_key="", base_url=DEAD_URL)
    try:
        dead.list_models()
        ev.append("O1e clean-error path -> no error raised (FAIL)")
        ok = False
    except RuntimeError as exc:
        friendly = "unreachable" in str(exc).lower()
        ok &= friendly
        ev.append(f"O1e unreachable host -> clean RuntimeError, friendly={friendly}: {str(exc)[:80]!r}")
    except Exception as exc:  # noqa: BLE001 - smoke must report any non-clean failure
        ev.append(f"O1e unreachable host -> non-clean {type(exc).__name__} (FAIL)")
        ok = False
 
    for line in ev:
        _dbg("OC_CHAT.CLIENT", func, line, 3)
    _dbg("OC_CHAT.CLIENT", func, f"O1 ok={ok}", 2)
    return ok, ev
 
 
def _smoke_o4(base_url: str, model: str) -> tuple[bool, list[str]]:
    """O4 — generation options (top_k) reach the live /api/chat request body."""
    func = "_smoke_o4"
    ev: list[str] = []
    from oc_chat.core.generation_options import GenerationOptions
    from providers.ollama.provider import OllamaProvider
    from providers.ollama import client as ollama_client
 
    provider = OllamaProvider()
    extra_body = provider.translate_generation_options(GenerationOptions(top_k=10))
    ev.append(f"O4a translate_generation_options(top_k=10) -> {extra_body}")
 
    captured: dict[str, Any] = {}
    original_post = ollama_client._post
 
    def _capturing_post(b_url: str, path: str, body: dict, timeout: int = 300) -> Any:
        if path == "/api/chat":
            captured.update(body)
        return original_post(b_url, path, body, timeout)
 
    ollama_client._post = _capturing_post  # type: ignore[assignment]
    try:
        cli = provider.build_client(api_key="", base_url=base_url)
        for _kind, _data in cli.stream_chat(
            [{"role": "user", "content": "hi"}], model,
            extra_body=extra_body, max_tokens=8,
        ):
            pass
    finally:
        ollama_client._post = original_post  # type: ignore[assignment]
 
    sent_top_k = (captured.get("options") or {}).get("top_k")
    ok = sent_top_k == 10
    ev.append(f"O4b live /api/chat body options={captured.get('options')!r} -> top_k={sent_top_k}")
    for line in ev:
        _dbg("OC_CHAT.CLIENT", func, line, 3)
    _dbg("OC_CHAT.CLIENT", func, f"O4 ok={ok}", 2)
    return ok, ev
 
 
def _smoke_o5(base_url: str, model: str) -> tuple[bool, list[str]]:
    """O5 — PerformanceStats / tok-s footer from real generation timing."""
    func = "_smoke_o5"
    ev: list[str] = []
    from providers.ollama.provider import OllamaProvider
 
    provider = OllamaProvider()
    client = provider.build_client(api_key="", base_url=base_url)
 
    usage: dict[str, Any] = {}
    for kind, data in client.stream_chat(
        [{"role": "user", "content": "Count to five."}], model, max_tokens=64,
    ):
        if kind == "usage":
            usage = data
    perf = provider.parse_performance(usage)
    ev.append(
        f"O5a parse_performance -> eval_count={perf.eval_count} "
        f"eval_duration_ns={perf.eval_duration_ns} is_default={perf.is_default()}"
    )
 
    tps = perf.tokens_per_second
    ok = (not perf.is_default()) and tps is not None and tps > 0
    if tps is not None:
        footer = f"{perf.eval_count}tok@{tps:.1f}tok/s"
        ev.append(f"O5b tok/s footer -> {footer!r}")
    else:
        ev.append("O5b tokens_per_second -> None (FAIL)")
    for line in ev:
        _dbg("OC_CHAT.CLIENT", func, line, 3)
    _dbg("OC_CHAT.CLIENT", func, f"O5 ok={ok}", 2)
    return ok, ev
 
 
def main(argv: list[str] | None = None) -> int:
    """Run the O1/O4/O5 smokes; return 0 when all pass."""
    func = "main"
    parser = argparse.ArgumentParser(description="M0.11 Ollama live-smoke harness.")
    parser.add_argument("--model", default="llama3.2:3b", help="Ollama model tag to smoke")
    args = parser.parse_args(argv)
 
    base_url = os.environ.get("OLLAMA_BASE_URL", "http://localhost:11434")
    print(f"=== M0.11 Ollama live smoke — base_url={base_url} model={args.model} ===\n")
 
    results: list[tuple[str, bool, list[str]]] = []
    for name, fn in (("O1", _smoke_o1), ("O4", _smoke_o4), ("O5", _smoke_o5)):
        try:
            ok, ev = fn(base_url, args.model)
        except Exception as exc:  # noqa: BLE001 - a crashed smoke is a FAIL, not a traceback
            ok, ev = False, [f"{name} raised {type(exc).__name__}: {exc}"]
        results.append((name, ok, ev))
        print(f"[{'PASS' if ok else 'FAIL'}] {name}")
        for line in ev:
            print(f"    {line}")
        print()
 
    all_ok = all(ok for _name, ok, _ev in results)
    print(f"=== {'ALL SMOKES PASS' if all_ok else 'SMOKE FAILURES PRESENT'} ===")
    _dbg("OC_CHAT.CLIENT", func, f"smoke complete all_ok={all_ok}", 2)
    return 0 if all_ok else 1
 
 
if __name__ == "__main__":
    sys.exit(main())