skillwiki/tests/test_skill_runtime_phase1.py

from __future__ import annotations

import asyncio
from types import SimpleNamespace

import pytest

from skillos.api.routes import execution
from skillos.api.schemas import ExecutePlanRequest
from skillos.layers.skill_repository.indexing import SearchResult
from skillos.layers.skill_runtime.executor import SkillExecutor
from skillos.layers.skill_runtime.planner import ExecutionPlan, PlanStep, StepStatus
from skillos.models.skill_model import (
    Skill,
    SkillEvaluation,
    SkillImplementation,
    SkillInterface,
    SkillState,
)


def make_skill(name: str = "fill_form") -> Skill:
    return Skill(
        name=name,
        description=f"{name} test skill",
        state=SkillState.RELEASED,
        interface=SkillInterface(
            input_schema={"type": "object", "properties": {}},
            output_schema={"type": "object", "properties": {}},
        ),
        implementation=SkillImplementation(code="output['ok'] = True"),
    )


@pytest.mark.asyncio
async def test_execute_plan_formats_match_reasons_and_records_metrics():
    skill = make_skill()
    app = FakeAppState(
        skills=[skill],
        search_results=[SearchResult(skill=skill, score=0.9, match_reasons=["exact name match", "state boost"])],
        plan_steps=[PlanStep(step_index=7, skill_id=skill.skill_id, skill_name=skill.name)],
    )

    result = await execution.execute_plan(ExecutePlanRequest(goal="fill form"), app=app)

    assert result.status == "success"
    assert result.retrieved_skills[0].match_reason == "exact name match; state boost"
    assert result.steps[0].step_index == 7
    assert result.steps[0].outputs == result.steps[0].result
    assert skill.metrics.usage_count == 1
    assert skill.metrics.success_count == 1
    assert app.recorded == [(skill.skill_id, True)]
    assert result.experience_recorded is True
    assert result.experience_unit is not None
    assert result.experience_unit.source_type == "agent_execution"
    assert result.experience_unit.source_execution_id == result.plan_id
    assert result.experience_unit.metadata["paper_backlog_task"] == "C-P1-2"
    assert result.experience_unit.metadata["paper_method"] == "XSkill action-level experience stream"
    assert result.experience_unit.normalized_actions[0]["skill_id"] == skill.skill_id


@pytest.mark.asyncio
async def test_execute_plan_no_skills_returns_failed_without_crashing():
    app = FakeAppState(skills=[], search_results=[], plan_steps=[])

    result = await execution.execute_plan(ExecutePlanRequest(goal="unknown task"), app=app)

    assert result.status == "failed"
    assert result.steps == []
    assert result.retrieved_skills == []
    assert result.verifier_summary is None


@pytest.mark.asyncio
async def test_execute_plan_attaches_deterministic_verifier_summary():
    skill = make_skill()
    skill.evaluation = SkillEvaluation(
        verifier_specs=[{"type": "json_equals", "path": "output.ok", "value": True}]
    )
    app = FakeAppState(
        skills=[skill],
        search_results=[SearchResult(skill=skill, score=0.9, match_reasons=["name match"])],
        plan_steps=[PlanStep(step_index=0, skill_id=skill.skill_id, skill_name=skill.name)],
    )

    result = await execution.execute_plan(ExecutePlanRequest(goal="fill form"), app=app)

    assert result.verifier_passed is True
    assert result.verifier_summary is not None
    assert result.verifier_summary["mode"] == "deterministic"
    assert result.verifier_summary["checked_skills"] == 1
    assert result.verifier_summary["results"][0]["skill_id"] == skill.skill_id


@pytest.mark.asyncio
async def test_execution_history_returns_items_in_reverse_order():
    original = list(execution._execution_history)
    execution._execution_history.clear()
    try:
        execution._execution_history.extend([
            {
                "execution_id": "old",
                "goal": "old goal",
                "status": "failed",
                "step_count": 0,
                "success_count": 0,
                "total_latency_ms": 1.0,
                "retrieved_skill_count": 0,
                "created_at": "2026-05-04T00:00:00",
            },
            {
                "execution_id": "new",
                "goal": "new goal",
                "status": "success",
                "step_count": 1,
                "success_count": 1,
                "total_latency_ms": 2.0,
                "retrieved_skill_count": 1,
                "created_at": "2026-05-04T00:01:00",
            },
        ])

        history = await execution.get_execution_history()

        assert [item["execution_id"] for item in history] == ["new", "old"]
    finally:
        execution._execution_history[:] = original


@pytest.mark.asyncio
async def test_execution_history_returns_full_experience_unit_for_plan():
    original = list(execution._execution_history)
    execution._execution_history.clear()
    try:
        skill = make_skill()
        app = FakeAppState(
            skills=[skill],
            search_results=[SearchResult(skill=skill, score=0.9, match_reasons=["name match"])],
            plan_steps=[PlanStep(
                step_index=0,
                skill_id=skill.skill_id,
                skill_name=skill.name,
                input_mapping={"field": "email"},
            )],
        )

        result = await execution.execute_plan(ExecutePlanRequest(goal="fill login form"), app=app)
        history = await execution.get_execution_history()
        unit = await execution.get_execution_experience(result.plan_id)

        assert history[0]["execution_id"] == result.plan_id
        assert history[0]["experience_unit_id"] == unit.unit_id
        assert history[0]["experience_source_type"] == "agent_execution"
        assert unit.source_execution_id == result.plan_id
        assert unit.source_type == "agent_execution"
        assert unit.normalized_actions[0]["input_mapping"] == {"field": "email"}
        assert unit.proposed_skill_name == "skill_from_fill_login_form"
        assert unit.metadata["paper_backlog_task"] == "C-P1-2"
    finally:
        execution._execution_history[:] = original


@pytest.mark.asyncio
async def test_executor_schedules_async_callbacks_and_ignores_callback_errors():
    executor = SkillExecutor()
    received: list[tuple[str, dict]] = []

    async def async_callback(event_type: str, data: dict) -> None:
        received.append((event_type, data))

    def failing_callback(event_type: str, data: dict) -> None:
        raise RuntimeError("callback failed")

    executor.add_event_callback(async_callback)
    executor.add_event_callback(failing_callback)
    executor._emit("plan_completed", {"plan_id": "plan-1"})
    await asyncio.sleep(0)

    assert received == [("plan_completed", {"plan_id": "plan-1"})]


class FakeAppState:
    def __init__(self, skills: list[Skill], search_results: list[SearchResult], plan_steps: list[PlanStep]) -> None:
        self.state_tracker = FakeStateTracker()
        self.wiki = FakeWiki(skills)
        self.search = FakeSearch(search_results)
        self.planner = FakePlanner(plan_steps)
        self.executor = FakeExecutor()
        self.recorded = self.wiki.recorded


class FakeStateTracker:
    def __init__(self) -> None:
        self.current: dict = {}

    def update(self, changes: dict) -> None:
        self.current.update(changes)


class FakeWiki:
    def __init__(self, skills: list[Skill]) -> None:
        self.skills = {skill.skill_id: skill for skill in skills}
        self.recorded: list[tuple[str, bool]] = []

    async def get_many(self, skill_ids: list[str]) -> dict[str, Skill | None]:
        return {skill_id: self.skills.get(skill_id) for skill_id in skill_ids}

    async def record_execution(self, skill_id: str, success: bool, latency_ms: float) -> None:
        self.recorded.append((skill_id, success))
        skill = self.skills.get(skill_id)
        if skill:
            skill.record_execution(success, latency_ms)


class FakeSearch:
    def __init__(self, results: list[SearchResult]) -> None:
        self.results = results

    async def search(self, query: object) -> list[SearchResult]:
        return self.results


class FakePlanner:
    def __init__(self, steps: list[PlanStep]) -> None:
        self.steps = steps

    async def plan(self, task_description: str, available_skills: list[Skill], current_state: dict) -> ExecutionPlan:
        return ExecutionPlan(plan_id="plan-1", task_id="plan-1", task_description=task_description, steps=self.steps)


class FakeExecutor:
    async def execute_plan(self, plan: ExecutionPlan, skill_map: dict[str, Skill], initial_state: dict) -> dict:
        for step in plan.steps:
            if step.skill_id in skill_map:
                step.status = StepStatus.SUCCESS
                step.result = {"ok": True}
            else:
                step.status = StepStatus.FAILED
                step.error = "missing skill"
        return {"done": True}