service-update-alerts/test_classifier.py

#!/usr/bin/env python3
"""
Test the Ollama model's ability to classify release notes.
Runs leave-one-out: for each example, uses the rest as few-shot context
and checks whether the model predicts the correct action.

Usage: python test_classifier.py [service_name]
  service_name defaults to "immich"
"""

import json
import os
import sys
from pathlib import Path
from urllib.request import Request, urlopen

import re as _re

OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://192.168.1.20:11434")
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "gemma4:26b-64k")
SERVICE_NAME = sys.argv[1] if len(sys.argv) > 1 else "immich"
EXAMPLES_DIR = Path(__file__).parent / "services" / SERVICE_NAME / "review-examples"


def guess_version_from_notes(notes: str) -> str:
    """Extract a version number from release notes (e.g. '# v2.7.0' or '### v8.1.5')."""
    match = _re.search(r"v?(\d+\.\d+\.\d+)", notes)
    return match.group(1) if match else "1.0.0"


def make_previous_version(version: str) -> str:
    """Decrement patch to simulate upgrading from the previous patch release."""
    parts = version.split(".")
    if len(parts) == 3 and parts[2].isdigit():
        patch = int(parts[2])
        if patch > 0:
            return f"{parts[0]}.{parts[1]}.{patch - 1}"
    return version


def load_examples(directory: Path) -> list[dict]:
    examples = []
    for path in sorted(directory.glob("*.txt")):
        text = path.read_text(encoding="utf-8")
        if "---" not in text:
            print(f"WARNING: {path.name} has no frontmatter separator, skipping")
            continue
        header, _, body = text.partition("---\n")
        action = None
        reason = None
        for line in header.strip().splitlines():
            if line.startswith("action:"):
                action = line.split(":", 1)[1].strip()
            elif line.startswith("reason:"):
                reason = line.split(":", 1)[1].strip()
        if action not in ("auto", "manual"):
            print(f"WARNING: {path.name} has invalid action '{action}', skipping")
            continue
        examples.append({
            "file": path.name,
            "action": action,
            "reason": reason or "",
            "notes": body.strip(),
        })
    return examples


def build_messages(examples: list[dict], target_notes: str, current: str = "1.0.0", latest: str = "2.0.0") -> list[dict]:
    context_file = EXAMPLES_DIR.parent / "context.txt"
    context = context_file.read_text(encoding="utf-8").strip() if context_file.is_file() else None

    system = (
        f"You are an update safety classifier for self-hosted {SERVICE_NAME} instances.\n"
        f"Given release notes for a {SERVICE_NAME} update, decide whether the update can be\n"
        "applied automatically (action: auto) or requires the admin to intervene\n"
        "manually before or during the update (action: manual).\n\n"
    )
    if context:
        system += f"Instance context: {context}\n\n"
    system += (
        "Rules:\n"
        "- Only flag as manual if the breaking changes actually affect this specific setup.\n"
        "- Classify only THIS release, not future or past releases mentioned in the notes.\n"
        "- Warnings or announcements about breaking changes in upcoming versions do NOT\n"
        "  make the current release manual.\n"
        "- Conditional notes like 'follow X instructions if upgrading from <= Y' only apply\n"
        "  if the current running version is <= Y. If we are already past Y, they do not apply.\n"
        "- Respond ONLY with a JSON object: {\"action\": \"auto\", \"reason\": \"...\"}\n"
        "  or {\"action\": \"manual\", \"reason\": \"...\"}\n"
        "- Keep reason under 60 words."
    )

    messages = [{"role": "system", "content": system}]

    for ex in examples:
        messages.append({"role": "user", "content": f"Release notes:\n{ex['notes']}"})
        messages.append({
            "role": "assistant",
            "content": json.dumps({"action": ex["action"], "reason": ex["reason"]}),
        })

    messages.append({"role": "user", "content": f"Upgrading from {current} to {latest}.\n\nRelease notes:\n{target_notes}"})
    return messages


def call_ollama(messages: list[dict]) -> dict:
    payload = json.dumps({
        "model": OLLAMA_MODEL,
        "messages": messages,
        "stream": False,
        "options": {"temperature": 0},
    }).encode("utf-8")
    req = Request(
        f"{OLLAMA_BASE_URL}/v1/chat/completions",
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(req, timeout=300) as resp:
        data = json.loads(resp.read())
    return data


def parse_response(text: str) -> dict:
    text = text.strip()
    # extract JSON from response (model may wrap it in markdown)
    start = text.find("{")
    end = text.rfind("}") + 1
    if start == -1 or end == 0:
        return {"action": "parse_error", "reason": text[:200]}
    try:
        return json.loads(text[start:end])
    except json.JSONDecodeError:
        return {"action": "parse_error", "reason": text[:200]}


def main():
    examples = load_examples(EXAMPLES_DIR)
    if not examples:
        print("No examples found.")
        sys.exit(1)

    print(f"Loaded {len(examples)} examples from {EXAMPLES_DIR}")
    print(f"Model: {OLLAMA_MODEL}")
    print(f"Running leave-one-out test ({len(examples)} calls)...\n")

    passed = 0
    failed = 0

    for i, target in enumerate(examples):
        few_shot = [ex for j, ex in enumerate(examples) if j != i]
        latest = guess_version_from_notes(target["notes"])
        current = make_previous_version(latest)
        messages = build_messages(few_shot, target["notes"], current, latest)

        try:
            response = call_ollama(messages)
            content = response["choices"][0]["message"]["content"]
            result = parse_response(content)
        except Exception as exc:
            result = {"action": "error", "reason": str(exc)}

        predicted = result.get("action", "error")
        expected = target["action"]
        ok = predicted == expected
        status = "PASS" if ok else "FAIL"

        if ok:
            passed += 1
        else:
            failed += 1

        print(f"[{status}] {target['file']}")
        print(f"       expected : {expected}")
        print(f"       predicted: {predicted}")
        print(f"       reason   : {result.get('reason', '')}")
        print()

    print(f"Results: {passed}/{len(examples)} passed", end="")
    if failed:
        print(f", {failed} failed")
    else:
        print(" — all correct!")

    sys.exit(0 if failed == 0 else 1)


if __name__ == "__main__":
    main()