from amplifier_collections import (
    CollectionResolver,
    discover_collection_resources,
    install_collection,
    CollectionLock,
)
from pathlib import Path
# Define search paths for your application
search_paths = [
    Path("/var/amplifier/system/collections"),  # Bundled (lowest)
    Path.home() / ".amplifier" / "collections",  # User
    Path(".amplifier/collections"),              # Project (highest)
]
# Create resolver
resolver = CollectionResolver(search_paths=search_paths)
# Resolve collection name to path
foundation_path = resolver.resolve("foundation")
# Discover resources
resources = discover_collection_resources(foundation_path)
print(f"Found {len(resources.profiles)} profiles")
print(f"Found {len(resources.agents)} agents")
# Install new collection
from amplifier_module_resolution import GitSource
lock = CollectionLock(lock_path=Path(".amplifier/collections.lock"))
source = GitSource("git+https://github.com/user/my-collection@v1.0.0")
await install_collection(
    source=source,
    target_dir=Path(".amplifier/collections/my-collection"),
    lock=lock
)

from amplifier_collections import CollectionMetadata
from pathlib import Path
class CollectionMetadata(BaseModel):
    """Collection metadata from pyproject.toml."""
    model_config = ConfigDict(frozen=True)
    # From [project] section
    name: str
    version: str
    description: str = ""
    # From [tool.amplifier.collection] section
    author: str = ""
    capabilities: list[str] = Field(default_factory=list)
    requires: dict[str, str] = Field(default_factory=dict)
    # From [project.urls] section
    homepage: str | None = None
    repository: str | None = None
# Load metadata from collection
metadata = CollectionMetadata.from_pyproject(
    Path("/path/to/collection/pyproject.toml")
)
print(f"{metadata.name} v{metadata.version}")
print(f"Author: {metadata.author}")
print(f"Capabilities: {', '.join(metadata.capabilities)}")

from amplifier_collections import CollectionResolver
from pathlib import Path
class CollectionResolver:
    """Resolve collection names to filesystem paths."""
    def __init__(self, search_paths: list[Path]):
        """Initialize with app-specific search paths.

        Args:
            search_paths: Paths to search in precedence order (lowest to highest)
        """
# Create resolver
resolver = CollectionResolver(search_paths=[...])
# Resolve collection name to path
path = resolver.resolve("foundation")
# Returns: Path | None (highest precedence match, or None if not found)
# List all available collections
collections = resolver.list_collections()
# Returns: list[tuple[str, Path]] with precedence resolved
# Example: [("foundation", Path(...)), ("custom", Path(...))]

from amplifier_collections import CollectionResources, discover_collection_resources
from pathlib import Path
class CollectionResources(BaseModel):
    """Discovered resources in a collection."""
    model_config = ConfigDict(frozen=True)
    profiles: list[Path] = Field(default_factory=list)
    agents: list[Path] = Field(default_factory=list)
    context: list[Path] = Field(default_factory=list)
    scenario_tools: list[Path] = Field(default_factory=list)
    modules: list[Path] = Field(default_factory=list)
# Discover resources by convention
resources = discover_collection_resources(Path("/path/to/collection"))
# Access discovered resources
for profile in resources.profiles:
    print(f"Profile: {profile.name}")
for agent in resources.agents:
    print(f"Agent: {agent.name}")

from amplifier_collections import list_profiles, list_agents
# Get profile names (without .md extension)
profile_names = list_profiles(Path("/path/to/collection"))
# Returns: ['base', 'foundation', 'production']
# Get agent names (without .md extension)
agent_names = list_agents(Path("/path/to/collection"))
# Returns: ['analyzer', 'optimizer']

from amplifier_collections import install_collection, CollectionLock, CollectionInstallError
from amplifier_module_resolution import GitSource
from pathlib import Path
# Create installation source
source = GitSource("git+https://github.com/user/collection@v1.0.0")
# Create lock file manager
lock = CollectionLock(lock_path=Path(".amplifier/collections.lock"))
# Install collection
try:
    metadata = await install_collection(
        source=source,
        target_dir=Path(".amplifier/collections/my-collection"),
        lock=lock
    )
    print(f"Installed {metadata.name} v{metadata.version}")
except CollectionInstallError as e:
    print(f"Installation failed: {e.message}")

from amplifier_collections import uninstall_collection
await uninstall_collection(
    collection_name="my-collection",
    collections_dir=Path(".amplifier/collections"),
    lock=lock
)

from amplifier_collections import CollectionLock, CollectionLockEntry
from pathlib import Path
class CollectionLock:
    """Manage collection lock file."""
    def __init__(self, lock_path: Path):
        """Initialize with lock file path."""
# Create lock manager
lock = CollectionLock(lock_path=Path(".amplifier/collections.lock"))
# Add entry (direct parameters - simpler than constructing object)
lock.add_entry(
    name="my-collection",
    source="git+https://github.com/user/my-collection@v1.0.0",
    commit="abc123def456",  # Git commit SHA (or None if not git)
    path=Path(".amplifier/collections/my-collection")
)
# Get entry
entry = lock.get_entry("my-collection")
if entry:
    print(f"Installed: {entry.installed_at}")
    print(f"Commit: {entry.commit}")
# List all entries
for entry in lock.list_entries():
    print(f"{entry.name} @ {entry.commit[:7]}")
# Remove entry
lock.remove_entry("my-collection")

from amplifier_collections import CollectionResolver, discover_collection_resources
from pathlib import Path
# CLI defines search paths
search_paths = [
    Path(__file__).parent / "data" / "collections",  # Bundled
    Path.home() / ".amplifier" / "collections",      # User
    Path(".amplifier/collections"),                   # Project
]
resolver = CollectionResolver(search_paths=search_paths)
# List available collections
for name, path in resolver.list_collections():
    metadata = CollectionMetadata.from_pyproject(path / "pyproject.toml")
    resources = discover_collection_resources(path)
    print(f"\n{name} v{metadata.version}")
    print(f"  Profiles: {len(resources.profiles)}")
    print(f"  Agents: {len(resources.agents)}")
    print(f"  Context files: {len(resources.context)}")
# Resolve specific collection
foundation = resolver.resolve("foundation")
if foundation:
    resources = discover_collection_resources(foundation)
    # Use resources...

from amplifier_collections import CollectionResolver, discover_collection_resources
class WebCollectionService:
    """Web service for collection management."""
    def __init__(self, workspace_id: str):
        # Web-specific paths
        self.search_paths = [
            Path("/var/amplifier/system/collections"),
            Path(f"/var/amplifier/workspaces/{workspace_id}/collections"),
        ]
        self.resolver = CollectionResolver(search_paths=self.search_paths)
    async def get_collection_info(self, name: str) -> dict:
        """Get collection information for API response."""
        path = self.resolver.resolve(name)
        if not path:
            return {"error": "Collection not found"}
        metadata = CollectionMetadata.from_pyproject(path / "pyproject.toml")
        resources = discover_collection_resources(path)
        return {
            "name": metadata.name,
            "version": metadata.version,
            "description": metadata.description,
            "author": metadata.author,
            "profiles": [p.stem for p in resources.profiles],
            "agents": [a.stem for a in resources.agents],
            "context_files": len(resources.context),
        }
    async def list_collections(self) -> list[dict]:
        """List all available collections."""
        results = []
        for name, path in self.resolver.list_collections():
            info = await self.get_collection_info(name)
            results.append(info)
        return results

from amplifier_collections import install_collection, CollectionLock, CollectionInstallError
import httpx
import zipfile
class HttpZipSource:
    """HTTP zip download installation source."""
    def __init__(self, url: str):
        self.url = url
    async def install_to(self, target_dir: Path) -> None:
        """Download and extract zip to target."""
        async with httpx.AsyncClient() as client:
            response = await client.get(self.url)
            response.raise_for_status()
            # Write to temporary file
            temp_zip = target_dir.parent / f"{target_dir.name}.zip"
            temp_zip.write_bytes(response.content)
            # Extract to target
            with zipfile.ZipFile(temp_zip) as zf:
                zf.extractall(target_dir)
            # Cleanup
            temp_zip.unlink()
# Use custom source
lock = CollectionLock(lock_path=Path(".amplifier/collections.lock"))
source = HttpZipSource("https://cdn.example.com/collections/my-collection-v1.0.0.zip")
await install_collection(
    source=source,
    target_dir=Path(".amplifier/collections/my-collection"),
    lock=lock
)

from amplifier_collections import CollectionResolver, discover_collection_resources
from pathlib import Path
import tempfile
def test_collection_discovery():
    """Test collection resource discovery."""
    with tempfile.TemporaryDirectory() as tmpdir:
        collection_path = Path(tmpdir) / "test-collection"
        collection_path.mkdir()
        # Create test structure
        (collection_path / "profiles").mkdir()
        (collection_path / "profiles" / "test.md").write_text("# Test profile")
        (collection_path / "agents").mkdir()
        (collection_path / "agents" / "helper.md").write_text("# Test agent")
        # Discover resources
        resources = discover_collection_resources(collection_path)
        assert len(resources.profiles) == 1
        assert len(resources.agents) == 1
        assert resources.profiles[0].name == "test.md"

from amplifier_collections import extract_collection_name_from_path
from pathlib import Path
# Extract collection name from path containing "/collections/"
path = Path("~/.amplifier/collections/foundation/profiles/base.md")
name = extract_collection_name_from_path(path)
# Returns: "foundation" (reads from pyproject.toml)

from amplifier_collections import (
    CollectionError,
    CollectionInstallError,
    CollectionMetadataError,
    CollectionNotFoundError,
)
# Base exception
class CollectionError(Exception):
    """Base exception for collection operations."""
    def __init__(self, message: str, context: dict | None = None):
        self.message = message
        self.context = context or {}
# Installation errors
class CollectionInstallError(CollectionError):
    """Collection installation failed."""
# Metadata errors
class CollectionMetadataError(CollectionError):
    """Invalid or missing collection metadata."""
# Not found errors
class CollectionNotFoundError(CollectionError):
    """Collection not found in search paths."""
# Usage
try:
    metadata = CollectionMetadata.from_pyproject(path / "pyproject.toml")
except CollectionMetadataError as e:
    print(f"Error: {e.message}")
    print(f"File: {e.context.get('file')}")

# Invalid collection (missing pyproject.toml)
resources = discover_collection_resources(Path("/invalid/path"))
# Raises: CollectionMetadataError("Collection missing pyproject.toml")
# Invalid metadata (missing required field)
# Raises: CollectionMetadataError("Required field missing: name")

class CollectionMetadata(BaseModel):
    model_config = ConfigDict(frozen=True)