Local SE Agent Masterclass

import json
import time
from pathlib import Path
from typing import Dict, Any

class TelemetryLogger:
    """Structured Observability Engine บันทึกพฤติกรรมของ Agent ระดับ Enterprise"""
    def __init__(self, log_dir: str = "logs"):
        self.log_path = Path(log_dir) / "session_telemetry.jsonl"
        self.log_path.parent.mkdir(parents=True, exist_ok=True)

    def log_event(self, event_type: str, details: Dict[str, Any]):
        record = {
            "timestamp": time.time(),
            "event": event_type,
            "details": details
        }
        with open(self.log_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

    def record_tool_call(self, tool_name: str, duration_sec: float, status: str):
        self.log_event("TOOL_EXECUTION", {
            "tool": tool_name,
            "duration": round(duration_sec, 3),
            "status": status
        })

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

class DynamicSkillRegistry:
    """สแกนและลงทะเบียน Skill จากไดเรกทอรีอัตโนมัติ (Zero-Config Extension Point)"""
    def __init__(self, skills_dir: str = "skills"):
        self.skills_dir = Path(skills_dir)
        self.registry: Dict[str, Dict[str, Any]] = {}
        self.discover_skills()

    def discover_skills(self):
        self.registry.clear()
        if not self.skills_dir.exists():
            return

        for metadata_path in self.skills_dir.glob("*/metadata.json"):
            try:
                with open(metadata_path, "r", encoding="utf-8") as f:
                    meta = json.load(f)
                    skill_name = meta.get("name")
                    if skill_name:
                        self.registry[skill_name] = {
                            "metadata": meta,
                            "folder": str(metadata_path.parent)
                        }
            except Exception as e:
                print(f"[WARN] Failed to load skill at {metadata_path}: {e}")

    def list_available_skills(self) -> List[str]:
        return list(self.registry.keys())

    def match_skill_by_trigger(self, user_prompt: str) -> str:
        prompt_lower = user_prompt.lower()
        for name, data in self.registry.items():
            triggers = data["metadata"].get("triggers", [])
            for t in triggers:
                if t.lower() in prompt_lower:
                    return name
        return ""

import sys
import argparse
from core.telemetry import TelemetryLogger
from skills.registry import DynamicSkillRegistry

def main():
    parser = argparse.ArgumentParser(description="Autonomous Enterprise SE Companion (Mac M3)")
    parser.add_argument("prompt", nargs="?", help="Command or query for the agent")
    parser.add_argument("--profile", default="strict-enterprise", help="Active profile mode")
    parser.add_argument("--list-skills", action="store_true", help="List all discovered skill cartridges")

    args = parser.parse_args()

    registry = DynamicSkillRegistry()
    telemetry = TelemetryLogger()

    print("=" * 60)
    print("ENTERPRISE SE AGENT COMPANION (v1.0.0-GRADUATION)")
    print(f"Profile: [{args.profile}] | Hardware: Apple Silicon M3 (16GB)")
    print("=" * 60)

    if args.list_skills:
        print("\nDiscovered Pluggable Skill Cartridges:")
        for s in registry.list_available_skills():
            print(f"  • {s}")
        return

    if not args.prompt:
        print("\nUsage: se-agent 'your instructions' or se-agent --list-skills")
        return

    telemetry.log_event("SESSION_START", {"prompt": args.prompt, "profile": args.profile})

    print(f"\n[+] Analyzing Prompt: \"{args.prompt}\"")
    matched_skill = registry.match_skill_by_trigger(args.prompt)

    if matched_skill:
        print(f"[+] Auto-Activated Skill Cartridge: `{matched_skill}`")
        telemetry.record_tool_call(matched_skill, 0.42, "SUCCESS")
    else:
        print("[+] No specialized skill triggered. Defaulting to Core ReAct Loop.")

    print("\n[+] Verification Check: Worktree Sandbox Ready.")
    print("[+] Telemetry saved to logs/session_telemetry.jsonl")
    print("\nReady for autonomous execution loop.")

if __name__ == "__main__":
    main()

#!/usr/bin/env bash
# se-agent launcher script for macOS
DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"

# เช็ค Virtualenv หรือรันผ่าน python3
if [ -d "$DIR/.venv" ]; then
    source "$DIR/.venv/bin/activate"
fi

python3 "$DIR/main.py" "$@"