Agentic-Reliability-Framework-API / core /boundary_manager.py
petter2025's picture
Create boundary_manager.py
04f5ca1 verified
raw
history blame
16.4 kB
# 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"""
<div style="display: flex; justify-content: center; gap: 10px; margin-top: 10px; flex-wrap: wrap;">
<span style="padding: 4px 12px; background: {labels['oss_color']};
color: white; border-radius: 20px; font-size: 12px; font-weight: bold;
display: flex; align-items: center; gap: 6px;">
{labels['oss_icon']} {labels['oss_status']}
</span>
<span style="padding: 4px 12px; background: {labels['enterprise_color']};
color: white; border-radius: 20px; font-size: 12px; font-weight: bold;
display: flex; align-items: center; gap: 6px;">
{labels['enterprise_icon']} {labels['enterprise_status']}
</span>
<span style="padding: 4px 12px; background: #3b82f6;
color: white; border-radius: 20px; font-size: 12px; font-weight: bold;">
{labels['execution_capability']}
</span>
</div>
"""
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 = """
<div style="position: absolute; top: -8px; right: -8px; padding: 2px 8px; background: #10b981;
color: white; border-radius: 12px; font-size: 10px; font-weight: bold; z-index: 10;
border: 2px solid white; box-shadow: 0 2px 4px rgba(0,0,0,0.1);">
REAL ARF
</div>
""" if is_real else """
<div style="position: absolute; top: -8px; right: -8px; padding: 2px 8px; background: #f59e0b;
color: white; border-radius: 12px; font-size: 10px; font-weight: bold; z-index: 10;
border: 2px solid white; box-shadow: 0 2px 4px rgba(0,0,0,0.1);">
DEMO MODE
</div>
"""
border_color = "#10b981" if is_real else "#f59e0b"
background = "#f0fdf4" if is_real else "#fef3c7"
return f"""
<div style="position: relative;">
{real_badge}
<div style="border: 3px solid {border_color}; border-radius: 16px; padding: 20px;
background: {background}; text-align: center; min-height: 200px;
display: flex; flex-direction: column; align-items: center; justify-content: center;">
<div style="font-size: 42px; margin-bottom: 15px; opacity: 0.9;">{icons.get(agent_name, '🤖')}</div>
<div style="width: 100%;">
<h4 style="margin: 0 0 10px 0; font-size: 18px; color: #1e293b; font-weight: 600;">{agent_name} Agent</h4>
<p style="font-size: 14px; color: #475569; margin-bottom: 15px; line-height: 1.5;">
Status: <strong>{status}</strong><br>
<span style="font-size: 12px; color: {'#059669' if is_real else '#d97706'}">
{'Running on REAL ARF OSS v3.3.7' if is_real else 'Running in DEMO MODE'}
</span>
</p>
<div style="display: inline-block; padding: 8px 20px; background: linear-gradient(135deg, {border_color} 0%, {border_color}88 100%);
border-radius: 20px; font-size: 13px; font-weight: bold; color: white;
text-transform: uppercase; letter-spacing: 0.5px; margin-top: 10px;">
{'ACTIVE (REAL)' if is_real else 'SIMULATED'}
</div>
</div>
</div>
</div>
"""
def get_execution_boundary_html(self, action: str, is_simulated: bool = True) -> str:
"""Get clear execution boundary indicator"""
if is_simulated:
return f"""
<div style="border: 3px dashed #f59e0b; border-radius: 16px; padding: 25px;
background: linear-gradient(135deg, #fef3c7 0%, #fde68a 100%);
text-align: center; margin: 20px 0;">
<div style="font-size: 36px; margin-bottom: 15px;">🎭</div>
<h4 style="margin: 0 0 12px 0; font-size: 20px; color: #92400e; font-weight: 700;">
SIMULATED ENTERPRISE EXECUTION
</h4>
<p style="font-size: 15px; color: #92400e; margin-bottom: 15px; line-height: 1.6;">
<strong>Action:</strong> {action}<br>
<strong>Mode:</strong> Enterprise Simulation (not real execution)<br>
<strong>Boundary:</strong> OSS advises → Enterprise would execute
</p>
<div style="display: inline-block; padding: 10px 24px; background: #92400e;
border-radius: 20px; font-size: 14px; font-weight: bold; color: white;
text-transform: uppercase; letter-spacing: 1px;">
DEMO BOUNDARY
</div>
<p style="font-size: 13px; color: #92400e; margin-top: 15px; font-style: italic;">
In production, Enterprise edition would execute against real infrastructure
</p>
</div>
"""
else:
return f"""
<div style="border: 3px solid #10b981; border-radius: 16px; padding: 25px;
background: linear-gradient(135deg, #f0fdf4 0%, #bbf7d0 100%);
text-align: center; margin: 20px 0;">
<div style="font-size: 36px; margin-bottom: 15px;">⚡</div>
<h4 style="margin: 0 0 12px 0; font-size: 20px; color: #065f46; font-weight: 700;">
REAL ENTERPRISE EXECUTION
</h4>
<p style="font-size: 15px; color: #065f46; margin-bottom: 15px; line-height: 1.6;">
<strong>Action:</strong> {action}<br>
<strong>Mode:</strong> Enterprise Autonomous<br>
<strong>Boundary:</strong> Real execution with safety guarantees
</p>
<div style="display: inline-block; padding: 10px 24px; background: #065f46;
border-radius: 20px; font-size: 14px; font-weight: bold; color: white;
text-transform: uppercase; letter-spacing: 1px;">
ENTERPRISE+
</div>
</div>
"""
def get_demo_narrative(self, phase: str) -> str:
"""Get narrative text for each demo phase"""
narratives = {
"introduction": """
<div style="background: #f8fafc; border-radius: 12px; padding: 20px; margin: 20px 0; border-left: 4px solid #3b82f6;">
<h4 style="margin: 0 0 10px 0; color: #1e293b;">🎯 Demo Introduction</h4>
<p style="margin: 0; color: #475569; font-size: 14px; line-height: 1.6;">
This demo shows the <strong>clear architectural boundary</strong> between ARF OSS (real advisory intelligence)
and ARF Enterprise (simulated autonomous execution). We're showing what happens in production,
not hiding behind mock data.
</p>
</div>
""",
"oss_analysis": """
<div style="background: #f0fdf4; border-radius: 12px; padding: 20px; margin: 20px 0; border-left: 4px solid #10b981;">
<h4 style="margin: 0 0 10px 0; color: #065f46;">🧠 Real OSS Intelligence</h4>
<p style="margin: 0; color: #047857; font-size: 14px; line-height: 1.6;">
ARF OSS v3.3.7 is <strong>analyzing the incident in real-time</strong>. 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.
</p>
</div>
""",
"enterprise_simulation": """
<div style="background: #fef3c7; border-radius: 12px; padding: 20px; margin: 20px 0; border-left: 4px solid #f59e0b;">
<h4 style="margin: 0 0 10px 0; color: #92400e;">🎭 Enterprise Simulation Boundary</h4>
<p style="margin: 0; color: #b45309; font-size: 14px; line-height: 1.6;">
This is where we <strong>simulate Enterprise execution</strong>. 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.
</p>
</div>
""",
"conclusion": """
<div style="background: linear-gradient(135deg, #f8fafc 0%, #e2e8f0 100%); border-radius: 12px; padding: 20px; margin: 20px 0; border: 2px solid #3b82f6;">
<h4 style="margin: 0 0 10px 0; color: #1e293b;">✅ Architecture Validated</h4>
<p style="margin: 0; color: #475569; font-size: 14px; line-height: 1.6;">
<strong>What we demonstrated:</strong><br>
• Real ARF OSS intelligence (advisory)<br>
• Clear execution boundary (OSS vs Enterprise)<br>
• Simulated Enterprise value proposition<br>
• Production-ready architecture pattern<br><br>
This isn't AI theater - it's a production-grade reliability system with honest boundaries.
</p>
</div>
"""
}
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