semif-agent/tests/integration/test_pipeline.py

"""End-to-end pipeline test. Run ONLY on the box with real SemIf + real LLM.

    python -m pytest tests/integration -q

The decision engine is real (llamacpp GGUF) and self-assessment is a real
local LLM endpoint. If either is unavailable this fails loudly — no mocking.
"""

import json
from pathlib import Path

import pytest

from semif_agent.cli import build_scheduler, load_config
from semif_agent.decisions import Request
from semif_agent.dream import dream
from semif_agent.engine import EngineUnavailable
from semif_agent.skills import CategoryDraft, SkillDraft, generate_category, generate_skill


def require_real(config: dict):
    from semif_agent.engine import SemIfEngine, EngineConfig

    engine = SemIfEngine(
        EngineConfig(
            backend=config.get("engine", {}).get("backend", "llamacpp"),
            source=config.get("engine", {}).get("source", ""),
            revision=config.get("engine", {}).get("revision", ""),
            gguf=config.get("engine", {}).get("gguf", ""),
            context_tokens=int(config.get("engine", {}).get("context_tokens", 4096)),
            threads=config.get("engine", {}).get("threads"),
        )
    )
    try:
        engine._ensure_loaded()
    except EngineUnavailable as exc:
        pytest.fail(f"real engine unavailable: {exc}")


def test_pipeline_end_to_end(tmp_path):
    config = load_config()
    require_real(config)
    config["log"] = str(tmp_path / "decisions.jsonl")
    config["trace"] = str(tmp_path / "runs.jsonl")
    scheduler, config = build_scheduler(config)

    inputs = [
        "send my girlfriend an email that says I'm going to be late to the party",
        "tell me if my package was delivered",
    ]
    for text in inputs:
        status, detail = scheduler.submit(text)
        print(f"[{status}] {detail}")
        assert status in ("running", "preempted", "queued", "rejected")

    scheduler.run_queue()

    rows = scheduler.log.read()
    assert len(rows) > 0, "expected SemIf decisions to be logged"

    phases = [r.get("extra", {}).get("phase") for r in rows]
    assert "navigate:category" in phases, "navigation decisions must be logged"
    assert "navigate:leaf" in phases, "navigation decisions must be logged"
    for row in rows:
        assert row.get("extra", {}).get("run_id"), "every decision must carry a run_id"

    trace_rows = scheduler.trace.read()
    assert any(r["kind"] == "submit" for r in trace_rows)
    assert any(r["kind"] == "assessed" for r in trace_rows)

    report = dream(scheduler.log)
    assert report.cross_entropy is not None
    print(report.render())

    skills = scheduler.status()
    assert "queue:" in skills

    contacts_path = Path(config.get("skills", {}).get("contacts", "data/contacts.json"))
    assert contacts_path.is_file()


def test_busy_choice_path(tmp_path):
    config = load_config()
    require_real(config)
    config["log"] = str(tmp_path / "decisions.jsonl")
    config["trace"] = str(tmp_path / "runs.jsonl")
    scheduler, config = build_scheduler(config)

    scheduler.busy("driving on the freeway", skill="driving")
    status, detail = scheduler.submit("send my girlfriend an email that says I'm going to be late")
    print(f"[{status}] {detail}")
    assert status in ("preempted", "queued", "dropped")
    scheduler.idle()


def test_engine_generation_normal_mode(tmp_path):
    """The pinned decision model must also generate text in the normal way."""
    config = load_config()
    require_real(config)
    scheduler, config = build_scheduler(config)
    out = scheduler.engine.generate(
        [
            {"role": "system", "content": "Reply with the single word ok."},
            {"role": "user", "content": "say ok"},
        ],
        max_tokens=16,
    )
    print(f"generation: {out!r}")
    assert isinstance(out, str) and out.strip()


def test_generate_category(tmp_path):
    """Authoring a category stub through the real decision model."""
    config = load_config()
    require_real(config)
    scheduler, config = build_scheduler(config)
    draft = generate_category(
        scheduler.engine,
        Request("tell me if my package was delivered"),
        scheduler.tree,
    )
    print(f"draft: {draft.name!r} — {draft.description!r}")
    assert isinstance(draft, CategoryDraft)
    assert draft.name and draft.description


def test_generate_skill(tmp_path):
    """Authoring a skill leaf stub through the real decision model."""
    config = load_config()
    require_real(config)
    scheduler, config = build_scheduler(config)
    draft = generate_skill(
        scheduler.engine,
        Request("tell me if my package was delivered"),
        "tracking",
        scheduler.tree,
    )
    print(f"draft: {draft.name!r} — {draft.description!r}")
    assert isinstance(draft, SkillDraft)
    assert draft.name and draft.description


def test_create_skill_empty_category_does_not_wedge(tmp_path):
    """A dispatch that lands on an empty category must not leave the scheduler wedged.

    Regression: navigation on a category with no skills produced a single-option
    SemIf decision, which the backend rejects; the exception unwound past the
    current-process reset, so every later request queued forever behind a phantom
    current. The empty category must now short-circuit straight to create_skill.
    """
    config = load_config()
    require_real(config)
    config["log"] = str(tmp_path / "decisions.jsonl")
    config["trace"] = str(tmp_path / "runs.jsonl")
    scheduler, config = build_scheduler(config)
    scheduler.tree["travel_planning"] = []

    for _ in range(2):
        status, detail = scheduler.submit("look up flights to japan for february")
        print(f"[{status}] {detail}")
        assert status in ("running", "preempted", "queued", "rejected", "error")
        assert scheduler.current is None, "scheduler must never stay wedged after a submit"

    status, detail = scheduler.submit("tell me if my package was delivered")
    print(f"[{status}] {detail}")
    assert scheduler.current is None