⚡ Bolt: Blockchain Escalation Audit Trail with Bounded O(1) Cache

.jules/bolt.md

@@ -25,3 +25,7 @@
 ## 2025-07-15 - Fast Bounding Box Pre-filter
 **Learning:** Calculating great circle distance (Haversine) for every issue against a target location is computationally expensive (O(N) with heavy math ops like sin, cos, atan2). In high-traffic aggregations, this can become a bottleneck.
 **Action:** Use a fast bounding box pre-filter (`get_bounding_box` with a 5% epsilon) to quickly discard issues that are definitely outside the search radius before running the expensive exact haversine distance calculation.
+
+## 2026-07-20 - Unbounded In-Memory Caches & Process Safety
+**Learning:** Class-level in-memory dictionaries used to achieve O(1) blockchain lookups can leak memory if they grow unboundedly in high-traffic applications. Additionally, single-column select queries in SQLAlchemy provide a highly performant and process-safe database-level alternative when multiple worker processes are involved.
+**Action:** Always bound in-memory cache growth (e.g., limit to 1000 items with eviction) and combine with optimized database column-selection queries to guarantee process-safety and prevent memory leaks.

backend/escalation_engine.py

@@ -4,6 +4,8 @@
 """
 
 import datetime
+import hashlib
+import threading
 from typing import List, Dict, Any, Optional
 from sqlalchemy.orm import Session
 from sqlalchemy import and_, or_
@@ -17,6 +19,11 @@ class EscalationEngine:
     Engine for handling grievance escalations based on SLA breaches and severity changes.
     """
 
+    # Cache for O(1) blockchain integrity hash lookups
+    # Stores {grievance_id: last_integrity_hash}
+    _audit_last_hash_cache = {}
+    _cache_lock = threading.Lock()
+
     def __init__(self, routing_service: RoutingService, sla_service: SLAConfigService,
                  rules_config: Dict[str, Any]):
         """
@@ -248,25 +255,68 @@ def _escalate_grievance(self, grievance: Grievance, reason: EscalationReason,
             # Recalculate SLA
             self._recalculate_sla(grievance, db)
 
-            # Create audit log
+            # Retrieve previous hash (O(1) from cache or O(log N) from indexed DB)
+            prev_hash = self._get_last_audit_hash(grievance.id, db)
+
+            # Calculate new integrity hash: SHA256(grievance_id | reason.value | prev_hash)
+            reason_val = reason.value if hasattr(reason, "value") else str(reason)
+            hash_input = f"{grievance.id}|{reason_val}|{prev_hash or 'GENESIS'}"
+            new_hash = hashlib.sha256(hash_input.encode()).hexdigest()
+
+            # Create audit log with blockchain integration
             audit_log = EscalationAudit(
                 grievance_id=grievance.id,
                 previous_authority=previous_authority,
                 new_authority=grievance.assigned_authority,
                 reason=reason,
-                notes=notes
+                notes=notes,
+                integrity_hash=new_hash,
+                previous_integrity_hash=prev_hash
             )
 
             db.add(audit_log)
             db.commit()
 
+            # Update O(1) cache for next escalation audit (with max size limit to prevent memory leak)
+            with self._cache_lock:
+                if len(self._audit_last_hash_cache) >= 1000:
+                    self._audit_last_hash_cache.clear()
+                self._audit_last_hash_cache[grievance.id] = new_hash
+
             return True
 
         except Exception as e:
             db.rollback()
             print(f"Error during escalation: {e}")
             return False
 
+    def _get_last_audit_hash(self, grievance_id: int, db: Session) -> Optional[str]:
+        """
+        Retrieves the last integrity hash for an escalation audit of a grievance.
+        Bolt Optimization: Uses thread-safe memory cache for O(1) lookup.
+        """
+        with self._cache_lock:
+            if grievance_id in self._audit_last_hash_cache:
+                return self._audit_last_hash_cache[grievance_id]
+
+        # Cache miss: Fallback to indexed DB query using optimized single-column selection
+        from sqlalchemy import desc
+        last_audit_hash = db.query(EscalationAudit.integrity_hash)\
+            .filter(EscalationAudit.grievance_id == grievance_id)\
+            .order_by(desc(EscalationAudit.id))\
+            .first()
+
+        last_hash = last_audit_hash[0] if last_audit_hash else None
+
+        # Update cache for next time (with max size limit to prevent memory leak)
+        if last_hash:
+            with self._cache_lock:
+                if len(self._audit_last_hash_cache) >= 1000:
+                    self._audit_last_hash_cache.clear()
+                self._audit_last_hash_cache[grievance_id] = last_hash
+
+        return last_hash
+
     def _recalculate_sla(self, grievance: Grievance, db: Session) -> None:
         """
         Recalculate SLA deadline for a grievance.

backend/grievance_service.py

@@ -12,7 +12,7 @@
 from sqlalchemy import and_, desc
 from datetime import datetime, timezone, timedelta
 
-from backend.models import Grievance, Jurisdiction, GrievanceStatus, SeverityLevel, GrievanceFollower
+from backend.models import Grievance, Jurisdiction, GrievanceStatus, SeverityLevel, GrievanceFollower, EscalationAudit
 from backend.database import SessionLocal
 from backend.routing_service import RoutingService
 from backend.sla_config_service import SLAConfigService
@@ -169,8 +169,10 @@ def follow_grievance(self, grievance_id: int, user_email: str, db: Session = Non
             db.commit()
             db.refresh(follower)
 
-            # Update O(1) cache for next follower
+            # Update O(1) cache for next follower (with max size limit to prevent memory leak)
             with self._cache_lock:
+                if len(self._follower_last_hash_cache) >= 1000:
+                    self._follower_last_hash_cache.clear()
                 self._follower_last_hash_cache[grievance_id] = new_hash
 
             return follower
@@ -192,17 +194,19 @@ def _get_last_integrity_hash(self, grievance_id: int, db: Session) -> Optional[s
             if grievance_id in self._follower_last_hash_cache:
                 return self._follower_last_hash_cache[grievance_id]
 
-        # Cache miss: Fallback to indexed DB query
-        last_follower = db.query(GrievanceFollower)\
+        # Cache miss: Fallback to indexed DB query using optimized single-column selection
+        last_follower_hash = db.query(GrievanceFollower.integrity_hash)\
             .filter(GrievanceFollower.grievance_id == grievance_id)\
             .order_by(desc(GrievanceFollower.id))\
             .first()
 
-        last_hash = last_follower.integrity_hash if last_follower else None
+        last_hash = last_follower_hash[0] if last_follower_hash else None
 
-        # Update cache for next time
+        # Update cache for next time (with max size limit to prevent memory leak)
         if last_hash:
             with self._cache_lock:
+                if len(self._follower_last_hash_cache) >= 1000:
+                    self._follower_last_hash_cache.clear()
                 self._follower_last_hash_cache[grievance_id] = last_hash
 
         return last_hash
@@ -238,6 +242,38 @@ def verify_follower_integrity(self, follower_id: int, db: Session = None) -> Dic
             if is_local_session:
                 db.close()
 
+    def verify_audit_integrity(self, audit_id: int, db: Session = None) -> Dict[str, Any]:
+        """
+        Verify the blockchain-style integrity of an escalation audit record.
+        """
+        is_local_session = False
+        if db is None:
+            db = SessionLocal()
+            is_local_session = True
+
+        try:
+            audit = db.query(EscalationAudit).filter(EscalationAudit.id == audit_id).first()
+            if not audit:
+                return {"is_valid": False, "message": "Audit record not found"}
+
+            # Re-calculate hash: SHA256(grievance_id | reason.value | prev_hash)
+            reason_val = audit.reason.value if hasattr(audit.reason, "value") else str(audit.reason)
+            hash_input = f"{audit.grievance_id}|{reason_val}|{audit.previous_integrity_hash or 'GENESIS'}"
+            calculated_hash = hashlib.sha256(hash_input.encode()).hexdigest()
+
+            is_valid = (calculated_hash == audit.integrity_hash)
+
+            return {
+                "is_valid": is_valid,
+                "current_hash": audit.integrity_hash,
+                "calculated_hash": calculated_hash,
+                "previous_hash": audit.previous_integrity_hash,
+                "message": "Integrity verified" if is_valid else "INTEGRITY BREACH DETECTED"
+            }
+        finally:
+            if is_local_session:
+                db.close()
+
     def get_grievance(self, grievance_id: int, db: Session = None) -> Optional[Grievance]:
         """
         Get a grievance by ID.
@@ -305,32 +341,34 @@ def update_grievance_status(self, grievance_id: int, status: GrievanceStatus,
                 db.close()
 
     def escalate_grievance_severity(self, grievance_id: int, new_severity: SeverityLevel,
-                                   reason: str = "") -> bool:
+                                   reason: str = "", db: Session = None) -> bool:
         """
         Escalate grievance severity.
 
         Args:
             grievance_id: Grievance ID
             new_severity: New severity level
             reason: Reason for escalation
+            db: Database session
 
         Returns:
             True if escalation successful
         """
-        return self.escalation_engine.escalate_grievance_severity(grievance_id, new_severity, reason)
+        return self.escalation_engine.escalate_grievance_severity(grievance_id, new_severity, reason, db)
 
-    def manual_escalate(self, grievance_id: int, reason: str = "") -> bool:
+    def manual_escalate(self, grievance_id: int, reason: str = "", db: Session = None) -> bool:
         """
         Manually escalate a grievance.
 
         Args:
             grievance_id: Grievance ID
             reason: Reason for escalation
+            db: Database session
 
         Returns:
             True if escalation successful
         """
-        return self.escalation_engine.manual_escalate(grievance_id, reason)
+        return self.escalation_engine.manual_escalate(grievance_id, reason, db)
 
     def run_escalation_check(self) -> Dict[str, int]:
         """