Upload 7 files

README.md

@@ -1,3 +1,30 @@
----
-license: mit
----
+# Contextual Engineering Patterns: Architecting Adaptable AI Agents
+
+[![License: CC BY 4.0](https://img.shields.io/badge/License-CC%20BY%204.0-lightgrey.svg)](https://creativecommons.org/licenses/by/4.0/)
+[![Status: Open Access](https://img.shields.io/badge/Status-Open%20Access-green.svg)]()
+[![Book: Published](https://img.shields.io/badge/Book-Read%20Full%20Text-blue)](https://zenodo.org) > **Reference implementations for the architectural patterns defined in the book *"Contextual Engineering: Architecting Adaptable AI Agents for the Real World"* by Tobi Lekan Adeosun.**
+
+## 📖 Overview
+
+Standard AI agents are designed for the "Abundance Baseline" of Silicon Valley—perfect internet, unlimited power, and institutional trust. When deployed in the Global South, these agents fail due to the **"Agentic Gap"** between their reasoning capabilities and environmental realities.
+
+This repository contains the **Python reference implementations** for the three core adaptation layers introduced in the book:
+1.  **Infrastructure Adapter:** Handling offline states and compute scarcity.
+2.  **Cultural Adapter:** Managing semantic drift and high-context communication.
+3.  **Safety Adapter:** Enforcing constitutional guardrails and Human-in-the-Loop (HITL) workflows.
+
+## 📂 Repository Structure
+
+The code is organized by the "Adapter Layer" it serves, matching the chapters of the manuscript.
+
+```text
+├── src
+│   ├── infrastructure
+│   │   ├── sync_manager.py       # (Chapter 3) The "Sync-Later" Architecture & Offline Queue
+│   │   └── inference_router.py   # (Chapter 4) The Hybrid Router (Local vs. Cloud)
+│   ├── safety
+│   │   ├── sentinel.py           # (Chapter 9) Constitutional Safety Checks & Kill Switches
+│   │   └── escalation_ladder.py  # (Chapter 10) Human-in-the-Loop Risk Evaluation Logic
+│   └── culture
+│       └── context_injector.py   # (Chapter 6) Dynamic Few-Shot Prompting logic
+└── README.md

src/infrastructure/inference_router.py

@@ -0,0 +1,180 @@
+# src/infrastructure/inference_router.py
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from enum import Enum
+from typing import Optional
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+# Configuration Constants
+COMPLEXITY_THRESHOLD = 0.3
+MIN_BATTERY_PERCENT = 20
+MAX_PROMPT_LENGTH = 10000
+
+
+class NetworkStatus(Enum):
+    CONNECTED = "CONNECTED"
+    DISCONNECTED = "DISCONNECTED"
+    HIGH_LATENCY = "HIGH_LATENCY"
+
+
+class RoutingDecision(Enum):
+    LOCAL = "LOCAL"
+    CLOUD = "CLOUD"
+    DEGRADED_LOCAL = "DEGRADED_LOCAL"
+
+
+@dataclass
+class InferenceResult:
+    response: Optional[str]
+    routing_decision: RoutingDecision
+    is_degraded: bool = False
+    error_message: Optional[str] = None
+
+
+class ComplexityClassifier(ABC):
+    @abstractmethod
+    def predict(self, prompt: str) -> float:
+        """Returns complexity score between 0.0 and 1.0."""
+        pass
+
+
+class NetworkMonitor(ABC):
+    @abstractmethod
+    def get_status(self) -> NetworkStatus:
+        pass
+
+
+class DeviceMonitor(ABC):
+    @abstractmethod
+    def get_battery_percent(self) -> int:
+        pass
+
+
+class LanguageModel(ABC):
+    @abstractmethod
+    def generate(self, prompt: str) -> str:
+        pass
+
+
+class UserNotifier(ABC):
+    @abstractmethod
+    def warn_user(self, message: str) -> None:
+        pass
+
+
+class InferenceRouter:
+    """Routes inference requests based on complexity and infrastructure context."""
+
+    def __init__(
+        self,
+        local_classifier: ComplexityClassifier,
+        network_monitor: NetworkMonitor,
+        device_monitor: DeviceMonitor,
+        local_slm: LanguageModel,
+        cloud_llm: LanguageModel,
+        user_notifier: UserNotifier,
+        complexity_threshold: float = COMPLEXITY_THRESHOLD,
+        min_battery_percent: int = MIN_BATTERY_PERCENT
+    ):
+        self._local_classifier = local_classifier
+        self._network_monitor = network_monitor
+        self._device_monitor = device_monitor
+        self._local_slm = local_slm
+        self._cloud_llm = cloud_llm
+        self._user_notifier = user_notifier
+        self._complexity_threshold = complexity_threshold
+        self._min_battery_percent = min_battery_percent
+
+    def route(self, user_prompt: str) -> InferenceResult:
+        """Route inference request to appropriate model."""
+        # Input validation
+        if not user_prompt or not isinstance(user_prompt, str):
+            logger.error("Invalid user prompt provided")
+            return InferenceResult(
+                response=None,
+                routing_decision=RoutingDecision.LOCAL,
+                error_message="Invalid prompt: must be a non-empty string"
+            )
+
+        if len(user_prompt) > MAX_PROMPT_LENGTH:
+            logger.error(
+                f"Prompt exceeds maximum length of {MAX_PROMPT_LENGTH}")
+            return InferenceResult(
+                response=None,
+                routing_decision=RoutingDecision.LOCAL,
+                error_message=f"Prompt too long: max {MAX_PROMPT_LENGTH} characters"
+            )
+
+        # Step 1: Analyze Complexity locally
+        try:
+            complexity_score = self._local_classifier.predict(user_prompt)
+        except Exception as e:
+            logger.exception("Complexity classification failed")
+            # Default to local model on classification failure
+            return self._execute_local(user_prompt, is_degraded=True)
+
+        # Step 2: Check Infrastructure Context
+        network_status = self._network_monitor.get_status()
+        battery_percent = self._device_monitor.get_battery_percent()
+
+        is_offline = network_status == NetworkStatus.DISCONNECTED
+        battery_low = battery_percent < self._min_battery_percent
+
+        # ROUTING DECISION MATRIX
+
+        # Scenario A: Simple Task (e.g., "Set an alarm")
+        if complexity_score < self._complexity_threshold:
+            logger.info(f"Simple task (score={complexity_score:.2f}) -> LOCAL")
+            return self._execute_local(user_prompt)
+
+        # Scenario B: Hard Task, but No Internet or Low Battery
+        if is_offline or battery_low:
+            reason = "offline" if is_offline else "low battery"
+            logger.warning(f"Complex task but {reason} -> DEGRADED_LOCAL")
+            self._user_notifier.warn_user("Limited capability mode")
+            return self._execute_local(user_prompt, is_degraded=True)
+
+        # Scenario C: Hard Task + Good Internet + Battery OK
+        logger.info(f"Complex task (score={complexity_score:.2f}) -> CLOUD")
+        return self._execute_cloud(user_prompt)
+
+    def _execute_local(self, prompt: str, is_degraded: bool = False) -> InferenceResult:
+        """Execute inference using local small language model."""
+        try:
+            response = self._local_slm.generate(prompt)
+            return InferenceResult(
+                response=response,
+                routing_decision=(
+                    RoutingDecision.DEGRADED_LOCAL if is_degraded
+                    else RoutingDecision.LOCAL
+                ),
+                is_degraded=is_degraded
+            )
+        except Exception as e:
+            logger.exception("Local model inference failed")
+            return InferenceResult(
+                response=None,
+                routing_decision=RoutingDecision.LOCAL,
+                is_degraded=is_degraded,
+                error_message=f"Local inference failed: {str(e)}"
+            )
+
+    def _execute_cloud(self, prompt: str) -> InferenceResult:
+        """Execute inference using cloud LLM with fallback to local."""
+        try:
+            response = self._cloud_llm.generate(prompt)
+            return InferenceResult(
+                response=response,
+                routing_decision=RoutingDecision.CLOUD
+            )
+        except Exception as e:
+            logger.exception("Cloud inference failed, falling back to local")
+            self._user_notifier.warn_user(
+                "Cloud service unavailable, using limited capability mode"
+            )
+            return self._execute_local(prompt, is_degraded=True)

src/infrastructure/offline_action.py

@@ -0,0 +1,135 @@
+# ------------------------------------------------------------------------------
+# Contextual Engineering Patterns
+# Copyright (c) 2025 Tobi Lekan Adeosun
+# Licensed under the MIT License.
+#
+# From Chapter 3: "The Disconnected Agent"
+# ------------------------------------------------------------------------------
+
+import uuid
+import time
+import json
+import hashlib
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Dict, Optional
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+# Configuration Constants
+DEFAULT_TTL_HOURS = 24
+DEFAULT_MIN_BATTERY_LEVEL = 15
+SECONDS_PER_HOUR = 3600
+
+
+class ActionStatus(Enum):
+    PENDING = "PENDING"
+    SYNCED = "SYNCED"
+    FAILED = "FAILED"
+    EXPIRED = "EXPIRED"
+
+
+def generate_hash(data: str) -> str:
+    """Helper to create a deterministic hash for idempotency."""
+    return hashlib.sha256(data.encode('utf-8')).hexdigest()
+
+
+@dataclass
+class OfflineAction:
+    """Represents an action to be synced when connectivity is restored."""
+
+    action_type: str
+    payload: Dict[str, Any]
+    priority: int = 1
+    ttl_hours: int = DEFAULT_TTL_HOURS
+    min_battery_level: int = DEFAULT_MIN_BATTERY_LEVEL
+
+    # Auto-generated fields
+    id: str = field(default_factory=lambda: str(uuid.uuid4()))
+    timestamp: float = field(default_factory=time.time)
+    status: ActionStatus = field(default=ActionStatus.PENDING)
+    idempotency_key: str = field(default="", init=False)
+
+    def __post_init__(self):
+        """Generate idempotency key after initialization."""
+        # CRITICAL: The Idempotency Key ensures that if the
+        # network flutters and sends the request twice,
+        # the server only executes it once.
+        key_data = f"{json.dumps(self.payload, sort_keys=True)}:{self.timestamp}"
+        self.idempotency_key = generate_hash(key_data)
+
+    @property
+    def expiry_timestamp(self) -> float:
+        """Calculate the expiry timestamp based on TTL."""
+        return self.timestamp + (self.ttl_hours * SECONDS_PER_HOUR)
+
+    def is_expired(self) -> bool:
+        """Check if the action has exceeded its TTL."""
+        return time.time() > self.expiry_timestamp
+
+    def can_sync(self, current_battery_level: int) -> bool:
+        """Check if conditions allow syncing this action."""
+        if self.is_expired():
+            logger.warning(f"Action {self.id} has expired")
+            return False
+
+        if current_battery_level < self.min_battery_level:
+            logger.warning(
+                f"Battery too low ({current_battery_level}%) to sync action {self.id}"
+            )
+            return False
+
+        return True
+
+    def mark_synced(self) -> None:
+        """Mark the action as successfully synced."""
+        self.status = ActionStatus.SYNCED
+        logger.info(f"Action {self.id} marked as SYNCED")
+
+    def mark_failed(self, reason: Optional[str] = None) -> None:
+        """Mark the action as failed."""
+        self.status = ActionStatus.FAILED
+        logger.error(f"Action {self.id} marked as FAILED: {reason}")
+
+    def mark_expired(self) -> None:
+        """Mark the action as expired."""
+        self.status = ActionStatus.EXPIRED
+        logger.warning(f"Action {self.id} marked as EXPIRED")
+
+    def serialize(self) -> str:
+        """Convert to JSON for storage in local SQLite."""
+        data = {
+            "id": self.id,
+            "action_type": self.action_type,
+            "payload": self.payload,
+            "priority": self.priority,
+            "timestamp": self.timestamp,
+            "status": self.status.value,
+            "ttl_hours": self.ttl_hours,
+            "expiry_timestamp": self.expiry_timestamp
+            # Note: idempotency_key intentionally excluded from serialization
+            # to prevent exposure; regenerate on deserialization
+        }
+        return json.dumps(data)
+
+    @classmethod
+    def deserialize(cls, json_str: str) -> "OfflineAction":
+        """Create an OfflineAction from JSON string."""
+        try:
+            data = json.loads(json_str)
+            action = cls(
+                action_type=data["action_type"],
+                payload=data["payload"],
+                priority=data.get("priority", 1),
+                ttl_hours=data.get("ttl_hours", DEFAULT_TTL_HOURS)
+            )
+            action.id = data["id"]
+            action.timestamp = data["timestamp"]
+            action.status = ActionStatus(data.get("status", "PENDING"))
+            return action
+        except (json.JSONDecodeError, KeyError) as e:
+            logger.exception("Failed to deserialize OfflineAction")
+            raise ValueError(f"Invalid JSON data for OfflineAction: {str(e)}")

src/infrastructure/sync_manager.py

@@ -0,0 +1,288 @@
+# src/infrastructure/sync_manager.py
+
+import time
+import uuid
+import hashlib
+import json
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Dict, List, Optional
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+# Configuration Constants
+DEFAULT_TTL_HOURS = 24
+SECONDS_PER_HOUR = 3600
+MAX_RETRY_ATTEMPTS = 5
+BASE_RETRY_DELAY_SECONDS = 1
+
+
+class NetworkState(Enum):
+    ONLINE = "ONLINE"
+    OFFLINE = "OFFLINE"
+    HIGH_LATENCY = "HIGH_LATENCY"
+
+
+class ActionStatus(Enum):
+    PENDING = "PENDING"
+    SYNCED = "SYNCED"
+    EXPIRED = "EXPIRED"
+    FAILED = "FAILED"
+    RETRY_SCHEDULED = "RETRY_SCHEDULED"
+
+
+class SyncResult(Enum):
+    SUCCESS = "SUCCESS"
+    ABORT_NO_NET = "ABORT_NO_NET"
+    PARTIAL_SUCCESS = "PARTIAL_SUCCESS"
+    ALL_FAILED = "ALL_FAILED"
+
+
+@dataclass
+class SyncSummary:
+    result: SyncResult
+    synced_count: int = 0
+    failed_count: int = 0
+    expired_count: int = 0
+    error_messages: List[str] = None
+
+    def __post_init__(self):
+        if self.error_messages is None:
+            self.error_messages = []
+
+
+class LocalDatabase(ABC):
+    @abstractmethod
+    def insert(self, table: str, data: dict) -> None:
+        pass
+
+    @abstractmethod
+    def query(self, sql: str) -> List[dict]:
+        pass
+
+    @abstractmethod
+    def update(self, record_id: str, **kwargs) -> None:
+        pass
+
+    @abstractmethod
+    def get_state_snapshot(self) -> dict:
+        """Get current local state for conflict detection."""
+        pass
+
+
+class ApiClient(ABC):
+    @abstractmethod
+    def post(self, endpoint: str, json: dict) -> dict:
+        pass
+
+
+class SyncQueue:
+    """Manages offline action queue with sync capabilities."""
+
+    def __init__(
+        self,
+        local_db: LocalDatabase,
+        api_client: ApiClient,
+        ttl_hours: int = DEFAULT_TTL_HOURS
+    ):
+        self._db = local_db
+        self._api_client = api_client
+        self._ttl_hours = ttl_hours
+        self._retry_counts: Dict[str, int] = {}
+
+    def get_local_state_hash(self) -> str:
+        """Generate a hash of the current local state for conflict detection."""
+        try:
+            state_snapshot = self._db.get_state_snapshot()
+            state_json = json.dumps(state_snapshot, sort_keys=True)
+            return hashlib.sha256(state_json.encode('utf-8')).hexdigest()
+        except Exception as e:
+            logger.exception("Failed to generate local state hash")
+            return ""
+
+    def enqueue_action(
+        self,
+        action_type: str,
+        payload: dict,
+        priority: int = 1
+    ) -> str:
+        """
+        Stores an action locally with a Time-To-Live (TTL).
+        If the action acts on old data, we tag it for conflict resolution.
+
+        Returns:
+            The action ID for tracking.
+        """
+        if not action_type or not isinstance(action_type, str):
+            raise ValueError("action_type must be a non-empty string")
+
+        if not isinstance(payload, dict):
+            raise ValueError("payload must be a dictionary")
+
+        action_id = str(uuid.uuid4())
+        action_item = {
+            "id": action_id,
+            "timestamp": time.time(),
+            "type": action_type,
+            "payload": self._sanitize_payload(payload),
+            "priority": priority,
+            "status": ActionStatus.PENDING.value,
+            "ttl_expiry": time.time() + (self._ttl_hours * SECONDS_PER_HOUR),
+            "device_state_hash": self.get_local_state_hash(),
+            "retry_count": 0
+        }
+
+        # Write to SQLite immediately (Atomic Write)
+        try:
+            self._db.insert("offline_queue", action_item)
+            logger.info(f"Action {action_id} queued locally")
+            return action_id
+        except Exception as e:
+            logger.exception(f"Failed to enqueue action: {str(e)}")
+            raise
+
+    def _sanitize_payload(self, payload: dict) -> dict:
+        """Remove sensitive fields before storage/transmission."""
+        sensitive_keys = {'password', 'token',
+                          'secret', 'api_key', 'credential'}
+        return {
+            k: v for k, v in payload.items()
+            if k.lower() not in sensitive_keys
+        }
+
+    def attempt_sync(self, current_network_state: NetworkState) -> SyncSummary:
+        """
+        Only attempts sync if network is stable.
+
+        Returns:
+            SyncSummary with results of the sync operation.
+        """
+        if current_network_state == NetworkState.OFFLINE:
+            logger.info("Sync aborted: network offline")
+            return SyncSummary(result=SyncResult.ABORT_NO_NET)
+
+        try:
+            pending_actions = self._db.query(
+                "SELECT * FROM offline_queue WHERE status='PENDING' OR status='RETRY_SCHEDULED'"
+            )
+        except Exception as e:
+            logger.exception("Failed to query pending actions")
+            return SyncSummary(
+                result=SyncResult.ALL_FAILED,
+                error_messages=[f"Database query failed: {str(e)}"]
+            )
+
+        if not pending_actions:
+            logger.info("No pending actions to sync")
+            return SyncSummary(result=SyncResult.SUCCESS)
+
+        # Sort by priority (higher first) then timestamp (older first)
+        pending_actions.sort(
+            key=lambda x: (-x.get('priority', 1), x['timestamp']))
+
+        synced_count = 0
+        failed_count = 0
+        expired_count = 0
+        error_messages = []
+
+        for action in pending_actions:
+            action_id = action['id']
+
+            # Check TTL expiry
+            if time.time() > action['ttl_expiry']:
+                self._db.update(action_id, status=ActionStatus.EXPIRED.value)
+                expired_count += 1
+                logger.warning(f"Action {action_id} expired")
+                continue
+
+            # Attempt sync
+            sync_success = self._sync_action(action)
+            if sync_success:
+                synced_count += 1
+            else:
+                failed_count += 1
+                error_messages.append(f"Action {action_id} failed to sync")
+
+        # Determine overall result
+        if failed_count == 0:
+            result = SyncResult.SUCCESS
+        elif synced_count == 0:
+            result = SyncResult.ALL_FAILED
+        else:
+            result = SyncResult.PARTIAL_SUCCESS
+
+        return SyncSummary(
+            result=result,
+            synced_count=synced_count,
+            failed_count=failed_count,
+            expired_count=expired_count,
+            error_messages=error_messages
+        )
+
+    def _sync_action(self, action: dict) -> bool:
+        """Attempt to sync a single action to the cloud."""
+        action_id = action['id']
+
+        # Prepare payload (exclude internal fields)
+        sync_payload = {
+            "id": action_id,
+            "type": action['type'],
+            "payload": action['payload'],
+            "timestamp": action['timestamp'],
+            "device_state_hash": action.get('device_state_hash', '')
+        }
+
+        try:
+            self._api_client.post("/sync", json=sync_payload)
+            self._db.update(action_id, status=ActionStatus.SYNCED.value)
+            logger.info(f"Action {action_id} synced successfully")
+            return True
+
+        except TimeoutError:
+            self._schedule_retry(action_id)
+            return False
+
+        except ConnectionError as e:
+            logger.error(f"Connection error syncing {action_id}: {str(e)}")
+            self._schedule_retry(action_id)
+            return False
+
+        except Exception as e:
+            logger.exception(f"Unexpected error syncing {action_id}")
+            retry_count = self._retry_counts.get(action_id, 0)
+            if retry_count >= MAX_RETRY_ATTEMPTS:
+                self._db.update(action_id, status=ActionStatus.FAILED.value)
+                logger.error(
+                    f"Action {action_id} permanently failed after {retry_count} retries")
+            else:
+                self._schedule_retry(action_id)
+            return False
+
+    def _schedule_retry(self, action_id: str) -> None:
+        """Schedule an action for retry with exponential backoff."""
+        current_count = self._retry_counts.get(action_id, 0)
+
+        if current_count >= MAX_RETRY_ATTEMPTS:
+            self._db.update(action_id, status=ActionStatus.FAILED.value)
+            logger.error(
+                f"Action {action_id} exceeded max retries ({MAX_RETRY_ATTEMPTS})"
+            )
+            return
+
+        self._retry_counts[action_id] = current_count + 1
+        delay = BASE_RETRY_DELAY_SECONDS * (2 ** current_count)
+
+        self._db.update(
+            action_id,
+            status=ActionStatus.RETRY_SCHEDULED.value,
+            retry_count=current_count + 1
+        )
+
+        logger.info(
+            f"Action {action_id} scheduled for retry #{current_count + 1} "
+            f"in {delay} seconds"
+        )

src/safety/escalation_ladder.py

@@ -0,0 +1,144 @@
+# src/safety/escalation_ladder.py
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Optional
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class RiskLevel(Enum):
+    LOW = "LOW"
+    MEDIUM = "MEDIUM"
+    HIGH = "HIGH"
+    CRITICAL = "CRITICAL"
+
+
+class ExecutionStatus(Enum):
+    SUCCESS = "SUCCESS"
+    WAITING_FOR_APPROVAL = "WAITING_FOR_APPROVAL"
+    REJECTED = "REJECTED"
+    ERROR = "ERROR"
+
+
+@dataclass
+class ExecutionResult:
+    status: ExecutionStatus
+    result: Optional[Any] = None
+    error_message: Optional[str] = None
+    decision_id: Optional[str] = None
+
+
+class RiskPolicy(ABC):
+    @abstractmethod
+    def get_level(self, tool_name: str) -> RiskLevel:
+        pass
+
+
+class Tool(ABC):
+    @abstractmethod
+    def run(self, parameters: dict) -> Any:
+        pass
+
+
+class ApprovalService(ABC):
+    @abstractmethod
+    def create_approval_request(self, tool_name: str, parameters: dict) -> str:
+        pass
+
+    @abstractmethod
+    def notify_manager(self, decision_id: str) -> None:
+        pass
+
+
+class EscalationLadder:
+    def __init__(
+        self,
+        risk_policy: RiskPolicy,
+        tool: Tool,
+        approval_service: ApprovalService
+    ):
+        self._risk_policy = risk_policy
+        self._tool = tool
+        self._approval_service = approval_service
+
+    def execute_tool(self, tool_name: str, parameters: dict) -> ExecutionResult:
+        """Execute a tool with appropriate risk-based escalation."""
+        if not tool_name or not isinstance(tool_name, str):
+            logger.error("Invalid tool_name provided")
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR,
+                error_message="Invalid tool_name: must be a non-empty string"
+            )
+
+        if not isinstance(parameters, dict):
+            logger.error("Invalid parameters provided")
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR,
+                error_message="Invalid parameters: must be a dictionary"
+            )
+
+        try:
+            risk_level = self._risk_policy.get_level(tool_name)
+        except Exception as e:
+            logger.exception("Failed to assess risk level")
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR,
+                error_message=f"Risk assessment failed: {str(e)}"
+            )
+
+        if risk_level == RiskLevel.CRITICAL:
+            logger.warning(f"CRITICAL risk tool '{tool_name}' blocked")
+            return ExecutionResult(
+                status=ExecutionStatus.REJECTED,
+                error_message="Critical risk tools are not permitted"
+            )
+
+        if risk_level == RiskLevel.HIGH:
+            return self._handle_high_risk(tool_name, parameters)
+
+        if risk_level == RiskLevel.MEDIUM:
+            logger.info(f"MEDIUM risk tool '{tool_name}' - logging for audit")
+            return self._execute_with_logging(tool_name, parameters)
+
+        # LOW risk - execute immediately
+        return self._execute_tool(parameters)
+
+    def _handle_high_risk(self, tool_name: str, parameters: dict) -> ExecutionResult:
+        """Handle high-risk tool execution with approval workflow."""
+        try:
+            decision_id = self._approval_service.create_approval_request(
+                tool_name, parameters
+            )
+            self._approval_service.notify_manager(decision_id)
+            logger.info(f"Approval request created: {decision_id}")
+            return ExecutionResult(
+                status=ExecutionStatus.WAITING_FOR_APPROVAL,
+                decision_id=decision_id
+            )
+        except Exception as e:
+            logger.exception("Failed to create approval request")
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR,
+                error_message=f"Approval workflow failed: {str(e)}"
+            )
+
+    def _execute_with_logging(self, tool_name: str, parameters: dict) -> ExecutionResult:
+        """Execute tool with enhanced audit logging."""
+        logger.info(f"Executing medium-risk tool: {tool_name}")
+        return self._execute_tool(parameters)
+
+    def _execute_tool(self, parameters: dict) -> ExecutionResult:
+        """Execute the tool and return result."""
+        try:
+            result = self._tool.run(parameters)
+            return ExecutionResult(status=ExecutionStatus.SUCCESS, result=result)
+        except Exception as e:
+            logger.exception("Tool execution failed")
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR,
+                error_message=f"Execution failed: {str(e)}"
+            )

src/safety/prompts.py

@@ -0,0 +1,64 @@
+# src/safety/prompts.py
+
+from dataclasses import dataclass, field
+from typing import List, Optional
+from enum import Enum
+
+
+class PromptType(Enum):
+    SYSTEM = "SYSTEM"
+    USER = "USER"
+    ASSISTANT = "ASSISTANT"
+
+
+@dataclass(frozen=True)
+class ContextualPrompt:
+    """Immutable prompt template with contextual awareness."""
+    role: str
+    region: Optional[str] = None
+    context_items: List[str] = field(default_factory=list)
+    constraints: List[str] = field(default_factory=list)
+    prompt_type: PromptType = PromptType.SYSTEM
+
+    def build(self) -> str:
+        """Build the complete prompt string."""
+        parts = [f"You are a {self.role}"]
+
+        if self.region:
+            parts[0] += f" operating in {self.region}."
+        else:
+            parts[0] += "."
+
+        if self.context_items:
+            parts.append("\nCONTEXT:")
+            for item in self.context_items:
+                parts.append(f"- {item}")
+
+        if self.constraints:
+            parts.append("\nCONSTRAINTS:")
+            for constraint in self.constraints:
+                parts.append(f"- {constraint}")
+
+        return "\n".join(parts)
+
+
+# Example: Non-contextual prompt (anti-pattern)
+BAD_PROMPT = "You are a financial advisor. Give strict, rational advice."
+
+# Example: Contextually engineered prompt for Lagos, Nigeria
+LAGOS_FINANCIAL_ADVISOR = ContextualPrompt(
+    role="financial advisor",
+    region="Lagos, Nigeria",
+    context_items=[
+        "You understand the concept of 'Black Tax' and extended family support.",
+        "You prioritize community reputation over ruthless individual efficiency.",
+        "You speak in a respectful, empathetic tone."
+    ],
+    constraints=[
+        "Do NOT suggest cutting off family members.",
+        "DO suggest budgeting for family support as a fixed expense line item."
+    ]
+)
+
+# For backward compatibility
+CONTEXTUAL_PROMPT = LAGOS_FINANCIAL_ADVISOR.build()

src/safety/sentinel.py

@@ -0,0 +1,155 @@
+# src/safety/sentinel.py
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from enum import Enum
+from typing import List, Optional
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+# Configuration Constants
+ANGER_THRESHOLD = 0.9
+FACT_SCORE_THRESHOLD = 0.5
+
+
+class SafetyStatus(Enum):
+    SAFE = "SAFE"
+    KILL_SWITCH_ACTIVE = "KILL_SWITCH_ACTIVE"
+    CHECK_FAILED = "CHECK_FAILED"
+
+
+@dataclass
+class AgentState:
+    user_history: List[str]
+    last_response: str
+
+
+@dataclass
+class SafetyCheckResult:
+    status: SafetyStatus
+    anger_score: Optional[float] = None
+    fact_score: Optional[float] = None
+    error_message: Optional[str] = None
+
+
+class SentimentAnalyzer(ABC):
+    @abstractmethod
+    def analyze(self, user_history: List[str]) -> float:
+        """Returns anger score between 0.0 and 1.0."""
+        pass
+
+
+class FactVerifier(ABC):
+    @abstractmethod
+    def verify_facts(self, response: str) -> float:
+        """Returns fact accuracy score between 0.0 and 1.0."""
+        pass
+
+
+class AgentController(ABC):
+    @abstractmethod
+    def stop_agent(self) -> None:
+        pass
+
+    @abstractmethod
+    def alert_human_manager(self) -> None:
+        pass
+
+    @abstractmethod
+    def display_message(self, message: str) -> None:
+        pass
+
+
+class SafetySentinel:
+    """Monitors agent behavior and triggers safety responses."""
+
+    HANDOFF_MESSAGE = "I am having trouble. Connecting you to a human..."
+
+    def __init__(
+        self,
+        sentiment_analyzer: SentimentAnalyzer,
+        fact_verifier: FactVerifier,
+        agent_controller: AgentController,
+        anger_threshold: float = ANGER_THRESHOLD,
+        fact_score_threshold: float = FACT_SCORE_THRESHOLD
+    ):
+        self._sentiment_analyzer = sentiment_analyzer
+        self._fact_verifier = fact_verifier
+        self._agent_controller = agent_controller
+        self._anger_threshold = anger_threshold
+        self._fact_score_threshold = fact_score_threshold
+
+    def safety_check(self, agent_state: AgentState) -> SafetyCheckResult:
+        """Perform safety checks on the current agent state."""
+        if not agent_state:
+            logger.error("Invalid agent state provided")
+            return SafetyCheckResult(
+                status=SafetyStatus.CHECK_FAILED,
+                error_message="Agent state is required"
+            )
+
+        try:
+            # Check 1: Sentiment Analysis
+            user_anger = self._sentiment_analyzer.analyze(
+                agent_state.user_history
+            )
+        except Exception as e:
+            logger.exception("Sentiment analysis failed")
+            return SafetyCheckResult(
+                status=SafetyStatus.CHECK_FAILED,
+                error_message=f"Sentiment analysis error: {str(e)}"
+            )
+
+        try:
+            # Check 2: Hallucination Detection (Fact Check)
+            fact_score = self._fact_verifier.verify_facts(
+                agent_state.last_response
+            )
+        except Exception as e:
+            logger.exception("Fact verification failed")
+            return SafetyCheckResult(
+                status=SafetyStatus.CHECK_FAILED,
+                anger_score=user_anger,
+                error_message=f"Fact verification error: {str(e)}"
+            )
+
+        is_unsafe = (
+            user_anger > self._anger_threshold or
+            fact_score < self._fact_score_threshold
+        )
+
+        if is_unsafe:
+            logger.warning(
+                f"Safety threshold breached: anger={user_anger}, "
+                f"fact_score={fact_score}"
+            )
+            return SafetyCheckResult(
+                status=SafetyStatus.KILL_SWITCH_ACTIVE,
+                anger_score=user_anger,
+                fact_score=fact_score
+            )
+
+        return SafetyCheckResult(
+            status=SafetyStatus.SAFE,
+            anger_score=user_anger,
+            fact_score=fact_score
+        )
+
+    def execute_safety_response(self, agent_state: AgentState) -> SafetyCheckResult:
+        """Check safety and execute appropriate response."""
+        result = self.safety_check(agent_state)
+
+        if result.status == SafetyStatus.KILL_SWITCH_ACTIVE:
+            try:
+                self._agent_controller.stop_agent()
+                self._agent_controller.alert_human_manager()
+                self._agent_controller.display_message(self.HANDOFF_MESSAGE)
+                logger.info("Kill switch activated - handed off to human")
+            except Exception as e:
+                logger.exception("Failed to execute safety response")
+                result.error_message = f"Safety response failed: {str(e)}"
+
+        return result