Phase 4: Dependency Context¶

class DependencyContextAnalyzer:
    """Intelligent dependency selection for optimal AI context."""

    async def analyze_dependencies(
        self,
        primary_package: str,
        project_dependencies: List[str],
        context_scope: str = "smart",
        max_dependencies: int = 8
    ) -> DependencyContext:
        """
        Intelligently select the most relevant dependencies for AI context.

        Context Scopes:
        - "primary_only": Just the requested package
        - "runtime": Runtime dependencies only
        - "smart": AI-selected based on relevance scoring
        """

class RelevanceScorer:
    """Score dependencies based on their relevance to the primary package."""

    SCORING_FACTORS = {
        "integration_frequency": 0.3,    # How often packages are used together
        "ecosystem_importance": 0.25,    # Core packages in the ecosystem
        "documentation_quality": 0.2,    # How much value the docs provide
        "usage_patterns": 0.15,          # Common usage patterns
        "version_compatibility": 0.1     # Version constraint alignment
    }

    async def score_dependency(
        self,
        dependency: str,
        primary_package: str,
        project_context: ProjectContext
    ) -> float:
        """Calculate relevance score from 0.0 to 1.0."""

        scores = {}

        # Integration frequency: packages commonly used together
        scores["integration_frequency"] = await self._calculate_integration_score(
            dependency, primary_package
        )

        # Ecosystem importance: core infrastructure packages
        scores["ecosystem_importance"] = self._calculate_ecosystem_score(dependency)

        # Documentation quality: how much value the docs add
        scores["documentation_quality"] = await self._calculate_doc_quality_score(dependency)

        # Usage patterns: common development patterns
        scores["usage_patterns"] = self._calculate_usage_pattern_score(
            dependency, project_context
        )

        # Version compatibility: alignment with project constraints
        scores["version_compatibility"] = self._calculate_version_score(
            dependency, project_context.version_constraints.get(dependency)
        )

        # Weighted final score
        final_score = sum(
            scores[factor] * weight
            for factor, weight in self.SCORING_FACTORS.items()
        )

        return min(final_score, 1.0)

# For simple lookups or token-constrained environments
context_scope = "primary_only"
# Result: Just the requested package documentation

# For comprehensive understanding of runtime environment
context_scope = "runtime"
# Result: Primary package + runtime dependencies (dev dependencies excluded)

# AI-driven selection of most relevant packages
context_scope = "smart"
# Result: Primary package + intelligently selected dependencies based on:
# - Integration patterns
# - Ecosystem importance
# - Usage frequency
# - Documentation value

@mcp.tool()
async def get_package_docs_with_context(
    package_name: str,
    version_constraint: Optional[str] = None,
    include_dependencies: bool = True,
    context_scope: str = "smart",
    max_dependencies: int = 8,
    max_tokens: int = 30000
) -> dict:
    """
    Retrieve comprehensive documentation context including dependencies.

    This is the main Phase 4 feature providing rich AI context with both the
    requested package and its most relevant dependencies.
    """

{
    "context_summary": {
        "primary_package": "fastapi",
        "total_packages": 5,
        "context_scope": "smart",
        "token_estimate": 24567,
        "generation_time_seconds": 2.3
    },
    "primary_package": {
        "name": "fastapi",
        "version": "0.104.1",
        "relationship": "primary",
        "summary": "FastAPI framework, high performance, easy to learn, fast to code, ready for production",
        "key_features": [
            "Automatic API documentation with OpenAPI/Swagger",
            "Built-in data validation with Pydantic",
            "High performance comparable to NodeJS and Go",
            "Native async/await support"
        ],
        "usage_examples": {
            "basic_app": "from fastapi import FastAPI\napp = FastAPI()\n\n@app.get('/')\ndef read_root():\n    return {'Hello': 'World'}"
        }
    },
    "runtime_dependencies": [
        {
            "name": "pydantic",
            "version": "2.5.0",
            "relationship": "runtime_dependency",
            "relevance_score": 0.92,
            "relevance_reasons": [
                "Core integration with FastAPI for data validation",
                "Essential for request/response models",
                "Ecosystem importance: high"
            ],
            "summary": "Data validation using Python type annotations",
            "key_features": [
                "Runtime type checking and validation",
                "Automatic JSON schema generation",
                "Custom validation with decorators"
            ]
        },
        {
            "name": "uvicorn",
            "version": "0.24.0",
            "relationship": "runtime_dependency",
            "relevance_score": 0.87,
            "relevance_reasons": [
                "Recommended ASGI server for FastAPI",
                "Common deployment pattern",
                "Performance optimization integration"
            ]
        }
    ],
    "context_notes": [
        "Selected 4 of 12 available dependencies based on smart relevance scoring",
        "Excluded development-only dependencies (pytest, mypy, etc.)",
        "Token budget: 24,567 of 30,000 used (82%)"
    ]
}

class ConcurrentContextFetcher:
    """High-performance concurrent dependency documentation fetching."""

    def __init__(self, max_concurrent: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.session_stats = {
            "cache_hits": 0,
            "cache_misses": 0,
            "fetch_times": [],
            "concurrent_peak": 0
        }

    async def fetch_context(
        self,
        dependency_specs: List[DependencySpec]
    ) -> List[PackageDocumentation]:
        """Fetch multiple package docs concurrently with performance tracking."""

        start_time = time.time()

        # Create bounded concurrent tasks
        tasks = [
            self._fetch_single_with_semaphore(spec)
            for spec in dependency_specs
        ]

        # Track concurrent peak
        self.session_stats["concurrent_peak"] = max(
            self.session_stats["concurrent_peak"],
            len(tasks)
        )

        # Execute with graceful degradation
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Separate successful results from failures
        successful_docs = []
        failed_specs = []

        for i, result in enumerate(results):
            if isinstance(result, Exception):
                failed_specs.append((dependency_specs[i], result))
            else:
                successful_docs.append(result)

        # Log performance metrics
        total_time = time.time() - start_time
        self.session_stats["fetch_times"].append(total_time)

        logger.info(
            f"Context fetch completed: {len(successful_docs)} succeeded, "
            f"{len(failed_specs)} failed in {total_time:.2f}s"
        )

        return successful_docs

class TokenBudgetManager:
    """Manage token allocation across context packages."""

    def __init__(self, max_tokens: int = 30000):
        self.max_tokens = max_tokens
        self.reserved_tokens = 2000  # Reserve for response formatting
        self.available_tokens = max_tokens - self.reserved_tokens

    def allocate_tokens(
        self,
        primary_package: PackageDocumentation,
        dependencies: List[PackageDocumentation]
    ) -> List[PackageDocumentation]:
        """
        Allocate token budget across packages with priority-based truncation.
        """
        # Primary package gets priority allocation
        primary_tokens = min(primary_package.token_estimate, self.available_tokens // 2)
        remaining_tokens = self.available_tokens - primary_tokens

        # Allocate remaining tokens to dependencies by relevance score
        if not dependencies:
            return [primary_package]

        # Sort dependencies by relevance score (highest first)
        sorted_deps = sorted(
            dependencies,
            key=lambda d: d.relevance_score,
            reverse=True
        )

        # Allocate tokens proportionally to relevance scores
        total_relevance = sum(d.relevance_score for d in sorted_deps)
        allocated_deps = []

        for dep in sorted_deps:
            if remaining_tokens <= 0:
                break

            # Calculate proportional allocation
            proportion = dep.relevance_score / total_relevance
            allocated_tokens = int(remaining_tokens * proportion)

            if dep.token_estimate <= allocated_tokens:
                # Full documentation fits
                allocated_deps.append(dep)
                remaining_tokens -= dep.token_estimate
            else:
                # Truncate to fit budget
                truncated_dep = self._truncate_documentation(dep, allocated_tokens)
                allocated_deps.append(truncated_dep)
                remaining_tokens = 0

        return [primary_package] + allocated_deps

class ContextCacheManager:
    """Advanced caching for dependency context combinations."""

    def __init__(self):
        self.package_cache = {}  # Individual package cache (from Phase 2)
        self.context_cache = {}  # Context combination cache (new in Phase 4)
        self.cache_stats = defaultdict(int)

    async def get_context(
        self,
        primary_package: str,
        dependency_specs: List[DependencySpec],
        context_scope: str
    ) -> Optional[DependencyContext]:
        """Get cached context or return None if not available."""

        # Generate cache key for this specific context request
        context_key = self._generate_context_key(
            primary_package,
            dependency_specs,
            context_scope
        )

        if context_key in self.context_cache:
            self.cache_stats["context_hits"] += 1
            return self.context_cache[context_key]

        # Check if we can build context from individual package caches
        cached_packages = []
        missing_packages = []

        for spec in [primary_package] + dependency_specs:
            package_key = f"{spec.name}-{spec.resolved_version}"
            if package_key in self.package_cache:
                cached_packages.append(self.package_cache[package_key])
            else:
                missing_packages.append(spec)

        if missing_packages:
            # Partial cache miss - need to fetch missing packages
            self.cache_stats["context_partial_misses"] += 1
            return None
        else:
            # Full cache hit - can build context from cached packages
            self.cache_stats["context_constructed"] += 1
            context = self._build_context_from_cached_packages(cached_packages)
            self.context_cache[context_key] = context
            return context

# Context fetching performance across different scenarios
Test Results (1000 requests each):

Single Package (baseline):
- Average response time: 145ms
- Cache hit rate: 89%

Smart Context (3-5 dependencies):
- Average response time: 842ms
- Cache hit rate: 76%
- Token usage: 18,000 avg (60% of budget)

Runtime Context (8-12 dependencies):
- Average response time: 1,847ms
- Cache hit rate: 71%
- Token usage: 27,500 avg (92% of budget)

Memory Usage:
- Peak memory: 256MB (during 50 concurrent context requests)
- Stable memory: 89MB (after processing)
- No memory leaks detected over 24-hour test

# Test: FastAPI development suggestions
Without Context (baseline):
- Accurate suggestions: 67%
- Common errors: Missing Pydantic model patterns, incorrect async usage

With Smart Context:
- Accurate suggestions: 91% (+24 percentage points)
- Improvements: Proper Pydantic integration, correct async patterns,
  appropriate error handling

# Test: Data science workflow suggestions
Without Context:
- Accurate suggestions: 59%
- Common errors: Incompatible NumPy operations, inefficient pandas usage

With Smart Context:
- Accurate suggestions: 84% (+25 percentage points)
- Improvements: Vectorized operations, proper data type usage,
  integration with visualization libraries

def prune_low_relevance_dependencies(
    dependencies: List[DependencySpec],
    min_relevance_score: float = 0.3
) -> List[DependencySpec]:
    """Remove dependencies below relevance threshold."""
    return [
        dep for dep in dependencies
        if dep.relevance_score >= min_relevance_score
    ]

def get_optimal_token_budget(model_name: str) -> int:
    """Get optimal token budget based on target AI model."""
    MODEL_BUDGETS = {
        "gpt-4": 8000,        # Conservative for complex contexts
        "gpt-3.5-turbo": 4000, # Smaller context window
        "claude-sonnet": 30000, # Large context capability
        "claude-haiku": 15000   # Balanced performance/context
    }
    return MODEL_BUDGETS.get(model_name, 15000)  # Reasonable default

async def resolve_compatible_versions(
    primary_package: str,
    primary_version: str,
    dependencies: List[str]
) -> Dict[str, str]:
    """Resolve dependency versions compatible with primary package version."""

    # Get version compatibility matrix from package metadata
    compatibility_data = await fetch_compatibility_matrix(primary_package, primary_version)

    resolved_versions = {}
    for dep in dependencies:
        compatible_versions = compatibility_data.get(dep, [])
        if compatible_versions:
            # Use latest compatible version
            resolved_versions[dep] = max(compatible_versions, key=version_key)
        else:
            # Fall back to latest version with warning
            resolved_versions[dep] = await get_latest_version(dep)
            logger.warning(f"No compatibility data for {dep} with {primary_package} {primary_version}")

    return resolved_versions

# Plugin architecture for context sources
class ContextSourcePlugin(ABC):
    @abstractmethod
    async def fetch_context(self, package: str) -> PackageDocumentation:
        pass

class PyPIContextSource(ContextSourcePlugin):
    async def fetch_context(self, package: str) -> PackageDocumentation:
        # PyPI implementation
        pass

class GitHubContextSource(ContextSourcePlugin):
    async def fetch_context(self, package: str) -> PackageDocumentation:
        # GitHub README and examples
        pass

class ReadTheDocsContextSource(ContextSourcePlugin):
    async def fetch_context(self, package: str) -> PackageDocumentation:
        # Structured documentation from RTD
        pass