Spaces:
Sleeping
Sleeping
| """ | |
| Multi-Tenant Configuration Manager (Phase 3B). | |
| Manages tenant-specific configuration overrides including: | |
| - Enabled wedges and journeys per tenant | |
| - Custom escalation rules (institution-specific clinical protocols) | |
| - Domain overrides (enable/disable specific domains) | |
| - Institution context (nurse line, ED name, timezone, branding) | |
| - FHIR data source configuration | |
| - Compliance settings (audit retention, PHI logging) | |
| Usage: | |
| manager = TenantManager(config) | |
| tenant = manager.get_tenant("example_health_system") | |
| merged_rules = manager.get_merged_escalation_rules("example_health_system") | |
| enabled_domains = manager.get_enabled_domains("example_health_system") | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from typing import Any, Dict, List, Optional, Set | |
| from decision.engine.config_loader import DecisionConfigLoader | |
| logger = logging.getLogger("decision.tenant_manager") | |
| class TenantConfig: | |
| """Parsed tenant configuration with convenience accessors.""" | |
| def __init__(self, tenant_id: str, raw: Dict[str, Any]): | |
| self.tenant_id = tenant_id | |
| self._raw = raw | |
| def name(self) -> str: | |
| return self._raw.get("institution_name", self.tenant_id) | |
| def type(self) -> str: | |
| return self._raw.get("institution_type", "unknown") | |
| def enabled_wedges(self) -> List[str]: | |
| return self._raw.get("enabled_wedges", []) | |
| def enabled_journeys(self) -> List[str]: | |
| return self._raw.get("enabled_journeys", self.enabled_wedges) | |
| def custom_escalation_rules(self) -> List[Dict[str, Any]]: | |
| return self._raw.get("custom_escalation_rules", []) | |
| def domain_overrides(self) -> Dict[str, Any]: | |
| return self._raw.get("domain_overrides", {}) | |
| def disabled_domains(self) -> Set[str]: | |
| overrides = self.domain_overrides | |
| return { | |
| domain | |
| for domain, cfg in overrides.items() | |
| if isinstance(cfg, dict) and not cfg.get("enabled", True) | |
| } | |
| def institution_context(self) -> Dict[str, Any]: | |
| return self._raw.get("institution_context", {}) | |
| def nurse_line_number(self) -> str: | |
| return self.institution_context.get("nurse_line_number", "") | |
| def ed_name(self) -> str: | |
| return self.institution_context.get("ed_name", "the emergency department") | |
| def timezone(self) -> str: | |
| return self.institution_context.get("timezone", "America/New_York") | |
| def operating_hours(self) -> Dict[str, str]: | |
| return self.institution_context.get("operating_hours", {}) | |
| def fhir_config(self) -> Dict[str, Any]: | |
| return self._raw.get("data_source", {}).get("fhir", {}) | |
| def branding(self) -> Dict[str, str]: | |
| return self._raw.get("branding", {}) | |
| def greeting_name(self) -> str: | |
| return self.branding.get("agent_introduction", "Avery, a care assistant") | |
| def compliance(self) -> Dict[str, Any]: | |
| return self._raw.get("compliance", {}) | |
| def audit_retention_years(self) -> int: | |
| return self.compliance.get("audit_retention_years", 7) | |
| def __repr__(self) -> str: | |
| return f"TenantConfig({self.tenant_id}, name={self.name}, wedges={self.enabled_wedges})" | |
| class TenantManager: | |
| """ | |
| Manages multi-tenant configurations and rule merging. | |
| Provides: | |
| - Tenant lookup and validation | |
| - Merged escalation rules (global + tenant custom) | |
| - Domain filtering (respect tenant disabled_domains) | |
| - Institution context for response generation | |
| """ | |
| def __init__(self, config: DecisionConfigLoader): | |
| self._config = config | |
| self._tenants: Dict[str, TenantConfig] = {} | |
| for tid, raw in config.tenants.items(): | |
| self._tenants[tid] = TenantConfig(tid, raw) | |
| logger.info("TenantManager initialized with %d tenants", len(self._tenants)) | |
| def tenant_ids(self) -> List[str]: | |
| return sorted(self._tenants.keys()) | |
| def get_tenant(self, tenant_id: str) -> Optional[TenantConfig]: | |
| return self._tenants.get(tenant_id) | |
| def get_merged_escalation_rules( | |
| self, tenant_id: str | |
| ) -> List[Dict[str, Any]]: | |
| """ | |
| Merge global escalation rules with tenant-specific custom rules. | |
| Tenant rules are appended AFTER global rules, so they take effect | |
| as additional checks. Tenant rules with the same ID as global rules | |
| override the global version. | |
| """ | |
| global_rules = list(self._config.risk_escalation_rules) | |
| tenant = self._tenants.get(tenant_id) | |
| if not tenant: | |
| return global_rules | |
| custom = tenant.custom_escalation_rules | |
| if not custom: | |
| return global_rules | |
| # Build a map of global rules by ID | |
| global_ids = {r.get("id"): i for i, r in enumerate(global_rules)} | |
| merged = list(global_rules) | |
| for rule in custom: | |
| rule_id = rule.get("id") | |
| if rule_id and rule_id in global_ids: | |
| # Override existing rule | |
| merged[global_ids[rule_id]] = rule | |
| logger.info( | |
| "Tenant %s overrides global rule: %s", tenant_id, rule_id | |
| ) | |
| else: | |
| # Append new rule | |
| merged.append(rule) | |
| logger.info( | |
| "Tenant %s adds custom rule: %s", tenant_id, rule_id | |
| ) | |
| return merged | |
| def get_enabled_domains( | |
| self, tenant_id: str, all_domains: Optional[List[str]] = None | |
| ) -> List[str]: | |
| """ | |
| Get domains enabled for a tenant (all domains minus disabled ones). | |
| """ | |
| if all_domains is None: | |
| all_domains = self._config.domain_names | |
| tenant = self._tenants.get(tenant_id) | |
| if not tenant: | |
| return all_domains | |
| disabled = tenant.disabled_domains | |
| if not disabled: | |
| return all_domains | |
| enabled = [d for d in all_domains if d not in disabled] | |
| logger.info( | |
| "Tenant %s: %d domains enabled (%d disabled: %s)", | |
| tenant_id, | |
| len(enabled), | |
| len(disabled), | |
| disabled, | |
| ) | |
| return enabled | |
| def is_wedge_enabled(self, tenant_id: str, wedge_type: str) -> bool: | |
| """Check if a wedge/journey is enabled for a tenant.""" | |
| tenant = self._tenants.get(tenant_id) | |
| if not tenant: | |
| return True # No tenant config = all enabled | |
| return wedge_type in tenant.enabled_wedges | |
| def get_institution_context_for_prompts( | |
| self, tenant_id: str | |
| ) -> Dict[str, str]: | |
| """ | |
| Get institution-specific context for LLM prompt injection. | |
| Returns strings that can be inserted into response specs. | |
| """ | |
| tenant = self._tenants.get(tenant_id) | |
| if not tenant: | |
| return { | |
| "agent_name": "Avery", | |
| "institution_name": "your healthcare provider", | |
| "nurse_line": "the nurse line", | |
| "ed_name": "the emergency department", | |
| } | |
| return { | |
| "agent_name": "Avery", | |
| "institution_name": tenant.name, | |
| "nurse_line": tenant.nurse_line_number or "the nurse line", | |
| "ed_name": tenant.ed_name, | |
| "greeting": tenant.greeting_name, | |
| } | |