Autonomous Goal Loops for AI Agents with MCP

Reactive ReAct Pattern (Single Turn / Fragile):
User Prompt ──▶ LLM Reasoning ──▶ Single MCP Tool Call ──▶ Human Inspection ──▶ Manual Re-prompt

Autonomous /goal Execution Harness:
┌────────────────────────────────────────────────────────────────────────┐
│                        OBJECTIVE SPECIFICATION                         │
│  Declarative Target + Invariant Constraints + Acceptance Predicates    │
└───────────────────────────────────┬────────────────────────────────────┘
                                    │
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│                         GOAL RUNTIME CONTROLLER                        │
│                                                                        │
│   ┌────────────────────┐     ┌───────────────────┐                     │
│   │ Hierarchical Task  │────▶│ Directed Acyclic  │                     │
│   │ Network (HTN)      │     │ Task Graph (DAG)  │                     │
│   └────────────────────┘     └─────────┬─────────┘                     │
│                                        │                               │
│                                        ▼                               │
│   ┌───────────────────────────────────────────────────────────────┐    │
│   │                     ITERATIVE EXECUTION LOOP                  │    │
│   │                                                               │    │
│   │   [Select Node] ──▶ [Compile Context] ──▶ [Dispatch MCP Tools]│    │
│   │         ▲                                        │            │    │
│   │         │                                        ▼            │    │
│   │   [Re-plan / Backoff] ◀── [Evaluate] ◀── [Observe Results]    │    │
│   └────────────────────────────────────┬──────────────────────────┘    │
│                                        │                               │
│                                        ▼                               │
│   ┌───────────────────────────────────────────────────────────────┐    │
│   │               PERSISTENT ATOMIC CHECKPOINT STORE              │    │
│   │            (State Hashes, AST Diffs, Disk Rollbacks)          │    │
│   └───────────────────────────────────────────────────────────────┘    │
└───────────────────────────────────┬────────────────────────────────────┘
                                    │
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│                    DETERMINISTIC VERIFICATION GATE                     │
│      Test Suite Pass + Static Analysis Clean + AST Contract Match      │
└────────────────────────────────────────────────────────────────────────┘

{
  "goal_id": "goal-auth-migration-8841",
  "objective": "Migrate Passport.js local strategy to Lucia Auth with Argon2id password hashing",
  "acceptance_criteria": [
    "lucia.config.ts exports initialized auth instance",
    "All authentication routes under /api/auth pass integration tests",
    "Zero occurrences of passport in package.json and src/",
    "npm run build succeeds with zero TypeScript diagnostics"
  ],
  "tasks": [
    {
      "id": "task-01",
      "title": "Audit dependency tree and uninstall Passport",
      "dependencies": [],
      "mcp_tools": ["ripgrep_search", "filesystem_replace", "bash_execute"],
      "verification": "npm ls passport returns non-zero"
    },
    {
      "id": "task-02",
      "title": "Install Lucia and Argon2 packages",
      "dependencies": ["task-01"],
      "mcp_tools": ["bash_execute"],
      "verification": "node -e 'require(\"lucia\")' returns 0"
    },
    {
      "id": "task-03",
      "title": "Generate Lucia core schema and adapter",
      "dependencies": ["task-02"],
      "mcp_tools": ["filesystem_write", "filesystem_view"],
      "verification": "test -f src/lib/auth.ts"
    },
    {
      "id": "task-04",
      "title": "Rewrite login and register route controllers",
      "dependencies": ["task-03"],
      "mcp_tools": ["filesystem_replace", "filesystem_view", "ripgrep_search"],
      "verification": "npm run test:auth"
    },
    {
      "id": "task-05",
      "title": "Full suite regression test and lint check",
      "dependencies": ["task-04"],
      "mcp_tools": ["bash_execute"],
      "verification": "npm run test && npm run lint"
    }
  ]
}

       ┌────────────────────────┐
       │      INITIALIZING      │
       └───────────┬────────────┘
                   │ Load Goal Spec & Scan Repo
                   ▼
       ┌────────────────────────┐
       │      DECOMPOSING       │
       └───────────┬────────────┘
                   │ Generate HTN Graph & Acceptance Criteria
                   ▼
┌──────▶┌────────────────────────┐
│       │      TASK_PENDING      │
│       └───────────┬────────────┘
│                   │ Select Next Unblocked DAG Node
│                   ▼
│       ┌────────────────────────┐
│       │       EXECUTING        │◀─────────────────────────┐
│       └───────────┬────────────┘                          │
│                   │ Tool Dispatch (MCP Calls)             │ Tool Error /
│                   ▼                                       │ Partial Failure
│       ┌────────────────────────┐                          │
│       │       EVALUATING       │                          │
│       └───────────┬────────────┘                          │
│                   │ Run Deterministic Assertion           │
│         ┌─────────┴─────────┐                             │
│         │ Pass              │ Fail (< Max Retries)        │
│         ▼                   ▼                             │
│ ┌───────────────┐   ┌───────────────┐                     │
│ │ STATE_COMMIT  │   │ REPAIR_LOOP   │─────────────────────┘
│ └───────┬───────┘   └───────┬───────┘
│         │                   │ Fail (>= Max Retries)
│         │ All Tasks Done    ▼
│         │           ┌───────────────┐
│         │           │   ESCALATED   │──▶ (Human Intervention)
│         ▼           └───────────────┘
│ ┌───────────────┐
└─│ LOOP_COMPLETE │
  └───────────────┘

import hashlib
import json
import os
import subprocess
import time

class GoalCheckpointManager:
    def __init__(self, workspace_path: str, goal_id: str):
        self.workspace_path = workspace_path
        self.goal_id = goal_id
        self.checkpoint_dir = os.path.join(workspace_path, ".goals", goal_id)
        os.makedirs(self.checkpoint_dir, exist_ok=True)

    def create_checkpoint(self, step_idx: int, task_id: str, state_metadata: dict) -> str:
        """Creates a git tree snapshot and serializes state metadata."""
        # Create lightweight git tree reference
        tree_sha = subprocess.check_output(
            ["git", "write-tree"], cwd=self.workspace_path, text=True
        ).strip()

        checkpoint_data = {
            "step_index": step_idx,
            "task_id": task_id,
            "timestamp": time.time(),
            "git_tree_sha": tree_sha,
            "metadata": state_metadata
        }

        filename = f"checkpoint_step_{step_idx:04d}_{task_id}.json"
        filepath = os.path.join(self.checkpoint_dir, filename)

        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(checkpoint_data, f, indent=2)

        return filepath

    def rollback_to_checkpoint(self, checkpoint_path: str):
        """Rolls back the working directory to the checkpoint's git tree."""
        with open(checkpoint_path, "r", encoding="utf-8") as f:
            data = json.load(f)

        tree_sha = data["git_tree_sha"]
        subprocess.check_call(
            ["git", "read-tree", tree_sha], cwd=self.workspace_path
        )
        subprocess.check_call(
            ["git", "checkout-index", "-a", "-f"], cwd=self.workspace_path
        )

┌────────────────────────────────────────────────────────┐
│                   TIER 1: WORKING STATE LEDGER         │
│  - Active Goal Objective & Constraints                │
│  - Completed Tasks vs. Pending Tasks                   │
│  - Modified Files Manifest (Current Paths & Hashes)    │
│  Token Cost: ~800 tokens (Constant Across Turns)       │
└───────────────────────────────────┬────────────────────┘
                                    │
                                    ▼
┌────────────────────────────────────────────────────────┐
│              TIER 2: SLIDING CONVERSATION WINDOW       │
│  - Current Subtask Instructions                        │
│  - Last 3 Tool Calls & Raw Observations                │
│  Token Cost: ~3,000 - 6,000 tokens (Bounded)           │
└───────────────────────────────────┬────────────────────┘
                                    │
                                    ▼
┌────────────────────────────────────────────────────────┐
│              TIER 3: EPHEMERAL SCRATCHPAD (MCP)        │
│  - Raw Compiler Logs, AST Dumps, Test Stack Traces     │
│  - Stored in Local Memory MCP Server / Temp Files      │
│  - LLM receives only synthesized extracts              │
│  Token Cost: 0 tokens in main prompt                   │
└────────────────────────────────────────────────────────┘

#!/usr/bin/env python3
"""
Autonomous Goal Loop Engine with Model Context Protocol (MCP) Integration.
Features:
- Structured Goal Specification
- Step-by-Step Task Loop
- Deterministic Tool Verification
- Self-Correction Backoff & Rollback
"""

import os
import sys
import json
import time
import subprocess
from typing import Dict, List, Any, Optional

class MCPToolClient:
    """Simulated MCP Client connecting to stdio MCP Servers."""
    def __init__(self, workspace_root: str):
        self.workspace_root = workspace_root

    def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Dispatch tool calls to MCP backend handlers."""
        if tool_name == "read_file":
            filepath = os.path.join(self.workspace_root, arguments["path"])
            if not os.path.exists(filepath):
                return {"error": f"File not found: {arguments['path']}"}
            with open(filepath, "r", encoding="utf-8") as f:
                return {"content": f.read()}

        elif tool_name == "write_file":
            filepath = os.path.join(self.workspace_root, arguments["path"])
            os.makedirs(os.path.dirname(filepath), exist_ok=True)
            with open(filepath, "w", encoding="utf-8") as f:
                f.write(arguments["content"])
            return {"status": "success", "bytes_written": len(arguments["content"])}

        elif tool_name == "run_test_assertion":
            command = arguments["command"]
            proc = subprocess.run(
                command,
                shell=True,
                cwd=self.workspace_root,
                capture_output=True,
                text=True
            )
            return {
                "exit_code": proc.returncode,
                "stdout": proc.stdout[-2000:],  # Bound token size
                "stderr": proc.stderr[-2000:]
            }
        else:
            return {"error": f"Unknown MCP tool: {tool_name}"}

class GoalTask:
    def __init__(self, task_id: str, description: str, verification_cmd: str):
        self.task_id = task_id
        self.description = description
        self.verification_cmd = verification_cmd
        self.status = "PENDING"  # PENDING, IN_PROGRESS, COMPLETED, FAILED
        self.retries = 0
        self.max_retries = 3

class GoalLoopOrchestrator:
    def __init__(self, goal_name: str, workspace_root: str, tasks: List[GoalTask]):
        self.goal_name = goal_name
        self.workspace_root = workspace_root
        self.tasks = tasks
        self.mcp = MCPToolClient(workspace_root)
        self.iteration_count = 0
        self.max_iterations = 25

    def verify_task_completion(self, task: GoalTask) -> bool:
        """Executes the deterministic verification predicate."""
        print(f"  🔍 Verifying task '{task.task_id}' using: {task.verification_cmd}")
        result = self.mcp.call_tool("run_test_assertion", {"command": task.verification_cmd})
        passed = (result.get("exit_code") == 0)
        if passed:
            print(f"  ✅ Verification PASSED for {task.task_id}")
        else:
            print(f"  ❌ Verification FAILED for {task.task_id} (exit code: {result.get('exit_code')})")
            print(f"     stderr: {result.get('stderr', '').strip()[:200]}")
        return passed

    def mock_agent_reasoning_and_action(self, task: GoalTask) -> Dict[str, Any]:
        """
        Simulates model generating actions based on current subtask.
        In production, this queries your LLM endpoint (Gemini / Claude / Codex)
        passing current workspace state and tool definitions.
        """
        # Emulating autonomous action dispatch based on task
        if "create auth config" in task.description.lower():
            return {
                "tool": "write_file",
                "args": {
                    "path": "src/config/auth.json",
                    "content": '{"provider": "lucia", "algorithm": "argon2id", "version": 2}'
                }
            }
        elif "create test suite" in task.description.lower():
            return {
                "tool": "write_file",
                "args": {
                    "path": "tests/test_auth_config.py",
                    "content": (
                        "import json, os\n"
                        "def test_auth():\n"
                        "    with open('src/config/auth.json') as f:\n"
                        "        d = json.load(f)\n"
                        "    assert d['provider'] == 'lucia'\n"
                        "if __name__ == '__main__':\n"
                        "    test_auth()\n"
                        "    print('ALL_PASSED')\n"
                    )
                }
            }
        return {"tool": "run_test_assertion", "args": {"command": "echo 'Noop action'"}}

    def run(self) -> bool:
        """Executes the main autonomous goal loop."""
        print(f"\n=======================================================")
        print(f"🚀 INITIATING AUTONOMOUS GOAL LOOP: {self.goal_name}")
        print(f"Workspace: {self.workspace_root}")
        print(f"Total Subtasks: {len(self.tasks)}")
        print(f"=======================================================\n")

        for task in self.tasks:
            task.status = "IN_PROGRESS"
            converged = False

            while not converged:
                self.iteration_count += 1
                if self.iteration_count > self.max_iterations:
                    print(f"🚨 GOAL ABORTED: Exceeded maximum global iterations ({self.max_iterations}).")
                    return False

                print(f"\n[Loop Step {self.iteration_count}] Working on '{task.task_id}': {task.description}")

                # 1. Synthesize Action Plan via LLM
                action = self.mock_agent_reasoning_and_action(task)
                tool_name = action["tool"]
                tool_args = action["args"]

                print(f"  ⚡ Dispatching MCP Tool: {tool_name}")
                tool_result = self.mcp.call_tool(tool_name, tool_args)

                # 2. Run Deterministic Verification
                passed = self.verify_task_completion(task)

                if passed:
                    task.status = "COMPLETED"
                    converged = True
                else:
                    task.retries += 1
                    print(f"  ⚠️ Step attempt {task.retries}/{task.max_retries} failed.")
                    if task.retries >= task.max_retries:
                        print(f"🚨 CRITICAL ERROR: Task '{task.task_id}' exceeded max retries. Halting goal loop.")
                        task.status = "FAILED"
                        return False
                    print(f"  🔄 Engaging automated repair loop for '{task.task_id}'...")

        print(f"\n=======================================================")
        print(f"🎉 GOAL ACHIEVED: All {len(self.tasks)} subtasks converged successfully.")
        print(f"Total Iterations: {self.iteration_count}")
        print(f"=======================================================\n")
        return True

if __name__ == "__main__":
    # Test execution in scratch directory
    workspace = os.path.abspath("./.goal_test_env")
    os.makedirs(workspace, exist_ok=True)

    test_tasks = [
        GoalTask(
            task_id="TASK-01",
            description="Create auth config file with Lucia provider",
            verification_cmd=f"python -c \"import json; d=json.load(open('{workspace}/src/config/auth.json')); assert d['provider']=='lucia'\""
        ),
        GoalTask(
            task_id="TASK-02",
            description="Create test suite and verify execution",
            verification_cmd=f"python {workspace}/tests/test_auth_config.py"
        )
    ]

    engine = GoalLoopOrchestrator(
        goal_name="Deploy-Lucia-Auth-System",
        workspace_root=workspace,
        tasks=test_tasks
    )

    success = engine.run()
    sys.exit(0 if success else 1)