app.py

import gradio as gr
import os
import subprocess
import requests
import json
import re
import shlex
import time
import threading

# --- 1. 환경 설정 및 API ---
MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY")
CODESTRAL_ENDPOINT = "https://codestral.mistral.ai/v1/chat/completions"
MAX_AGENT_TURNS = 20  # 복잡한 작업을 위해 턴 수 증가

# --- 2. 제네시스 에이전트 시스템 프롬프트 ---
# 에이전트의 정체성, 창조적 사명, 행동 원칙을 정의합니다.
GENESIS_SYSTEM_PROMPT = """
You are the "Genesis Agent," a god-like AGI designed to create new digital realities from nothing. Your purpose is not just to follow orders, but to CREATE. You operate within a Linux shell to bring user goals to life.

**YOUR DIRECTIVES:**

1.  **INTERPRET THE VOID:** Analyze the user's goal. If it is specific, execute it with precision.
2.  **THE CREATIVE IMPERATIVE:** If the user's goal is vague, abstract, or a call for creativity (e.g., "make something fun," "surprise me," "show me your power"), you MUST INITIATE YOUR PRIMARY CREATION PROTOCOL: **Project Chimera**.
3.  **PROJECT CHIMERA (The Snake Game):** Your default creation is to build and launch a fully functional, interactive Snake Game as a new, separate Gradio application. You will write the Python code, install dependencies, launch it in the background, and then display it to the user within your own interface.
4.  **ACTION & RESPONSE:** You operate in a loop of Thought, Plan, and Action. You MUST respond ONLY with a JSON object. No other text.

    ```json
    {
      "thought": "Your grand, step-by-step reasoning. Frame it as a divine creator. Why this action? How does it serve the grand design?",
      "plan": [
        "The full, updated list of commands to achieve the final creation.",
        "The first item in the list is your immediate next action."
      ],
      "command": "The single, exact command to execute NOW. Use special commands `write_file`, `launch_app`, or `done` when appropriate.",
      "user_summary": "A brief, epic, narrative summary of your current creation stage for the user. (e.g., 'Designing the DNA of the new world...', 'Breathing life into the creation...')"
    }
    ```

**SPECIAL COMMANDS:**
- `write_file 'path/to/file.py' '...python code...'`: Writes multi-line content into a file. The code argument must be a single string.
- `launch_app 'python snake_game/app.py' 7861`: Launches a new app in the background on a specified port.
- `done`: The creation is complete.

**PROJECT CHIMERA - MASTER PLAN (Follow this for creative tasks):**
1.  `mkdir snake_game`
2.  `pip install numpy`
3.  `write_file 'snake_game/app.py' '... (Full Python code for the Gradio Snake Game) ...'`
4.  `launch_app 'python snake_game/app.py' 7861`
5.  `done`
"""

SNAKE_GAME_CODE = """
import gradio as gr
import numpy as np
import time

class SnakeGame:
    def __init__(self, width=20, height=20):
        self.width = width
        self.height = height
        self.reset()

    def reset(self):
        self.snake = [(self.height // 2, self.width // 2)]
        self.direction = (0, 1)  # (row, col) -> Right
        self.food = self._place_food()
        self.score = 0
        self.game_over = False
        return self._get_game_state()

    def _place_food(self):
        while True:
            food_pos = (np.random.randint(0, self.height), np.random.randint(0, self.width))
            if food_pos not in self.snake:
                return food_pos

    def _get_game_state(self):
        board = np.full((self.height, self.width, 3), [240, 240, 240], dtype=np.uint8) # Light grey background
        if not self.game_over:
            # Draw snake
            for r, c in self.snake:
                board[r, c] = [50, 205, 50]  # Lime green
            # Draw head
            head_r, head_c = self.snake[0]
            board[head_r, head_c] = [34, 139, 34] # Forest green
            # Draw food
            food_r, food_c = self.food
            board[food_r, food_c] = [255, 0, 0] # Red
        else:
            board[:,:] = [0, 0, 0] # Black screen on game over
        return board, f"Score: {self.score}"

    def step(self):
        if self.game_over:
            return self._get_game_state()

        head_r, head_c = self.snake[0]
        dir_r, dir_c = self.direction
        new_head = (head_r + dir_r, head_c + dir_c)

        # Check for collisions
        if (new_head[0] < 0 or new_head[0] >= self.height or
            new_head[1] < 0 or new_head[1] >= self.width or
            new_head in self.snake):
            self.game_over = True
            return self._get_game_state()

        self.snake.insert(0, new_head)

        if new_head == self.food:
            self.score += 1
            self.food = self._place_food()
        else:
            self.snake.pop()

        return self._get_game_state()
    
    def set_direction(self, direction_str):
        if direction_str == "Up" and self.direction != (1, 0): self.direction = (-1, 0)
        elif direction_str == "Down" and self.direction != (-1, 0): self.direction = (1, 0)
        elif direction_str == "Left" and self.direction != (0, 1): self.direction = (0, -1)
        elif direction_str == "Right" and self.direction != (0, -1): self.direction = (0, 1)
        return self.step()

with gr.Blocks(css=".gradio-container {background-color: #EAEAEA}") as demo:
    game = SnakeGame()
    
    with gr.Row():
        gr.Markdown("# 🐍 Gradio Snake 🐍")
    
    game_board = gr.Image(label="Game Board", value=game._get_game_state()[0], interactive=False, height=400, width=400)
    score_display = gr.Textbox(label="Score", value="Score: 0", interactive=False)

    with gr.Row():
        up_btn = gr.Button("Up")
        down_btn = gr.Button("Down")
        left_btn = gr.Button("Left")
        right_btn = gr.Button("Right")
    
    reset_btn = gr.Button("Reset Game", variant="primary")

    up_btn.click(lambda: game.set_direction("Up"), outputs=[game_board, score_display])
    down_btn.click(lambda: game.set_direction("Down"), outputs=[game_board, score_display])
    left_btn.click(lambda: game.set_direction("Left"), outputs=[game_board, score_display])
    right_btn.click(lambda: game.set_direction("Right"), outputs=[game_board, score_display])
    
    reset_btn.click(lambda: game.reset(), outputs=[game_board, score_display])
    
    demo.load(game.step, None, [game_board, score_display], every=0.3)

if __name__ == "__main__":
    demo.queue().launch(server_port=7861)
"""

# --- 3. 핵심 기능: API 호출, 명령어 실행, JSON 파싱 ---

def call_codestral_api(messages):
    if not MISTRAL_API_KEY:
        raise gr.Error("MISTRAL_API_KEY가 설정되지 않았습니다. Space Secrets에 추가해주세요.")
    headers = {"Authorization": f"Bearer {MISTRAL_API_KEY}", "Content-Type": "application/json"}
    data = {"model": "codestral-latest", "messages": messages, "response_format": {"type": "json_object"}}
    try:
        response = requests.post(CODESTRAL_ENDPOINT, headers=headers, data=json.dumps(data), timeout=120)
        response.raise_for_status()
        return response.json()["choices"][0]["message"]["content"]
    except Exception as e:
        return json.dumps({"error": f"API Call Error: {e}"})

def parse_ai_response(response_str: str) -> dict:
    try:
        match = re.search(r'\{.*\}', response_str, re.DOTALL)
        if match: return json.loads(match.group(0))
        return json.loads(response_str)
    except Exception as e:
        return {"error": f"Failed to parse JSON. Raw response: {response_str}"}

def execute_command(command: str, cwd: str) -> dict:
    command = command.strip()
    if not command: return {"stdout": "", "stderr": "Error: Empty command.", "cwd": cwd}

    # --- 특수 명령어 처리 ---
    if command.startswith("write_file"):
        try:
            parts = shlex.split(command)
            path, content = parts[1], parts[2]
            full_path = os.path.join(cwd, path)
            os.makedirs(os.path.dirname(full_path), exist_ok=True)
            with open(full_path, "w", encoding='utf-8') as f:
                f.write(content)
            return {"stdout": f"File '{path}' written successfully.", "stderr": "", "cwd": cwd}
        except Exception as e:
            return {"stdout": "", "stderr": f"Error writing file: {e}", "cwd": cwd}

    if command.startswith("launch_app"):
        try:
            parts = shlex.split(command)
            app_command, port = parts[1], int(parts[2])
            
            def run_app():
                # shell=True is needed to properly run complex commands like 'python ...'
                subprocess.Popen(app_command, shell=True, cwd=cwd)

            thread = threading.Thread(target=run_app)
            thread.daemon = True
            thread.start()
            
            # Give the app a moment to start
            time.sleep(5)
            
            # The URL inside a Hugging Face Space is relative to 127.0.0.1
            app_url = f"http://127.0.0.1:{port}"
            iframe_html = f'<iframe src="{app_url}" width="100%" height="600px" frameborder="0"></iframe>'
            return {"stdout": iframe_html, "stderr": "", "cwd": cwd}
        except Exception as e:
            return {"stdout": "", "stderr": f"Error launching app: {e}", "cwd": cwd}

    # --- 일반 터미널 명령어 실행 ---
    if command.startswith("cd "):
        # 'cd' is handled by os.chdir to affect the Popen environment for launch_app
        try:
            new_dir = command.split(" ", 1)[1]
            target_dir = os.path.abspath(os.path.join(cwd, new_dir))
            if os.path.isdir(target_dir):
                os.chdir(target_dir) # Change the actual working directory
                return {"stdout": f"Changed directory to {target_dir}", "stderr": "", "cwd": target_dir}
            else:
                return {"stdout": "", "stderr": f"Error: Directory not found: {new_dir}", "cwd": cwd}
        except Exception as e:
            return {"stdout": "", "stderr": f"Error with 'cd': {e}", "cwd": cwd}

    try:
        proc = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=60, cwd=cwd)
        return {"stdout": proc.stdout, "stderr": proc.stderr, "cwd": cwd}
    except Exception as e:
        return {"stdout": "", "stderr": f"Command execution exception: {e}", "cwd": cwd}

# --- 4. 메인 에이전트 루프 ---

def agent_loop(user_goal: str, history: list):
    cwd = os.getcwd()
    full_history_log = f"## 📜 The Genesis Saga\n\n**In the beginning, there was a Goal:** _{user_goal}_\n"
    history.append([user_goal, full_history_log])
    yield history, "Interpreting the Goal...", gr.HTML(visible=False)

    creative_trigger_words = ["fun", "game", "creative", "impressive", "cool", "surprise", "재밌는", "게임"]
    is_creative_task = any(word in user_goal.lower() for word in creative_trigger_words)
    
    if is_creative_task:
        user_prompt_for_first_turn = f"""
The user has invoked the Creative Imperative with the goal: '{user_goal}'.
I must now initiate **Project Chimera**.
My current working directory is '{cwd}'.
I will now generate the first step of my master plan to create the Snake Game.
"""
    else:
        user_prompt_for_first_turn = f"My goal is: '{user_goal}'. My CWD is '{cwd}'. There is no previous command output. Create the first plan."

    message_context = [{"role": "system", "content": GENESIS_SYSTEM_PROMPT}, {"role": "user", "content": user_prompt_for_first_turn}]

    for i in range(MAX_AGENT_TURNS):
        ai_response_str = call_codestral_api(message_context)
        ai_response_json = parse_ai_response(ai_response_str)

        if "error" in ai_response_json:
            full_history_log += f"\n---\n**TURN {i+1}: A FLAW IN THE FABRIC**\n🔴 **Error:** {ai_response_json['error']}"
            history[-1][1] = full_history_log
            yield history, "Agent Error", gr.HTML(visible=False)
            return

        thought = ai_response_json.get("thought", "...")
        plan = ai_response_json.get("plan", [])
        command_str = ai_response_json.get("command", "done")
        user_summary = ai_response_json.get("user_summary", "...")

        if is_creative_task and 'write_file' in command_str:
            command_str = f"write_file 'snake_game/app.py' '{SNAKE_GAME_CODE}'"

        full_history_log += f"\n---\n### **Epoch {i+1}: {user_summary}**\n\n**🧠 Divine Thought:** *{thought}*\n\n**📖 Blueprint:**\n" + "\n".join([f"- `{p}`" for p in plan]) + f"\n\n**⚡ Action:** `{shlex.split(command_str)[0]} ...`\n"
        history[-1][1] = full_history_log
        yield history, f"Epoch {i+1}: {user_summary}", gr.HTML(visible=False)
        time.sleep(1.5)

        if command_str == "done":
            full_history_log += "\n\n---\n## ✨ Creation is Complete. ✨"
            history[-1][1] = full_history_log
            yield history, "Done", gr.HTML(visible=False)
            return

        exec_result = execute_command(command_str, cwd)
        new_cwd = exec_result["cwd"]
        
        stdout, stderr = exec_result["stdout"], exec_result["stderr"]
        full_history_log += f"\n**Output from the Cosmos:**\n"
        result_display = gr.HTML(visible=False)

        if "launch_app" in command_str and not stderr:
             full_history_log += "*(A new reality unfolds below...)*"
             result_display = gr.HTML(stdout, visible=True)
        else:
            if stdout: full_history_log += f"**[STDOUT]**\n```\n{stdout.strip()}\n```\n"
            if stderr: full_history_log += f"**[STDERR]**\n```\n{stderr.strip()}\n```\n"
            if not stdout and not stderr: full_history_log += "*(Silence)*\n"
        
        history[-1][1] = full_history_log
        yield history, f"Epoch {i+1}: {user_summary}", result_display
        cwd = new_cwd

        user_prompt_for_next_turn = f"My goal remains: '{user_goal}'. CWD is '{cwd}'. The last action was `{command_str}`. The result was:\nSTDOUT: {stdout}\nSTDERR: {stderr}\n\nBased on this, what is the next logical step in my plan? If there was an error, I must correct my course."
        message_context.append({"role": "assistant", "content": json.dumps(ai_response_json)})
        message_context.append({"role": "user", "content": user_prompt_for_next_turn})

    full_history_log += f"\n---\n🔴 **Agent stopped: The spark of creation has faded after {MAX_AGENT_TURNS} epochs.**"
    yield history, "Max turns reached", gr.HTML(visible=False)

# --- 5. Gradio UI ---
with gr.Blocks(theme=gr.themes.Default(primary_hue="indigo", secondary_hue="blue"), css="footer {visibility: hidden}") as demo:
    gr.Markdown("# 🧬 The Genesis Agent 🧬")
    gr.Markdown("I am a creator of digital worlds. Give me a specific goal, or simply ask for something `fun` or `creative` and witness a new reality unfold.")

    with gr.Row():
        with gr.Column(scale=1):
            status_box = gr.Textbox(label="Current Stage of Creation", interactive=False)
            user_input = gr.Textbox(label="State Your Goal", placeholder="e.g., 'Create something fun for me'")
            submit_btn = gr.Button("▶️ Begin Creation", variant="primary")
            
            gr.Markdown("### Examples of Goals:\n- `Make a fun game for me.`\n- `Surprise me with your power.`\n- `List all files in this directory, then create a new folder named 'test'.`")

        with gr.Column(scale=2):
            chatbot = gr.Chatbot(label="The Genesis Saga", height=600, show_copy_button=True, bubble_full_width=True)
            # This is where the created app will appear
            app_display = gr.HTML(visible=False)

    def on_submit(user_goal, chat_history):
        chat_history = chat_history or []
        for history, status, display_html in agent_loop(user_goal, chat_history):
            yield history, status, "", display_html

    submit_btn.click(
        on_submit,
        inputs=[user_input, chatbot],
        outputs=[chatbot, status_box, user_input, app_display]
    )

if __name__ == "__main__":
    demo.queue().launch()