ouroboros/tests/test_skill_widget_surface.py

"""Skill-widget surface regressions: preflight entry truth, reconcile receipts, catalogue liveness.

Single home for this stream's Python pins so no existing oversized test module
grows. No network, no real LLM calls; every case runs against an isolated
tmp_path drive root.
"""
from __future__ import annotations

import json
import pathlib

import pytest

from ouroboros.tools.registry import ToolContext
from tests._governance_docs_shared import architecture_text


def _make_ctx(tmp_path: pathlib.Path) -> ToolContext:
    repo_dir = tmp_path / "repo"
    repo_dir.mkdir()
    drive_root = tmp_path / "drive"
    drive_root.mkdir()
    return ToolContext(repo_dir=repo_dir, drive_root=drive_root)


def _extension_manifest(name: str = "alpha", *, ui_tab: str = "") -> str:
    return (
        "---\n"
        f"name: {name}\n"
        "description: widget surface test\n"
        "version: 0.1.0\n"
        "type: extension\n"
        "entry: plugin.py\n"
        "permissions: [widget]\n"
        f"{ui_tab}"
        "---\n"
        "body\n"
    )


def _module_ui_tab(entry: str = "widget.js") -> str:
    return (
        "ui_tab:\n"
        "  id: main\n"
        "  title: Main\n"
        "  render:\n"
        "    kind: module\n"
        f"    entry: {entry}\n"
        "    height: 480\n"
    )


def _make_skill(tmp_path: pathlib.Path, monkeypatch, manifest: str, plugin: str) -> pathlib.Path:
    skills_root = tmp_path / "skills"
    skills_root.mkdir(exist_ok=True)
    monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
    skill_dir = skills_root / "alpha"
    skill_dir.mkdir()
    (skill_dir / "SKILL.md").write_text(manifest, encoding="utf-8")
    (skill_dir / "plugin.py").write_text(plugin, encoding="utf-8")
    return skill_dir


_TRIVIAL_PLUGIN = "def register(api):\n    pass\n"


# --------------------------------------------------------------------------- F12


def test_skill_preflight_flags_missing_module_widget_entry(tmp_path, monkeypatch):
    """A declared module entry that is not on disk must not read as verified ok."""
    ctx = _make_ctx(tmp_path)
    skill_dir = _make_skill(
        tmp_path,
        monkeypatch,
        _extension_manifest(ui_tab=_module_ui_tab()),
        _TRIVIAL_PLUGIN,
    )

    from ouroboros.tools import skill_preflight as sp

    result = json.loads(sp._handle_skill_preflight(ctx, skill="alpha"))
    entry_rows = [row for row in result["widgets"] if row["item"] == "widget_entry_exists"]
    assert entry_rows, result["widgets"]
    assert result["ok"] is False
    assert entry_rows[0]["ok"] is False
    assert "widget.js" in entry_rows[0]["detail"]
    assert entry_rows[0]["source"] == "manifest.ui_tab.render"

    # Same declaration, file now present: the row flips and names the entry.
    (skill_dir / "widget.js").write_text("const a = 1;\n", encoding="utf-8")
    result = json.loads(sp._handle_skill_preflight(ctx, skill="alpha"))
    entry_rows = [row for row in result["widgets"] if row["item"] == "widget_entry_exists"]
    assert entry_rows[0]["ok"] is True
    assert entry_rows[0]["detail"] == "widget.js"


def test_skill_preflight_checks_plugin_registered_module_entry(tmp_path, monkeypatch):
    """The plugin.py register_ui_tab path is covered without touching the AST walker."""
    ctx = _make_ctx(tmp_path)
    _make_skill(
        tmp_path,
        monkeypatch,
        _extension_manifest(),
        "_UI_RENDER = {'kind': 'module', 'entry': 'missing.js', 'height': 400}\n"
        "def register(api):\n"
        "    api.register_ui_tab('main', 'Main', render=_UI_RENDER)\n",
    )

    from ouroboros.tools import skill_preflight as sp

    result = json.loads(sp._handle_skill_preflight(ctx, skill="alpha"))
    entry_rows = [row for row in result["widgets"] if row["item"] == "widget_entry_exists"]
    assert entry_rows, result["widgets"]
    assert result["ok"] is False
    assert entry_rows[0]["ok"] is False
    assert "missing.js" in entry_rows[0]["detail"]
    assert entry_rows[0]["source"].startswith("plugin.py:")


def test_noncanonical_module_entry_fails_preflight_and_runtime_validation(
    tmp_path, monkeypatch
):
    from ouroboros.contracts.plugin_api import ExtensionRegistrationError
    from ouroboros.extension_ui_validation import validate_ui_render
    from ouroboros.tools import skill_preflight as sp

    ctx = _make_ctx(tmp_path)
    _make_skill(
        tmp_path,
        monkeypatch,
        _extension_manifest(ui_tab=_module_ui_tab("widget space.js")),
        _TRIVIAL_PLUGIN,
    )

    result = json.loads(sp._handle_skill_preflight(ctx, skill="alpha"))
    rows = [row for row in result["widgets"] if row["item"] == "widget_schema"]
    assert result["ok"] is False
    assert rows[0]["ok"] is False
    assert rows[0]["verified"] is True
    assert "browser-safe" in rows[0]["detail"]
    with pytest.raises(ExtensionRegistrationError, match="browser-safe"):
        validate_ui_render({"kind": "module", "entry": "widget space.js"})


# --------------------------------------------------------------------------- F13


@pytest.mark.parametrize(
    ("source", "expect_block"),
    [("export const a = 1;\n", True), ("const a = 1;\n", False)],
)
def test_skill_preflight_parses_module_entry_as_classic_script(
    tmp_path, monkeypatch, source, expect_block
):
    """The declared entry is checked in the grammar the frame actually runs it in."""
    ctx = _make_ctx(tmp_path)
    skill_dir = _make_skill(
        tmp_path,
        monkeypatch,
        _extension_manifest(ui_tab=_module_ui_tab()),
        _TRIVIAL_PLUGIN,
    )
    (skill_dir / "widget.js").write_text(source, encoding="utf-8")

    from ouroboros.tools import skill_preflight as sp

    result = json.loads(sp._handle_skill_preflight(ctx, skill="alpha"))
    rows = [row for row in result["files"] if row["path"] == "widget.js"]
    assert rows, result["files"]
    row = rows[0]
    assert row["grammar"] == "classic_script"
    if row.get("skipped"):
        # No usable node runtime: the honest skip branch, never a verdict. The
        # grammar assertion above still held, so the selection is covered.
        assert row["skip_reason"] in {"runtime_unavailable", "validator_killed", "validator_timeout"}
        assert result["degraded"] is True
        pytest.skip(f"no usable node runtime for the classic-script check ({row['skip_reason']})")
    if expect_block:
        assert row["ok"] is False
        assert result["ok"] is False
    else:
        assert row["ok"] is True


def test_skill_preflight_validator_env_keeps_windows_process_base_keys(tmp_path, monkeypatch):
    """The scrubbed validator env still lets a Windows child start.

    A node started without SystemRoot aborts before it reads the script, so the
    valid entry above read as a syntax error on windows-latest (7.0.0-rc.9).
    Pinned on every host by simulating that environment: the process-base keys
    are forwarded, everything else stays scrubbed, a POSIX env is byte-identical
    to before, and the validator's pipes stay BYTES decoded as UTF-8 with
    replacement -- the 0x8f that kills a locale-decoded (cp1252) reader thread
    is inert here.
    """
    from ouroboros.tools import skill_preflight as sp

    seen: dict = {}

    class _FakeProc:
        returncode = 0
        pid = 4242

        def communicate(self, timeout=None):
            return b"", b"\x8f\xff not utf-8"

    def _fake_popen(cmd, **kwargs):
        seen["cmd"] = list(cmd)
        seen["kwargs"] = kwargs
        return _FakeProc()

    monkeypatch.setattr(sp, "Popen", _fake_popen)
    for key in sp._WINDOWS_BASE_ENV_KEYS:
        monkeypatch.delenv(key, raising=False)
    monkeypatch.setenv("OPENROUTER_API_KEY", "must-not-reach-the-validator")

    result = sp._run_check(["node", "--check", "widget.js"], cwd=tmp_path)
    posix_env = seen["kwargs"]["env"]
    assert set(posix_env) == {"PATH", "HOME", "LANG"}
    assert posix_env["LANG"] == "C.UTF-8"
    assert not any(key in seen["kwargs"] for key in ("text", "universal_newlines", "encoding"))
    assert result["returncode"] == 0
    assert result["stderr"] == "\ufffd\ufffd not utf-8"

    monkeypatch.setenv("SYSTEMROOT", "C:\\Windows")
    monkeypatch.setenv("TEMP", "C:\\Users\\runneradmin\\AppData\\Local\\Temp")
    sp._run_check(["node", "--check", "widget.js"], cwd=tmp_path)
    windows_env = seen["kwargs"]["env"]
    assert windows_env["SYSTEMROOT"] == "C:\\Windows"
    assert windows_env["TEMP"] == "C:\\Users\\runneradmin\\AppData\\Local\\Temp"
    assert set(windows_env) == {"PATH", "HOME", "LANG", "SYSTEMROOT", "TEMP"}
    assert "OPENROUTER_API_KEY" not in windows_env


# ------------------------------------------------------------- S1-07 / S1-02 / F14


def _prepare_live_extension(tmp_path: pathlib.Path, name: str = "extlive"):
    """Write, enable and PASS-review one extension so the loader will accept it."""
    from ouroboros.skill_loader import SkillReviewState, find_skill, save_enabled, save_review_state

    repo_root = tmp_path / "skills"
    drive_root = tmp_path / "drive"
    drive_root.mkdir(exist_ok=True)
    skill_dir = repo_root / name
    skill_dir.mkdir(parents=True)
    (skill_dir / "SKILL.md").write_text(
        "---\n"
        f"name: {name}\n"
        "description: Live extension.\n"
        "version: 0.1.0\n"
        "type: extension\n"
        "entry: plugin.py\n"
        'permissions: ["tool"]\n'
        "---\n"
        "body\n",
        encoding="utf-8",
    )
    (skill_dir / "plugin.py").write_text(
        "def register(api):\n"
        "    api.register_tool('ping', lambda ctx: 'pong', description='Ping.', schema={})\n",
        encoding="utf-8",
    )
    loaded = find_skill(drive_root, name, repo_path=str(repo_root))
    assert loaded is not None
    save_enabled(drive_root, name, True, actor="test_fixture")
    save_review_state(drive_root, name, SkillReviewState(status="pass", content_hash=loaded.content_hash))
    loaded = find_skill(drive_root, name, repo_path=str(repo_root))
    assert loaded is not None
    return loaded, repo_root, drive_root


@pytest.fixture(autouse=True)
def _clean_loader_state(monkeypatch):
    from tests._shared import clean_extension_runtime_state

    monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "advanced")
    clean_extension_runtime_state()
    yield
    clean_extension_runtime_state()



def _set_process_role(monkeypatch, server: bool) -> None:
    """Pin which process "answers" a reconcile.

    v7 split: the loader stamps its own receipts through its imported binding,
    while the reconcile queue and the liveness projection read the OWNER
    (``extension_companion.is_server_process``) at call time — so the pin must
    land on both, or the receipt vocabulary (``requested`` / ``request_failed``)
    is decided by whatever process-pid state an earlier test left behind."""
    import ouroboros.extension_companion as extension_companion
    import ouroboros.extension_loader as extension_loader

    monkeypatch.setattr(extension_companion, "is_server_process", lambda: server)
    monkeypatch.setattr(extension_loader, "is_server_process", lambda: server)

def test_reconcile_receipt_names_the_answering_process(tmp_path, monkeypatch):
    """A reconcile receipt says which process answered and whether the marker landed."""
    from ouroboros import extension_loader

    loaded, repo_root, drive_root = _prepare_live_extension(tmp_path)

    _set_process_role(monkeypatch, True)
    state = extension_loader.reconcile_extension(
        loaded.name, drive_root, lambda: {}, repo_path=str(repo_root)
    )
    assert state["process"] == "server"
    assert state["server_reconcile"] == ""

    extension_loader.unload_extension(loaded.name)
    _set_process_role(monkeypatch, False)
    state = extension_loader.reconcile_extension(
        loaded.name, drive_root, lambda: {}, repo_path=str(repo_root)
    )
    assert state["process"] == "worker"
    assert state["server_reconcile"] == "requested"
    assert list((drive_root / "state" / "extension_reconcile").glob("*")), "no marker written"


def test_reconcile_records_health_for_the_resulting_runtime_state(tmp_path, monkeypatch):
    """Each reconcile receipt updates the durable health projection."""
    from ouroboros import extension_health, extension_loader
    from ouroboros.skill_loader import save_enabled

    loaded, repo_root, drive_root = _prepare_live_extension(tmp_path)
    _set_process_role(monkeypatch, True)

    state = extension_loader.reconcile_extension(
        loaded.name, drive_root, lambda: {}, repo_path=str(repo_root)
    )
    assert state["live_loaded"] is True
    assert (extension_health.read_extension_health(drive_root, loaded.name) or {})["status"] == "live"

    save_enabled(drive_root, loaded.name, False, actor="test_fixture")
    state = extension_loader.reconcile_extension(
        loaded.name, drive_root, lambda: {}, repo_path=str(repo_root)
    )
    assert state["desired_live"] is False
    assert (extension_health.read_extension_health(drive_root, loaded.name) or {})["status"] == "inactive"


def test_reconcile_receipt_reports_a_failed_marker_request(tmp_path, monkeypatch):
    """A failed worker handoff cannot replace the authoritative server health."""
    from ouroboros import extension_health, extension_loader
    from ouroboros import extension_reconcile_queue

    loaded, repo_root, drive_root = _prepare_live_extension(tmp_path)
    extension_health.record_extension_health(
        drive_root, loaded.name, status="live", version="0.0.1", sha="server-good",
    )
    extension_health.record_extension_health(
        drive_root, loaded.name, status="broken", version="0.0.2", sha="shared-sha",
        reason="load_error", load_error="server import failed",
    )
    _set_process_role(monkeypatch, False)

    def boom(*_a, **_k):
        raise OSError("marker directory is read-only")

    monkeypatch.setattr(extension_reconcile_queue, "request_extension_reconcile", boom)
    state = extension_loader.reconcile_extension(
        loaded.name, drive_root, lambda: {}, repo_path=str(repo_root),
        health_stamp=("0.0.2", "shared-sha"),
    )
    health = extension_health.read_extension_health(drive_root, loaded.name) or {}
    server = health.get("last_observed") or {}
    worker = (health.get("observations") or {}).get("worker") or {}

    assert state["process"] == "worker"
    assert state["server_reconcile"] == "request_failed"
    assert state["action"] == "extension_loaded"
    assert health["status"] == "broken"
    assert health["regressed"] is True
    assert health["last_known_good"]["sha"] == "server-good"
    assert server["status"] == "broken"
    assert server["sha"] == "shared-sha"
    assert worker["status"] == "live"
    assert worker["sha"] == "shared-sha"
    assert worker["server_reconcile"] == "request_failed"
    monkeypatch.setattr("ouroboros.skill_loader.find_skill", lambda *_a, **_k: object())
    monkeypatch.setattr("ouroboros.skill_loader.load_enabled", lambda *_a, **_k: True)
    assert [row["skill"] for row in extension_health.regressed_extensions(drive_root)] == [
        loaded.name
    ]


def test_toggle_skill_receipt_carries_process_and_marker_outcome(tmp_path, monkeypatch):
    """The agent-facing toggle receipt is where the 'already_live/ready' misreading happened."""
    from ouroboros.tools import skill_exec as skill_exec_mod

    loaded, repo_root, drive_root = _prepare_live_extension(tmp_path)
    monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(repo_root))
    ctx = ToolContext(repo_dir=tmp_path / "repo2", drive_root=drive_root)
    (tmp_path / "repo2").mkdir()

    payload = json.loads(skill_exec_mod._handle_toggle_skill(ctx, skill=loaded.name, enabled=True))
    assert payload["process"] in {"server", "worker"}
    assert payload["server_reconcile"] in {"", "requested", "request_failed"}


def test_save_enabled_appends_one_typed_actor_row(tmp_path):
    """Enablement changes leave a durable, non-rotating, actor-attributed record."""
    from ouroboros.skill_loader import save_enabled

    drive_root = tmp_path / "drive"
    drive_root.mkdir()
    save_enabled(drive_root, "alpha", True, actor="owner_ui", reason="client_host=127.0.0.1")
    save_enabled(drive_root, "alpha", False, actor="agent_tool", reason="task-7")
    save_enabled(drive_root, "alpha", True)

    rows = [
        json.loads(line)
        for line in (drive_root / "logs" / "events.jsonl").read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]
    rows = [row for row in rows if row.get("type") == "skill_enabled_changed"]
    assert [(r["enabled"], r["previous"], r["actor"]) for r in rows] == [
        (True, False, "owner_ui"),
        (False, True, "agent_tool"),
        (True, False, ""),
    ]
    assert rows[0]["reason"] == "client_host=127.0.0.1"


def test_save_enabled_row_is_disclosure_never_a_gate(tmp_path):
    """An unwritable logs path must not fail the enablement write."""
    from ouroboros.skill_loader import load_enabled, save_enabled

    drive_root = tmp_path / "drive"
    drive_root.mkdir()
    (drive_root / "logs").write_text("not a directory", encoding="utf-8")
    save_enabled(drive_root, "alpha", True, actor="owner_ui")
    assert load_enabled(drive_root, "alpha") is True


def test_save_enabled_best_effort_disclosure_contract_is_documented():
    from ouroboros.skill_loader import save_enabled

    architecture = architecture_text()
    assert "best-effort" in str(save_enabled.__doc__)
    assert "append failure is logged and never blocks the enablement change" in architecture


def test_api_skill_toggle_records_the_owner_ui_actor(tmp_path, monkeypatch):
    """The HTTP owner toggle labels itself, so the incident class is reconstructible."""
    from ouroboros.skill_loader import SkillReviewState, compute_content_hash, save_review_state
    from tests.test_extensions_api import _make_client, _stop_patches, _write_ext

    skills_root = tmp_path / "skills"
    skill_dir = _write_ext(
        skills_root,
        "ext_actor",
        permissions=["tool"],
        plugin="def register(api):\n    api.register_tool('t', lambda ctx: 'ok', description='', schema={})\n",
    )
    monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(skills_root))
    client, drive_root, _patches = _make_client(tmp_path, monkeypatch)
    try:
        save_review_state(
            drive_root,
            "ext_actor",
            SkillReviewState(
                status="pass",
                content_hash=compute_content_hash(skill_dir, manifest_entry="plugin.py"),
            ),
        )
        resp = client.post("/api/skills/ext_actor/toggle", json={"enabled": True})
        assert resp.status_code == 200, resp.text
    finally:
        client.close()
        _stop_patches(_patches)   # the started patches (the password resolver among them) must not outlive the test

    rows = [
        json.loads(line)
        for line in (drive_root / "logs" / "events.jsonl").read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]
    rows = [r for r in rows if r.get("type") == "skill_enabled_changed" and r.get("skill") == "ext_actor"]
    assert rows, "the owner toggle left no enablement row"
    assert rows[-1]["actor"] == "owner_ui"
    assert rows[-1]["enabled"] is True
    # Shared lifecycle owner keeps transport provenance in its structured
    # owner_api_action row; skill_enabled_changed remains the state-change fact.
    action_rows = [json.loads(line) for line in (drive_root / "logs" / "events.jsonl").read_text().splitlines()
                   if line.strip()]
    action_rows = [row for row in action_rows if row.get("type") == "owner_api_action"
                   and row.get("skill") == "ext_actor" and row.get("action") == "skill_enable"]
    assert len(action_rows) == 1
    assert {key: action_rows[0][key] for key in ("actor", "skill", "action", "client_host", "ok")} == {
        "actor": "owner_ui", "skill": "ext_actor", "action": "skill_enable",
        "client_host": "testclient", "ok": True,
    }


def test_summarize_skills_projects_live_extension_facts(tmp_path, monkeypatch):
    """The catalogue reports the same live facts as /api/extensions, not worker guesses."""
    from ouroboros import skill_loader
    from ouroboros.skill_loader import summarize_skills

    loaded, repo_root, drive_root = _prepare_live_extension(tmp_path)
    monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(repo_root))

    calls = {"n": 0}
    real_discover = skill_loader.discover_skills

    def counting(*a, **k):
        calls["n"] += 1
        return real_discover(*a, **k)

    monkeypatch.setattr(skill_loader, "discover_skills", counting)
    summary = summarize_skills(drive_root)
    # One walk for the catalogue itself; `skills=` must stop the per-row re-walk.
    assert calls["n"] == 1, f"discover_skills ran {calls['n']} times"

    row = next(r for r in summary["skills"] if r["name"] == loaded.name)
    assert row["desired_live"] is True
    assert row["live_loaded"] is False
    assert row["live_reason"]
    assert row["process"] in {"server", "worker"}
    assert row["available_for_execution"] is False


def test_skill_exec_extension_message_reports_typed_liveness(tmp_path, monkeypatch):
    """skill_exec no longer asserts register(api) ran for an extension that never loaded."""
    from ouroboros.tools import skill_exec as skill_exec_mod

    loaded, repo_root, drive_root = _prepare_live_extension(tmp_path)
    monkeypatch.setenv("OUROBOROS_SKILLS_REPO_PATH", str(repo_root))
    ctx = ToolContext(repo_dir=tmp_path / "repo3", drive_root=drive_root)
    (tmp_path / "repo3").mkdir()

    out = skill_exec_mod._handle_skill_exec(ctx, skill=loaded.name, script="x.py")
    assert "SKILL_EXEC_EXTENSION" in out
    assert "live_loaded=False" in out
    assert "has already been called" not in out


@pytest.mark.parametrize('registration', [True, False])
def test_documented_module_widget_recipe_creates_the_registered_tab(tmp_path, registration):
    """Execute the documentation example; manifest metadata alone stays metadata."""
    import re
    from tests._extension_loader_shared import _write_ext_skill
    from ouroboros import extension_loader
    from ouroboros.skill_loader import find_skill, save_enabled, save_review_state, SkillReviewState
    from tests._shared import clean_extension_runtime_state

    text = (pathlib.Path(__file__).resolve().parents[1] / 'docs/CREATING_SKILLS.md').read_text(encoding='utf-8')
    section = text.split('### `kind: "module"` widgets', 1)[1].split('#### The in-frame bridge', 1)[0]
    declaration = re.search(r'```yaml\n(.*?)```', section, re.S).group(1)
    plugin = re.search(r'```python\n(.*?)```', section, re.S).group(1)
    root, skills = tmp_path / 'drive', tmp_path / 'skills'
    skill_dir = _write_ext_skill(skills, 'recipe', permissions=['widget'],
        plugin_body=plugin if registration else 'def register(api):\n    pass\n',
        extra_frontmatter=declaration)
    widget = "document.getElementById('root').textContent = 'Recipe widget ✓';"
    (skill_dir / 'widget.js').write_text(widget, encoding='utf-8')
    assert (skill_dir / 'widget.js').read_bytes() == widget.encode('utf-8')
    loaded = find_skill(root, 'recipe', repo_path=str(skills))
    save_enabled(root, 'recipe', True)
    save_review_state(root, 'recipe', SkillReviewState(status='pass', content_hash=loaded.content_hash))
    loaded = find_skill(root, 'recipe', repo_path=str(skills))
    clean_extension_runtime_state()
    try:
        assert extension_loader.load_extension(loaded, lambda: {}, drive_root=root, repo_path=str(skills)) is None
        rows = extension_loader.live_widget_projection('recipe')
        assert len(rows) == (1 if registration else 0)
        if registration:
            assert rows[0]['tab']['render']['entry'] == 'widget.js'
            assert rows[0]['tab']['title'] == 'Editor'
    finally:
        clean_extension_runtime_state()