Local SE Agent Masterclass

{
  "model_name": "qwen2.5-coder:7b",
  "max_iterations": 6,
  "memory_window": 6,
  "temperature": 0.1,
  "auto_backup": true,
  "audit_logging": true
}

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

class WorkspaceManager:
    """
    ผู้จัดการสภาพแวดล้อมโปรเจกต์ (Workspace Manager)
    ตรวจหา Root Directory ของโปรเจกต์ที่ผู้ใช้กำลังเปิดอยู่ และจัดเก็บบันทึก Audit Trail
    """
    ROOT_INDICATORS = ["pom.xml", "package.json", "build.gradle", "pyproject.toml", "Cargo.toml", ".git"]

    @classmethod
    def find_project_root(cls, start_path: Optional[str] = None) -> Path:
        """
        ไต่ระดับโฟลเดอร์ย้อนกลับขึ้นไป (Upward Traversal) เพื่อหา Root Directory ที่แท้จริง
        """
        current = Path(start_path).resolve() if start_path else Path.cwd().resolve()
        
        # วนลูปตรวจสอบ current และ parent directories
        for directory in [current] + list(current.parents):
            for indicator in cls.ROOT_INDICATORS:
                if (directory / indicator).exists():
                    return directory
                    
        # หากไม่พบ indicator ใดๆ เลย ให้ใช้ไดเรกทอรีปัจจุบัน
        return current

    @classmethod
    def load_config(cls, agent_home: Path) -> Dict[str, Any]:
        """โหลดไฟล์คอนฟิกจากโฟลเดอร์ติดตั้งหลักของ Agent"""
        config_file = agent_home / "config.json"
        default_config = {
            "model_name": "qwen2.5-coder:7b",
            "max_iterations": 6,
            "memory_window": 6,
            "temperature": 0.1,
            "auto_backup": true,
            "audit_logging": true
        }
        if config_file.exists():
            try:
                data = json.loads(config_file.read_text(encoding="utf-8"))
                default_config.update(data)
            except Exception:
                pass
        return default_config

    @classmethod
    def log_session(cls, workspace_root: Path, task: str, final_answer: str, steps: int) -> None:
        """บันทึกประวัติการทำงานลงไฟล์ .agent_history.jsonl ในโปรเจกต์"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "workspace": str(workspace_root),
            "task": task,
            "steps": steps,
            "summary": final_answer[:300]
        }
        log_path = workspace_root / ".agent_history.jsonl"
        try:
            with open(log_path, "a", encoding="utf-8") as f:
                f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
        except Exception:
            pass

import sys
import argparse
from pathlib import Path
from core import (
    OllamaLLMClient,
    ToolDispatcher,
    ReActAgentEngine,
    MaxIterationsReachedError,
    TerminalUI
)
from core.workspace import WorkspaceManager
from core.tools import read_file, run_maven_test, apply_code_patch, rollback_file

AGENT_HOME = Path(__file__).resolve().parent

DEFAULT_SYSTEM_PROMPT = (
    "You are an Autonomous Senior Software Engineering Agent running locally on Apple Silicon Mac.\n"
    "You operate in a strict ReAct loop (Reason, Act, Observe).\n"
    "You have access to tools to inspect local code, run Maven tests, and safely apply refactoring patches.\n\n"
    "Available Tools:\n"
    "1. read_file(filepath: string): Reads file content safely from workspace.\n"
    "2. run_maven_test(project_dir: string): Runs Maven unit tests for a Java project.\n"
    "3. apply_code_patch(filepath: string, new_code: string): Safely applies code patch with automatic backup and diff.\n"
    "4. rollback_file(filepath: string): Instantly restores original file from .bak backup.\n\n"
    "Response Rules:\n"
    "- When taking action, reply strictly with JSON:\n"
    "{\n"
    '  "thought": "Your reasoning why you take this step",\n'
    '  "tool": "tool_name",\n'
    '  "arguments": { ... }\n'
    "}\n\n"
    "- When task is completed, reply strictly with JSON:\n"
    "{\n"
    '  "thought": "Summary of achievements",\n'
    '  "final_answer": "Comprehensive conclusion and explanation for human developer"\n'
    "}\n\n"
    "Respond ONLY with valid JSON."
)

def create_agent(config: dict, workspace_root: Path, ui: TerminalUI) -> ReActAgentEngine:
    """Factory Function: ประกอบร่างระบบและล็อกขอบเขตความปลอดภัยของ Workspace"""
    with ui.status(f"กำลังเชื่อมต่อ Ollama Daemon และตรวจสอบโมเดล '{config['model_name']}'..."):
        client = OllamaLLMClient(model_name=config["model_name"])
        if not client.check_health():
            ui.print_error(f"ไม่พบโมเดล '{config['model_name']}' หรือ Ollama Daemon ไม่ได้เปิดทำงาน!\nกรุณารันคำสั่ง 'ollama serve' หรือ 'ollama list'")
            sys.exit(1)

    dispatcher = ToolDispatcher(workspace_root=str(workspace_root))
    dispatcher.register("read_file", lambda filepath: read_file(filepath, workspace_root=str(workspace_root)))
    dispatcher.register("run_maven_test", lambda project_dir=".": run_maven_test(project_dir=str(workspace_root / project_dir)))
    dispatcher.register("apply_code_patch", lambda filepath, new_code: apply_code_patch(filepath, new_code, workspace_root=str(workspace_root)))
    dispatcher.register("rollback_file", lambda filepath: rollback_file(filepath, workspace_root=str(workspace_root)))

    return ReActAgentEngine(
        client=client,
        dispatcher=dispatcher,
        max_iterations=config["max_iterations"],
        memory_window=config["memory_window"]
    )

def run_task(agent: ReActAgentEngine, task: str, workspace_root: Path, config: dict, ui: TerminalUI) -> None:
    ui.console.print(f"\n[bold cyan]🎯 ปฏิบัติการในโฟลเดอร์:[/bold cyan] [underline white]{workspace_root}[/underline white]")
    ui.console.print(f"[bold cyan]💬 ภารกิจ:[/bold cyan] [italic white]\"{task}\"[/italic white]\n" + "─" * 60)
    try:
        final_answer = agent.run(user_prompt=task, system_prompt=DEFAULT_SYSTEM_PROMPT)
        ui.print_final_answer(final_answer)
        if config.get("audit_logging", True):
            WorkspaceManager.log_session(workspace_root, task, final_answer, steps=len(agent.messages))
    except MaxIterationsReachedError as err:
        ui.print_error(f"Guardrail Alert: {err}")
    except KeyboardInterrupt:
        ui.console.print("\n[yellow]⚠️ ผู้ใช้สั่งยกเลิกงานกลางคัน (Ctrl+C)[/yellow]")
    except Exception as err:
        ui.print_error(f"เกิดข้อผิดพลาดไม่คาดคิด: {err}")

def main():
    parser = argparse.ArgumentParser(description="Local SE Agent Personal Developer Workspace")
    parser.add_argument("task_pos", nargs="?", type=str, help="คำสั่งงานด่วน เช่น: se-agent 'ตรวจบั๊กในโปรเจกต์'")
    parser.add_argument("-t", "--task", type=str, help="คำสั่งงานเดี่ยว")
    parser.add_argument("-i", "--interactive", action="store_true", help="เปิดโหมด Interactive REPL Prompt")
    parser.add_argument("-w", "--workspace", type=str, help="กำหนด Workspace Root ด้วยตนเอง")

    args = parser.parse_args()
    ui = TerminalUI()

    # 1. โหลดคอนฟิก
    config = WorkspaceManager.load_config(AGENT_HOME)
    
    # 2. ค้นหา Project Root อัตโนมัติจากโฟลเดอร์ที่ผู้ใช้สั่งรัน
    workspace_root = Path(args.workspace).resolve() if args.workspace else WorkspaceManager.find_project_root()

    # 3. แสดง Banner
    ui.print_banner(model_name=config["model_name"], max_iterations=config["max_iterations"])

    # 4. สร้าง Agent
    agent = create_agent(config=config, workspace_root=workspace_root, ui=ui)

    task = args.task_pos or args.task
    if task:
        run_task(agent, task, workspace_root, config, ui)
    else:
        ui.console.print(f"\n[bold magenta]💬 เข้าสู่โหมด Interactive Shell[/bold magenta] (พิกัด: [underline white]{workspace_root}[/underline white])")
        ui.console.print("[slate-500]พิมพ์ 'exit' หรือ 'quit' เพื่อออก[/slate-500]\n" + "─" * 60)
        while True:
            user_input = ui.prompt_user()
            if not user_input:
                continue
            if user_input.lower() in ["exit", "quit", "q"]:
                ui.console.print("\n[bold green]👋 ลาก่อนครับ! ขอบคุณที่ใช้งาน Local SE Agent Workspace[/bold green]")
                break
            run_task(agent, user_input, workspace_root, config, ui)

if __name__ == "__main__":
    main()

#!/usr/bin/env bash
# ค้นหาพิกัดที่แท้จริงของโฟลเดอร์ se_agent บนเครื่อง Mac
AGENT_DIR="$HOME/se_agent"
VENV_PYTHON="$AGENT_DIR/.venv/bin/python3"

if [ ! -f "$VENV_PYTHON" ]; then
    echo "❌ ไม่พบ Virtual Environment ที่ $VENV_PYTHON"
    echo "👉 กรุณารัน: cd ~/se_agent && python3 -m venv .venv && pip install -r requirements.txt"
    exit 1
fi

# รัน main.py พร้อมส่งต่อ Arguments ทั้งหมด ($@)
exec "$VENV_PYTHON" "$AGENT_DIR/main.py" "$@"