Task Routing & Scheduling

async def web_search(
    self, 
    query: Annotated[str, "Query for general web search"]
) -> str:
    """Perform general web search for real-time information"""
    return await self.repo_searcher.deep_search(query)

# Automatically detect repository type
if repo_type is None:
    if repository.startswith(('http://', 'https://')) and 'github.com' in repository:
        repo_type = 'github'
    elif os.path.exists(repository):
        repo_type = 'local'
    else:
        # Try to determine if it's a GitHub URL format
        if repository.startswith(('http://', 'https://')) or repository.count('/') >= 1:
            repo_type = 'github'
        else:
            raise ValueError(f"Unable to determine repository type: {repository}")

def run_general_code_assistant(
    self,
    task_description: Annotated[str, "Programming task"],
    work_directory: Annotated[Optional[str], "Working directory"] = None
):
    """Provide general programming assistance"""
    explorer = CodeExplorer(
        local_repo_path=None,  # No repository
        work_dir=work_dir,
        task_type="general",
        use_venv=True,
        is_cleanup_venv=False
    )
    
    result = asyncio.run(explorer.a_code_analysis(enhanced_task, max_turns=20))
    return result

@staticmethod
def initialize_tasks(args, root_path='coding'):
    """Initialize task environment and task list"""
    # Create unique working directory
    work_task_path = PathManager.create_unique_path(root_path)
    
    # Extract task configuration
    task_info = args.config_data
    task_id = "repo_master"
    
    # Build output configuration
    out_task_info = {
        'repo': task_info['repo'],
        'task_description': task_info['task_description'],
        'task_prompt': task_info.get('task_prompt', 
                                     TaskManager.get_task_prompt()),
        'input_data': task_info['input_data'],
        'parameters': task_info.get('parameters', {}),
        'root_path': root_path,
        'work_task_path': work_task_path,
        'task_id': task_id
    }
    
    return out_task_info

def get_task_prompt():
    return """### Task Description
{task_description}

#### Repository Path (Absolute Path): 

Understanding Guide: ['Read README.md to understand basic functionality']

#### File Paths
- Input file paths and descriptions:
{input_data}

- Output file directory: 
Results must be saved in the {output_dir_path} directory.

#### Additional Notes
**Core Objective**: Quickly understand and analyze the code repository, 
generate and execute necessary code to complete user-specified tasks.
"""

placeholders = {
    '{repo_path}': target_repo_path,
    '{input_data}': target_input_data,
    '{output_dir_path}': target_output_path,
    '{task_description}': task_info.get('task_description', '')
}
for placeholder, value in placeholders.items():
    task_desc = task_desc.replace(placeholder, value)

elif repo_type == 'github':
    repo_url = repo_info['url']
    repo_name = repo_url.split('/')[-1].replace('.git', '')
    target_repo_path = f"{work_dir}/{repo_name}"
    
    if not os.path.exists(target_repo_path):
        clone_cmd = f"git clone {repo_url} {target_repo_path}"
        subprocess.run(clone_cmd, shell=True, check=True)

if repo_type == 'local':
    source_repo_path = repo_info['path']
    repo_name = Path(source_repo_path).name
    target_repo_path = f"{work_dir}/{repo_name}"
    
    if not os.path.exists(target_repo_path):
        os.system(f"cp -a {source_repo_path} {target_repo_path}")

data_info = task_info.get('input_data', [])
try:
    new_data_info = []
    for data in data_info:
        data_path = data.get('path')
        data_desc = data.get('description')
        if data_path and os.path.exists(data_path):
            new_data_path = DataProcessor.copy_dataset(
                data_path, target_input_path
            )
            new_data_info.append({
                'path': new_data_path,
                'description': data_desc
            })
except Exception as e:
    print(f"Error occurred while copying dataset: {e}")

1. Repository Search: Use github_repo_search to find relevant repositories
2. Sequential Execution: Select most promising repository
3. Result Evaluation: Critically evaluate if result satisfies requirements
4. Switching: If failed, select next best repository and retry
5. Continue: Until success or all options exhausted

# 1. Search for repositories
repo_list = github_repo_search(task)

# 2. Try each repository sequentially
for repo in repo_list:
    result = run_repository_agent(
        task_description=task,
        repository=repo['repo_url']
    )
    
    # 3. Evaluate result
    if result_satisfies_requirements(result):
        return result
    
    # 4. Try next repository
    continue

@staticmethod
def run_agent(task_info, retry_times=2, work_dir=None):
    """Run Code Agent to execute tasks"""
    task_id = task_info['task_id']
    work_task_path = task_info['work_task_path']
    
    # Create working directory
    work_dir = work_dir if work_dir else f'{work_task_path}/{task_id}/workspace'
    os.makedirs(work_dir, exist_ok=True)
    
    # Prepare environment
    target_output_path, target_input_data, target_repo_path = \
        DataProcessor.setup_task_environment(task_info, work_dir)
    
    # Generate task description
    task = TaskManager.prepare_task_description(
        task_info, target_output_path, target_input_data, target_repo_path
    )
    
    # Run code agent
    explorer = CodeExplorer(
        target_repo_path,
        work_dir=work_dir,
        task_type="gitbench",
        use_venv=True,
        is_cleanup_venv=False
    )
    
    answer = asyncio.run(explorer.a_code_analysis(task, max_turns=20))
    
    # Retry logic
    if not os.path.exists(target_output_path) and retry_times > 0:
        return AgentRunner.run_agent(task_info, retry_times - 1)
    
    return answer

@staticmethod
def generate_task_id():
    """Generate random task ID"""
    date_str = datetime.now().strftime("%m%d_%H%M")
    return f'gitbench_{date_str}'

@staticmethod
def create_unique_dir(base_path, prefix):
    """Create a unique directory that doesn't exist"""
    path = f'{base_path}/{prefix}'
    while os.path.exists(path):
        path = f'{path}_{random.randint(1, 10)}'
    os.makedirs(path, exist_ok=True)
    return path

if not os.path.exists(target_output_path) and retry_times > 0:
    print(f"---Task {task_id} submission failed, retrying {retry_times} times---")
    return AgentRunner.run_agent(task_info, retry_times - 1)

3. Sequential Execution and Fallback:
   - If one approach doesn't yield a solution, consider trying an 
     alternative mode (e.g., switching to General Code Assistant Mode)
   - Be persistent in finding a solution
   - If one repository doesn't work, try another

task_trace_dir = f"res/trace/code_analysis_{self.task_id}"

if os.path.exists(self.work_dir):
    with open(f"{self.work_dir}/trace_{timestamp}.txt", "w") as f:
        f.write(json.dumps(messages, ensure_ascii=False, indent=2))

def _extract_final_answer(self, chat_result) -> str:
    """Extract final answer from chat history"""
    final_answer = chat_result.summary
    
    if isinstance(final_answer, dict):
        final_answer = final_answer['content']
    
    if final_answer is None:
        final_answer = ""
    final_answer = final_answer.strip().lstrip()
    
    # Fallback to last message content
    messages = chat_result.chat_history
    final_content = messages[-1].get("content", "")
    
    if final_answer == "":
        final_answer = final_content
    
    return final_answer

coding/
└── gitbench_0304_1423/
    └── repo_master/
        └── workspace/
            ├── repository_name/     # Cloned/copied repository
            ├── input_dataset/       # Input data files
            ├── output_result/       # Generated outputs
            ├── task_info.json       # Task configuration
            └── trace_*.txt          # Conversation history