124 lines
4.2 KiB
Python
124 lines
4.2 KiB
Python
"""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, generate_category
|
|
|
|
|
|
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 |