Architectural Fitness Functions and Governance: Measuring and Evolving Architecture Quality¶

# fitness_functions/modularity.py
def test_schema_modularity():
    """Ensure schemas don't have excessive cross-schema dependencies"""
    cross_schema_deps = count_cross_schema_dependencies()
    assert cross_schema_deps < 10, f"Too many cross-schema dependencies: {cross_schema_deps}"

# Good: Low coupling
class UserService:
    def get_user(self, user_id: str):
        # Self-contained, no direct DB access
        return self.user_repository.find(user_id)

# Bad: High coupling
class UserService:
    def get_user(self, user_id: str):
        # Direct database access, tight coupling
        return db.execute("SELECT * FROM users WHERE id = %s", user_id)

# fitness_functions/coupling.py
def test_service_coupling():
    """Ensure services don't have excessive dependencies"""
    dependency_graph = build_dependency_graph()
    max_dependencies = max(len(deps) for deps in dependency_graph.values())
    assert max_dependencies < 5, f"Service has too many dependencies: {max_dependencies}"

# Resilience pattern
from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=60)
def call_external_service():
    """Resilient service call with circuit breaker"""
    return external_service.call()

# fitness_functions/resilience.py
def test_circuit_breaker_coverage():
    """Ensure all external calls have circuit breakers"""
    external_calls = find_external_service_calls()
    protected_calls = find_circuit_breaker_protected_calls()
    coverage = len(protected_calls) / len(external_calls)
    assert coverage > 0.9, f"Circuit breaker coverage too low: {coverage}"

# Observability pattern
import structlog

logger = structlog.get_logger()

def process_request(request):
    logger.info("request_received", request_id=request.id, user_id=request.user_id)
    # Process request
    logger.info("request_processed", request_id=request.id, duration=duration)

# fitness_functions/observability.py
def test_logging_coverage():
    """Ensure critical paths have logging"""
    critical_functions = find_critical_functions()
    logged_functions = find_logged_functions()
    coverage = len(logged_functions) / len(critical_functions)
    assert coverage > 0.8, f"Logging coverage too low: {coverage}"

# fitness_functions/scalability.py
def test_horizontal_scaling():
    """Ensure services support horizontal scaling"""
    deployments = get_kubernetes_deployments()
    for deployment in deployments:
        assert deployment.replicas >= 2, f"{deployment.name} not horizontally scalable"
        assert has_hpa(deployment), f"{deployment.name} missing HPA"

# Data integrity pattern
@transactional
def transfer_funds(from_account, to_account, amount):
    """Atomic fund transfer"""
    from_account.debit(amount)
    to_account.credit(amount)

# fitness_functions/data_integrity.py
def test_transaction_coverage():
    """Ensure data modifications are transactional"""
    data_modifications = find_data_modifications()
    transactional_modifications = find_transactional_modifications()
    coverage = len(transactional_modifications) / len(data_modifications)
    assert coverage > 0.95, f"Transaction coverage too low: {coverage}"

# Domain isolation
class UserDomain:
    """User domain with clear boundaries"""
    def create_user(self, user_data):
        # User domain logic only
        pass

class OrderDomain:
    """Order domain with clear boundaries"""
    def create_order(self, order_data):
        # Order domain logic only
        pass

# fitness_functions/domain_isolation.py
def test_domain_boundaries():
    """Ensure domains don't leak into each other"""
    user_domain_code = get_domain_code("user")
    order_domain_code = get_domain_code("order")

    # Check for domain leakage
    user_in_order = check_code_references(user_domain_code, order_domain_code)
    assert not user_in_order, "Domain boundary violation detected"

# Strong interface contract
from pydantic import BaseModel

class UserV1(BaseModel):
    id: int
    name: str

class UserV2(BaseModel):
    id: int
    name: str
    email: str  # Backward compatible addition

# fitness_functions/interface_contracts.py
def test_api_versioning():
    """Ensure APIs are versioned"""
    api_endpoints = get_api_endpoints()
    versioned_endpoints = [ep for ep in api_endpoints if has_version(ep)]
    coverage = len(versioned_endpoints) / len(api_endpoints)
    assert coverage > 0.9, f"API versioning coverage too low: {coverage}"

# Latency tolerance
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def call_service_with_retry():
    """Service call with exponential backoff"""
    return service.call()

# fitness_functions/latency.py
def test_p95_latency():
    """Ensure P95 latency is within SLA"""
    latency_metrics = get_latency_metrics()
    p95_latency = calculate_percentile(latency_metrics, 95)
    assert p95_latency < 500, f"P95 latency too high: {p95_latency}ms"

# Time governance
from datetime import datetime
import ntplib

def get_synchronized_time():
    """Get NTP-synchronized time"""
    ntp_client = ntplib.NTPClient()
    response = ntp_client.request('pool.ntp.org')
    return datetime.fromtimestamp(response.tx_time)

# fitness_functions/time_governance.py
def test_time_synchronization():
    """Ensure all nodes are time-synchronized"""
    nodes = get_cluster_nodes()
    for node in nodes:
        time_skew = get_time_skew(node)
        assert time_skew < 0.1, f"Node {node} time skew too high: {time_skew}s"

# Predictable, idempotent operation
def create_user(user_id: str, user_data: dict):
    """Idempotent user creation"""
    if user_exists(user_id):
        return get_user(user_id)
    return create_new_user(user_id, user_data)

# fitness_functions/predictability.py
def test_idempotency():
    """Ensure critical operations are idempotent"""
    critical_operations = find_critical_operations()
    idempotent_operations = find_idempotent_operations()
    coverage = len(idempotent_operations) / len(critical_operations)
    assert coverage > 0.8, f"Idempotency coverage too low: {coverage}"

# Autonomous, stateless service
class StatelessService:
    def process_request(self, request):
        # No shared state, fully autonomous
        return self.process(request)

# fitness_functions/autonomy.py
def test_stateless_services():
    """Ensure services are stateless"""
    services = get_services()
    for service in services:
        assert is_stateless(service), f"Service {service.name} is not stateless"

# Complexity management
def simple_function(data):
    """Low complexity function"""
    if data:
        return process(data)
    return None

# Bad: High complexity
def complex_function(data):
    """High complexity function - should be refactored"""
    if data:
        if data.type == "A":
            if data.value > 10:
                # ... many nested conditions
                pass

# fitness_functions/complexity.py
def test_cyclomatic_complexity():
    """Ensure functions don't exceed complexity threshold"""
    functions = get_all_functions()
    for func in functions:
        complexity = calculate_cyclomatic_complexity(func)
        assert complexity < 10, f"Function {func.name} too complex: {complexity}"

# fitness_functions/structural/boundary_integrity.py
def test_layer_boundaries():
    """Ensure layers don't violate boundaries"""
    # Presentation layer should not import from data layer
    presentation_imports = get_imports("presentation")
    data_layer_imports = [imp for imp in presentation_imports if "data" in imp]
    assert len(data_layer_imports) == 0, f"Boundary violation: {data_layer_imports}"

# fitness_functions/structural/dependency_limits.py
def test_module_dependencies():
    """Ensure modules don't have excessive dependencies"""
    modules = get_modules()
    for module in modules:
        dependencies = get_dependencies(module)
        assert len(dependencies) < 10, f"Module {module} has too many dependencies: {len(dependencies)}"

# fitness_functions/behavioral/latency.py
def test_api_latency():
    """Ensure API latency is within SLA"""
    endpoints = get_api_endpoints()
    for endpoint in endpoints:
        latency = measure_latency(endpoint)
        assert latency.p95 < 500, f"Endpoint {endpoint} P95 latency too high: {latency.p95}ms"

# fitness_functions/behavioral/throughput.py
def test_throughput():
    """Ensure system meets minimum throughput"""
    throughput = measure_throughput()
    assert throughput > 1000, f"Throughput too low: {throughput} req/s"

# fitness_functions/behavioral/tail_latency.py
def test_tail_latency():
    """Ensure P99 latency is acceptable"""
    latency_metrics = get_latency_metrics()
    p99_latency = calculate_percentile(latency_metrics, 99)
    assert p99_latency < 1000, f"P99 latency too high: {p99_latency}ms"

# fitness_functions/data/schema_drift.py
def test_schema_consistency():
    """Ensure schemas don't drift"""
    expected_schema = load_expected_schema()
    actual_schema = get_actual_schema()

    drift = calculate_schema_drift(expected_schema, actual_schema)
    assert drift < 0.1, f"Schema drift detected: {drift}"

# fitness_functions/data/quality.py
def test_data_quality():
    """Ensure data meets quality standards"""
    data = load_data()

    # Check null rates
    null_rates = calculate_null_rates(data)
    for column, rate in null_rates.items():
        assert rate < 0.05, f"Column {column} has too many nulls: {rate}"

    # Check value ranges
    ranges = get_expected_ranges()
    for column, expected_range in ranges.items():
        actual_range = get_actual_range(data[column])
        assert actual_range.within(expected_range), f"Column {column} out of range"

# fitness_functions/security/dependencies.py
def test_dependency_vulnerabilities():
    """Ensure no known vulnerabilities in dependencies"""
    vulnerabilities = scan_dependencies()
    critical_vulns = [v for v in vulnerabilities if v.severity == "critical"]
    assert len(critical_vulns) == 0, f"Critical vulnerabilities found: {critical_vulns}"

# fitness_functions/security/secrets.py
def test_secret_exposure():
    """Ensure no secrets in code"""
    secrets = scan_for_secrets()
    assert len(secrets) == 0, f"Secrets found in code: {secrets}"

# fitness_functions/resilience/circuit_breakers.py
def test_circuit_breaker_coverage():
    """Ensure external calls have circuit breakers"""
    external_calls = find_external_calls()
    protected_calls = find_circuit_breaker_protected_calls()
    coverage = len(protected_calls) / len(external_calls)
    assert coverage > 0.9, f"Circuit breaker coverage too low: {coverage}"

# fitness_functions/resilience/retry.py
def test_retry_coverage():
    """Ensure transient failures have retry logic"""
    external_calls = find_external_calls()
    retry_protected = find_retry_protected_calls()
    coverage = len(retry_protected) / len(external_calls)
    assert coverage > 0.8, f"Retry coverage too low: {coverage}"

# fitness_functions/ml/reproducibility.py
def test_model_signature_consistency():
    """Ensure model signatures are consistent"""
    model_versions = get_model_versions()
    for version in model_versions:
        signature = get_model_signature(version)
        expected_signature = get_expected_signature()
        assert signature == expected_signature, f"Model {version} signature mismatch"

# fitness_functions/ml/feature_parity.py
def test_feature_parity():
    """Ensure training and inference features match"""
    training_features = get_training_features()
    inference_features = get_inference_features()

    assert set(training_features) == set(inference_features), "Feature mismatch"

# fitness_functions/geospatial/crs_consistency.py
def test_crs_consistency():
    """Ensure all geometries use consistent CRS"""
    geometries = get_geometries()
    crs_counts = count_crs(geometries)

    # Should have one dominant CRS
    dominant_crs_ratio = max(crs_counts.values()) / sum(crs_counts.values())
    assert dominant_crs_ratio > 0.95, f"CRS inconsistency: {crs_counts}"

# fitness_functions/geospatial/validity.py
def test_geometry_validity():
    """Ensure all geometries are valid"""
    geometries = get_geometries()
    invalid = [g for g in geometries if not g.is_valid]

    invalid_ratio = len(invalid) / len(geometries)
    assert invalid_ratio < 0.01, f"Too many invalid geometries: {invalid_ratio}"

# fitness_functions/api/stability.py
def test_api_breaking_changes():
    """Ensure no breaking API changes"""
    current_api = get_current_api_schema()
    previous_api = get_previous_api_schema()

    breaking_changes = detect_breaking_changes(current_api, previous_api)
    assert len(breaking_changes) == 0, f"Breaking changes detected: {breaking_changes}"

# fitness_functions/api/versioning.py
def test_api_versioning():
    """Ensure APIs are properly versioned"""
    endpoints = get_api_endpoints()
    versioned_endpoints = [ep for ep in endpoints if has_version(ep)]

    coverage = len(versioned_endpoints) / len(endpoints)
    assert coverage > 0.9, f"API versioning coverage too low: {coverage}"

# fitness_functions/governance/adr.py
def test_adr_consistency():
    """Ensure ADRs are consistent with implementation"""
    adrs = load_adrs()
    implementations = get_implementations()

    for adr in adrs:
        implementation = find_implementation(adr)
        assert implementation.matches(adr), f"ADR {adr.id} not consistent with implementation"

# fitness_functions/governance/versioning.py
def test_versioning_discipline():
    """Ensure all components are versioned"""
    components = get_components()
    versioned_components = [c for c in components if has_version(c)]

    coverage = len(versioned_components) / len(components)
    assert coverage > 0.95, f"Versioning coverage too low: {coverage}"

# fitness_functions/governance/repo_structure.py
def test_repo_structure():
    """Ensure repos follow standard structure"""
    repos = get_repos()
    for repo in repos:
        structure = get_repo_structure(repo)
        expected_structure = get_standard_structure()

        drift = calculate_structure_drift(structure, expected_structure)
        assert drift < 0.1, f"Repo {repo} structure drift: {drift}"

# drift_detection/architectural_drift.py
class ArchitecturalDriftDetector:
    def detect_drift(self, baseline: dict, current: dict) -> dict:
        """Detect architectural drift"""
        drift = {
            'structural': self.detect_structural_drift(baseline, current),
            'behavioral': self.detect_behavioral_drift(baseline, current),
            'governance': self.detect_governance_drift(baseline, current)
        }
        return drift

# debt_tracking/architectural_debt.py
class ArchitecturalDebtTracker:
    def calculate_debt(self) -> dict:
        """Calculate architectural technical debt"""
        return {
            'structural': self.calculate_structural_debt(),
            'performance': self.calculate_performance_debt(),
            'security': self.calculate_security_debt(),
            'governance': self.calculate_governance_debt()
        }

# governance/maturity_rating.py
class ComponentMaturityRater:
    def rate_component(self, component: str) -> int:
        """Rate component maturity"""
        scores = {
            'standards': self.assess_standards(component),
            'metrics': self.assess_metrics(component),
            'automation': self.assess_automation(component),
            'documentation': self.assess_documentation(component)
        }

        average_score = sum(scores.values()) / len(scores)
        return int(average_score)  # 1-5 scale

# guardrails/architectural_guardrails.py
class ArchitecturalGuardrails:
    def enforce_guardrails(self, change: dict) -> bool:
        """Enforce architectural guardrails"""
        # Check layer boundaries
        if not self.check_layer_boundaries(change):
            return False

        # Check dependency limits
        if not self.check_dependency_limits(change):
            return False

        # Check complexity limits
        if not self.check_complexity_limits(change):
            return False

        return True

# fitness_functions/postgres/table_count.py
def test_schema_table_count():
    """Ensure schemas don't have excessive tables"""
    schemas = get_postgres_schemas()
    for schema in schemas:
        table_count = count_tables(schema)
        assert table_count < 50, f"Schema {schema} has too many tables: {table_count}"

# fitness_functions/postgres/relationship_density.py
def test_relationship_density():
    """Ensure relationship density is reasonable"""
    tables = get_postgres_tables()
    for table in tables:
        relationships = count_relationships(table)
        density = relationships / count_columns(table)
        assert density < 0.5, f"Table {table} relationship density too high: {density}"

# fitness_functions/postgres/wal_growth.py
def test_wal_growth():
    """Ensure WAL growth is within limits"""
    wal_size = get_wal_size()
    wal_growth_rate = get_wal_growth_rate()

    assert wal_size < 100 * 1024 * 1024 * 1024, f"WAL size too large: {wal_size}"
    assert wal_growth_rate < 1 * 1024 * 1024 * 1024, f"WAL growth rate too high: {wal_growth_rate}"

# fitness_functions/postgres/geospatial_index.py
def test_geospatial_index_quality():
    """Ensure geospatial indexes are effective"""
    spatial_tables = get_spatial_tables()
    for table in spatial_tables:
        index_usage = get_index_usage(table)
        assert index_usage > 0.8, f"Table {table} geospatial index usage too low: {index_usage}"

# fitness_functions/kubernetes/pod_startup.py
def test_pod_startup_time():
    """Ensure pods start within SLA"""
    pods = get_kubernetes_pods()
    for pod in pods:
        startup_time = measure_pod_startup_time(pod)
        assert startup_time < 30, f"Pod {pod.name} startup time too high: {startup_time}s"

# fitness_functions/kubernetes/dependency_graph.py
def test_service_dependencies():
    """Ensure service dependency graph is acyclic"""
    dependency_graph = build_service_dependency_graph()
    cycles = detect_cycles(dependency_graph)
    assert len(cycles) == 0, f"Cyclic dependencies detected: {cycles}"

# fitness_functions/kubernetes/hop_count.py
def test_interservice_hop_count():
    """Ensure interservice calls don't exceed hop limit"""
    service_calls = trace_service_calls()
    for call in service_calls:
        hop_count = count_hops(call)
        assert hop_count < 5, f"Service call {call} has too many hops: {hop_count}"

# fitness_functions/redis/stream_lag.py
def test_stream_lag():
    """Ensure Redis stream lag is within limits"""
    streams = get_redis_streams()
    for stream in streams:
        lag = get_stream_lag(stream)
        assert lag < 1000, f"Stream {stream} lag too high: {lag} messages"

# fitness_functions/redis/memory_fragmentation.py
def test_memory_fragmentation():
    """Ensure memory fragmentation is acceptable"""
    fragmentation = get_redis_memory_fragmentation()
    assert fragmentation < 1.5, f"Memory fragmentation too high: {fragmentation}"

# fitness_functions/nicegui/latency.py
def test_ui_backend_latency():
    """Ensure UI to backend latency is acceptable"""
    ui_actions = get_ui_actions()
    for action in ui_actions:
        latency = measure_ui_backend_latency(action)
        assert latency < 200, f"UI action {action} latency too high: {latency}ms"

# fitness_functions/nicegui/coupling.py
def test_component_coupling():
    """Ensure UI components aren't overly coupled"""
    components = get_nicegui_components()
    for component in components:
        dependencies = get_component_dependencies(component)
        assert len(dependencies) < 5, f"Component {component} has too many dependencies: {len(dependencies)}"

# fitness_functions/ml/load_time.py
def test_model_load_time():
    """Ensure model load time is acceptable"""
    models = get_ml_models()
    for model in models:
        load_time = measure_model_load_time(model)
        assert load_time < 5, f"Model {model} load time too high: {load_time}s"

# fitness_functions/ml/onnx_shape.py
def test_onnx_shape_consistency():
    """Ensure ONNX model shapes are consistent"""
    models = get_onnx_models()
    for model in models:
        input_shape = get_model_input_shape(model)
        expected_shape = get_expected_input_shape(model)
        assert input_shape == expected_shape, f"Model {model} shape mismatch"

# fitness_functions/ml/reproducibility.py
def test_ml_reproducibility():
    """Ensure ML models are reproducible"""
    models = get_ml_models()
    for model in models:
        reproducibility_score = calculate_reproducibility_score(model)
        assert reproducibility_score > 0.95, f"Model {model} reproducibility too low: {reproducibility_score}"

# fitness_functions/pipelines/schema_drift.py
def test_schema_drift():
    """Ensure schemas don't drift beyond acceptable limits"""
    pipelines = get_data_pipelines()
    for pipeline in pipelines:
        drift = calculate_schema_drift(pipeline)
        assert drift < 0.1, f"Pipeline {pipeline} schema drift too high: {drift}"

# fitness_functions/pipelines/idempotency.py
def test_transformation_idempotency():
    """Ensure transformations are idempotent"""
    transformations = get_transformations()
    for transformation in transformations:
        is_idempotent = test_idempotency(transformation)
        assert is_idempotent, f"Transformation {transformation} is not idempotent"

# fitness_functions/lakehouse/row_group_consistency.py
def test_parquet_row_group_consistency():
    """Ensure Parquet row groups are consistent"""
    parquet_files = get_parquet_files()
    for file in parquet_files:
        row_groups = get_row_groups(file)
        schemas = [get_row_group_schema(rg) for rg in row_groups]

        # All row groups should have same schema
        assert all(s == schemas[0] for s in schemas), f"File {file} has inconsistent row group schemas"

# fitness_functions/lakehouse/metadata_locks.py
def test_metadata_evolution():
    """Ensure metadata evolution is backward compatible"""
    metadata_versions = get_metadata_versions()
    for i in range(1, len(metadata_versions)):
        current = metadata_versions[i]
        previous = metadata_versions[i-1]

        assert is_backward_compatible(current, previous), f"Metadata version {i} not backward compatible"

# observability/heat_maps.py
class ArchitectureHeatMap:
    def generate_coupling_heatmap(self):
        """Generate coupling heat map"""
        components = get_components()
        coupling_matrix = build_coupling_matrix(components)

        # Visualize as heat map
        import matplotlib.pyplot as plt
        plt.imshow(coupling_matrix, cmap='hot', interpolation='nearest')
        plt.colorbar()
        plt.title('Component Coupling Heat Map')
        plt.show()

# observability/dependency_graph.py
class DependencyGraphMonitor:
    def monitor_dependencies(self):
        """Monitor dependency graph for changes"""
        current_graph = build_dependency_graph()
        baseline_graph = load_baseline_graph()

        changes = detect_graph_changes(current_graph, baseline_graph)
        if changes:
            alert_on_dependency_changes(changes)

# observability/complexity_tracking.py
class ComplexityTracker:
    def track_complexity(self):
        """Track complexity over time"""
        complexity_metrics = {
            'cyclomatic': calculate_cyclomatic_complexity(),
            'coupling': calculate_coupling(),
            'cohesion': calculate_cohesion(),
            'lines_of_code': count_lines_of_code()
        }

        store_complexity_metrics(complexity_metrics)
        plot_complexity_trends(complexity_metrics)

# observability/domain_drift.py
class DomainDriftDetector:
    def detect_domain_drift(self):
        """Detect drift from intended domain boundaries"""
        intended_boundaries = load_intended_boundaries()
        actual_boundaries = detect_actual_boundaries()

        drift = calculate_boundary_drift(intended_boundaries, actual_boundaries)
        if drift > threshold:
            alert_on_domain_drift(drift)

# observability/health_scoring.py
class ArchitectureHealthScorer:
    def calculate_health_score(self, component: str) -> float:
        """Calculate architecture health score"""
        metrics = {
            'modularity': self.measure_modularity(component),
            'coupling': self.measure_coupling(component),
            'cohesion': self.measure_cohesion(component),
            'resilience': self.measure_resilience(component),
            'observability': self.measure_observability(component)
        }

        # Weighted average
        weights = {
            'modularity': 0.2,
            'coupling': 0.2,
            'cohesion': 0.2,
            'resilience': 0.2,
            'observability': 0.2
        }

        health_score = sum(metrics[key] * weights[key] for key in metrics)
        return health_score

# anti_patterns/big_ball_of_mud.py
def detect_big_ball_of_mud():
    """Detect Big Ball of Mud anti-pattern"""
    dependency_graph = build_dependency_graph()

    # Check for high average coupling
    avg_coupling = calculate_average_coupling(dependency_graph)
    if avg_coupling > 0.8:
        return True, "High average coupling indicates Big Ball of Mud"

    return False, None

# anti_patterns/invisible_architecture.py
def detect_invisible_architecture():
    """Detect invisible architecture anti-pattern"""
    adr_count = count_adrs()
    component_count = count_components()

    if adr_count / component_count < 0.5:
        return True, "Low ADR coverage indicates invisible architecture"

    return False, None

# anti_patterns/geospatial_coupling.py
def detect_overcoupled_geospatial():
    """Detect overcoupled geospatial pipelines"""
    pipelines = get_geospatial_pipelines()
    for pipeline in pipelines:
        dependencies = get_pipeline_dependencies(pipeline)
        if len(dependencies) > 10:
            return True, f"Pipeline {pipeline} has too many dependencies"

    return False, None

# anti_patterns/monolithic_schema.py
def detect_monolithic_schema():
    """Detect monolithic Postgres schema"""
    schemas = get_postgres_schemas()
    for schema in schemas:
        table_count = count_tables(schema)
        if table_count > 50:
            return True, f"Schema {schema} is monolithic: {table_count} tables"

    return False, None

# anti_patterns/mlflow_sprawl.py
def detect_mlflow_sprawl():
    """Detect MLflow artifact sprawl"""
    artifacts = get_mlflow_artifacts()
    organized_artifacts = count_organized_artifacts(artifacts)

    organization_ratio = organized_artifacts / len(artifacts)
    if organization_ratio < 0.7:
        return True, f"Low artifact organization: {organization_ratio}"

    return False, None

# anti_patterns/cyclic_dependencies.py
def detect_cyclic_dependencies():
    """Detect cyclic dependencies"""
    dependency_graph = build_dependency_graph()
    cycles = detect_cycles(dependency_graph)

    if len(cycles) > 0:
        return True, f"Cyclic dependencies detected: {cycles}"

    return False, None

# anti_patterns/ui_db_coupling.py
def detect_ui_db_coupling():
    """Detect UI to DB direct coupling"""
    ui_components = get_ui_components()
    for component in ui_components:
        if has_direct_db_access(component):
            return True, f"UI component {component} directly accesses DB"

    return False, None

# anti_patterns/distributed_monolith.py
def detect_distributed_monolith():
    """Detect distributed monolith anti-pattern"""
    services = get_microservices()
    dependency_graph = build_service_dependency_graph()

    avg_dependencies = calculate_average_dependencies(dependency_graph)
    if avg_dependencies > 5:
        return True, f"High service coupling indicates distributed monolith"

    return False, None

# anti_patterns/hard_coded_spatial.py
def detect_hard_coded_spatial():
    """Detect hard-coded spatial assumptions"""
    code = get_codebase()
    hard_coded_crs = find_hard_coded_crs(code)
    hard_coded_bounds = find_hard_coded_bounds(code)

    if len(hard_coded_crs) > 0 or len(hard_coded_bounds) > 0:
        return True, "Hard-coded spatial assumptions detected"

    return False, None

# anti_patterns/versionless_contracts.py
def detect_versionless_contracts():
    """Detect versionless domain contracts"""
    contracts = get_domain_contracts()
    versioned_contracts = [c for c in contracts if has_version(c)]

    coverage = len(versioned_contracts) / len(contracts)
    if coverage < 0.8:
        return True, f"Low contract versioning coverage: {coverage}"

    return False, None

# fitness_functions/complete_example.py
import pytest
from architecture_metrics import ArchitectureMetrics

class TestServiceArchitecture:
    def test_boundary_integrity(self):
        """Ensure service boundaries are maintained"""
        service_code = get_service_code("user-service")

        # Check for direct database access
        db_imports = find_imports(service_code, "psycopg", "sqlalchemy")
        assert len(db_imports) == 0, "Service directly accesses database"

        # Check for UI dependencies
        ui_imports = find_imports(service_code, "nicegui", "react")
        assert len(ui_imports) == 0, "Service depends on UI components"

    def test_latency_sla(self):
        """Ensure service meets latency SLA"""
        metrics = ArchitectureMetrics()
        p95_latency = metrics.get_p95_latency("user-service")
        assert p95_latency < 500, f"P95 latency {p95_latency}ms exceeds SLA"

    def test_circuit_breaker_coverage(self):
        """Ensure external calls have circuit breakers"""
        external_calls = find_external_calls("user-service")
        protected_calls = find_circuit_breaker_protected_calls("user-service")

        coverage = len(protected_calls) / len(external_calls) if external_calls else 1.0
        assert coverage > 0.9, f"Circuit breaker coverage {coverage} too low"

# drift_measurement/quarterly_drift.py
class QuarterlyDriftMeasurement:
    def measure_drift(self):
        """Measure architecture drift quarterly"""
        baseline = load_baseline_architecture()
        current = capture_current_architecture()

        drift_report = {
            'structural_drift': self.measure_structural_drift(baseline, current),
            'behavioral_drift': self.measure_behavioral_drift(baseline, current),
            'governance_drift': self.measure_governance_drift(baseline, current),
            'recommendations': self.generate_recommendations(baseline, current)
        }

        save_drift_report(drift_report)
        return drift_report

# pre_commit/architecture_guardrails.py
def pre_commit_architecture_check():
    """Reject harmful architectural changes"""
    changes = get_staged_changes()

    # Check for boundary violations
    if violates_boundaries(changes):
        print("ERROR: Boundary violation detected")
        return False

    # Check for dependency violations
    if violates_dependency_rules(changes):
        print("ERROR: Dependency rule violation")
        return False

    # Check for complexity increases
    if increases_complexity(changes):
        print("WARNING: Complexity increase detected")
        # Allow but warn

    return True

# fail_safe/architecture_failures.py
class ArchitectureFailSafe:
    def handle_architecture_failure(self, failure: dict):
        """Handle architecture failure safely"""
        # Log failure
        log_architecture_failure(failure)

        # Alert team
        alert_architecture_team(failure)

        # Attempt automatic remediation
        if can_remediate_automatically(failure):
            remediate(failure)
        else:
            # Escalate to manual intervention
            escalate_to_manual(failure)

# llm/fitness_function_generator.py
class LLMFitnessFunctionGenerator:
    def generate_fitness_function(self, quality_attribute: str, component: str) -> str:
        """Generate fitness function using LLM"""
        prompt = f"""
        Generate a fitness function for {quality_attribute} in {component}.

        Requirements:
        1. Measurable and automatable
        2. Tests architectural quality, not functionality
        3. Includes assertions with thresholds
        4. Can run in CI/CD

        Provide Python code with pytest.
        """

        response = self.llm_client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an architecture fitness function expert."},
                {"role": "user", "content": prompt}
            ]
        )

        return response.choices[0].message.content

# llm/coupling_detector.py
class LLMCouplingDetector:
    def detect_excessive_coupling(self, codebase: dict) -> dict:
        """Detect excessive coupling using LLM"""
        prompt = f"""
        Analyze this codebase for excessive coupling:

        {json.dumps(codebase, indent=2)}

        Identify:
        1. Highly coupled components
        2. Coupling patterns
        3. Recommendations for reducing coupling
        """

        response = self.llm_client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are a coupling analysis expert."},
                {"role": "user", "content": prompt}
            ]
        )

        return json.loads(response.choices[0].message.content)

# llm/architecture_improver.py
class LLMArchitectureImprover:
    def propose_improvements(self, current_architecture: dict, metrics: dict) -> List[dict]:
        """Propose architecture improvements using LLM"""
        prompt = f"""
        Propose architecture improvements based on:

        Current architecture:
        {json.dumps(current_architecture, indent=2)}

        Metrics:
        {json.dumps(metrics, indent=2)}

        Provide:
        1. Improvement recommendations
        2. Expected impact
        3. Implementation steps
        4. Priority levels
        """

        response = self.llm_client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an architecture improvement expert."},
                {"role": "user", "content": prompt}
            ]
        )

        return json.loads(response.choices[0].message.content)

# llm/dependency_analyzer.py
class LLMDependencyAnalyzer:
    def analyze_dependency_graph(self, graph: dict) -> dict:
        """Analyze dependency graph using LLM"""
        prompt = f"""
        Analyze this dependency graph:

        {json.dumps(graph, indent=2)}

        Identify:
        1. Cyclic dependencies
        2. Overly complex dependencies
        3. Missing abstractions
        4. Refactoring opportunities
        """

        response = self.llm_client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are a dependency graph analysis expert."},
                {"role": "user", "content": prompt}
            ]
        )

        return json.loads(response.choices[0].message.content)

# llm/adr_to_standards.py
class LLMADRToStandardsConverter:
    def convert_adr_to_standards(self, adr: dict) -> dict:
        """Convert ADR to measurable standards using LLM"""
        prompt = f"""
        Convert this ADR to measurable architectural standards:

        {json.dumps(adr, indent=2)}

        Provide:
        1. Measurable criteria
        2. Fitness functions
        3. Acceptance thresholds
        4. Monitoring metrics
        """

        response = self.llm_client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an ADR to standards conversion expert."},
                {"role": "user", "content": prompt}
            ]
        )

        return json.loads(response.choices[0].message.content)