OpenSpace/tests/skill_engine/test_skill_trust_lifecycle.py

import asyncio
import sqlite3
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace

import pytest

from openspace.skill_engine.evidence import EvidenceStore
from openspace.skill_engine.evidence.types import (
    EvidencePacket,
    EvidenceScope,
    PacketBudget,
    ResourceRef,
)
from openspace.skill_engine.evolution.candidates import EvolutionCandidateStore
from openspace.skill_engine.evolution.admission import EvolutionAdmission
from openspace.skill_engine.evolution.audit import EvolutionAuditService
from openspace.skill_engine.evolution.engine import EvolutionCommitter
from openspace.skill_engine.protocol import SkillDiscoveryService
from openspace.skill_engine.registry import SkillMeta
from openspace.skill_engine.store import SkillStore
from openspace.skill_engine.triggers import TriggerJobSpec, TriggerStore, default_policies
from openspace.skill_engine.types import (
    ExecutionAnalysis,
    SkillJudgment,
    SkillCategory,
    SkillLineage,
    SkillOrigin,
    SkillRecord,
    SkillTrustState,
)


def _save(store: SkillStore, record: SkillRecord) -> None:
    asyncio.run(store.save_record(record))


def _captured_record(skill_id: str = "captured__v0_test") -> SkillRecord:
    return SkillRecord(
        skill_id=skill_id,
        name="captured",
        description="Captured workflow",
        enabled=True,
        trust_state=SkillTrustState.PROVISIONAL,
        lineage=SkillLineage(
            origin=SkillOrigin.CAPTURED,
            source_task_id="task_origin",
        ),
    )


def _analysis(
    task_id: str,
    skill_id: str,
    *,
    completed: bool,
    phase_failed: bool = False,
) -> ExecutionAnalysis:
    return ExecutionAnalysis(
        task_id=task_id,
        timestamp=datetime.now(),
        task_completed=completed,
        skill_judgments=[
            SkillJudgment(
                skill_id=skill_id,
                skill_applied=True,
                note="workflow applied",
            )
        ],
        skill_phase_failed_skill_ids=[skill_id] if phase_failed else [],
    )


def _candidate_packet(task_id: str) -> EvidencePacket:
    runtime_ref = ResourceRef(
        ref_id=f"runtime:{task_id}",
        ref_type="runtime_snapshot",
        task_id=task_id,
        preview="task completed",
        metadata={"status": "success", "final_response_preview": "done"},
    )
    tool_ref = ResourceRef(
        ref_id=f"tool:{task_id}",
        ref_type="tool_event",
        task_id=task_id,
        preview="bash workflow completed",
        metadata={"tool_key": "shell:default:bash", "status": "success"},
    )
    validation_ref = ResourceRef(
        ref_id=f"tool:verify:{task_id}",
        ref_type="tool_event",
        task_id=task_id,
        preview="independent workflow check passed",
        metadata={"tool_key": "shell:default:bash", "status": "success"},
    )
    return EvidencePacket(
        packet_id=f"packet:{task_id}",
        trigger_job_id=f"trigger:{task_id}",
        packet_type="analysis",
        profile_name="analysis_current_task",
        subprofile="task_finished",
        manifest_watermark=1,
        scope=EvidenceScope(task_id=task_id, source_task_ids=(task_id,)),
        selected_refs={
            "runtime_snapshot": [runtime_ref],
            "tool_event": [tool_ref, validation_ref],
        },
        expanded_snippets=[],
        readable_paths=[],
        instructions={},
        budget=PacketBudget(max_chars=1000, used_chars=10),
        redaction_status="ok",
        build_status="ok",
        missing_ref_types=[],
    )


def _candidate_decision(decision_id: str, summary: str) -> SimpleNamespace:
    return SimpleNamespace(
        decision_id=decision_id,
        proposed_action="CAPTURED",
        target_skill_ids=[],
        reason_summary=summary,
        reason_tags=["captured", "candidate_default"],
        local_category_path="local/workflow/test",
        evidence_claims=[
            SimpleNamespace(
                claim="workflow evidence",
                refs=["tool:task_one", "tool:verify:task_one"],
            )
        ],
        proposal_contract={
            "capability": "Run and verify a reusable workflow.",
            "preconditions": [],
            "procedure_refs": ["tool:task_one"],
            "validation_refs": ["tool:verify:task_one"],
            "validation_summary": "A separate command verified the postcondition.",
            "limitations": [],
        },
    )


def _candidate_admission(admission_id: str) -> SimpleNamespace:
    return SimpleNamespace(
        admission_id=admission_id,
        warnings=["provisional_evolution_disabled"],
        required_refs_checked=[],
    )


def test_origin_plus_one_cross_task_success_promotes_trust(tmp_path: Path) -> None:
    store = SkillStore(tmp_path / "skills.db")
    try:
        record = _captured_record()
        _save(store, record)
        asyncio.run(
            store.record_trust_observation(
                record.skill_id,
                "task:task_origin",
                "success",
                task_id="task_origin",
                source="evolution_origin",
            )
        )

        provisional = store.load_record(record.skill_id)
        assert provisional is not None
        assert provisional.trust_state == SkillTrustState.PROVISIONAL
        assert provisional.trust_successes == 1

        asyncio.run(
            store.record_analysis(
                _analysis("task_reuse", record.skill_id, completed=True)
            )
        )

        trusted = store.load_record(record.skill_id)
        assert trusted is not None
        assert trusted.trust_state == SkillTrustState.TRUSTED
        assert trusted.trust_successes == 2
        observations = store.load_trust_observations(record.skill_id)
        assert [item["task_id"] for item in observations] == [
            "task_origin",
            "task_reuse",
        ]
        assert observations[-1]["evidence_refs"] == ["analysis:task_reuse"]
    finally:
        store.close()


def test_committer_builds_evolved_revision_as_provisional(tmp_path: Path) -> None:
    store = SkillStore(tmp_path / "skills.db")
    try:
        committer = EvolutionCommitter(
            evidence_store=SimpleNamespace(),
            skill_store=store,
            registry=SimpleNamespace(),
        )
        record = committer._build_skill_record(
            action=SimpleNamespace(action_id="action_one", decision_id="decision_one"),
            staged=SimpleNamespace(
                proposed_name="captured",
                proposed_description="Captured workflow",
                content_snapshot={"SKILL.md": "# Captured\n"},
                content_diff="",
                tool_dependencies=[],
                critical_tools=[],
                apply_metadata={},
            ),
            authoring=SimpleNamespace(model="test-model"),
            decision=SimpleNamespace(
                reason_summary="Capture reusable workflow",
                local_category_path="local/workflow/test",
            ),
            action_packet=SimpleNamespace(
                scope=SimpleNamespace(task_id="task_origin")
            ),
            action_type="CAPTURED",
            target_dir=tmp_path / "captured",
            parent_skill_ids=[],
            evidence_refs=["runtime:task_origin"],
            skill_id="captured__v0_test",
        )

        assert record.trust_state == SkillTrustState.PROVISIONAL
        assert record.enabled is True
    finally:
        store.close()


def test_registry_sync_refreshes_existing_record_with_category(
    tmp_path: Path,
) -> None:
    skill_path = tmp_path / "remember" / "SKILL.md"
    skill_path.parent.mkdir()
    skill_path.write_text(
        "---\nname: remember\ndescription: Updated description.\n---\n",
        encoding="utf-8",
    )
    store = SkillStore(tmp_path / "skills.db")
    try:
        store._save_record_sync(
            SkillRecord(
                skill_id="remember__test",
                name="remember",
                description="Old description.",
                path=str(skill_path),
                category=SkillCategory.WORKFLOW,
                lineage=SkillLineage(origin=SkillOrigin.IMPORTED),
            )
        )

        created, _ = store._sync_from_registry_sync(
            [
                SkillMeta(
                    skill_id="remember__test",
                    name="remember",
                    description="Updated description.",
                    path=skill_path,
                )
            ]
        )

        refreshed = store.load_record("remember__test")
        assert created == 0
        assert refreshed is not None
        assert refreshed.description == "Updated description."
    finally:
        store.close()


def test_existing_skill_store_schema_migrates_to_trusted_enabled(tmp_path: Path) -> None:
    db_path = tmp_path / "legacy.db"
    now = datetime.now().isoformat()
    conn = sqlite3.connect(db_path)
    conn.executescript(
        """
        CREATE TABLE skill_records (
            skill_id TEXT PRIMARY KEY,
            name TEXT NOT NULL,
            description TEXT NOT NULL DEFAULT '',
            path TEXT NOT NULL DEFAULT '',
            is_active INTEGER NOT NULL DEFAULT 1,
            category TEXT NOT NULL DEFAULT 'workflow',
            visibility TEXT NOT NULL DEFAULT 'private',
            creator_id TEXT NOT NULL DEFAULT '',
            lineage_origin TEXT NOT NULL DEFAULT 'imported',
            lineage_revision_id TEXT NOT NULL DEFAULT '',
            lineage_generation INTEGER NOT NULL DEFAULT 0,
            lineage_parent_revision_ids_json TEXT NOT NULL DEFAULT '[]',
            lineage_source_task_id TEXT,
            lineage_change_summary TEXT NOT NULL DEFAULT '',
            lineage_content_hash TEXT NOT NULL DEFAULT '',
            lineage_evolution_action_id TEXT,
            lineage_provenance_refs_json TEXT NOT NULL DEFAULT '[]',
            lineage_revision_metadata_json TEXT NOT NULL DEFAULT '{}',
            lineage_content_diff TEXT NOT NULL DEFAULT '',
            lineage_content_snapshot TEXT NOT NULL DEFAULT '{}',
            lineage_created_at TEXT NOT NULL,
            lineage_created_by TEXT NOT NULL DEFAULT '',
            total_selections INTEGER NOT NULL DEFAULT 0,
            total_applied INTEGER NOT NULL DEFAULT 0,
            total_completions INTEGER NOT NULL DEFAULT 0,
            total_fallbacks INTEGER NOT NULL DEFAULT 0,
            first_seen TEXT NOT NULL,
            last_updated TEXT NOT NULL
        );
        """
    )
    conn.execute(
        """
        INSERT INTO skill_records (
            skill_id, name, lineage_created_at, first_seen, last_updated
        ) VALUES (?, ?, ?, ?, ?)
        """,
        ("legacy", "legacy", now, now, now),
    )
    conn.commit()
    conn.close()

    store = SkillStore(db_path)
    try:
        migrated = store.load_record("legacy")
        assert migrated is not None
        assert migrated.enabled is True
        assert migrated.trust_state == SkillTrustState.TRUSTED
    finally:
        store.close()


def test_duplicate_observation_is_idempotent(tmp_path: Path) -> None:
    store = SkillStore(tmp_path / "skills.db")
    try:
        record = _captured_record()
        _save(store, record)
        for _ in range(2):
            asyncio.run(
                store.record_trust_observation(
                    record.skill_id,
                    "task:task_origin",
                    "success",
                    task_id="task_origin",
                )
            )

        loaded = store.load_record(record.skill_id)
        assert loaded is not None
        assert loaded.trust_successes == 1
        assert loaded.trust_state == SkillTrustState.PROVISIONAL
    finally:
        store.close()


def test_trust_promotion_threshold_is_configurable(tmp_path: Path) -> None:
    store = SkillStore(
        tmp_path / "skills.db",
        trust_promotion_min_independent_successes=3,
    )
    try:
        record = _captured_record()
        _save(store, record)
        for task_id in ("task_origin", "task_reuse_one"):
            asyncio.run(
                store.record_trust_observation(
                    record.skill_id,
                    f"task:{task_id}",
                    "success",
                    task_id=task_id,
                )
            )
        assert store.load_record(record.skill_id).trust_state == SkillTrustState.PROVISIONAL

        asyncio.run(
            store.record_trust_observation(
                record.skill_id,
                "task:task_reuse_two",
                "success",
                task_id="task_reuse_two",
            )
        )
        assert store.load_record(record.skill_id).trust_state == SkillTrustState.TRUSTED
    finally:
        store.close()


def test_only_attributable_failure_demotes_trusted_skill(tmp_path: Path) -> None:
    store = SkillStore(tmp_path / "skills.db")
    try:
        record = SkillRecord(
            skill_id="trusted__v0_test",
            name="trusted",
            description="Trusted workflow",
            trust_state=SkillTrustState.TRUSTED,
        )
        _save(store, record)

        asyncio.run(
            store.record_analysis(
                _analysis("task_generic_failure", record.skill_id, completed=False)
            )
        )
        still_trusted = store.load_record(record.skill_id)
        assert still_trusted is not None
        assert still_trusted.trust_state == SkillTrustState.TRUSTED
        assert still_trusted.trust_failures == 0

        asyncio.run(
            store.record_analysis(
                _analysis(
                    "task_skill_failure",
                    record.skill_id,
                    completed=False,
                    phase_failed=True,
                )
            )
        )
        demoted = store.load_record(record.skill_id)
        assert demoted is not None
        assert demoted.trust_state == SkillTrustState.PROVISIONAL
        assert demoted.trust_failures == 1
    finally:
        store.close()


def test_repromotion_requires_fresh_successes_after_failure(tmp_path: Path) -> None:
    store = SkillStore(tmp_path / "skills.db")
    try:
        record = SkillRecord(
            skill_id="trusted__v0_test",
            name="trusted",
            description="Trusted workflow",
            trust_state=SkillTrustState.TRUSTED,
        )
        _save(store, record)
        asyncio.run(
            store.record_analysis(
                _analysis(
                    "task_failure",
                    record.skill_id,
                    completed=False,
                    phase_failed=True,
                )
            )
        )
        asyncio.run(
            store.record_analysis(
                _analysis("task_recovery_one", record.skill_id, completed=True)
            )
        )
        assert store.load_record(record.skill_id).trust_state == SkillTrustState.PROVISIONAL

        asyncio.run(
            store.record_analysis(
                _analysis("task_recovery_two", record.skill_id, completed=True)
            )
        )
        assert store.load_record(record.skill_id).trust_state == SkillTrustState.TRUSTED
    finally:
        store.close()


def test_enabled_is_independent_from_trust(tmp_path: Path) -> None:
    store = SkillStore(tmp_path / "skills.db")
    try:
        record = _captured_record()
        _save(store, record)

        assert asyncio.run(store.set_skill_enabled(record.skill_id, False)) is True
        disabled = store.load_record(record.skill_id)
        assert disabled is not None
        assert disabled.enabled is False
        assert disabled.trust_state == SkillTrustState.PROVISIONAL
        assert store.is_skill_enabled(record.skill_id) is False
        meta = SimpleNamespace(
            skill_id=record.skill_id,
            name=record.name,
            disable_model_invocation=False,
            conditional_paths=[],
        )
        registry = SimpleNamespace(list_skills=lambda: [meta])
        discovery = SkillDiscoveryService(registry, store=store)
        assert discovery._candidate_skills(None) == []
    finally:
        store.close()


def test_candidate_recurrence_counts_distinct_tasks_not_admissions(
    tmp_path: Path,
) -> None:
    store = EvolutionCandidateStore(tmp_path / "evidence.db")
    try:
        packet = _candidate_packet("task_one")
        first = store.create_or_merge(
            _candidate_decision("decision_one", "Capture the same workflow"),
            _candidate_admission("admission_one"),
            packet,
        )
        second = store.create_or_merge(
            _candidate_decision("decision_two", "Capture the same workflow"),
            _candidate_admission("admission_two"),
            packet,
        )

        assert first.candidate_id == second.candidate_id
        assert second.recurrence_count == 1
        assert second.recurrence == "single"
        assert not hasattr(store, "request_recheck")
        assert not hasattr(store, "mark_promoted")
        with pytest.raises(ValueError, match="only be rejected or superseded"):
            store.update_candidate_status(
                second.candidate_id,
                "promoted",
            )
    finally:
        store.close()


def test_semantic_validation_candidate_records_why_it_is_pending(
    tmp_path: Path,
) -> None:
    store = EvolutionCandidateStore(tmp_path / "evidence.db")
    try:
        candidate = store.create_or_merge(
            _candidate_decision("decision_one", "Capture the same workflow"),
            SimpleNamespace(
                admission_id="admission_one",
                outcome="direct",
                warnings=[],
                hard_failures=[],
                required_refs_checked=[],
            ),
            _candidate_packet("task_one"),
            reason="semantic_validation_failed",
        )

        assert candidate.status == "pending"
        assert candidate.blocked_reason == "validation_failed:semantic"
        assert candidate.needed_evidence == [
            "narrower_source_supported_capability_or_artifact_repair"
        ]
        assert candidate.decision_snapshot["candidate_reason"] == (
            "semantic_validation_failed"
        )
    finally:
        store.close()


def test_candidate_recheck_trigger_is_not_supported(tmp_path: Path) -> None:
    store = TriggerStore(db_path=tmp_path / "evidence.db")
    try:
        job = store.create_job(
            TriggerJobSpec(
                trigger_type="CANDIDATE_RECHECK",
                reason="legacy_candidate_recheck",
                scope=EvidenceScope(),
                idempotency_key="legacy_candidate_recheck:test",
                evidence_profile="candidate_recheck",
                subprofile="candidate_recheck",
            ),
            manifest_watermark=0,
        )

        assert job.status == "rejected"
        assert job.error == "unknown trigger_type: CANDIDATE_RECHECK"
        assert "CANDIDATE_RECHECK" not in {
            policy.trigger_type for policy in default_policies()
        }
    finally:
        store.close()


def test_legacy_open_candidate_recheck_job_is_superseded(tmp_path: Path) -> None:
    db_path = tmp_path / "evidence.db"
    store = TriggerStore(db_path=db_path)
    try:
        job = store.create_job(
            TriggerJobSpec(
                trigger_type="CANDIDATE_RECHECK",
                reason="legacy_candidate_recheck",
                scope=EvidenceScope(),
                idempotency_key="legacy_candidate_recheck:open",
                evidence_profile="candidate_recheck",
                subprofile="candidate_recheck",
            ),
            manifest_watermark=0,
        )
    finally:
        store.close()

    conn = sqlite3.connect(db_path)
    conn.execute(
        "UPDATE trigger_jobs SET status='pending', completed_at=NULL, error=NULL "
        "WHERE job_id=?",
        (job.job_id,),
    )
    conn.commit()
    conn.close()

    migrated_store = TriggerStore(db_path=db_path)
    try:
        migrated = migrated_store.get_job(job.job_id)
        assert migrated is not None
        assert migrated.status == "superseded"
        assert migrated.completed_at is not None
        assert migrated.error == (
            "candidate recheck retired; evolution candidates are audit-only"
        )
    finally:
        migrated_store.close()


def test_candidate_review_items_are_inspect_only(tmp_path: Path) -> None:
    evidence_store = EvidenceStore(tmp_path / "evidence.db")
    candidate_store = EvolutionCandidateStore(evidence_store=evidence_store)
    try:
        candidate = candidate_store.create_or_merge(
            _candidate_decision("decision_one", "Capture the same workflow"),
            _candidate_admission("admission_one"),
            _candidate_packet("task_one"),
        )
        audit = EvolutionAuditService(
            evidence_store,
            candidate_store=candidate_store,
        )

        items = audit.list_review_items()

        assert len(items) == 1
        assert items[0]["candidate_id"] == candidate.candidate_id
        assert items[0]["action_kind"] == "inspect"
        assert items[0]["approval_available"] is False
        assert "never auto-promotes" in items[0]["review_note"]
    finally:
        candidate_store.close()
        evidence_store.close()


def test_single_observation_capture_is_admitted_as_provisional_by_default() -> None:
    packet = _candidate_packet("task_one")
    decision = _candidate_decision(
        "decision_one",
        "Capture this reusable multi-step workflow for future tasks",
    )

    result = EvolutionAdmission().admit(decision, packet)

    assert result.outcome == "direct"
    assert "single_observation_allowed" in result.warnings
    assert result.source_validation_passed is True


def test_stale_repeated_metadata_does_not_override_single_source_task() -> None:
    packet = _candidate_packet("task_one")
    selected_refs = dict(packet.selected_refs)
    selected_refs["evolution_candidate_ref"] = [
        ResourceRef(
            ref_id="candidate:stale",
            ref_type="evolution_candidate_ref",
            task_id="task_one",
            preview="stale repeated candidate",
            metadata={
                "candidate_id": "stale",
                "recurrence": "repeated",
                "recurrence_count": 2,
                "source_task_ids": ["task_one"],
            },
        )
    ]
    packet = EvidencePacket(
        packet_id=packet.packet_id,
        trigger_job_id=packet.trigger_job_id,
        packet_type=packet.packet_type,
        profile_name=packet.profile_name,
        subprofile=packet.subprofile,
        manifest_watermark=packet.manifest_watermark,
        scope=packet.scope,
        selected_refs=selected_refs,
        expanded_snippets=packet.expanded_snippets,
        readable_paths=packet.readable_paths,
        instructions=packet.instructions,
        budget=packet.budget,
        redaction_status=packet.redaction_status,
        build_status=packet.build_status,
        missing_ref_types=packet.missing_ref_types,
    )

    result = EvolutionAdmission(
        allow_single_observation_capture=False
    ).admit(
        _candidate_decision(
            "decision_one",
            "Capture this reusable multi-step workflow for future tasks",
        ),
        packet,
    )

    assert result.outcome == "candidate"
    assert "provisional_evolution_disabled" in result.warnings


def test_only_structured_manual_evidence_bypasses_disabled_provisional_policy() -> None:
    packet = _candidate_packet("task_one")
    decision = _candidate_decision(
        "decision_one",
        "Capture this explicit reusable multi-step workflow for future tasks",
    )
    admission = EvolutionAdmission(allow_single_observation_capture=False)

    text_only_result = admission.admit(decision, packet)

    assert text_only_result.outcome == "candidate"
    selected_refs = dict(packet.selected_refs)
    selected_refs["manual_request_ref"] = [
        ResourceRef(
            ref_id="manual:capture",
            ref_type="manual_request_ref",
            task_id="task_one",
            preview="Capture this workflow",
        )
    ]
    manual_packet = EvidencePacket(
        packet_id="packet_manual",
        trigger_job_id=packet.trigger_job_id,
        packet_type=packet.packet_type,
        profile_name=packet.profile_name,
        subprofile=packet.subprofile,
        manifest_watermark=packet.manifest_watermark,
        scope=packet.scope,
        selected_refs=selected_refs,
        expanded_snippets=packet.expanded_snippets,
        readable_paths=packet.readable_paths,
        instructions=packet.instructions,
        budget=packet.budget,
        redaction_status=packet.redaction_status,
        build_status=packet.build_status,
        missing_ref_types=packet.missing_ref_types,
    )

    manual_result = admission.admit(decision, manual_packet)

    assert manual_result.outcome == "direct"
    assert "provisional_evolution_disabled" not in manual_result.warnings


def test_candidate_identity_keeps_unrelated_workflows_separate(tmp_path: Path) -> None:
    store = EvolutionCandidateStore(tmp_path / "evidence.db")
    try:
        packet = _candidate_packet("task_one")
        store.create_or_merge(
            _candidate_decision("decision_one", "Bound large shell output"),
            _candidate_admission("admission_one"),
            packet,
        )
        store.create_or_merge(
            _candidate_decision("decision_two", "Decode constants from an ELF"),
            _candidate_admission("admission_two"),
            packet,
        )

        candidates = store.list_candidates()
        assert len(candidates) == 2
        assert {candidate.recurrence_count for candidate in candidates} == {1}
    finally:
        store.close()


def test_candidate_store_repairs_legacy_false_recurrence(tmp_path: Path) -> None:
    db_path = tmp_path / "evidence.db"
    store = EvolutionCandidateStore(db_path)
    try:
        candidate = store.create_or_merge(
            _candidate_decision("decision_one", "Capture the same workflow"),
            _candidate_admission("admission_one"),
            _candidate_packet("task_one"),
        )
    finally:
        store.close()

    conn = sqlite3.connect(db_path)
    conn.execute(
        "UPDATE evolution_candidates SET recurrence='repeated', recurrence_count=2 "
        "WHERE candidate_id=?",
        (candidate.candidate_id,),
    )
    conn.commit()
    conn.close()

    repaired_store = EvolutionCandidateStore(db_path)
    try:
        repaired = repaired_store.load_candidate(candidate.candidate_id)
        assert repaired is not None
        assert repaired.recurrence == "single"
        assert repaired.recurrence_count == 1
    finally:
        repaired_store.close()