Thales Maciel
663241d5d2
Make the local chat-host loop explicit and cheap so users can warm the machine once instead of rediscovering environment and guest setup on every session. Add cache-backed daily-loop manifests plus the new `pyro prepare` flow, extend `pyro doctor --environment` with warm/cold/stale readiness reporting, and add `make smoke-daily-loop` to prove the warmed repro-fix reset path end to end. Also fix `python -m pyro_mcp.cli` to invoke `main()` so the new smoke and `dist-check` actually exercise the CLI module, and update the docs/roadmap to present `doctor -> prepare -> connect host -> reset` as the recommended daily path. Validation: `uv lock`, `UV_OFFLINE=1 UV_CACHE_DIR=.uv-cache make check`, `UV_OFFLINE=1 UV_CACHE_DIR=.uv-cache make dist-check`, and `UV_OFFLINE=1 UV_CACHE_DIR=.uv-cache make smoke-daily-loop`.
359 lines
11 KiB
Python
359 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from pyro_mcp.daily_loop import (
|
|
DailyLoopManifest,
|
|
evaluate_daily_loop_status,
|
|
load_prepare_manifest,
|
|
prepare_manifest_path,
|
|
prepare_request_is_satisfied,
|
|
)
|
|
from pyro_mcp.runtime import RuntimeCapabilities
|
|
from pyro_mcp.vm_manager import VmManager
|
|
|
|
|
|
def test_prepare_daily_loop_executes_then_reuses(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
manager = VmManager(
|
|
backend_name="mock",
|
|
base_dir=tmp_path / "state",
|
|
cache_dir=tmp_path / "cache",
|
|
)
|
|
monkeypatch.setattr(manager, "_backend_name", "firecracker")
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"_runtime_capabilities",
|
|
RuntimeCapabilities(
|
|
supports_vm_boot=True,
|
|
supports_guest_exec=True,
|
|
supports_guest_network=True,
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"inspect_environment",
|
|
lambda environment: {"installed": True},
|
|
)
|
|
monkeypatch.setattr(
|
|
manager._environment_store,
|
|
"ensure_installed",
|
|
lambda environment: object(),
|
|
)
|
|
|
|
observed: dict[str, object] = {}
|
|
|
|
def fake_create_workspace(**kwargs: object) -> dict[str, object]:
|
|
observed["network_policy"] = kwargs["network_policy"]
|
|
return {"workspace_id": "ws-123"}
|
|
|
|
def fake_exec_workspace(
|
|
workspace_id: str,
|
|
*,
|
|
command: str,
|
|
timeout_seconds: int = 30,
|
|
secret_env: dict[str, str] | None = None,
|
|
) -> dict[str, object]:
|
|
observed["exec"] = {
|
|
"workspace_id": workspace_id,
|
|
"command": command,
|
|
"timeout_seconds": timeout_seconds,
|
|
"secret_env": secret_env,
|
|
}
|
|
return {
|
|
"workspace_id": workspace_id,
|
|
"stdout": "/workspace\n",
|
|
"stderr": "",
|
|
"exit_code": 0,
|
|
"duration_ms": 1,
|
|
"execution_mode": "guest_vsock",
|
|
}
|
|
|
|
def fake_reset_workspace(
|
|
workspace_id: str,
|
|
*,
|
|
snapshot: str = "baseline",
|
|
) -> dict[str, object]:
|
|
observed["reset"] = {"workspace_id": workspace_id, "snapshot": snapshot}
|
|
return {"workspace_id": workspace_id}
|
|
|
|
def fake_delete_workspace(
|
|
workspace_id: str,
|
|
*,
|
|
reason: str = "explicit_delete",
|
|
) -> dict[str, object]:
|
|
observed["delete"] = {"workspace_id": workspace_id, "reason": reason}
|
|
return {"workspace_id": workspace_id, "deleted": True}
|
|
|
|
monkeypatch.setattr(manager, "create_workspace", fake_create_workspace)
|
|
monkeypatch.setattr(manager, "exec_workspace", fake_exec_workspace)
|
|
monkeypatch.setattr(manager, "reset_workspace", fake_reset_workspace)
|
|
monkeypatch.setattr(manager, "delete_workspace", fake_delete_workspace)
|
|
|
|
first = manager.prepare_daily_loop("debian:12")
|
|
assert first["prepared"] is True
|
|
assert first["executed"] is True
|
|
assert first["reused"] is False
|
|
assert first["network_prepared"] is False
|
|
assert first["execution_mode"] == "guest_vsock"
|
|
assert observed["network_policy"] == "off"
|
|
assert observed["exec"] == {
|
|
"workspace_id": "ws-123",
|
|
"command": "pwd",
|
|
"timeout_seconds": 30,
|
|
"secret_env": None,
|
|
}
|
|
assert observed["reset"] == {"workspace_id": "ws-123", "snapshot": "baseline"}
|
|
assert observed["delete"] == {"workspace_id": "ws-123", "reason": "prepare_cleanup"}
|
|
|
|
second = manager.prepare_daily_loop("debian:12")
|
|
assert second["prepared"] is True
|
|
assert second["executed"] is False
|
|
assert second["reused"] is True
|
|
assert second["reason"] == "reused existing warm manifest"
|
|
|
|
|
|
def test_prepare_daily_loop_force_and_network_upgrade_manifest(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
manager = VmManager(
|
|
backend_name="mock",
|
|
base_dir=tmp_path / "state",
|
|
cache_dir=tmp_path / "cache",
|
|
)
|
|
monkeypatch.setattr(manager, "_backend_name", "firecracker")
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"_runtime_capabilities",
|
|
RuntimeCapabilities(
|
|
supports_vm_boot=True,
|
|
supports_guest_exec=True,
|
|
supports_guest_network=True,
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"inspect_environment",
|
|
lambda environment: {"installed": True},
|
|
)
|
|
monkeypatch.setattr(
|
|
manager._environment_store,
|
|
"ensure_installed",
|
|
lambda environment: object(),
|
|
)
|
|
|
|
observed_policies: list[str] = []
|
|
|
|
def fake_create_workspace(**kwargs: object) -> dict[str, object]:
|
|
observed_policies.append(str(kwargs["network_policy"]))
|
|
return {"workspace_id": "ws-1"}
|
|
|
|
monkeypatch.setattr(manager, "create_workspace", fake_create_workspace)
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"exec_workspace",
|
|
lambda workspace_id, **kwargs: {
|
|
"workspace_id": workspace_id,
|
|
"stdout": "/workspace\n",
|
|
"stderr": "",
|
|
"exit_code": 0,
|
|
"duration_ms": 1,
|
|
"execution_mode": "guest_vsock",
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"reset_workspace",
|
|
lambda workspace_id, **kwargs: {"workspace_id": workspace_id},
|
|
)
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"delete_workspace",
|
|
lambda workspace_id, **kwargs: {"workspace_id": workspace_id, "deleted": True},
|
|
)
|
|
|
|
manager.prepare_daily_loop("debian:12")
|
|
payload = manager.prepare_daily_loop("debian:12", network=True, force=True)
|
|
assert payload["executed"] is True
|
|
assert payload["network_prepared"] is True
|
|
assert observed_policies == ["off", "egress"]
|
|
|
|
manifest_path = prepare_manifest_path(
|
|
tmp_path / "cache",
|
|
platform="linux-x86_64",
|
|
environment="debian:12",
|
|
)
|
|
manifest, manifest_error = load_prepare_manifest(manifest_path)
|
|
assert manifest_error is None
|
|
if manifest is None:
|
|
raise AssertionError("expected prepare manifest")
|
|
assert manifest.network_prepared is True
|
|
|
|
|
|
def test_prepare_daily_loop_requires_guest_capabilities(tmp_path: Path) -> None:
|
|
manager = VmManager(
|
|
backend_name="mock",
|
|
base_dir=tmp_path / "state",
|
|
cache_dir=tmp_path / "cache",
|
|
)
|
|
with pytest.raises(RuntimeError, match="guest-backed runtime"):
|
|
manager.prepare_daily_loop("debian:12")
|
|
|
|
|
|
def test_load_prepare_manifest_reports_invalid_json(tmp_path: Path) -> None:
|
|
manifest_path = prepare_manifest_path(
|
|
tmp_path,
|
|
platform="linux-x86_64",
|
|
environment="debian:12",
|
|
)
|
|
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
manifest_path.write_text("{broken", encoding="utf-8")
|
|
|
|
manifest, error = load_prepare_manifest(manifest_path)
|
|
assert manifest is None
|
|
assert error is not None
|
|
|
|
|
|
def test_prepare_manifest_round_trip(tmp_path: Path) -> None:
|
|
manifest_path = prepare_manifest_path(
|
|
tmp_path,
|
|
platform="linux-x86_64",
|
|
environment="debian:12",
|
|
)
|
|
manifest = DailyLoopManifest(
|
|
environment="debian:12",
|
|
environment_version="1.0.0",
|
|
platform="linux-x86_64",
|
|
catalog_version="4.5.0",
|
|
bundle_version="bundle-1",
|
|
prepared_at=123.0,
|
|
network_prepared=True,
|
|
last_prepare_duration_ms=456,
|
|
)
|
|
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
manifest_path.write_text(json.dumps(manifest.to_payload()), encoding="utf-8")
|
|
|
|
loaded, error = load_prepare_manifest(manifest_path)
|
|
assert error is None
|
|
assert loaded == manifest
|
|
|
|
|
|
def test_load_prepare_manifest_rejects_non_object(tmp_path: Path) -> None:
|
|
manifest_path = prepare_manifest_path(
|
|
tmp_path,
|
|
platform="linux-x86_64",
|
|
environment="debian:12",
|
|
)
|
|
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
manifest_path.write_text('["not-an-object"]', encoding="utf-8")
|
|
|
|
manifest, error = load_prepare_manifest(manifest_path)
|
|
assert manifest is None
|
|
assert error == "prepare manifest is not a JSON object"
|
|
|
|
|
|
def test_load_prepare_manifest_rejects_invalid_payload(tmp_path: Path) -> None:
|
|
manifest_path = prepare_manifest_path(
|
|
tmp_path,
|
|
platform="linux-x86_64",
|
|
environment="debian:12",
|
|
)
|
|
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
manifest_path.write_text(json.dumps({"environment": "debian:12"}), encoding="utf-8")
|
|
|
|
manifest, error = load_prepare_manifest(manifest_path)
|
|
assert manifest is None
|
|
assert error is not None
|
|
assert "prepare manifest is invalid" in error
|
|
|
|
|
|
def test_evaluate_daily_loop_status_edge_cases() -> None:
|
|
manifest = DailyLoopManifest(
|
|
environment="debian:12",
|
|
environment_version="1.0.0",
|
|
platform="linux-x86_64",
|
|
catalog_version="4.5.0",
|
|
bundle_version="bundle-1",
|
|
prepared_at=1.0,
|
|
network_prepared=False,
|
|
last_prepare_duration_ms=2,
|
|
)
|
|
|
|
assert evaluate_daily_loop_status(
|
|
environment="debian:12",
|
|
environment_version="1.0.0",
|
|
platform="linux-x86_64",
|
|
catalog_version="4.5.0",
|
|
bundle_version="bundle-1",
|
|
installed=True,
|
|
manifest=manifest,
|
|
manifest_error="broken manifest",
|
|
) == ("stale", "broken manifest")
|
|
assert evaluate_daily_loop_status(
|
|
environment="debian:12",
|
|
environment_version="1.0.0",
|
|
platform="linux-x86_64",
|
|
catalog_version="4.5.0",
|
|
bundle_version="bundle-1",
|
|
installed=False,
|
|
manifest=manifest,
|
|
) == ("stale", "environment install is missing")
|
|
assert evaluate_daily_loop_status(
|
|
environment="debian:12-build",
|
|
environment_version="1.0.0",
|
|
platform="linux-x86_64",
|
|
catalog_version="4.5.0",
|
|
bundle_version="bundle-1",
|
|
installed=True,
|
|
manifest=manifest,
|
|
) == ("stale", "prepare manifest environment does not match the selected environment")
|
|
assert evaluate_daily_loop_status(
|
|
environment="debian:12",
|
|
environment_version="2.0.0",
|
|
platform="linux-x86_64",
|
|
catalog_version="4.5.0",
|
|
bundle_version="bundle-1",
|
|
installed=True,
|
|
manifest=manifest,
|
|
) == ("stale", "environment version changed since the last prepare run")
|
|
assert evaluate_daily_loop_status(
|
|
environment="debian:12",
|
|
environment_version="1.0.0",
|
|
platform="linux-aarch64",
|
|
catalog_version="4.5.0",
|
|
bundle_version="bundle-1",
|
|
installed=True,
|
|
manifest=manifest,
|
|
) == ("stale", "platform changed since the last prepare run")
|
|
assert evaluate_daily_loop_status(
|
|
environment="debian:12",
|
|
environment_version="1.0.0",
|
|
platform="linux-x86_64",
|
|
catalog_version="4.5.0",
|
|
bundle_version="bundle-2",
|
|
installed=True,
|
|
manifest=manifest,
|
|
) == ("stale", "runtime bundle version changed since the last prepare run")
|
|
|
|
|
|
def test_prepare_request_is_satisfied_network_gate() -> None:
|
|
manifest = DailyLoopManifest(
|
|
environment="debian:12",
|
|
environment_version="1.0.0",
|
|
platform="linux-x86_64",
|
|
catalog_version="4.5.0",
|
|
bundle_version="bundle-1",
|
|
prepared_at=1.0,
|
|
network_prepared=False,
|
|
last_prepare_duration_ms=2,
|
|
)
|
|
|
|
assert prepare_request_is_satisfied(None, require_network=False) is False
|
|
assert prepare_request_is_satisfied(manifest, require_network=True) is False
|
|
assert prepare_request_is_satisfied(manifest, require_network=False) is True
|