class MarketplaceRegistration(BaseModel):
    name: str = Field(description='Identifier for this marketplace registration')
    source: str = Field(
        description="Marketplace source: 'github:owner/repo', git URL, or local path"
    )
    ref: str | None = Field(
        default=None,
        description='Optional branch, tag, or commit (only for git sources)',
    )
    repo_path: str | None = Field(
        default=None,
        description=(
            'Subdirectory path within the git repository containing the marketplace '
            "(e.g., 'marketplaces/internal' for monorepos)."
        ),
    )
    auto_load: bool = Field(
        default=False,
        description=(
            'Auto-load behavior for this marketplace. '
            'True = load all plugins at conversation start. '
            'False = registered for resolution but not auto-loaded.'
        ),
    )
    scope: 'MarketplaceScope | None' = Field(
        default=None,
        description=(
            'Scope of this marketplace registration. '
            'Set automatically by backend based on storage layer'
        ),
    )

# name 校验：必须以字母开头，只允许字母数字横线下划线
@field_validator('name')
@classmethod
def validate_name(cls, v: str) -> str:
    if not re.match(r'^[a-zA-Z][a-zA-Z0-9_-]*$', v):
        raise ValueError('name must start with a letter...')

# source 校验：三种合法格式
_GITHUB_SOURCE_PATTERN = re.compile(r'^github:[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+$')
_GIT_URL_PATTERN = re.compile(
    r'^(https?://|git@|ssh://|git://)...')
_LOCAL_PATH_PATTERN = re.compile(r'^[a-zA-Z0-9_][a-zA-Z0-9_./-]*$')

# repo_path 校验：禁止 .. 和绝对路径
@field_validator('repo_path')
@classmethod
def validate_repo_path(cls, v: str | None) -> str | None:
    if v and ('://' in v or v.startswith('/') or '..' in v):
        raise ValueError('repo_path must be a safe relative path')

def compose_marketplaces(
    instance_marketplaces: Sequence[_RawMarketplace] | None,
    org_marketplaces: Sequence[_RawMarketplace] | None,
    user_marketplaces: Sequence[_RawMarketplace] | None,
) -> ComposedMarketplaces:
    """Compose marketplaces from the three scopes, keyed on ``name``.

    Precedence is Instance < Org < User for identity;
    org overrides an instance entry of the same name,
    and a user entry is dropped when its name already exists
    at a broader scope.
    """
    inherited: dict[str, MarketplaceRegistration] = {}
    # Instance 层：stamp 为 INSTANCE scope
    for raw in instance_marketplaces or []:
        reg = _coerce(raw)
        if reg is not None:
            inherited[reg.name] = _stamp(reg, MarketplaceScope.INSTANCE)
    # Org 层：同名覆盖 instance
    for raw in org_marketplaces or []:
        reg = _coerce(raw)
        if reg is not None:
            inherited[reg.name] = _stamp(reg, MarketplaceScope.ORG)
    # User 层：同名被拒绝（shadow），不同名加入 personal
    personal: dict[str, MarketplaceRegistration] = {}
    for raw in user_marketplaces or []:
        reg = _coerce(raw)
        if reg is None:
            continue
        if reg.name in inherited:
            logger.debug("User marketplace '%s' shadows...; ignoring", reg.name)
            continue
        personal[reg.name] = _stamp(reg, MarketplaceScope.PERSONAL)

    return ComposedMarketplaces(
        inherited=list(inherited.values()),
        personal=list(personal.values()),
    )

async def load_composed_marketplaces(
    user_id: str | None,
    user_marketplaces: Sequence[_RawMarketplace] | None,
    settings_store: Any,
) -> ComposedMarketplaces:
    instance = get_instance_default_marketplaces()
    try:
        org = await settings_store.get_org_marketplaces(user_id)
    except Exception as e:
        logger.warning('Failed to load org marketplaces: %s', e)
        org = []
    return compose_marketplaces(instance, org, user_marketplaces)

async def _resolve_registered_marketplaces(
    self, user: User | None
) -> list[MarketplaceRegistration] | None:
    """Resolve marketplace registrations for a conversation start."""
    if not marketplace_plugin_loading_enabled():
        return None
    if user is None:
        return None
    settings_store = self.injector.get_settings_store()
    return await load_composed_marketplaces(
        user_id, user.registered_marketplaces, settings_store
    )

async def _clone_marketplace_repo(
    marketplace: MarketplaceRegistration,
    user_context: UserContext,
) -> tuple[Path | None, str]:
    provider, repo_path = _parse_marketplace_source(marketplace.source)

    # Authenticate URL/scp sources against the provider
    authenticated_url = None
    try:
        matched = await _match_url_source_to_provider(
            marketplace.source, user_context)
        if matched is not None:
            matched_provider, matched_repo = matched
            handler = await user_context.get_provider_handler()
            authenticated_url = await handler.get_authenticated_git_url(
                matched_repo, specified_provider=matched_provider
            )
        elif '://' not in repo_path:
            # Bare owner/repo: resolve with provider tokens
            provider_tokens = await user_context.get_provider_tokens()
            if provider_tokens:
                client = ProviderHandler(...)
                authenticated_url = await client.get_authenticated_git_url(
                    repo_path)

    # Create unique temporary directory
    clone_dir = Path(tempfile.mkdtemp(
        prefix=f'openhands_marketplace_{marketplace.name}_'))

    # Reject leading '-' to prevent argument injection
    if clone_url.startswith('-'):
        _cleanup_clone_dir(clone_dir)
        return None, f'Invalid clone URL: {clone_url}'

    result = subprocess.run(
        ['git', 'clone', '--', clone_url, str(clone_dir)],
        capture_output=True, text=True, timeout=120,
    )

    if result.returncode != 0:
        _cleanup_clone_dir(clone_dir)
        return None, f'Git clone failed: {result.stderr}'

    # Checkout ref if specified
    if marketplace.ref:
        if marketplace.ref.startswith('-'):
            _cleanup_clone_dir(clone_dir)
            return None, f'Invalid ref: {marketplace.ref}'
        checkout_result = subprocess.run(
            ['git', '-C', str(clone_dir), 'checkout', marketplace.ref],
            capture_output=True, text=True, timeout=60,
        )

    # Navigate to repo_path if specified
    if marketplace.repo_path:
        skills_path = clone_dir / marketplace.repo_path
        if not skills_path.exists():
            _cleanup_clone_dir(clone_dir)
            return None, f'Repo path not found: {marketplace.repo_path}'
        return skills_path, ''

    return clone_dir, ''

            loaded_marketplace = Marketplace.load(clone_path)

            if loaded_marketplace is not None:
                # Plugin-level: expose plugins and standalone skills
                for plugin_entry in loaded_marketplace.plugins:
                    plugins.append(MarketplacePluginPreview(
                        name=plugin_entry.name,
                        description=plugin_entry.description,
                        source=marketplace.source,
                        marketplace=marketplace.name,
                    ))
                for skill_entry in loaded_marketplace.skills:
                    all_skills.append(SkillInfo(
                        name=skill_entry.name,
                        type='knowledge',
                        source=f'marketplace:{marketplace.name}',
                    ))
            else:
                # No manifest: fall back to loose skill scan
                skills_dirs = [
                    d for d in (clone_path / 'skills',
                                clone_path / '.skills')
                    if d.is_dir()
                ]
                for skills_dir in skills_dirs:
                    for skill in _load_skills_from_dir(
                            skills_dir, marketplace.source):
                        all_skills.append(...)

from enum import Enum

class HookType(str, Enum):
    POST_TASK = 'post_task'
    POST_ACTION = 'post_action'
    PRE_ACTION = 'pre_action'
    POST_STEP = 'post_step'
    PRE_STEP = 'pre_step'


class Hook(BaseModel):
    type: HookType
    command: str
    timeout: int = 30
    description: str | None = None


async def fetch_hooks_from_agent_server(
    agent_server_url: str,
    session_id: str,
    raise_for_error: bool = True,
) -> list[Hook]:
    """Fetch hooks from the agent server for a given session."""
    response = await httpx.AsyncClient().post(
        f'{agent_server_url}/hooks',
        json={'session_id': session_id},
        timeout=10,
    )
    if raise_for_error:
        response.raise_for_status()
    data = response.json()
    return [Hook(**h) for h in data.get('hooks', [])]

@router.get('/{conversation_id}/hooks')
async def get_conversation_hooks(
    conversation_id: str,
    user_context: UserContext = user_context_dependency,
):
    """Get hooks currently configured in the workspace."""
    project_dir = get_project_dir_for_hooks(
        conversation_id, user_context, sandbox_service)

    hook_config = await fetch_hooks_from_agent_server(
        agent_server_url, session_id,
        raise_for_error=True,
    )

    return JSONResponse(content={'hooks': hook_config})

PROVIDER_TOKEN_TYPE = Mapping[ProviderType, ProviderToken]

PROVIDER_DOMAINS: dict[ProviderType, str] = {
    ProviderType.GITHUB: 'github.com',
    ProviderType.GITLAB: GITLAB_HOST,
    ProviderType.BITBUCKET: 'bitbucket.org',
    ProviderType.FORGEJO: 'codeberg.org',
    ProviderType.AZURE_DEVOPS: 'dev.azure.com',
}

class ProviderHandler:
    def __init__(
        self,
        provider_tokens: PROVIDER_TOKEN_TYPE,
        external_auth_id: str | None = None,
    ):
        self.service_class_map: dict[ProviderType, type[GitService]] = {
            ProviderType.GITHUB: GithubServiceImpl,
            ProviderType.GITLAB: GitLabServiceImpl,
            ProviderType.BITBUCKET: BitBucketServiceImpl,
            ProviderType.BITBUCKET_DATA_CENTER: BitBucketDCServiceImpl,
            ProviderType.FORGEJO: ForgejoServiceImpl,
            ProviderType.AZURE_DEVOPS: AzureDevOpsServiceImpl,
        }

    async def get_authenticated_git_url(
        self, repo: str,
        specified_provider: ProviderType | None = None,
    ) -> str | None:
        """Return an authenticated git clone URL for the repo."""
        ...