ADR-010: Configuration Access Patterns

Status

ACCEPTED - July 21, 2025

Context

PM-015 Group 2 analysis revealed configuration pattern inconsistency across services that blocks 100% test success rate:

While achieving 91% test success rate demonstrates architectural robustness, the mixed patterns pose long-term maintainability risks, testing inconsistencies, and developer confusion about configuration access approaches.

Business Impact: Configuration inconsistency affects:

Technical Debt: Two GitHub issues created for systematic resolution:

Decision

We will implement Hybrid Configuration Access with Clean Abstractions using layer-appropriate patterns that balance pragmatism with architectural consistency.

Layer-Specific Rules

  1. Application/Domain Layers: Use ConfigService exclusively
    • WorkflowService, IntentService, OrchestrationEngine
    • Business logic requiring configuration decisions
    • All user-facing feature behavior
  2. Infrastructure Layer: ConfigService preferred, direct environment access allowed for:
    • Feature flags and toggles (ENABLE_MCP_FILE_SEARCH)
    • Runtime detection (e.g., MCP availability)
    • Emergency overrides and circuit breakers
    • Container orchestration variables
  3. Testing: Mock ConfigService, not environment variables
    • Consistent test behavior across environments
    • Elimination of environment-dependent test failures
    • Cleaner test setup and isolation

Strategic Rationale

Approved Patterns with Code Examples

PRIMARY PATTERN - Application/Domain Services

# ✅ APPROVED: Service layer with ConfigService injection
class WorkflowService:
    def __init__(self, config_service: ConfigService):
        self.config = config_service

    async def process_workflow(self, workflow_type: WorkflowType):
        # Application configuration through service
        if self.config.feature_enabled("advanced_workflows"):
            return await self._process_advanced_workflow(workflow_type)
        return await self._process_basic_workflow(workflow_type)

    def get_timeout_config(self) -> int:
        return self.config.get_timeout("workflow_execution", default=300)

INFRASTRUCTURE PATTERN - Repository Layer

# ✅ APPROVED: Repository with ConfigService + infrastructure utilities
from services.infrastructure.config.feature_flags import FeatureFlags

class FileRepository(BaseRepository):
    def __init__(self, session: AsyncSession, config_service: ConfigService):
        super().__init__(session)
        self.config = config_service

    async def search_files_with_content(self, session_id: str, query: str, limit: int = 10):
        # Get database results first (always available)
        filename_matches = await self.search_files_by_name(session_id, query)

        # Infrastructure layer feature detection
        if FeatureFlags.is_mcp_content_search_enabled():
            try:
                return await self._enhanced_mcp_search(session_id, query, filename_matches, limit)
            except Exception as e:
                logger.warning(f"MCP search failed, falling back: {e}")

        return filename_matches[:limit]

    def _get_repository_config(self) -> dict:
        # Repository-specific configuration through ConfigService
        return {
            "cache_ttl": self.config.get_int("file_cache_ttl", 300),
            "max_results": self.config.get_int("max_file_results", 1000)
        }

FEATURE FLAGS UTILITY (New Infrastructure Component)

# ✅ APPROVED: Infrastructure utility for feature flags
# File: services/infrastructure/config/feature_flags.py

import os
import logging
from typing import Optional

logger = logging.getLogger(__name__)

class FeatureFlags:
    """
    Static utility for feature flag access in infrastructure layer.

    Use this for infrastructure-level toggles, runtime detection,
    and emergency overrides. Application logic should use ConfigService.
    """

    @staticmethod
    def is_mcp_content_search_enabled() -> bool:
        """Check if MCP content search is enabled"""
        return FeatureFlags._get_boolean_flag("ENABLE_MCP_FILE_SEARCH", False)

    @staticmethod
    def is_mcp_connection_pooling_enabled() -> bool:
        """Check if MCP connection pooling is enabled"""
        return FeatureFlags._get_boolean_flag("USE_MCP_POOL", False)

    @staticmethod
    def is_debug_mode_enabled() -> bool:
        """Check if debug mode is enabled (infrastructure logging)"""
        return FeatureFlags._get_boolean_flag("DEBUG", False)

    @staticmethod
    def get_max_connection_pool_size() -> int:
        """Get infrastructure connection pool configuration"""
        return FeatureFlags._get_int_flag("MAX_CONNECTION_POOL_SIZE", 5)

    @staticmethod
    def _get_boolean_flag(flag_name: str, default: bool = False) -> bool:
        """Internal: Get boolean environment variable"""
        value = os.getenv(flag_name, str(default)).lower()
        return value in ("true", "1", "yes", "on")

    @staticmethod
    def _get_int_flag(flag_name: str, default: int) -> int:
        """Internal: Get integer environment variable"""
        try:
            return int(os.getenv(flag_name, str(default)))
        except (ValueError, TypeError):
            logger.warning(f"Invalid integer value for {flag_name}, using default {default}")
            return default

ENHANCED CONFIGSERVICE PATTERN

# ✅ APPROVED: Enhanced ConfigService usage
class MCPResourceManager:
    def __init__(self, client_config: Optional[Dict[str, Any]] = None,
                 config_service: Optional[ConfigService] = None):
        self.client_config = client_config or self._get_default_client_config()
        self.config_service = config_service or get_config_service()

        # Application configuration through service
        self.cache_ttl = self.config_service.get_int("mcp_cache_ttl", 300)
        self.timeout = self.config_service.get_float("mcp_timeout", 5.0)

        # Infrastructure feature detection
        self.use_connection_pool = (
            FeatureFlags.is_mcp_connection_pooling_enabled() and
            self._is_pool_infrastructure_available()
        )

    def _get_default_client_config(self) -> Dict[str, Any]:
        # Configuration service for application config
        return {
            "url": self.config_service.get_string("mcp_server_url", "stdio://./scripts/mcp_file_server.py"),
            "timeout": self.config_service.get_float("mcp_timeout", 5.0)
        }

    def _is_pool_infrastructure_available(self) -> bool:
        # Infrastructure runtime detection
        try:
            from services.infrastructure.mcp.connection_pool import MCPConnectionPool
            return True
        except ImportError:
            return False

Counter-Examples (What NOT to Do)

❌ AVOID: Direct environment access in domain/application layers

# ❌ BAD: Domain service accessing environment directly
class WorkflowService:
    async def process_workflow(self, workflow_type: WorkflowType):
        # This violates layer boundaries - domain logic shouldn't know about environment
        if os.getenv("ENABLE_ADVANCED_WORKFLOWS", "false").lower() == "true":
            return await self._process_advanced_workflow(workflow_type)
        return await self._process_basic_workflow(workflow_type)

❌ AVOID: Mixed patterns in same service

# ❌ BAD: Inconsistent configuration access
class MCPResourceManager:
    def __init__(self):
        # Sometimes use ConfigService
        if CONFIG_SERVICE_AVAILABLE:
            config = get_config()
            self.timeout = config.mcp_timeout
        else:
            # Sometimes use environment - creates confusion and testing issues
            self.timeout = float(os.getenv("MCP_TIMEOUT", "5.0"))
            self.use_pool = os.getenv("USE_MCP_POOL", "false").lower() == "true"

❌ AVOID: Testing environment variables directly

# ❌ BAD: Environment patching in tests creates brittle, environment-dependent tests
@pytest.fixture
def test_with_mcp_enabled():
    original = os.environ.get("ENABLE_MCP_FILE_SEARCH")
    os.environ["ENABLE_MCP_FILE_SEARCH"] = "true"
    yield
    if original is None:
        del os.environ["ENABLE_MCP_FILE_SEARCH"]
    else:
        os.environ["ENABLE_MCP_FILE_SEARCH"] = original

def test_mcp_search_enabled(test_with_mcp_enabled):
    # Test becomes environment-dependent and brittle
    result = repository.search_files_with_content("session", "query")

✅ CORRECT: Mock configuration service

# ✅ GOOD: Test configuration behavior through service mocking
@pytest.fixture
def mock_config_with_mcp_enabled():
    mock_config = Mock(spec=ConfigService)
    mock_config.feature_enabled.return_value = True
    mock_config.get_int.return_value = 300
    return mock_config

def test_mcp_search_enabled(mock_config_with_mcp_enabled):
    # Clean, predictable test that focuses on behavior
    repository = FileRepository(session, mock_config_with_mcp_enabled)
    result = await repository.search_files_with_content("session", "query")
    assert len(result) > 0

Migration Strategy

Phase 1: Document and Establish Patterns (This Week - July 21-25) ✅ COMPLETE

Phase 2: Gradual Service Migration (Next 2 Weeks - July 28 - August 8)

Priority 1: MCPResourceManager (GitHub Issue #39)

Priority 2: FileRepository (GitHub Issue #40)

Phase 3: Enforcement and Automation (August 2025)

ConfigService vs Environment Variables Guidelines

ConfigService Should Handle:

Direct Environment Access Allowed For:

Migration Checklist Template

For each service requiring configuration updates:

Analysis Phase:

Implementation Phase:

Validation Phase:

Success Metrics and Validation

Technical Metrics

Implementation Tracking

GitHub Issues Progress:

Test Health Validation:

Quality Assurance

Code Review Verification:

Automated Enforcement:

Long-term Benefits

Developer Experience:

Architectural Health:

PIPER.user.md Configuration Loading Pattern

Status: ✅ IMPLEMENTED (October 17, 2025) Reference Implementation: CalendarConfigService (CORE-MCP-MIGRATION #198)

Pattern Description

Integration services support user-specific configuration via config/PIPER.user.md, a gitignored markdown file containing YAML configuration blocks. This enables per-user customization without committing credentials or personal preferences to the repository.

Configuration Priority Order

All integrations following this pattern implement three-layer configuration:

  1. Environment variables (highest priority) - Overrides everything
  2. PIPER.user.md (middle priority) - User-specific defaults
  3. Hardcoded defaults (lowest priority) - Fallback values

Implementation Pattern

# ✅ APPROVED: Integration config service with PIPER.user.md loading
# File: services/integrations/calendar/config_service.py

import os
import re
import yaml
from pathlib import Path
from typing import Any, Dict

class CalendarConfigService:
    """
    Calendar configuration service with PIPER.user.md support.

    Implements three-layer configuration priority:
    1. Environment variables (highest)
    2. PIPER.user.md calendar section (middle)
    3. Hardcoded defaults (lowest)
    """

    def _load_from_user_config(self) -> Dict[str, Any]:
        """
        Load calendar configuration from PIPER.user.md.

        Returns:
            Dict with calendar configuration, or empty dict if not found/invalid
        """
        try:
            user_config_path = Path("config/PIPER.user.md")
            if not user_config_path.exists():
                return {}

            # Read markdown file
            content = user_config_path.read_text()

            # Find Calendar Integration section by heading
            calendar_section_match = re.search(
                r"##\s+📅\s+Calendar Integration(.*?)(?=##\s+|$)",
                content,
                re.DOTALL
            )

            if not calendar_section_match:
                return {}

            section_content = calendar_section_match.group(1)

            # Extract YAML from markdown code blocks
            yaml_match = re.search(r"```yaml.*?```", section_content, re.DOTALL)
            if not yaml_match:
                return {}

            # Parse YAML content
            full_yaml_block = yaml_match.group(0)
            yaml_content = (
                full_yaml_block.replace("```yaml", "")
                              .replace("```", "")
                              .strip()
            )
            config_data = yaml.safe_load(yaml_content)

            if config_data and "calendar" in config_data:
                return config_data["calendar"]

            return {}

        except Exception as e:
            # Graceful fallback - don't crash on config errors
            print(f"Warning: Could not load calendar config from PIPER.user.md: {e}")
            return {}

    def _load_config(self) -> CalendarConfig:
        """
        Load configuration with priority: env vars > PIPER.user.md > defaults.

        Returns:
            CalendarConfig: Configuration with priority order applied
        """
        # Load from PIPER.user.md first (base layer)
        user_config = self._load_from_user_config()

        # Environment variables override user config
        return CalendarConfig(
            client_secrets_file=os.getenv(
                "GOOGLE_CLIENT_SECRETS_FILE",
                user_config.get("client_secrets_file", "credentials.json"),
            ),
            token_file=os.getenv(
                "GOOGLE_TOKEN_FILE",
                user_config.get("token_file", "token.json")
            ),
            calendar_id=os.getenv(
                "GOOGLE_CALENDAR_ID",
                user_config.get("calendar_id", "primary")
            ),
            timeout_seconds=int(
                os.getenv(
                    "GOOGLE_CALENDAR_TIMEOUT",
                    str(user_config.get("timeout_seconds", 30))
                )
            ),
            # ... additional config fields
        )

PIPER.user.md Format

## 📅 Calendar Integration

Configure Google Calendar integration with OAuth 2.0 credentials.

```yaml
calendar:
  # OAuth 2.0 Configuration
  client_secrets_file: "credentials.json"
  token_file: "token.json"

  # API Configuration
  calendar_id: "primary"
  scopes:
    - "https://www.googleapis.com/auth/calendar.readonly"

  # Timeouts & Circuit Breaker
  timeout_seconds: 30
  circuit_timeout: 300
  error_threshold: 5

  # Feature Flags
  enable_spatial_mapping: true

Configuration Priority:

  1. Environment variables (highest - overrides everything)
  2. PIPER.user.md (middle - overrides defaults)
  3. Hardcoded defaults (lowest - fallback) ```

Testing Pattern

# ✅ APPROVED: Test PIPER.user.md loading with Path mocking
def test_loads_from_piper_user_md(tmp_path):
    """Test that config loads from PIPER.user.md when present."""
    piper_config = tmp_path / "PIPER.user.md"
    piper_config.write_text("""
## 📅 Calendar Integration

```yaml
calendar:
  client_secrets_file: "test_credentials.json"
  calendar_id: "test@calendar.com"
  timeout_seconds: 60

""")

with patch.object(Path, "exists", return_value=True):
    with patch.object(Path, "read_text", return_value=piper_config.read_text()):
        service = CalendarConfigService()
        config = service.get_config()

        # Verify values from PIPER.user.md
        assert config.client_secrets_file == "test_credentials.json"
        assert config.calendar_id == "test@calendar.com"
        assert config.timeout_seconds == 60

def test_env_vars_override_user_config(tmp_path): “"”Test that environment variables override PIPER.user.md.””” piper_config = tmp_path / “PIPER.user.md” piper_config.write_text(“””

calendar:
  calendar_id: "user_config@calendar.com"

""")

os.environ["GOOGLE_CALENDAR_ID"] = "env_override@calendar.com"

try:
    with patch.object(Path, "exists", return_value=True):
        with patch.object(Path, "read_text", return_value=piper_config.read_text()):
            service = CalendarConfigService()
            config = service.get_config()

            # Env var wins over user config
            assert config.calendar_id == "env_override@calendar.com"
finally:
    os.environ.pop("GOOGLE_CALENDAR_ID", None) ```

Implementation Requirements

  1. Graceful Fallback: Config loading must never crash the application
    • Missing PIPER.user.md → Return empty dict
    • Malformed YAML → Return empty dict and log warning
    • Missing section → Return empty dict
  2. Priority Order Enforcement: Environment variables ALWAYS override PIPER.user.md
    • Check if ENV_VAR in os.environ for explicit override detection
    • Use os.getenv(ENV_VAR, user_config.get(key, default)) pattern
  3. Section Identification: Use regex to find service-specific sections
    • Pattern: r"##\s+📅\s+Calendar Integration(.*?)(?=##\s+|$)"
    • Emoji markers provide visual organization in PIPER.user.md
    • DOTALL flag to capture multiline content
  4. YAML Extraction: Parse YAML from markdown code blocks
    • Pattern: r"yaml.*?" with DOTALL flag
    • Remove markdown markers before yaml.safe_load()
    • Look for service key in parsed dict (e.g., “calendar”)

Slack Configuration Service

Location: services/integrations/slack/config_service.py

Pattern: Service injection with PIPER.user.md YAML parsing

Architecture Note: Slack uses direct spatial (ADR-039), NOT tool-based MCP

Configuration Priority:

  1. Environment variables (highest priority)
    • SLACK_BOT_TOKEN
    • SLACK_APP_TOKEN
    • SLACK_SIGNING_SECRET
    • SLACK_API_BASE_URL
    • SLACK_TIMEOUT_SECONDS
    • SLACK_MAX_RETRIES
    • SLACK_ENVIRONMENT
    • SLACK_DEFAULT_CHANNEL
    • SLACK_RATE_LIMIT_RPM
    • SLACK_BURST_LIMIT
    • SLACK_WEBHOOK_URL
    • SLACK_CLIENT_ID
    • SLACK_CLIENT_SECRET
    • SLACK_REDIRECT_URI
  2. PIPER.user.md configuration (middle priority) 
    slack:
  authentication:
    bot_token: "xoxb-..."
    app_token: "xapp-..."
    signing_secret: "..."

  api:
    base_url: "https://slack.com/api"
    timeout_seconds: 30
    max_retries: 3
    environment: "development"  # development|staging|production

  behavior:
    default_channel: "general"
    rate_limit_per_minute: 60
    burst_limit: 10
    webhook_url: "https://hooks.slack.com/..."

  features:
    enable_webhooks: true
    enable_socket_mode: false
    enable_spatial_mapping: true

  oauth:
    client_id: "..."
    client_secret: "..."
    redirect_uri: "..."
  3. Hardcoded defaults (lowest priority)
    • Empty tokens
    • Default behavior settings
    • 60/min rate limit
    • 3 retries
    • 30s timeout

Implementation:

class SlackConfigService:
    def _load_from_user_config(self) -> Dict[str, Any]:
        """Load Slack configuration from PIPER.user.md"""
        # Parse YAML from markdown
        # Supports two patterns: ## 💬 Slack Integration and slack:
        # Return slack: section configuration

    def _load_config(self) -> SlackConfig:
        """Load with 3-layer priority"""
        user_config = self._load_from_user_config()
        auth = user_config.get("authentication", {})
        api = user_config.get("api", {})
        behavior = user_config.get("behavior", {})
        features = user_config.get("features", {})
        oauth = user_config.get("oauth", {})

        return SlackConfig(
            bot_token=os.getenv("SLACK_BOT_TOKEN", auth.get("bot_token", "")),
            app_token=os.getenv("SLACK_APP_TOKEN", api.get("app_token", "")),
            # ... other fields with same 3-layer pattern
        )

Test Coverage: 36 comprehensive tests:

Status: ✅ Implemented (October 2025, Phase 3)

Architectural Difference: Unlike Calendar/Notion (tool-based MCP), Slack uses direct spatial architecture per ADR-039.

Integrations Using This Pattern

Benefits

User Experience:

Testing:

Maintenance:

Configuration Pattern Summary

Integration Config Service PIPER.user.md Env Vars Tests Architecture Status
Calendar ✅ Yes ✅ Yes ✅ Yes 8 Tool-based 100%
GitHub ✅ Yes ✅ Yes ✅ Yes 16 Delegated 100%
Notion ✅ Yes ✅ Yes ✅ Yes 19 Tool-based 100%
Slack ✅ Yes ✅ Yes ✅ Yes 36 Direct 100%

Pattern Consistency: All MCP integrations follow the same configuration approach:

Architectural Diversity: While configuration patterns are consistent, integration architectures vary based on requirements:

Implementation History

Phase 1: Calendar (October 17, 2025)

Phase 1: GitHub (October 17, 2025)

Phase 2: Notion (October 18, 2025)

Phase 3: Slack (October 18, 2025)

Pattern Evolution: Configuration approach proven across 4 integrations with 3 different architectural patterns (tool-based, delegated, direct spatial).

Decision Rationale: This ADR provides the architectural foundation needed to eliminate configuration-related technical debt while supporting the pragmatic needs of different service layers. The hybrid approach balances architectural purity with development productivity, ensuring long-term maintainability without requiring disruptive wholesale rewrites.

Next Steps: Implement FeatureFlags utility, update pattern catalog, and begin Phase 2 service migrations per GitHub issues #39 and #40.