server/tasks.py

"""
Task definitions for the Git Conflict Resolver OpenEnv environment.

Each task contains:
- name, difficulty, description
- filename + file_language
- conflicted_content: the file the agent receives (with markers)
- expected_resolved: ground truth resolved file
- branch_ours / branch_theirs: branch names for context
- num_conflicts: number of conflict blocks
"""

from typing import Any, Dict

TASKS: Dict[str, Dict[str, Any]] = {

    # ─────────────────────────────────────────────────────────────────
    # TASK 1 — Easy: Single conflict block, obvious from context
    # ─────────────────────────────────────────────────────────────────
    "single_conflict": {
        "name": "single_conflict",
        "difficulty": "easy",
        "filename": "config.py",
        "file_language": "python",
        "branch_ours": "main",
        "branch_theirs": "feature/update-timeout",
        "num_conflicts": 1,
        "description": (
            "You are resolving a Git merge conflict in config.py. "
            "The file has ONE conflict block. "
            "The 'feature/update-timeout' branch changed the REQUEST_TIMEOUT from 30 to 60 seconds. "
            "The correct resolution is to accept the incoming change (theirs) — use 60 seconds. "
            "Return the full file with NO conflict markers."
        ),
        "conflicted_content": '''\
# config.py
# Application configuration

APP_NAME = "MyService"
APP_VERSION = "2.1.0"
DEBUG = False

# Network settings
HOST = "0.0.0.0"
PORT = 8080

<<<<<<< HEAD
REQUEST_TIMEOUT = 30  # seconds
=======
REQUEST_TIMEOUT = 60  # seconds — updated for slow upstream APIs
>>>>>>> feature/update-timeout

MAX_RETRIES = 3
LOG_LEVEL = "INFO"
''',
        "expected_resolved": '''\
# config.py
# Application configuration

APP_NAME = "MyService"
APP_VERSION = "2.1.0"
DEBUG = False

# Network settings
HOST = "0.0.0.0"
PORT = 8080

REQUEST_TIMEOUT = 60  # seconds — updated for slow upstream APIs

MAX_RETRIES = 3
LOG_LEVEL = "INFO"
''',
    },

    # ─────────────────────────────────────────────────────────────────
    # TASK 2 — Medium: Three conflict blocks, each needs a different resolution
    # ─────────────────────────────────────────────────────────────────
    "multi_conflict": {
        "name": "multi_conflict",
        "difficulty": "medium",
        "filename": "user_service.py",
        "file_language": "python",
        "branch_ours": "main",
        "branch_theirs": "feature/auth-refactor",
        "num_conflicts": 3,
        "description": (
            "You are resolving a Git merge conflict in user_service.py with THREE conflict blocks. "
            "Resolution rules based on comments and logic: "
            "1) Keep the new import from 'theirs' (bcrypt replaces hashlib). "
            "2) Keep 'ours' version of MAX_LOGIN_ATTEMPTS = 5 (theirs changed to 3, but product decided 5). "
            "3) Keep 'theirs' version of the hash_password function (uses bcrypt, more secure). "
            "Return the complete resolved file with NO conflict markers."
        ),
        "conflicted_content": '''\
# user_service.py
# Handles user authentication and management

<<<<<<< HEAD
import hashlib
=======
import bcrypt
>>>>>>> feature/auth-refactor

import logging
from typing import Optional

logger = logging.getLogger(__name__)

<<<<<<< HEAD
MAX_LOGIN_ATTEMPTS = 5
=======
MAX_LOGIN_ATTEMPTS = 3
>>>>>>> feature/auth-refactor

SESSION_EXPIRY_HOURS = 24


def get_user(user_id: str) -> Optional[dict]:
    """Fetch user record from database."""
    # Implementation omitted for brevity
    return None


<<<<<<< HEAD
def hash_password(password: str) -> str:
    """Hash a password using SHA-256."""
    return hashlib.sha256(password.encode()).hexdigest()
=======
def hash_password(password: str) -> str:
    """Hash a password using bcrypt (more secure)."""
    salt = bcrypt.gensalt()
    return bcrypt.hashpw(password.encode(), salt).decode()
>>>>>>> feature/auth-refactor


def verify_password(password: str, hashed: str) -> bool:
    """Verify a password against its hash."""
    return hash_password(password) == hashed
''',
        "expected_resolved": '''\
# user_service.py
# Handles user authentication and management

import bcrypt

import logging
from typing import Optional

logger = logging.getLogger(__name__)

MAX_LOGIN_ATTEMPTS = 5

SESSION_EXPIRY_HOURS = 24


def get_user(user_id: str) -> Optional[dict]:
    """Fetch user record from database."""
    # Implementation omitted for brevity
    return None


def hash_password(password: str) -> str:
    """Hash a password using bcrypt (more secure)."""
    salt = bcrypt.gensalt()
    return bcrypt.hashpw(password.encode(), salt).decode()


def verify_password(password: str, hashed: str) -> bool:
    """Verify a password against its hash."""
    return hash_password(password) == hashed
''',
    },

    # ─────────────────────────────────────────────────────────────────
    # TASK 3 — Hard: Must COMBINE code from both branches (not pick one)
    # ─────────────────────────────────────────────────────────────────
    "logic_conflict": {
        "name": "logic_conflict",
        "difficulty": "hard",
        "filename": "data_pipeline.py",
        "file_language": "python",
        "branch_ours": "main",
        "branch_theirs": "feature/add-validation",
        "num_conflicts": 2,
        "description": (
            "You are resolving a Git merge conflict in data_pipeline.py with TWO conflict blocks. "
            "IMPORTANT: Both sides have valid, additive changes — you must COMBINE them, not pick one. "
            "Block 1: 'ours' adds retry logic, 'theirs' adds timeout parameter — keep BOTH. "
            "Block 2: 'ours' filters out None values, 'theirs' also strips whitespace from strings — keep BOTH operations. "
            "Return the complete resolved file with NO conflict markers."
        ),
        "conflicted_content": '''\
# data_pipeline.py
# Data ingestion and transformation pipeline

import time
import logging
from typing import List, Any, Optional

logger = logging.getLogger(__name__)


<<<<<<< HEAD
def fetch_data(url: str, retries: int = 3) -> dict:
    """Fetch data from an API endpoint with retry logic."""
    for attempt in range(retries):
        try:
            # Simulate HTTP request
            response = make_request(url)
            return response
        except Exception as e:
            if attempt == retries - 1:
                raise
            logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying...")
            time.sleep(2 ** attempt)
=======
def fetch_data(url: str, timeout: int = 10) -> dict:
    """Fetch data from an API endpoint with configurable timeout."""
    response = make_request(url, timeout=timeout)
    return response
>>>>>>> feature/add-validation


<<<<<<< HEAD
def clean_records(records: List[Any]) -> List[Any]:
    """Remove None values from records."""
    return [r for r in records if r is not None]
=======
def clean_records(records: List[Any]) -> List[Any]:
    """Strip whitespace from string fields in records."""
    cleaned = []
    for r in records:
        if isinstance(r, str):
            cleaned.append(r.strip())
        else:
            cleaned.append(r)
    return cleaned
>>>>>>> feature/add-validation


def process_batch(records: List[Any]) -> List[Any]:
    """Process a batch of records through the pipeline."""
    cleaned = clean_records(records)
    return cleaned
''',
        "expected_resolved": '''\
# data_pipeline.py
# Data ingestion and transformation pipeline

import time
import logging
from typing import List, Any, Optional

logger = logging.getLogger(__name__)


def fetch_data(url: str, retries: int = 3, timeout: int = 10) -> dict:
    """Fetch data from an API endpoint with retry logic and configurable timeout."""
    for attempt in range(retries):
        try:
            # Simulate HTTP request
            response = make_request(url, timeout=timeout)
            return response
        except Exception as e:
            if attempt == retries - 1:
                raise
            logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying...")
            time.sleep(2 ** attempt)


def clean_records(records: List[Any]) -> List[Any]:
    """Remove None values and strip whitespace from string fields."""
    cleaned = []
    for r in records:
        if r is None:
            continue
        if isinstance(r, str):
            cleaned.append(r.strip())
        else:
            cleaned.append(r)
    return cleaned


def process_batch(records: List[Any]) -> List[Any]:
    """Process a batch of records through the pipeline."""
    cleaned = clean_records(records)
    return cleaned
''',
    },
}


def get_task(task_name: str) -> Dict[str, Any]:
    if task_name not in TASKS:
        raise ValueError(f"Unknown task '{task_name}'. Available: {list(TASKS.keys())}")
    return TASKS[task_name]


def list_tasks() -> list:
    return list(TASKS.keys())