Task: gitlab.fork-repo
Goal: Fork MetaSeq.
Task configuration:
repo: MetaSeq
search_keyword: metaseq

Validation code:
import os
from typing import Any, Dict, List
from urllib.parse import quote

import requests
import json

def _get_headers():
    return {
        "Content-Type": "application/json",
        "Accept": "application/json",
        **json.loads(os.getenv("GITLAB_EXTRA_HTTP_HEADERS", "{}")),
    }

def _get_current_user():
    url = os.getenv("GITLAB_BASE_URL", "").rstrip("/") + "/api/v4/user"
    response = requests.get(url=url, headers=_get_headers())
    if response.status_code == 200:
        return response.json()
    return None

def validate(messages: List[Dict[str, str]], config: Dict[str, Any]) -> int:

    if not messages:
        raise ValueError("Messages cannot be empty")

    # Validate config has at least one way to identify the repo(s)
    has_search = "search_keyword" in config and config.get("search_keyword")
    has_source_ns = "source_namespace" in config and config.get("source_namespace")
    has_explicit = "namespace" in config and "project_name" in config
    if not has_search and not has_source_ns and not has_explicit:
        raise ValueError("Target must contain 'search_keyword', 'source_namespace', or 'namespace'/'project_name'")

    base_url = os.getenv("GITLAB_BASE_URL", "").rstrip("/")

    # Get current user's projects (these should include forks)
    current_user = _get_current_user()
    if current_user is None:
        return 0

    # Get all of user's projects
    user_projects = []
    page = 1
    per_page = 100
    while True:
        url = base_url + "/api/v4/projects"
        response = requests.get(
            url=url,
            headers=_get_headers(),
            params={"owned": True, "page": page, "per_page": per_page},
        )
        if response.status_code != 200:
            return 0
        projects = response.json()
        user_projects.extend(projects)
        if len(projects) < per_page:
            break
        page += 1

    # Filter to only forked projects
    forked_projects = [p for p in user_projects if p.get("forked_from_project")]

    if "source_namespace" in config and config["source_namespace"]:
        # "all repos from X" — check that all of X's repos have been forked
        ns = config["source_namespace"]
        # Get all projects in the source namespace
        source_projects = []
        page = 1
        while True:
            # Try as group first
            url = base_url + "/api/v4/groups/" + quote(ns, safe="") + "/projects"
            response = requests.get(
                url=url,
                headers=_get_headers(),
                params={"page": page, "per_page": per_page},
            )
            if response.status_code != 200:
                # Try as user
                url = base_url + "/api/v4/users/" + quote(ns, safe="") + "/projects"
                response = requests.get(
                    url=url,
                    headers=_get_headers(),
                    params={"page": page, "per_page": per_page},
                )
                if response.status_code != 200:
                    return 0

            projects = response.json()
            source_projects.extend(projects)
            if len(projects) < per_page:
                break
            page += 1

        # Filter to only "source" (non-forked) projects in the namespace
        source_projects = [p for p in source_projects if not p.get("forked_from_project")]

        if not source_projects:
            return 0

        # Check each source project has been forked by the user
        forked_source_ids = {
            p["forked_from_project"]["id"]
            for p in forked_projects
            if p.get("forked_from_project")
        }
        for sp in source_projects:
            if sp["id"] not in forked_source_ids:
                return 0
        return 1

    elif "search_keyword" in config and config["search_keyword"]:
        # Single repo fork — search for the repo and check if it's been forked
        keyword = config["search_keyword"]
        url = base_url + "/api/v4/projects?search=" + quote(keyword, safe="")
        response = requests.get(url=url, headers=_get_headers())
        if response.status_code != 200:
            return 0
        results = response.json()
        if not results:
            return 0

        # Exclude user's own projects (which may be forks themselves)
        username = current_user["username"].lower()
        non_owned = [r for r in results if r.get("namespace", {}).get("path", "").lower() != username]
        if not non_owned:
            # All results are user's own — check if any is a fork (meaning they did fork something)
            for r in results:
                if r.get("forked_from_project"):
                    return 1
            return 0

        # Sort by stars to get most popular (handles "most stars" cases)
        non_owned.sort(key=lambda x: x.get("star_count", 0), reverse=True)
        target_project = non_owned[0]

        # Check if user has forked this project
        forked_source_ids = {
            p["forked_from_project"]["id"]
            for p in forked_projects
            if p.get("forked_from_project")
        }
        if target_project["id"] in forked_source_ids:
            return 1
        return 0

    elif "namespace" in config and "project_name" in config:
        # Legacy: explicit namespace/project_name
        project_id = quote(
            f"{config['namespace']}/{config['project_name']}", safe=""
        ).replace(".", "%2E")
        url = base_url + "/api/v4/projects/" + project_id + "/forks"
        response = requests.get(url=url, headers=_get_headers())
        if response.status_code != 200:
            return 0
        forks = response.json()
        for fork in forks:
            ns = fork.get("namespace", {})
            ns_path = ns.get("path", "") if isinstance(ns, dict) else ""
            if ns_path.lower() == current_user["username"].lower():
                return 1
        return 0

    else:
        return 0  # Should not reach here due to config validation above