mirror of
https://github.com/razzant/ouroboros.git
synced 2026-08-04 16:19:50 +00:00
fedd7d9fe6
Seed the combined Telegram bridge and Mini App as a native skill, add symmetric skill conflicts and managed repair routing, preserve the standalone PoC process boundary, and wire native delivery through Colab and packaged bootstrap.
1558 lines
60 KiB
Python
1558 lines
60 KiB
Python
"""Phase 5 regression tests for the extension HTTP surface.
|
|
|
|
Covers:
|
|
- ``GET /api/extensions`` catalogue snapshot
|
|
- ``GET /api/extensions/<skill>/manifest``
|
|
- ``ALL /api/extensions/<skill>/<rest>`` dispatcher
|
|
- ``POST /api/skills/<skill>/toggle`` UI-facing enable/disable
|
|
|
|
Uses Starlette TestClient so the full request path is exercised.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pathlib
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
|
|
from tests._shared import clean_extension_runtime_state
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_extensions():
|
|
clean_extension_runtime_state()
|
|
yield
|
|
clean_extension_runtime_state()
|
|
|
|
|
|
def _write_ext(
|
|
repo_root: pathlib.Path,
|
|
name: str,
|
|
*,
|
|
permissions: list[str],
|
|
plugin: str,
|
|
env_from_settings: list[str] | None = None,
|
|
conflicts: list[str] | None = None,
|
|
) -> pathlib.Path:
|
|
skill_dir = repo_root / name
|
|
skill_dir.mkdir(parents=True, exist_ok=True)
|
|
perms_yaml = json.dumps(permissions)
|
|
env_yaml = json.dumps(env_from_settings or [])
|
|
conflicts_yaml = json.dumps(conflicts or [])
|
|
(skill_dir / "SKILL.md").write_text(
|
|
(
|
|
"---\n"
|
|
f"name: {name}\n"
|
|
"description: Test ext.\n"
|
|
"version: 0.1.0\n"
|
|
"type: extension\n"
|
|
"entry: plugin.py\n"
|
|
f"permissions: {perms_yaml}\n"
|
|
f"env_from_settings: {env_yaml}\n"
|
|
f"conflicts: {conflicts_yaml}\n"
|
|
"---\n"
|
|
"body\n"
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(skill_dir / "plugin.py").write_text(plugin, encoding="utf-8")
|
|
return skill_dir
|
|
|
|
|
|
def _make_client(tmp_path: pathlib.Path, monkeypatch):
|
|
"""Return ``(client, drive_root, patches)`` — Starlette TestClient with drive_root pinned.
|
|
|
|
Tests that prefer the auto-cleanup variant should use the ``client_env``
|
|
fixture below instead of calling this directly.
|
|
"""
|
|
from unittest.mock import patch
|
|
from starlette.testclient import TestClient
|
|
|
|
import server as srv
|
|
|
|
monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "advanced")
|
|
drive_root = tmp_path / "drive"
|
|
drive_root.mkdir()
|
|
# ``srv.app`` is the NetworkAuthGate wrapper; the inner Starlette is at
|
|
# ``srv.app.app``. Pin ``drive_root`` / ``repo_dir`` on the inner state.
|
|
srv.app.app.state.drive_root = drive_root # type: ignore[attr-defined]
|
|
srv.app.app.state.repo_dir = tmp_path / "repo" # type: ignore[attr-defined]
|
|
|
|
patches = [
|
|
patch.object(srv, "_start_supervisor_if_needed", lambda *_a, **_k: None),
|
|
patch.object(srv, "_apply_settings_to_env", lambda *_a, **_k: None),
|
|
patch.object(srv, "apply_runtime_provider_defaults", lambda s: (s, False, [])),
|
|
patch("ouroboros.server_auth.get_configured_network_password", return_value=""),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
client = TestClient(srv.app)
|
|
return client, drive_root, patches
|
|
|
|
|
|
def _stop_patches(patches):
|
|
for p in patches:
|
|
try:
|
|
p.stop()
|
|
except RuntimeError:
|
|
pass
|
|
|
|
|
|
class _FakeUvicornServer:
|
|
def __init__(self, _config):
|
|
self.should_exit = False
|
|
|
|
async def serve(self):
|
|
await asyncio.sleep(0)
|
|
|
|
|
|
def _patch_lifespan_for_drive_root_test(monkeypatch, srv, settings: dict):
|
|
monkeypatch.setattr(srv, "load_settings", lambda: dict(settings))
|
|
monkeypatch.setattr(srv, "save_settings", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(srv, "apply_runtime_provider_defaults", lambda s: (s, False, []))
|
|
monkeypatch.setattr(srv, "_apply_settings_to_env", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(srv, "has_startup_ready_provider", lambda *_a, **_k: False)
|
|
# has_local_routing now lives only in server_runtime (server.py stopped
|
|
# importing it after the provider-check consolidation).
|
|
monkeypatch.setattr(
|
|
"ouroboros.server_runtime.has_local_routing", lambda *_a, **_k: False
|
|
)
|
|
monkeypatch.setattr(srv, "_start_supervisor_if_needed", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(srv.uvicorn, "Server", _FakeUvicornServer)
|
|
monkeypatch.setattr("ouroboros.launcher_bootstrap.ensure_data_skills_seeded", lambda: None)
|
|
monkeypatch.setattr("ouroboros.server_auth.get_configured_network_password", lambda: "")
|
|
|
|
|
|
def test_testclient_lifespan_reload_all_uses_app_state_drive_root(tmp_path, monkeypatch):
|
|
from starlette.testclient import TestClient
|
|
import server as srv
|
|
from ouroboros import extension_loader
|
|
|
|
drive_root = tmp_path / "drive"
|
|
repo_root = tmp_path / "skills"
|
|
drive_root.mkdir()
|
|
repo_root.mkdir()
|
|
srv.app.app.state.drive_root = drive_root # type: ignore[attr-defined]
|
|
srv.app.app.state.repo_dir = tmp_path / "repo" # type: ignore[attr-defined]
|
|
_patch_lifespan_for_drive_root_test(
|
|
monkeypatch,
|
|
srv,
|
|
{"OUROBOROS_SKILLS_REPO_PATH": str(repo_root), "OUROBOROS_RUNTIME_MODE": "advanced"},
|
|
)
|
|
monkeypatch.setattr("ouroboros.config.get_skills_repo_path", lambda: str(repo_root))
|
|
calls: list[tuple[pathlib.Path, str | None]] = []
|
|
monkeypatch.setattr(
|
|
extension_loader,
|
|
"reload_all",
|
|
lambda root, _reader, *, repo_path=None: calls.append((pathlib.Path(root), repo_path)) or {},
|
|
)
|
|
|
|
with TestClient(srv.app):
|
|
pass
|
|
|
|
assert calls == [(drive_root, str(repo_root))]
|
|
|
|
|
|
def test_testclient_settings_hot_reload_uses_app_state_drive_root(tmp_path, monkeypatch):
|
|
from starlette.testclient import TestClient
|
|
import server as srv
|
|
from ouroboros import extension_loader
|
|
|
|
drive_root = tmp_path / "drive"
|
|
old_repo = tmp_path / "skills-old"
|
|
new_repo = tmp_path / "skills-new"
|
|
drive_root.mkdir()
|
|
old_repo.mkdir()
|
|
new_repo.mkdir()
|
|
srv.app.app.state.drive_root = drive_root # type: ignore[attr-defined]
|
|
srv.app.app.state.repo_dir = tmp_path / "repo" # type: ignore[attr-defined]
|
|
settings = {"OUROBOROS_SKILLS_REPO_PATH": str(old_repo), "OUROBOROS_RUNTIME_MODE": "advanced"}
|
|
_patch_lifespan_for_drive_root_test(monkeypatch, srv, settings)
|
|
monkeypatch.setattr("ouroboros.config.get_skills_repo_path", lambda: str(old_repo))
|
|
calls: list[tuple[pathlib.Path, str | None]] = []
|
|
monkeypatch.setattr(
|
|
extension_loader,
|
|
"reload_all",
|
|
lambda root, _reader, *, repo_path=None: calls.append((pathlib.Path(root), repo_path)) or {},
|
|
)
|
|
|
|
with TestClient(srv.app) as client:
|
|
response = client.post("/api/settings", json={"OUROBOROS_SKILLS_REPO_PATH": str(new_repo)})
|
|
|
|
assert response.status_code == 200, response.text
|
|
assert calls
|
|
assert all(root == drive_root for root, _repo_path in calls)
|
|
assert (drive_root, str(new_repo)) in calls
|
|
|
|
|
|
@pytest.fixture
|
|
def client_env(tmp_path, monkeypatch):
|
|
"""Yield ``(client, drive_root)`` and stop lifecycle patches at teardown."""
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
yield client, drive_root
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extensions_index_lists_extension_skills(tmp_path, monkeypatch):
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"def register(api):\n"
|
|
" api.register_tool('t', lambda ctx: 'ok', description='', schema={})\n"
|
|
)
|
|
_write_ext(skills_root, "ext_a", permissions=["tool"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
broadcasts = []
|
|
client.app.app.state.broadcast_ws_sync = lambda payload: broadcasts.append(payload) # type: ignore[attr-defined]
|
|
try:
|
|
resp = client.get("/api/extensions")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
names = {s["name"] for s in data.get("skills", [])}
|
|
assert "ext_a" in names
|
|
assert "live" in data
|
|
ext_meta = next(s for s in data["skills"] if s["name"] == "ext_a")
|
|
assert ext_meta["live_reason"] == "disabled"
|
|
assert ext_meta["executable_review"] is False
|
|
assert ext_meta["review_gate"]["blocking_reason"] == "review_pending"
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_manifest_returns_metadata(tmp_path, monkeypatch):
|
|
skills_root = tmp_path / "skills"
|
|
plugin = "def register(api):\n pass\n"
|
|
_write_ext(skills_root, "ext_b", permissions=[], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
resp = client.get("/api/extensions/ext_b/manifest")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["name"] == "ext_b"
|
|
assert data["manifest"]["type"] == "extension"
|
|
assert data["executable_review"] is False
|
|
assert data["review_gate"]["blocking_reason"] == "review_pending"
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_manifest_prefers_runtime_load_error(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
find_skill,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
skill_dir = _write_ext(
|
|
skills_root,
|
|
"ext_manifest_error",
|
|
permissions=["route"],
|
|
plugin=(
|
|
"def _hello(request):\n"
|
|
" return {'hello': 'world'}\n"
|
|
"def register(api):\n"
|
|
" api.register_route('/absolute', _hello, methods=('GET',))\n"
|
|
),
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
broadcasts = []
|
|
client.app.app.state.broadcast_ws_sync = lambda payload: broadcasts.append(payload) # type: ignore[attr-defined]
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_manifest_error", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_manifest_error",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
loaded = find_skill(drive_root, "ext_manifest_error", repo_path=str(skills_root))
|
|
assert loaded is not None
|
|
state = extension_loader.reconcile_extension(
|
|
"ext_manifest_error",
|
|
drive_root,
|
|
lambda: {},
|
|
repo_path=str(skills_root),
|
|
retry_load_error=True,
|
|
)
|
|
assert state["action"] == "extension_load_error"
|
|
|
|
resp = client.get("/api/extensions/ext_manifest_error/manifest")
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert "absolute" in str(data["load_error"])
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extensions_index_marks_widget_only_extensions_as_ui_pending(
|
|
tmp_path, monkeypatch
|
|
):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
find_skill,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
skill_dir = _write_ext(
|
|
skills_root,
|
|
"ext_widget",
|
|
permissions=["widget"],
|
|
plugin=(
|
|
"def register(api):\n"
|
|
" api.register_ui_tab('weather', 'Weather', render={'kind': 'declarative', 'schema_version': 1, 'components': [{'type': 'markdown', 'text': 'ok'}]})\n"
|
|
),
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_widget", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_widget",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
loaded = find_skill(drive_root, "ext_widget", repo_path=str(skills_root))
|
|
assert loaded is not None
|
|
err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root)
|
|
assert err is None, err
|
|
|
|
resp = client.get("/api/extensions")
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
entry = next(s for s in data["skills"] if s["name"] == "ext_widget")
|
|
assert entry["live_loaded"] is True
|
|
assert entry["dispatch_live"] is False
|
|
assert entry["ui_tabs_pending"] == []
|
|
assert data["live"]["ui_tabs"][0]["key"] == "ext_widget:weather"
|
|
assert data["live"]["ui_tabs"][0]["render"]["kind"] == "declarative"
|
|
assert data["live"]["ui_tabs_pending"] == []
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_toggle_enables_and_loads_extension(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import SkillReviewState, save_review_state
|
|
from ouroboros.skill_loader import compute_content_hash
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"def register(api):\n"
|
|
" api.register_tool('t', lambda ctx: 'ok', description='', schema={})\n"
|
|
)
|
|
skill_dir = _write_ext(skills_root, "ext_toggle", permissions=["tool"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
broadcasts = []
|
|
client.app.app.state.broadcast_ws_sync = lambda payload: broadcasts.append(payload) # type: ignore[attr-defined]
|
|
try:
|
|
# Pre-mark review PASS so enable actually loads.
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_toggle",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
resp = client.post(
|
|
"/api/skills/ext_toggle/toggle",
|
|
json={"enabled": True},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["enabled"] is True
|
|
assert data["extension_action"] == "extension_loaded"
|
|
assert broadcasts[-1]["type"] == "extension_lifecycle"
|
|
assert broadcasts[-1]["skill"] == "ext_toggle"
|
|
assert broadcasts[-1]["action"] == "extension_loaded"
|
|
assert "ext_toggle" in extension_loader.snapshot()["extensions"]
|
|
|
|
# Disable → unload.
|
|
resp = client.post(
|
|
"/api/skills/ext_toggle/toggle",
|
|
json={"enabled": False},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["enabled"] is False
|
|
assert data["extension_action"] == "extension_unloaded"
|
|
assert broadcasts[-1]["action"] == "extension_unloaded"
|
|
assert "ext_toggle" not in extension_loader.snapshot()["extensions"]
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_projects_conflict_and_refuses_enable_until_peer_is_disabled(
|
|
tmp_path, monkeypatch
|
|
):
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = "def register(api):\n pass\n"
|
|
telegram_dir = _write_ext(
|
|
skills_root,
|
|
"telegram",
|
|
permissions=[],
|
|
plugin=plugin,
|
|
conflicts=["telegram-bridge"],
|
|
)
|
|
_write_ext(
|
|
skills_root,
|
|
"telegram-bridge",
|
|
permissions=[],
|
|
plugin=plugin,
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
save_review_state(
|
|
drive_root,
|
|
"telegram",
|
|
SkillReviewState(
|
|
status="pass",
|
|
content_hash=compute_content_hash(telegram_dir, manifest_entry="plugin.py"),
|
|
),
|
|
)
|
|
save_enabled(drive_root, "telegram-bridge", True)
|
|
|
|
index = client.get("/api/extensions")
|
|
assert index.status_code == 200, index.text
|
|
row = next(item for item in index.json()["skills"] if item["name"] == "telegram")
|
|
assert row["conflicts"] == ["telegram-bridge"]
|
|
assert row["conflict"] == {
|
|
"code": "skill_conflict",
|
|
"skills": ["telegram-bridge"],
|
|
"omitted": 0,
|
|
}
|
|
|
|
blocked = client.post("/api/skills/telegram/toggle", json={"enabled": True})
|
|
assert blocked.status_code == 409, blocked.text
|
|
assert blocked.json()["conflict"] == row["conflict"]
|
|
|
|
disabled = client.post(
|
|
"/api/skills/telegram-bridge/toggle",
|
|
json={"enabled": False},
|
|
)
|
|
assert disabled.status_code == 200, disabled.text
|
|
enabled = client.post("/api/skills/telegram/toggle", json={"enabled": True})
|
|
assert enabled.status_code == 200, enabled.text
|
|
assert enabled.json()["enabled"] is True
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_delete_removes_external_payload_state_and_unloads(client_env):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import SkillReviewState, compute_content_hash, save_review_state
|
|
|
|
client, drive_root = client_env
|
|
skill_dir = _write_ext(
|
|
drive_root / "skills" / "external",
|
|
"local_delete",
|
|
permissions=["tool"],
|
|
plugin="def register(api):\n api.register_tool('t', lambda ctx: 'ok', description='', schema={})\n",
|
|
)
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_review_state(drive_root, "local_delete", SkillReviewState(status="pass", content_hash=content_hash))
|
|
|
|
enabled = client.post("/api/skills/local_delete/toggle", json={"enabled": True})
|
|
assert enabled.status_code == 200, enabled.text
|
|
assert "local_delete" in extension_loader.snapshot()["extensions"]
|
|
assert (drive_root / "state" / "skills" / "local_delete").is_dir()
|
|
|
|
resp = client.post("/api/skills/local_delete/delete", json={"payload_root": "skills/external/local_delete"})
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["ok"] is True
|
|
assert data["deleted_payload_root"] == "skills/external/local_delete"
|
|
assert not skill_dir.exists()
|
|
assert not (drive_root / "state" / "skills" / "local_delete").exists()
|
|
assert "local_delete" not in extension_loader.snapshot()["extensions"]
|
|
|
|
hub_skill_dir = _write_ext(
|
|
drive_root / "skills" / "clawhub",
|
|
"hub_delete",
|
|
permissions=[],
|
|
plugin="def register(api):\n pass\n",
|
|
)
|
|
(hub_skill_dir / ".clawhub.json").write_text("{}", encoding="utf-8")
|
|
|
|
resp = client.post("/api/skills/hub_delete/delete", json={"payload_root": "skills/clawhub/hub_delete"})
|
|
|
|
assert resp.status_code == 403
|
|
assert hub_skill_dir.exists()
|
|
|
|
|
|
def test_api_skill_delete_rejects_external_symlink_bucket(client_env, tmp_path):
|
|
client, drive_root = client_env
|
|
external_target = tmp_path / "outside-external"
|
|
_write_ext(
|
|
external_target,
|
|
"symlink_delete",
|
|
permissions=[],
|
|
plugin="def register(api):\n pass\n",
|
|
)
|
|
skills_root = drive_root / "skills"
|
|
skills_root.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
(skills_root / "external").symlink_to(external_target, target_is_directory=True)
|
|
except (OSError, NotImplementedError) as exc:
|
|
pytest.skip(f"directory symlinks unavailable in this environment: {exc}")
|
|
|
|
resp = client.post(
|
|
"/api/skills/symlink_delete/delete",
|
|
json={"payload_root": "skills/external/symlink_delete"},
|
|
)
|
|
|
|
assert resp.status_code == 403
|
|
assert (external_target / "symlink_delete").exists()
|
|
|
|
|
|
def test_api_skill_delete_rejects_name_collision_before_state_delete(client_env):
|
|
client, drive_root = client_env
|
|
external_dir = _write_ext(
|
|
drive_root / "skills" / "external",
|
|
"collide_delete",
|
|
permissions=[],
|
|
plugin="def register(api):\n pass\n",
|
|
)
|
|
native_dir = _write_ext(
|
|
drive_root / "skills" / "native",
|
|
"collide_delete",
|
|
permissions=[],
|
|
plugin="def register(api):\n pass\n",
|
|
)
|
|
state_dir = drive_root / "state" / "skills" / "collide_delete"
|
|
state_dir.mkdir(parents=True)
|
|
(state_dir / "enabled.json").write_text('{"enabled": true}', encoding="utf-8")
|
|
|
|
resp = client.post(
|
|
"/api/skills/collide_delete/delete",
|
|
json={"payload_root": "skills/external/collide_delete"},
|
|
)
|
|
|
|
assert resp.status_code == 409
|
|
assert external_dir.exists()
|
|
assert native_dir.exists()
|
|
assert state_dir.exists()
|
|
|
|
|
|
def test_api_skill_delete_accepts_unsanitized_external_directory_leaf(client_env):
|
|
client, drive_root = client_env
|
|
skill_dir = _write_ext(
|
|
drive_root / "skills" / "external",
|
|
"hello world",
|
|
permissions=[],
|
|
plugin="def register(api):\n pass\n",
|
|
)
|
|
state_dir = drive_root / "state" / "skills" / "hello_world"
|
|
state_dir.mkdir(parents=True)
|
|
|
|
resp = client.post(
|
|
"/api/skills/hello_world/delete",
|
|
json={"payload_root": "skills/external/hello world"},
|
|
)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["deleted_payload_root"] == "skills/external/hello world"
|
|
assert not skill_dir.exists()
|
|
assert not state_dir.exists()
|
|
|
|
|
|
def test_api_skill_toggle_allows_warnings_review(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import SkillReviewState, save_review_state
|
|
from ouroboros.skill_loader import compute_content_hash
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"def register(api):\n"
|
|
" api.register_tool('t', lambda ctx: 'ok', description='', schema={})\n"
|
|
)
|
|
skill_dir = _write_ext(skills_root, "ext_advisory", permissions=["tool"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_advisory",
|
|
SkillReviewState(status="warnings", content_hash=content_hash),
|
|
)
|
|
resp = client.post("/api/skills/ext_advisory/toggle", json={"enabled": True})
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["enabled"] is True
|
|
assert data["review_status"] == "warnings"
|
|
assert data["extension_action"] == "extension_loaded"
|
|
assert "ext_advisory" in extension_loader.snapshot()["extensions"]
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_toggle_allows_warnings_under_blocking(tmp_path, monkeypatch):
|
|
from ouroboros.skill_loader import SkillReviewState, save_review_state, compute_content_hash
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"def register(api):\n"
|
|
" api.register_tool('t', lambda ctx: 'ok', description='', schema={})\n"
|
|
)
|
|
skill_dir = _write_ext(skills_root, "ext_blocked", permissions=["tool"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
monkeypatch.setenv("OUROBOROS_REVIEW_ENFORCEMENT", "blocking")
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_blocked",
|
|
SkillReviewState(status="warnings", content_hash=content_hash),
|
|
)
|
|
resp = client.post("/api/skills/ext_blocked/toggle", json={"enabled": True})
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["executable_review"] is True
|
|
assert data["review_gate"]["blocking_reason"] == "warnings_do_not_block_execution"
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_toggle_blocks_missing_isolated_deps_env(tmp_path, monkeypatch):
|
|
from ouroboros.marketplace.install_specs import install_specs_hash
|
|
from ouroboros.marketplace.isolated_deps import DEPS_STATE_FILENAME
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
save_review_state,
|
|
skill_state_dir,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = "def register(api):\n api.register_tool('t', lambda ctx: 'ok', description='', schema={})\n"
|
|
skill_dir = _write_ext(skills_root, "ext_deps", permissions=["tool"], plugin=plugin)
|
|
manifest = (skill_dir / "SKILL.md").read_text(encoding="utf-8")
|
|
(skill_dir / "SKILL.md").write_text(
|
|
manifest.replace(
|
|
"permissions: [\"tool\"]\n",
|
|
"permissions: [\"tool\"]\n"
|
|
"install_specs:\n"
|
|
" - kind: pip\n"
|
|
" package: wheel\n",
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_deps",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
state_dir = skill_state_dir(drive_root, "ext_deps")
|
|
state_dir.mkdir(parents=True, exist_ok=True)
|
|
specs = [{"kind": "pip", "package": "wheel"}]
|
|
(state_dir / DEPS_STATE_FILENAME).write_text(
|
|
json.dumps({"status": "installed", "specs_hash": install_specs_hash(specs)}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
resp = client.post("/api/skills/ext_deps/toggle", json={"enabled": True})
|
|
|
|
assert resp.status_code == 409, resp.text
|
|
data = resp.json()
|
|
assert data["deps_status"] == "missing"
|
|
assert not (state_dir / "enabled.json").exists()
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_toggle_collision_disable_does_not_write_shared_state(
|
|
tmp_path, monkeypatch
|
|
):
|
|
skills_root = tmp_path / "skills"
|
|
plugin = "def register(api):\n return None\n"
|
|
_write_ext(skills_root, "hello world", permissions=[], plugin=plugin)
|
|
_write_ext(skills_root, "hello_world", permissions=[], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
resp = client.post("/api/skills/hello_world/toggle", json={"enabled": False})
|
|
assert resp.status_code == 400, resp.text
|
|
data = resp.json()
|
|
assert data["extension_reason"] == "name_collision"
|
|
state_file = drive_root / "state" / "skills" / "hello_world" / "enabled.json"
|
|
assert not state_file.exists()
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_dispatcher_routes_to_registered_handler(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"from starlette.responses import JSONResponse\n"
|
|
"def _hello(request):\n"
|
|
" return JSONResponse({'hello': 'world'})\n"
|
|
"def register(api):\n"
|
|
" api.register_route('greet', _hello, methods=('GET',))\n"
|
|
)
|
|
skill_dir = _write_ext(skills_root, "ext_route", permissions=["route"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_route", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_route",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
from ouroboros.skill_loader import find_skill
|
|
from ouroboros.config import load_settings
|
|
refreshed = find_skill(drive_root, "ext_route", repo_path=str(skills_root))
|
|
err = extension_loader.load_extension(refreshed, load_settings, drive_root=drive_root)
|
|
assert err is None, err
|
|
|
|
resp = client.get("/api/extensions/ext_route/greet")
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json() == {"hello": "world"}
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_module_serves_only_live_declared_entry(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
find_skill,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"def register(api):\n"
|
|
" api.register_ui_tab('module', 'Module', render={'kind': 'module', 'entry': 'widget.js'})\n"
|
|
)
|
|
skill_dir = _write_ext(skills_root, "ext_module", permissions=["widget"], plugin=plugin)
|
|
(skill_dir / "widget.js").write_text("window.__ok = true;\n", encoding="utf-8")
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_module", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_module",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
loaded = find_skill(drive_root, "ext_module", repo_path=str(skills_root))
|
|
assert loaded is not None
|
|
err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root)
|
|
assert err is None, err
|
|
|
|
ok = client.get("/api/extensions/ext_module/module/widget.js")
|
|
assert ok.status_code == 200, ok.text
|
|
assert "window.__ok" in ok.text
|
|
assert ok.headers["cache-control"] == "no-store"
|
|
|
|
assert client.get("/api/extensions/ext_module/module/other.js").status_code == 404
|
|
assert client.get("/api/extensions/ext_module/module/../widget.js").status_code in {400, 404}
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_module_rejects_non_live_extension(tmp_path, monkeypatch):
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"def register(api):\n"
|
|
" api.register_ui_tab('module', 'Module', render={'kind': 'module', 'entry': 'widget.js'})\n"
|
|
)
|
|
_write_ext(skills_root, "ext_module", permissions=["widget"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, _, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
resp = client.get("/api/extensions/ext_module/module/widget.js")
|
|
assert resp.status_code == 409
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_settings_section_returns_only_requested_skill(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
find_skill,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin_a = (
|
|
"def register(api):\n"
|
|
" api.register_settings_section('config', 'Config A', schema={'components': [\n"
|
|
" {'type': 'markdown', 'text': 'A'}\n"
|
|
" ]})\n"
|
|
)
|
|
plugin_b = (
|
|
"def register(api):\n"
|
|
" api.register_settings_section('config', 'Config B', schema={'components': [\n"
|
|
" {'type': 'markdown', 'text': 'B'}\n"
|
|
" ]})\n"
|
|
)
|
|
skill_a = _write_ext(skills_root, "settings_a", permissions=["widget"], plugin=plugin_a)
|
|
skill_b = _write_ext(skills_root, "settings_b", permissions=["widget"], plugin=plugin_b)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
for name, skill_dir in {"settings_a": skill_a, "settings_b": skill_b}.items():
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, name, True)
|
|
save_review_state(drive_root, name, SkillReviewState(status="pass", content_hash=content_hash))
|
|
loaded = find_skill(drive_root, name, repo_path=str(skills_root))
|
|
assert loaded is not None
|
|
err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root)
|
|
assert err is None, err
|
|
|
|
resp = client.get("/api/extensions/settings_a/settings_section")
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["skill"] == "settings_a"
|
|
assert [section["skill"] for section in data["sections"]] == ["settings_a"]
|
|
assert data["sections"][0]["title"] == "Config A"
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_dispatcher_allows_head_for_get_route(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
find_skill,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
skill_dir = _write_ext(
|
|
skills_root,
|
|
"ext_head",
|
|
permissions=["route"],
|
|
plugin=(
|
|
"from starlette.responses import JSONResponse\n"
|
|
"def _hello(request):\n"
|
|
" return JSONResponse({'hello': 'world'})\n"
|
|
"def register(api):\n"
|
|
" api.register_route('greet', _hello, methods=('GET',))\n"
|
|
),
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_head", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_head",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
loaded = find_skill(drive_root, "ext_head", repo_path=str(skills_root))
|
|
assert loaded is not None
|
|
err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root)
|
|
assert err is None, err
|
|
|
|
resp = client.head("/api/extensions/ext_head/greet")
|
|
assert resp.status_code == 200, resp.text
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_dispatcher_404_for_unknown_route(tmp_path, monkeypatch):
|
|
client, _, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
resp = client.get("/api/extensions/nope/xyz")
|
|
assert resp.status_code == 404
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_dispatcher_surfaces_lazy_load_error(tmp_path, monkeypatch):
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"def _hello(request):\n"
|
|
" return {'hello': 'world'}\n"
|
|
"def register(api):\n"
|
|
" api.register_route('/absolute', _hello, methods=('GET',))\n"
|
|
)
|
|
skill_dir = _write_ext(skills_root, "ext_broken", permissions=["route"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_broken", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_broken",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
|
|
resp = client.get("/api/extensions/ext_broken/greet")
|
|
assert resp.status_code == 409, resp.text
|
|
data = resp.json()
|
|
assert data["state"]["action"] == "extension_load_error"
|
|
assert data["state"]["reason"] == "load_error"
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_dispatcher_rejects_not_live_route(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"from starlette.responses import JSONResponse\n"
|
|
"def _hello(request):\n"
|
|
" return JSONResponse({'hello': 'world'})\n"
|
|
"def register(api):\n"
|
|
" api.register_route('greet', _hello, methods=('GET',))\n"
|
|
)
|
|
skill_dir = _write_ext(skills_root, "ext_guarded", permissions=["route"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_guarded", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_guarded",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
from ouroboros.skill_loader import find_skill
|
|
|
|
loaded = find_skill(drive_root, "ext_guarded", repo_path=str(skills_root))
|
|
assert loaded is not None
|
|
err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root)
|
|
assert err is None, err
|
|
assert "ext_guarded" in extension_loader.snapshot()["extensions"]
|
|
|
|
# Leave stale registrations in memory but mark the skill disabled on disk.
|
|
save_enabled(drive_root, "ext_guarded", False)
|
|
|
|
resp = client.get("/api/extensions/ext_guarded/greet")
|
|
assert resp.status_code == 409, resp.text
|
|
data = resp.json()
|
|
assert data["state"]["reason"] == "disabled"
|
|
assert "ext_guarded" not in extension_loader.snapshot()["extensions"]
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_extension_dispatcher_reloads_stale_live_route(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
find_skill,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
skill_dir = _write_ext(
|
|
skills_root,
|
|
"ext_route_reload",
|
|
permissions=["route"],
|
|
plugin=(
|
|
"from starlette.responses import JSONResponse\n"
|
|
"def _hello(request):\n"
|
|
" return JSONResponse({'hello': 'v1'})\n"
|
|
"def register(api):\n"
|
|
" api.register_route('greet', _hello, methods=('GET',))\n"
|
|
),
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_route_reload", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_route_reload",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
loaded = find_skill(drive_root, "ext_route_reload", repo_path=str(skills_root))
|
|
assert loaded is not None
|
|
err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root)
|
|
assert err is None, err
|
|
|
|
(skill_dir / "plugin.py").write_text(
|
|
(
|
|
"from starlette.responses import JSONResponse\n"
|
|
"def _hello(request):\n"
|
|
" return JSONResponse({'hello': 'v2'})\n"
|
|
"def register(api):\n"
|
|
" api.register_route('greet', _hello, methods=('GET',))\n"
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
refreshed = find_skill(drive_root, "ext_route_reload", repo_path=str(skills_root))
|
|
assert refreshed is not None
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_route_reload",
|
|
SkillReviewState(status="pass", content_hash=refreshed.content_hash),
|
|
)
|
|
|
|
resp = client.get("/api/extensions/ext_route_reload/greet")
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json() == {"hello": "v2"}
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_toggle_rejects_non_boolean_enabled(tmp_path, monkeypatch):
|
|
skills_root = tmp_path / "skills"
|
|
plugin = "def register(api):\n pass\n"
|
|
_write_ext(skills_root, "ext_toggle_bad", permissions=[], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, _, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
resp = client.post("/api/skills/ext_toggle_bad/toggle", json={"enabled": "definitely"})
|
|
assert resp.status_code == 400
|
|
assert "boolean" in resp.text
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_grants_saves_keys_and_permissions(tmp_path, monkeypatch):
|
|
from ouroboros.skill_loader import SkillReviewState, compute_content_hash, load_skill_grants, save_review_state
|
|
|
|
skills_root = tmp_path / "skills"
|
|
skill_dir = _write_ext(
|
|
skills_root,
|
|
"grant_api",
|
|
permissions=["tool", "read_settings", "inject_chat"],
|
|
plugin="def register(api):\n pass\n",
|
|
env_from_settings=["OPENROUTER_API_KEY"],
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_review_state(
|
|
drive_root,
|
|
"grant_api",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
resp = client.post(
|
|
"/api/skills/grant_api/grants",
|
|
json={"items": ["OPENROUTER_API_KEY", "inject_chat"]},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["ok"] is True
|
|
assert data["granted_keys"] == ["OPENROUTER_API_KEY"]
|
|
assert data["granted_permissions"] == ["inject_chat"]
|
|
grants = load_skill_grants(drive_root, "grant_api")
|
|
assert grants["granted_keys"] == ["OPENROUTER_API_KEY"]
|
|
assert grants["granted_permissions"] == ["inject_chat"]
|
|
assert data["extension_reason"] in {"disabled", "not_extension", "name_collision", None}
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_grants_soft_fails_extension_reconcile_after_persist(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import SkillReviewState, compute_content_hash, load_skill_grants, save_review_state
|
|
|
|
skills_root = tmp_path / "skills"
|
|
skill_dir = _write_ext(
|
|
skills_root,
|
|
"grant_reconcile_soft_fail",
|
|
permissions=["inject_chat"],
|
|
plugin="def register(api):\n pass\n",
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_review_state(
|
|
drive_root,
|
|
"grant_reconcile_soft_fail",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
|
|
def fail_reconcile(*_args, **_kwargs):
|
|
raise RuntimeError("reconcile exploded")
|
|
|
|
monkeypatch.setattr(extension_loader, "reconcile_extension", fail_reconcile)
|
|
resp = client.post(
|
|
"/api/skills/grant_reconcile_soft_fail/grants",
|
|
json={"items": ["inject_chat"]},
|
|
)
|
|
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["ok"] is True
|
|
assert data["extension_reason"] == "reconcile_call_failed"
|
|
assert "reconcile exploded" in data["load_error"]
|
|
grants = load_skill_grants(drive_root, "grant_reconcile_soft_fail")
|
|
assert grants["granted_permissions"] == ["inject_chat"]
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_grants_rejects_blocking_blocker_review(tmp_path, monkeypatch):
|
|
from ouroboros.skill_loader import SkillReviewState, compute_content_hash, save_review_state
|
|
|
|
skills_root = tmp_path / "skills"
|
|
skill_dir = _write_ext(
|
|
skills_root,
|
|
"grant_blocked",
|
|
permissions=["tool", "read_settings"],
|
|
plugin="def register(api):\n pass\n",
|
|
env_from_settings=["OPENROUTER_API_KEY"],
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
monkeypatch.setenv("OUROBOROS_REVIEW_ENFORCEMENT", "blocking")
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_review_state(
|
|
drive_root,
|
|
"grant_blocked",
|
|
SkillReviewState(status="blockers", content_hash=content_hash),
|
|
)
|
|
resp = client.post("/api/skills/grant_blocked/grants", json={"items": ["OPENROUTER_API_KEY"]})
|
|
assert resp.status_code == 409
|
|
assert "fresh executable review" in resp.json()["error"]
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_reconcile_clears_cached_load_error(tmp_path, monkeypatch):
|
|
"""v5.2.2 dual-track grants: ``POST /api/skills/<name>/reconcile``
|
|
is the loopback endpoint the desktop launcher pings after a
|
|
successful core-key grant. It must clear the server's cached
|
|
``_load_failures`` entry and re-run ``load_extension`` so the
|
|
plugin picks up the freshly-granted key without forcing the user
|
|
to disable/enable.
|
|
"""
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
find_skill,
|
|
save_enabled,
|
|
save_review_state,
|
|
save_skill_grants,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"def register(api):\n"
|
|
" api.register_tool('n', lambda ctx: 'ok', description='n', schema={})\n"
|
|
)
|
|
_write_ext(
|
|
skills_root,
|
|
"reconcile_demo",
|
|
permissions=["tool", "read_settings"],
|
|
plugin=plugin,
|
|
env_from_settings=["OPENROUTER_API_KEY"],
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
broadcasts = []
|
|
client.app.app.state.broadcast_ws_sync = lambda payload: broadcasts.append(payload) # type: ignore[attr-defined]
|
|
try:
|
|
first = find_skill(drive_root, "reconcile_demo", repo_path=str(skills_root))
|
|
assert first is not None
|
|
save_enabled(drive_root, "reconcile_demo", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"reconcile_demo",
|
|
SkillReviewState(status="pass", content_hash=first.content_hash),
|
|
)
|
|
loaded = find_skill(drive_root, "reconcile_demo", repo_path=str(skills_root))
|
|
assert loaded is not None and loaded.enabled
|
|
|
|
# First load attempt — no grant on disk → fails with the new
|
|
# informative error and seeds ``_load_failures``.
|
|
err = extension_loader.load_extension(
|
|
loaded, lambda: {"OPENROUTER_API_KEY": "sk-secret"}, drive_root=drive_root,
|
|
)
|
|
assert err is not None
|
|
assert "missing owner grants" in err
|
|
with extension_loader._lock:
|
|
extension_loader._load_failures["reconcile_demo"] = (
|
|
extension_loader._ExtensionLoadFailure(
|
|
content_hash=loaded.content_hash,
|
|
skill_dir=str(loaded.skill_dir.resolve()),
|
|
error=err,
|
|
)
|
|
)
|
|
|
|
# Owner grants → simulate the launcher writing grants.json.
|
|
save_skill_grants(
|
|
drive_root,
|
|
"reconcile_demo",
|
|
["OPENROUTER_API_KEY"],
|
|
content_hash=loaded.content_hash,
|
|
requested_keys=["OPENROUTER_API_KEY"],
|
|
)
|
|
|
|
# The endpoint must clear the cached failure and load the plugin.
|
|
resp = client.post("/api/skills/reconcile_demo/reconcile")
|
|
assert resp.status_code == 200, resp.text
|
|
payload = resp.json()
|
|
assert payload["skill"] == "reconcile_demo"
|
|
assert payload["live_loaded"] is True
|
|
assert payload["extension_action"] == "extension_loaded"
|
|
assert broadcasts[-1]["type"] == "extension_lifecycle"
|
|
assert broadcasts[-1]["skill"] == "reconcile_demo"
|
|
assert broadcasts[-1]["action"] == "extension_loaded"
|
|
with extension_loader._lock:
|
|
assert "reconcile_demo" in extension_loader._extensions
|
|
assert "reconcile_demo" not in extension_loader._load_failures
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_reconcile_rejects_missing_skill_name(tmp_path, monkeypatch):
|
|
client, _drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
# Starlette path params with empty trailing segment → 404 path,
|
|
# but explicit empty skill via direct call returns 400 from the
|
|
# endpoint's own validation.
|
|
resp = client.post("/api/skills/ /reconcile")
|
|
# Whitespace-only path param hits the endpoint with stripped
|
|
# empty name → 400.
|
|
assert resp.status_code == 400
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_api_skill_review_offloads_to_thread_and_returns_outcome(tmp_path, monkeypatch):
|
|
"""Phase 5 regression: ``POST /api/skills/<skill>/review`` must
|
|
trigger the tri-model review and return the outcome. The async
|
|
Starlette endpoint offloads to ``asyncio.to_thread`` so the event
|
|
loop stays responsive."""
|
|
from unittest.mock import patch
|
|
|
|
from ouroboros.skill_review import SkillReviewOutcome
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = "def register(api): pass\n"
|
|
_write_ext(skills_root, "ext_r", permissions=[], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
canned = SkillReviewOutcome(
|
|
skill_name="ext_r",
|
|
status="pass",
|
|
findings=[{"item": "manifest_schema", "verdict": "PASS"}],
|
|
reviewer_models=["openai/gpt-5.5"],
|
|
content_hash="abcd",
|
|
error="",
|
|
)
|
|
with patch(
|
|
"ouroboros.gateway.extensions._review_skill_impl",
|
|
create=True,
|
|
return_value=canned,
|
|
), patch(
|
|
"ouroboros.skill_review.review_skill", return_value=canned,
|
|
):
|
|
resp = client.post("/api/skills/ext_r/review", json={})
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["status"] == "clean"
|
|
assert data["skill"] == "ext_r"
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_lifecycle_queue_endpoint_marks_stale_review_job_interrupted(tmp_path, monkeypatch):
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
job_dir = drive_root / "state" / "skills" / "alpha"
|
|
job_dir.mkdir(parents=True)
|
|
job_path = job_dir / "review_job.json"
|
|
job_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "running",
|
|
"skill": "alpha",
|
|
"content_hash": "abc",
|
|
"job_id": "skill-job-old",
|
|
"started_at": "2026-01-01T00:00:00+00:00",
|
|
"last_heartbeat_at": "2026-01-01T00:00:00+00:00",
|
|
"pid": 123456,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr("ouroboros.skill_review_runner._pid_alive", lambda _pid: False)
|
|
try:
|
|
resp = client.get("/api/skills/lifecycle-queue")
|
|
assert resp.status_code == 200
|
|
data = json.loads(job_path.read_text(encoding="utf-8"))
|
|
assert data["status"] == "interrupted"
|
|
assert data["interrupt_reason"] == "owner_process_exited"
|
|
progress = [
|
|
json.loads(line)
|
|
for line in (drive_root / "logs" / "progress.jsonl").read_text(encoding="utf-8").splitlines()
|
|
if line.strip()
|
|
]
|
|
assert progress[-1]["lifecycle"]["status"] == "interrupted"
|
|
assert progress[-1]["task_id"] == "skill_lifecycle_review_alpha_skill-job-old"
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_ws_endpoint_dispatches_ext_prefixed_messages():
|
|
"""Phase 5 regression: gateway.ws::ws_endpoint must route
|
|
provider-safe extension WS messages through ``extension_loader.list_ws_handlers()``.
|
|
AST-level check — the full runtime round-trip requires a live
|
|
supervisor which is out of scope for this file."""
|
|
import ast
|
|
src = (
|
|
pathlib.Path(__file__).resolve().parent.parent
|
|
/ "ouroboros"
|
|
/ "gateway"
|
|
/ "ws.py"
|
|
).read_text(encoding="utf-8")
|
|
assert "parse_extension_surface_name" in src, "gateway WS module has no extension dispatch branch"
|
|
assert "list_ws_handlers" in src, (
|
|
"gateway WS module does not look up extension WS handlers via "
|
|
"``extension_loader.list_ws_handlers``."
|
|
)
|
|
tree = ast.parse(src)
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.AsyncFunctionDef) and node.name == "ws_endpoint":
|
|
return
|
|
assert False, "ws_endpoint not found in gateway/ws.py"
|
|
|
|
|
|
def test_ws_endpoint_reconciles_and_unloads_not_live_extension(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
find_skill,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"async def _handler(payload):\n"
|
|
" return {'acked': True}\n"
|
|
"def register(api):\n"
|
|
" api.register_ws_handler('message', _handler)\n"
|
|
)
|
|
skill_dir = _write_ext(skills_root, "ext_ws_guarded", permissions=["ws_handler"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_ws_guarded", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_ws_guarded",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
loaded = find_skill(drive_root, "ext_ws_guarded", repo_path=str(skills_root))
|
|
assert loaded is not None
|
|
err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root)
|
|
assert err is None, err
|
|
assert "ext_ws_guarded" in extension_loader.snapshot()["extensions"]
|
|
|
|
save_enabled(drive_root, "ext_ws_guarded", False)
|
|
|
|
with client.websocket_connect("/ws") as ws:
|
|
ws.send_text(json.dumps({"type": extension_loader.extension_surface_name("ext_ws_guarded", "message")}))
|
|
reply = json.loads(ws.receive_text())
|
|
assert reply["type"] == "log"
|
|
assert "not live" in reply["data"]["message"]
|
|
assert "ext_ws_guarded" not in extension_loader.snapshot()["extensions"]
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_ws_endpoint_dispatches_first_message_after_lazy_load(tmp_path, monkeypatch):
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
plugin = (
|
|
"async def _handler(payload):\n"
|
|
" return {'acked': payload.get('payload')}\n"
|
|
"def register(api):\n"
|
|
" api.register_ws_handler('message', _handler)\n"
|
|
)
|
|
skill_dir = _write_ext(skills_root, "ext_ws_lazy", permissions=["ws_handler"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_ws_lazy", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_ws_lazy",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
extension_loader.unload_extension("ext_ws_lazy")
|
|
msg_type = extension_loader.extension_surface_name("ext_ws_lazy", "message")
|
|
with client.websocket_connect("/ws") as ws:
|
|
ws.send_text(json.dumps({"type": msg_type, "payload": "first"}))
|
|
reply = json.loads(ws.receive_text())
|
|
assert reply == {"type": f"{msg_type}.reply", "data": {"acked": "first"}}
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_ws_endpoint_surfaces_extension_load_error(tmp_path, monkeypatch):
|
|
skills_root = tmp_path / "skills"
|
|
skill_dir = _write_ext(
|
|
skills_root,
|
|
"ext_ws_broken",
|
|
permissions=["ws_handler"],
|
|
plugin=(
|
|
"async def _handler(payload):\n"
|
|
" return {'acked': True}\n"
|
|
"def register(api):\n"
|
|
" api.register_ws_handler('bad-type', _handler)\n"
|
|
),
|
|
)
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
client, drive_root, patches = _make_client(tmp_path, monkeypatch)
|
|
try:
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import SkillReviewState, compute_content_hash, save_enabled, save_review_state
|
|
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "ext_ws_broken", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"ext_ws_broken",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
|
|
with client.websocket_connect("/ws") as ws:
|
|
ws.send_text(json.dumps({"type": extension_loader.extension_surface_name("ext_ws_broken", "message")}))
|
|
reply = json.loads(ws.receive_text())
|
|
assert reply["type"] == "log"
|
|
assert "failed to go live" in reply["data"]["message"]
|
|
finally:
|
|
_stop_patches(patches)
|
|
|
|
|
|
def test_tool_registry_execute_dispatches_ext_tool(tmp_path, monkeypatch):
|
|
"""Phase 5 regression: ``ToolRegistry.execute`` falls back to
|
|
``extension_loader.get_tool`` for extension names, but only for
|
|
reviewed/live extensions that are surfaced through the normal
|
|
registry schema lookup."""
|
|
from ouroboros.tools import registry as tools_registry
|
|
from ouroboros import extension_loader
|
|
from ouroboros.skill_loader import (
|
|
SkillReviewState,
|
|
compute_content_hash,
|
|
find_skill,
|
|
save_enabled,
|
|
save_review_state,
|
|
)
|
|
|
|
skills_root = tmp_path / "skills"
|
|
drive_root = tmp_path / "drive"
|
|
drive_root.mkdir()
|
|
plugin = (
|
|
"def _echo(ctx, who='world'):\n"
|
|
" return f'hello {who}'\n"
|
|
"def register(api):\n"
|
|
" api.register_tool('echo', _echo, description='echo', schema={}, timeout_sec=10)\n"
|
|
)
|
|
skill_dir = _write_ext(skills_root, "testskill", permissions=["tool"], plugin=plugin)
|
|
monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "advanced")
|
|
monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
|
|
content_hash = compute_content_hash(skill_dir, manifest_entry="plugin.py")
|
|
save_enabled(drive_root, "testskill", True)
|
|
save_review_state(
|
|
drive_root,
|
|
"testskill",
|
|
SkillReviewState(status="pass", content_hash=content_hash),
|
|
)
|
|
loaded = find_skill(drive_root, "testskill", repo_path=str(skills_root))
|
|
assert loaded is not None
|
|
err = extension_loader.load_extension(loaded, lambda: {}, drive_root=drive_root)
|
|
assert err is None, err
|
|
try:
|
|
tmp_reg = tools_registry.ToolRegistry(repo_dir=tmp_path, drive_root=drive_root)
|
|
tool_name = extension_loader.extension_surface_name("testskill", "echo")
|
|
schema = tmp_reg.get_schema_by_name(tool_name)
|
|
assert schema is not None
|
|
assert schema["function"]["name"] == tool_name
|
|
result = tmp_reg.execute(tool_name, {"who": "phase5"})
|
|
# v5.1.2 iter-2: extension dispatch now goes through
|
|
# ``ouroboros.safety.check_safety``. In test envs without a
|
|
# safety backend, the supervisor returns a visible
|
|
# ``SAFETY_WARNING`` prefix while still letting the call run
|
|
# (fail-open). Assert the handler ran and produced its output;
|
|
# the warning prefix is acceptable.
|
|
assert "hello phase5" in result, result
|
|
# get_timeout honours the extension's declared timeout plus the v5.7.0
|
|
# cleanup buffer used by async handlers (so the outer tool executor
|
|
# does not time out before inner wait_for cancellation can finish).
|
|
assert tmp_reg.get_timeout(tool_name) == 13
|
|
finally:
|
|
extension_loader.unload_extension("testskill")
|