# core/boundary_manager.py
"""
Boundary Manager for ARF Demo - Enforces clear separation between real OSS and simulated Enterprise
Ensures the audience always knows what's real vs simulated
"""
import logging
from typing import Dict, Any, Tuple
from dataclasses import dataclass
from enum import Enum
logger = logging.getLogger(__name__)
class SystemMode(Enum):
"""Clear system mode definitions"""
REAL_OSS_ADVISORY = "real_oss_advisory" # Real ARF OSS package
SIMULATED_ENTERPRISE = "simulated_enterprise" # Enterprise simulation
MOCK_FALLBACK = "mock_fallback" # Mock when nothing is available
@dataclass
class SystemBoundary:
"""Clear boundary definition with labels"""
mode: SystemMode
real_components: list[str]
simulated_components: list[str]
oss_license: str
enterprise_license: str
execution_allowed: bool
def get_display_labels(self) -> Dict[str, str]:
"""Get clear display labels for UI"""
base_labels = {
"system_mode": self.mode.value,
"oss_status": f"✅ REAL ARF OSS v3.3.7" if self.mode == SystemMode.REAL_OSS_ADVISORY else "⚠️ MOCK ARF",
"enterprise_status": f"🎭 SIMULATED Enterprise" if self.mode == SystemMode.SIMULATED_ENTERPRISE else "⚠️ MOCK Enterprise",
"execution_capability": "Advisory Only (Apache 2.0)" if not self.execution_allowed else "Autonomous Execution (Enterprise)",
"license_display": f"OSS: {self.oss_license} | Enterprise: {self.enterprise_license}",
"boundary_note": "OSS advises, Enterprise executes" if self.mode == SystemMode.SIMULATED_ENTERPRISE else "Mock mode for demo"
}
# Add color coding
if self.mode == SystemMode.REAL_OSS_ADVISORY:
base_labels.update({
"oss_color": "#10b981", # Green for real
"enterprise_color": "#f59e0b", # Amber for simulated
"oss_icon": "✅",
"enterprise_icon": "🎭"
})
else:
base_labels.update({
"oss_color": "#64748b", # Gray for mock
"enterprise_color": "#64748b", # Gray for mock
"oss_icon": "⚠️",
"enterprise_icon": "⚠️"
})
return base_labels
class BoundaryManager:
"""Manages system boundaries and ensures clear labeling"""
def __init__(self):
self.current_boundary = None
self.installation_status = self._check_installation()
self._initialize_boundary()
def _check_installation(self) -> Dict[str, Any]:
"""Check what's really installed"""
# Simplified version - in real app this checks actual packages
try:
from core.true_arf_oss import TrueARFOSS
oss_available = True
except ImportError:
oss_available = False
try:
from core.enterprise_simulation import EnterpriseFeatureSimulation
enterprise_available = True
except ImportError:
enterprise_available = False
return {
"oss_available": oss_available,
"enterprise_available": enterprise_available,
"true_arf_version": "3.3.7" if oss_available else "mock"
}
def _initialize_boundary(self):
"""Initialize the system boundary based on what's available"""
installation = self.installation_status
if installation["oss_available"]:
# Real OSS + Simulated Enterprise
self.current_boundary = SystemBoundary(
mode=SystemMode.REAL_OSS_ADVISORY,
real_components=["TrueARFOSS", "Detection Agent", "Recall Agent", "Decision Agent"],
simulated_components=["EnterpriseExecution", "RollbackGuarantees", "NovelExecutionProtocols"],
oss_license="Apache 2.0",
enterprise_license="SIMULATED (requires Commercial)",
execution_allowed=False # OSS is advisory only
)
logger.info("✅ System initialized with REAL ARF OSS + SIMULATED Enterprise")
elif installation["enterprise_available"]:
# Mock OSS + Simulated Enterprise (unlikely but possible)
self.current_boundary = SystemBoundary(
mode=SystemMode.SIMULATED_ENTERPRISE,
real_components=[],
simulated_components=["EnterpriseFeatures", "ExecutionProtocols"],
oss_license="MOCK",
enterprise_license="SIMULATED",
execution_allowed=True # Simulated execution
)
logger.info("⚠️ System initialized with MOCK OSS + SIMULATED Enterprise")
else:
# Complete mock mode
self.current_boundary = SystemBoundary(
mode=SystemMode.MOCK_FALLBACK,
real_components=[],
simulated_components=["AllComponents"],
oss_license="MOCK",
enterprise_license="MOCK",
execution_allowed=False
)
logger.info("⚠️ System initialized in MOCK FALLBACK mode")
def get_boundary_badges(self) -> str:
"""Get HTML badges showing clear boundaries"""
labels = self.current_boundary.get_display_labels()
return f"""
{labels['oss_icon']} {labels['oss_status']}
{labels['enterprise_icon']} {labels['enterprise_status']}
{labels['execution_capability']}
"""
def get_agent_html(self, agent_name: str, is_real: bool = True, status: str = "Active") -> str:
"""Get agent HTML with clear boundary indicators"""
icons = {
"Detection": "🕵️♂️",
"Recall": "🧠",
"Decision": "🎯"
}
real_badge = """
REAL ARF
""" if is_real else """
DEMO MODE
"""
border_color = "#10b981" if is_real else "#f59e0b"
background = "#f0fdf4" if is_real else "#fef3c7"
return f"""
{real_badge}
{icons.get(agent_name, '🤖')}
{agent_name} Agent
Status: {status}
{'Running on REAL ARF OSS v3.3.7' if is_real else 'Running in DEMO MODE'}
{'ACTIVE (REAL)' if is_real else 'SIMULATED'}
🎭
SIMULATED ENTERPRISE EXECUTION
Action: {action}
Mode: Enterprise Simulation (not real execution)
Boundary: OSS advises → Enterprise would execute
DEMO BOUNDARY
In production, Enterprise edition would execute against real infrastructure
"""
else:
return f"""
⚡
REAL ENTERPRISE EXECUTION
Action: {action}
Mode: Enterprise Autonomous
Boundary: Real execution with safety guarantees
ENTERPRISE+
"""
def get_demo_narrative(self, phase: str) -> str:
"""Get narrative text for each demo phase"""
narratives = {
"introduction": """
🎯 Demo Introduction
This demo shows the clear architectural boundary between ARF OSS (real advisory intelligence)
and ARF Enterprise (simulated autonomous execution). We're showing what happens in production,
not hiding behind mock data.
""",
"oss_analysis": """
🧠 Real OSS Intelligence
ARF OSS v3.3.7 is analyzing the incident in real-time. This is not a mock - it's the actual
ARF OSS package running detection, recall, and decision agents. Notice the confidence scores
and reasoning chain.
""",
"enterprise_simulation": """
🎭 Enterprise Simulation Boundary
This is where we simulate Enterprise execution. In production, Enterprise would:
1. Validate safety constraints, 2. Execute with rollback guarantees, 3. Update the learning engine.
We're showing the value proposition without real infrastructure access.
""",
"conclusion": """
✅ Architecture Validated
What we demonstrated:
• Real ARF OSS intelligence (advisory)
• Clear execution boundary (OSS vs Enterprise)
• Simulated Enterprise value proposition
• Production-ready architecture pattern
This isn't AI theater - it's a production-grade reliability system with honest boundaries.
"""
}
return narratives.get(phase, "")
def validate_transition(self, from_mode: SystemMode, to_mode: SystemMode) -> Tuple[bool, str]:
"""Validate mode transitions (e.g., can't go from mock to real execution)"""
transitions = {
(SystemMode.REAL_OSS_ADVISORY, SystemMode.SIMULATED_ENTERPRISE): (True, "Valid: Real OSS to Simulated Enterprise"),
(SystemMode.MOCK_FALLBACK, SystemMode.SIMULATED_ENTERPRISE): (True, "Valid: Mock to Simulated Enterprise"),
(SystemMode.SIMULATED_ENTERPRISE, SystemMode.REAL_OSS_ADVISORY): (False, "Invalid: Can't go from Enterprise simulation back to OSS in demo"),
}
result = transitions.get((from_mode, to_mode), (True, "Valid transition"))
return result
# Singleton instance
_boundary_manager = None
def get_boundary_manager() -> BoundaryManager:
"""Get singleton BoundaryManager instance"""
global _boundary_manager
if _boundary_manager is None:
_boundary_manager = BoundaryManager()
return _boundary_manager