Local SE Agent Masterclass

import json
from pathlib import Path
from typing import Dict, Any, List, Optional

class SkillRegistry:
    """
    ศูนย์ลงทะเบียนและสแกนค้นหาตลับสกิล (Skill Registry)
    ทำหน้าที่สแกนไดเรกทอรี skills/ รวบรวม Metadata และสร้าง Catalog สารบัญย่อ
    """
    def __init__(self, skills_dir: Optional[str] = None):
        # 1. กำหนด Root Directory ของโฟลเดอร์ skills
        self.skills_dir = Path(skills_dir).resolve() if skills_dir else Path(__file__).resolve().parent
        # เก็บข้อมูลสารบัญ: key = skill_id, value = dict(metadata, folder_path)
        self.registry: Dict[str, Dict[str, Any]] = {}
        # ดำเนินการสแกนอัตโนมัติทันทีที่สร้าง Object
        self.discover_skills()

    def discover_skills(self) -> int:
        """
        สแกนโฟลเดอร์ย่อยใน skills/ โดยอัตโนมัติ (Auto-Discovery)
        :return: จำนวนสกิลที่ลงทะเบียนสำเร็จ
        """
        self.registry.clear()
        if not self.skills_dir.exists():
            return 0

        # วนลูปตรวจสอบทุกไดเรกทอรีย่อยใน skills_dir
        for folder in self.skills_dir.iterdir():
            if folder.is_dir():
                meta_file = folder / "metadata.json"
                if meta_file.exists():
                    try:
                        metadata = json.loads(meta_file.read_text(encoding="utf-8"))
                        skill_id = metadata.get("id", folder.name)
                        
                        self.registry[skill_id] = {
                            "metadata": metadata,
                            "folder_path": folder,
                            "instructions_file": folder / "instructions.md"
                        }
                    except Exception as err:
                        print(f"⚠️ คำเตือน: ไม่สามารถอ่าน metadata ของสกิล '{folder.name}': {err}")

        return len(self.registry)

    def get_catalog_summary(self) -> str:
        """
        [Stage 1 Low-Footprint Catalog]:
        สร้างสารบัญย่อสำหรับนำไปใส่ใน System Prompt (~20-25 tokens ต่อสกิล)
        """
        if not self.registry:
            return "No specialized skills available."

        catalog_lines = ["AVAILABLE SKILLS CATALOG:"]
        for skill_id, info in self.registry.items():
            meta = info["metadata"]
            name = meta.get("name", skill_id)
            desc = meta.get("description", "")
            catalog_lines.append(f"- [{skill_id}] {name}: {desc}")

        catalog_lines.append(
            "\nWhen a specialized task matching a skill arrives, the operational manual will be dynamically expanded."
        )
        return "\n".join(catalog_lines)

    def find_skill_by_intent(self, user_prompt: str) -> Optional[str]:
        """
        ค้นหา skill_id ที่สอดคล้องกับเจตนาของผู้ใช้จาก triggers หรือ keywords
        """
        prompt_lower = user_prompt.lower()
        for skill_id, info in self.registry.items():
            triggers = info["metadata"].get("triggers", [])
            for trigger in triggers:
                if trigger.lower() in prompt_lower:
                    return skill_id
        return None

    def get_skill_info(self, skill_id: str) -> Optional[Dict[str, Any]]:
        """ดึงข้อมูลสารบัญของสกิลตาม ID"""
        return self.registry.get(skill_id)

from typing import Optional, Dict, Any
from skills.registry import SkillRegistry

class SkillLifecycleManager:
    """
    ผู้จัดการวงจรชีวิตของทักษะ (Skill Lifecycle Manager)
    ควบคุมการโหลดคู่มือตัวเต็ม (Activation) และการปลดคืนพื้นที่ (Eviction)
    ยึดหลัก One-Active-Skill Policy เพื่อประหยัด RAM Mac M3 16GB
    """
    def __init__(self, registry: SkillRegistry):
        self.registry = registry
        self.active_skill_id: Optional[str] = None
        self.active_instructions: Optional[str] = None

    def activate_skill(self, skill_id: str) -> str:
        """
        [Stage 2 On-Demand Expansion]:
        เปิดใช้งานสกิล อ่าน instructions.md เข้าสู่ Working Memory และปลดสกิลเก่าออกทันที
        :return: เนื้อหาคู่มือ instructions.md
        """
        skill_info = self.registry.get_skill_info(skill_id)
        if not skill_info:
            raise ValueError(f"ไม่พบสกิล ID: '{skill_id}' ใน Registry")

        # ปลดสกิลเก่าออกก่อนเสมอ (Evict Previous Skill)
        if self.active_skill_id and self.active_skill_id != skill_id:
            self.evict_active_skill()

        inst_path = skill_info["instructions_file"]
        if inst_path.exists():
            self.active_instructions = inst_path.read_text(encoding="utf-8")
        else:
            self.active_instructions = f"Specialized mode: {skill_id}"

        self.active_skill_id = skill_id
        return self.active_instructions

    def evict_active_skill(self) -> None:
        """
        [Context Eviction]:
        ปลดคู่มือของสกิลที่ Active อยู่ปัจจุบันออกจาก Working Memory ทันที คืนความสดใหม่ให้ Context
        """
        self.active_skill_id = None
        self.active_instructions = None

    def get_active_context_injection(self) -> str:
        """
        ส่งคืนข้อความคู่มือเฉพาะกิจสำหรับนำไปฉีดเข้า Context Window ชั่วคราว
        """
        if not self.active_skill_id or not self.active_instructions:
            return ""

        return (
            f"\n\n--- [ACTIVE SPECIALIZED SKILL: {self.active_skill_id}] ---\n"
            f"{self.active_instructions}\n"
            f"--- [END SPECIALIZED MANUAL] ---\n"
        )

    def is_skill_active(self, skill_id: str) -> bool:
        """ตรวจสอบว่าสกิลเป้าหมายกำลังอยู่ในสถานะ Active หรือไม่"""
        return self.active_skill_id == skill_id

"""
Package บริหารจัดการ Pluggable Skills และ Dynamic Two-Stage Disclosure
[อัปเดต Day 16]: เพิ่ม SkillRegistry และ SkillLifecycleManager
"""
from skills.registry import SkillRegistry
from skills.lifecycle import SkillLifecycleManager

__all__ = ["SkillRegistry", "SkillLifecycleManager"]

import sys
from pathlib import Path
from skills.registry import SkillRegistry
from skills.lifecycle import SkillLifecycleManager

def main():
    print("=" * 70)
    print("🚀 เริ่มทดสอบ Day 16: Building the Dynamic Skill Engine ใน Python")
    print("=" * 70)

    # 1. ทดสอบ SkillRegistry Auto-Discovery
    skills_folder = Path("./skills").resolve()
    registry = SkillRegistry(skills_dir=str(skills_folder))
    discovered_count = len(registry.registry)

    print(f"\n📦 1. ตรวจพบตลับสกิลทั้งหมด: {discovered_count} สกิล")
    for s_id, info in registry.registry.items():
        print(f"   👉 [{s_id}] - {info['metadata']['name']}")

    assert discovered_count >= 2, "❌ ล้มเหลว: ต้องพบตลับสกิลอย่างน้อย 2 สกิลจาก Day 15!"
    print("✅ Auto-Discovery ผ่านฉลุย: สแกนโฟลเดอร์พบอัตโนมัติ 100%!")

    # 2. ทดสอบการสร้าง Stage 1 Catalog Summary
    print("\n" + "-" * 50)
    print("📋 2. ตรวจสอบสารบัญ Catalog ย่อสำหรับ System Prompt:")
    print("-" * 50)
    catalog = registry.get_catalog_summary()
    print(catalog)
    print("-" * 50)
    assert "unit_test_synthesizer" in catalog
    assert "git_release_manager" in catalog
    print("✅ Stage 1 Catalog สร้างสำเร็จ: กระชับ ไม่กิน Token เริ่มต้น!")

    # 3. ทดสอบการจับคู่ Intent (Intent Matching)
    test_user_prompt = "ช่วยเขียน junit test สำหรับ UserService ให้หน่อย"
    matched_skill = registry.find_skill_by_intent(test_user_prompt)
    print(f"\n💬 คำสั่งผู้ใช้: \"{test_user_prompt}\"")
    print(f"🎯 สกิลที่ตรงกับเจตนา: {matched_skill}")
    assert matched_skill == "unit_test_synthesizer"
    print("✅ Intent Matching สำเร็จ: ตรวจจับ Trigger 'junit' ตรงเผง!")

    # 4. ทดสอบ SkillLifecycleManager: Activation (Stage 2 Expansion)
    lifecycle = SkillLifecycleManager(registry)
    print(f"\n⚡ 3. สั่ง Activate สกิล: '{matched_skill}'")
    instructions = lifecycle.activate_skill(matched_skill)
    assert lifecycle.is_skill_active("unit_test_synthesizer")
    print(f"📜 ตัวอย่างคู่มือที่โหลดเข้า Memory (บางส่วน):\n{instructions[:200]}...")
    print("✅ Activation สำเร็จ: ดึง instructions.md เข้า Working Memory เรียบร้อย!")

    # 5. ทดสอบการฉีด Context Injection
    injection = lifecycle.get_active_context_injection()
    assert "ACTIVE SPECIALIZED SKILL: unit_test_synthesizer" in injection
    print("✅ Context Injection สำเร็จ: เตรียมพร้อมส่งต่อเข้า ReAct Loop ได้ทันที!")

    # 6. ทดสอบการสลับสกิลและการ Evict (Context Eviction)
    print(f"\n🔄 4. จำลองงานใหม่: สลับไปใช้ 'git_release_manager'")
    lifecycle.activate_skill("git_release_manager")
    assert not lifecycle.is_skill_active("unit_test_synthesizer"), "❌ สกิลเก่าต้องถูก Evict ออก!"
    assert lifecycle.is_skill_active("git_release_manager"), "❌ สกิลใหม่ต้อง Active!"
    print("✅ สลับสกิลสำเร็จ: คู่มือเก่าถูกถอดออกทันทีตามกฎ One-Active-Skill Policy!")

    # 7. ทดสอบการปลดคืนพื้นที่สมบูรณ์แบบ
    lifecycle.evict_active_skill()
    assert lifecycle.active_skill_id is None
    assert lifecycle.get_active_context_injection() == ""
    print("✅ Eviction สำเร็จ: เคลียร์ Working Memory คืนความสะอาดให้ Context 100%!")

    print("\n" + "=" * 70)
    print("🎉 ผ่านการทดสอบเครื่องยนต์ Dynamic Skill Engine ครบทุกมิติ 100%!")
    print("🏆 สถาปัตยกรรม Microkernel ของคุณพร้อมสำหรับ Day 17: Developing Your First Enterprise Skill!")
    print("=" * 70)

if __name__ == "__main__":
    main()