tests/test_model_only_runtime.py

from __future__ import annotations

import json
from pathlib import Path
from types import SimpleNamespace
import sys

ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / 'src'
for candidate in (ROOT, SRC):
    if str(candidate) not in sys.path:
        sys.path.insert(0, str(candidate))

import pytest

from app_kit.model_runtime import LoadedModel
from app_kit.project import processor_p1


class MockStore:
    def __init__(self) -> None:
        self.records: list[dict[str, object]] = []
        self.embeddings: list[tuple[str, str, str]] = []

    def store_record(self, project, pack_id, title, primary_text, payload, status='stored', record_id=None):
        self.records.append(
            {
                'project': project,
                'pack_id': pack_id,
                'title': title,
                'primary_text': primary_text,
                'payload': payload,
                'status': status,
            }
        )
        return record_id or 'record-1'

    def store_embedding(self, record_id, project, text, metadata=None):
        self.embeddings.append((record_id, project, text))


class MockPack:
    def __init__(self, dir_path: Path) -> None:
        self.path = dir_path
        self.pack_id = 'mock_123'
        self.expected_signals = {'triage': 'important'}
        self.manifest = {'inputs': [{'path': 'sample.txt', 'kind': 'document'}]}


class FakeLLM:
    def __init__(self, responses: list[str]) -> None:
        self.responses = responses
        self.calls: list[dict[str, object]] = []

    def create_completion(self, **kwargs):
        self.calls.append(kwargs)
        if not self.responses:
            raise RuntimeError('unexpected extra model call')
        content = self.responses.pop(0)
        return {
            'choices': [{'text': content}],
            'usage': {'prompt_tokens': 123, 'completion_tokens': 42, 'total_tokens': 165},
        }

    def create_chat_completion(self, **kwargs):
        self.calls.append(kwargs)
        if not self.responses:
            raise RuntimeError('unexpected extra model call')
        content = self.responses.pop(0)
        return {
            'choices': [{'message': {'content': content}}],
            'usage': {'prompt_tokens': 123, 'completion_tokens': 42, 'total_tokens': 165},
        }


@pytest.fixture()
def sample_text() -> str:
    return (
        'Notice of benefits renewal is attached.\n'
        'Please call the office by June 20 to confirm your appointment.\n'
        'If you have questions, keep the reference number handy.'
    )


def test_resolve_model_path_fail_fast_when_missing(monkeypatch, tmp_path):
    from app_kit import model_runtime

    monkeypatch.setattr(model_runtime, '_candidate_roots', lambda: [tmp_path / 'empty-cache'])
    monkeypatch.setenv('P1_ALLOW_MODEL_DOWNLOAD', '0')
    monkeypatch.delenv('P1_MODEL_PATH', raising=False)

    with pytest.raises(FileNotFoundError):
        model_runtime.resolve_model_path()


def test_processor_p1_uses_model_output_and_logs_metadata(monkeypatch, sample_text, tmp_path):
    from app_kit import project

    responses = [
        'important',
        'A benefits renewal notice asks the user to call the office before the deadline.',
        'missing info likely',
        'It is a benefits renewal notice.',
        'Call the office to confirm the appointment.',
        'Yes, June 20.',
        'A reference number should be kept handy.',
    ]
    fake_llm = FakeLLM(responses[:])
    loaded_model = LoadedModel(
        model_id='Abiray/MiniCPM5-1B-GGUF:Q4_K_M',
        model_path=tmp_path / 'minicpm5-1b-Q4_K_M.gguf',
        source='local-cache',
    )

    monkeypatch.setattr(project, 'read_text_inputs', lambda pack: sample_text)
    monkeypatch.setattr(project, 'resolve_model_path', lambda **kwargs: loaded_model)
    monkeypatch.setattr(project, 'load_llama', lambda model_path: fake_llm)

    pack = MockPack(tmp_path)
    store = MockStore()
    result = project.processor_p1(pack, store, SimpleNamespace())

    assert result['model_id'] == loaded_model.model_id
    assert result['adapter_name'] == 'llama-cpp-python'
    assert result['generation_stats']['triage']['prompt_tokens'] == 123
    assert result['generation_stats']['summary']['completion_tokens'] == 42
    assert result['triage'] == 'important'
    assert result['summary'].startswith('A benefits renewal notice')
    assert result['qa'][0]['question'] == 'What is this document about?'
    assert result['citations'][0]['snippet'] == 'Notice of benefits renewal is attached.'
    assert store.records and store.records[0]['payload']['model_id'] == loaded_model.model_id