Building a Self-Managing Autonomous Agent Execution Loop

Implementing Autonomous Execution Mode - Self-Managing Agent

One of the most interesting features of Claude Control is autonomous execution mode — instead of a human entering prompts each time, the agent plans and executes tasks on its own. On February 8th, I implemented the core loop for this feature.

The Autonomous Execution Loop

The heart of autonomous execution is a "plan → execute → evaluate → repeat" loop.

import asyncio
from enum import Enum
from typing import Optional

class ExecutionState(str, Enum):
    PLANNING = "planning"
    EXECUTING = "executing"
    EVALUATING = "evaluating"
    COMPLETED = "completed"
    PAUSED = "paused"
    ERROR = "error"

class AutonomousExecutor:
    def __init__(self, session_manager, logger):
        self.session_manager = session_manager
        self.logger = logger
        self.state = ExecutionState.PLANNING
        self.max_iterations = 50
        self.timeout_seconds = 3600  # 1 hour

    async def run(self, session_id: str, task: str):
        iteration = 0
        start_time = asyncio.get_event_loop().time()

        await self.logger.log_system(session_id, f"자율 실행 시작: {task}")

        while iteration < self.max_iterations:
            elapsed = asyncio.get_event_loop().time() - start_time
            if elapsed > self.timeout_seconds:
                await self.logger.log_system(session_id, "타임아웃 도달")
                break

            self.state = ExecutionState.PLANNING
            plan = await self._create_plan(session_id, task, iteration)

            if plan.get("completed"):
                self.state = ExecutionState.COMPLETED
                break

            self.state = ExecutionState.EXECUTING
            result = await self._execute_step(session_id, plan["next_step"])

            self.state = ExecutionState.EVALUATING
            evaluation = await self._evaluate_result(session_id, result)

            if evaluation.get("needs_retry"):
                continue

            iteration += 1

        return {"iterations": iteration, "state": self.state.value}

The Planning Phase

At each iteration, the next step is determined based on progress so far.

async def _create_plan(self, session_id: str, task: str, iteration: int):
    context = await self._get_execution_context(session_id)
    planning_prompt = f"""현재 태스크: {task}
진행 반복: {iteration}
이전 결과: {context.get('last_result', '없음')}

다음에 수행할 단계를 JSON으로 출력하세요.
완료되었으면 {{"completed": true}}를 출력하세요."""

    response = await self.session_manager.send_prompt(
        session_id, planning_prompt
    )
    return json.loads(response)

Timeouts and Safety Guards

In autonomous execution, safety guards are paramount. You have to prevent the system from getting stuck in infinite loops or running dangerous commands.

BLOCKED_COMMANDS = ["rm -rf /", "DROP DATABASE", "format", "shutdown"]

async def _safety_check(self, command: str) -> bool:
    for blocked in BLOCKED_COMMANDS:
        if blocked.lower() in command.lower():
            await self.logger.log_error(
                self.current_session,
                f"차단된 명령 감지: {command}"
            )
            return False
    return True

Auto-Continue

I also added a feature that automatically resumes execution when a session stalls mid-task or a response gets cut off because it's too long.

async def _execute_with_auto_continue(self, session_id: str, prompt: str):
    full_response = ""
    while True:
        response = await self.session_manager.send_prompt(session_id, prompt)
        full_response += response
        if not response.endswith("[CONTINUE]"):
            break
        prompt = "계속"
    return full_response

Takeaways

Building autonomous execution mode showed me the potential of the "AI managing AI" pattern. It's not perfect yet, but it handles straightforward repetitive tasks well enough to automate. Without safety guards like timeouts and iteration limits, this would have been genuinely dangerous.