wehub-resource-sync
116 lines
3.9 KiB
Python
116 lines
3.9 KiB
Python
"""``opensquilla router calibrate`` CLI.
|
|
|
|
Seeds a synthetic ``router_decisions`` table (hand-created — never real state)
|
|
and drives the command through Typer's ``CliRunner``. The output file always
|
|
lands under a monkeypatched ``OPENSQUILLA_STATE_DIR`` temp dir, never the real
|
|
home.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from opensquilla.cli.main import app
|
|
from opensquilla.persistence.router_decision_writer import RouterDecisionWriter
|
|
|
|
runner = CliRunner()
|
|
|
|
_CREATE_TABLE = (
|
|
"CREATE TABLE router_decisions ("
|
|
" decision_id TEXT PRIMARY KEY, session_key TEXT NOT NULL,"
|
|
" turn_index INTEGER, ts_ms INTEGER NOT NULL, classifier TEXT,"
|
|
" proposed_tier TEXT, confidence REAL, probs TEXT, flags TEXT,"
|
|
" final_tier TEXT, provider TEXT, model TEXT, thinking_level TEXT,"
|
|
" source TEXT, trail TEXT, baseline_model TEXT, savings_pct REAL,"
|
|
" executed_kind TEXT, ensemble_profile TEXT,"
|
|
" fallback_hops INTEGER NOT NULL DEFAULT 0)"
|
|
)
|
|
|
|
|
|
def _seed_db(path: Path, *, count: int = 40) -> None:
|
|
conn = sqlite3.connect(str(path), check_same_thread=False, isolation_level=None)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute(_CREATE_TABLE)
|
|
writer = RouterDecisionWriter(conn)
|
|
for index in range(count):
|
|
writer.record_decision(
|
|
{
|
|
"decision_id": f"c{index}",
|
|
"session_key": "agent:calib:main",
|
|
"turn_index": index,
|
|
"proposed_tier": "c2",
|
|
"final_tier": "c1",
|
|
"source": "v4_phase3",
|
|
"trail": [{"stage": "confidence_gate", "applied": True}],
|
|
}
|
|
)
|
|
writer.close()
|
|
|
|
|
|
def _env(monkeypatch: Any, tmp_path: Path, db: Path | None) -> None:
|
|
monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(tmp_path / "home"))
|
|
if db is not None:
|
|
monkeypatch.setenv("OPENSQUILLA_ROUTER_DECISIONS_DB", str(db))
|
|
else:
|
|
monkeypatch.delenv("OPENSQUILLA_ROUTER_DECISIONS_DB", raising=False)
|
|
|
|
|
|
def _output_file(tmp_path: Path) -> Path:
|
|
return tmp_path / "home" / "state" / "router_calibration.json"
|
|
|
|
|
|
def test_calibrate_dry_run_prints_without_writing(tmp_path: Path, monkeypatch: Any) -> None:
|
|
db = tmp_path / "sessions.db"
|
|
_seed_db(db)
|
|
_env(monkeypatch, tmp_path, db)
|
|
|
|
result = runner.invoke(app, ["router", "calibrate", "--dry-run"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "samples:" in result.output
|
|
assert "threshold_adjust:" in result.output
|
|
assert "dry-run" in result.output
|
|
assert not _output_file(tmp_path).exists()
|
|
|
|
|
|
def test_calibrate_writes_file(tmp_path: Path, monkeypatch: Any) -> None:
|
|
db = tmp_path / "sessions.db"
|
|
_seed_db(db)
|
|
_env(monkeypatch, tmp_path, db)
|
|
|
|
result = runner.invoke(app, ["router", "calibrate"])
|
|
assert result.exit_code == 0, result.output
|
|
|
|
out = _output_file(tmp_path)
|
|
assert out.exists()
|
|
payload = json.loads(out.read_text(encoding="utf-8"))
|
|
assert payload["sample_count"] == 40
|
|
# 40 clean c2 downgrades -> bias c2 down (clamped), threshold up.
|
|
assert payload["per_class_bias"] == {"c2": -0.15}
|
|
assert payload["threshold_adjust"] > 0.0
|
|
|
|
|
|
def test_calibrate_json_output(tmp_path: Path, monkeypatch: Any) -> None:
|
|
db = tmp_path / "sessions.db"
|
|
_seed_db(db)
|
|
_env(monkeypatch, tmp_path, db)
|
|
|
|
result = runner.invoke(app, ["router", "calibrate", "--json"])
|
|
assert result.exit_code == 0, result.output
|
|
doc = json.loads(result.output)
|
|
assert doc["wrote"] is True
|
|
assert doc["calibration"]["sample_count"] == 40
|
|
|
|
|
|
def test_calibrate_missing_db_is_neutral_no_crash(tmp_path: Path, monkeypatch: Any) -> None:
|
|
_env(monkeypatch, tmp_path, tmp_path / "does-not-exist.db")
|
|
|
|
result = runner.invoke(app, ["router", "calibrate", "--dry-run"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "samples: 0" in result.output
|
|
assert not _output_file(tmp_path).exists()
|