Black Hole Context Manager
← Corben Papers and Architecture Sources
This page publishes Corben Sorenson’s original source manuscript so readers can inspect the ideas that preceded or informed the living book. The text may contain historical terminology, claims, confidence, citations, or implementation status that the book later narrows, revises, tests, or rejects. Publication here establishes provenance and access—not correctness, novelty, replication, or support-state promotion.
Publication and provenance
| Field | Record |
|---|---|
| Source ID | black_hole_context_manager |
| Source class | author_technical_spec_family |
| Library class | technical_specification |
| Manuscript date | 2026-07-31 |
| Inventory updated | 2026-07-31 |
| Exact published-source SHA-256 | 8db214a8752aabbed1e8661701daa3bfaef6c96d09b1c65c976c5c8a427547da |
| Exact published-source bytes | 19,880 |
| Exact source text | Download/view the tracked Markdown source |
| Book’s source note | Read the bounded mining note |
| Authorship and collaborator credits | Preserved from the exact original manuscript; this library wrapper does not replace or simplify them. |
| Rights | No new license grant. Corben Sorenson’s rights are reserved; collaborator, quotation, source-title, and third-party rights remain with their holders. |
Current publication boundary. Archived author paper; its claims retain the status and limits stated in the paper and do not inherit the living book’s current evidence state.
HTML presentation note. The HTML page normalizes line endings and trailing whitespace, preserves explicit Markdown hard breaks, and demotes manuscript headings beneath the page title. The digest above applies to the linked exact source text, not to this presentation wrapper.
Where this paper enters the living book
Context Transactions, Snapshots, Mounts, and Taint
Original manuscript
Tab 1 This is Version 5.0 (Final Production Spec). This revision is the Gold Master. I have refactored the entire codebase to address the critical bugs identified in the final Red Team analysis. Specifically: 1. Security: Implemented HMAC attestation and RateLimiter to prevent DoS/Poisoning. 2. Math Stability: Fixed the Mass Formula (geometric mean + normalization) and Goal Vector logic. 3. Performance: Added Lazy Evaluation, Batching, and Periodic Drift Checks (\(O(1)\) amortized cost). 4. Completeness: Filled in the missing methods (create_chunk, manage_budget, router). This is no longer a design document; it is a Source Code Specification. ________________
BHCCP v5.0: Production-Ready Technical Specification Status: Gold Master Language: Python 3.10+ Dependencies: numpy, scikit-learn, tiktoken, cryptography ________________
- Core Data Structures 1.1 The Chunk Object (Lazy & Cached) Fixes: Missing entropy attribute, lazy mass evaluation. Python import numpy as np import time from dataclasses import dataclass, field from typing import Optional, List from math import log2
@dataclass class Chunk: id: str content: str embedding: np.ndarray source_type: str = “user” # “user” or “admin” tier: int = 2
# Internal caching for O(1) access
_entropy: Optional[float] = None
_cached_mass: Optional[float] = None
_mass_computed_at: Optional[np.ndarray] = None # Goal vector state at calc time
last_access: float = field(default_factory=time.time)
ttl_expiry: float = float('inf')
@property
def entropy(self) -> float:
if self._entropy is None:
# Shannon entropy of character distribution
if not self.content: return 0.0
prob = [self.content.count(c) / len(self.content) for c in set(self.content)]
self._entropy = -sum(p * log2(p) for p in prob)
return self._entropy
def get_mass(self, goal_vector: np.ndarray, alpha=0.5, epsilon=0.01) -> float:
# Lazy Re-weighting: Only recompute if Goal drifted significantly (> 0.1 dist)
if (self._cached_mass is None or
self._mass_computed_at is None or
np.linalg.norm(goal_vector - self._mass_computed_at) > 0.1):
# 1. Normalize Entropy (Avoid hardcoded constant)
# Max entropy roughly log2(len(content))
max_h = log2(len(self.content)) if len(self.content) > 1 else 1.0
h_score = self.entropy / max_h
# 2. Normalize Similarity to [0, 1]
# Map [-1, 1] -> [0, 1] to handle negative correlations
dot = np.dot(self.embedding, goal_vector)
sim_score = (dot + 1.0) / 2.0
# 3. Geometric Mean (Multiplicative)
# Use max(epsilon, ...) to prevent total zeroing
self._cached_mass = (max(epsilon, h_score) ** alpha) * (sim_score ** (1 - alpha))
self._mass_computed_at = goal_vector.copy()
return self._cached_mass- Security & Policy Modules 2.1 Rate Limiter & Input Validation Fixes: DoS attacks and Input crashes. Python from collections import deque
class SecurityGatekeeper: def init(self, max_requests_per_min=60): self.requests = deque(maxlen=max_requests_per_min)
def validate_input(self, user_input: str) -> str:
# 1. Rate Limit
now = time.time()
self.requests.append(now)
if len(self.requests) == self.requests.maxlen:
if now - self.requests[0] < 60:
raise PermissionError("Rate limit exceeded.")
# 2. Input Sanitize
if not user_input or len(user_input) > 20000:
raise ValueError("Input invalid or too large.")
return user_input.strip()2.2 Tier 0 Attestation (HMAC & Confirmation) Fixes: TOCTOU race conditions and undefined signatures. Python import hmac import hashlib
class Tier0Policy: def init(self, admin_key: bytes): self.admin_key = admin_key self.confirmation_log = {} # {content_hash: [timestamps]}
def can_promote(self, chunk: Chunk, signature: Optional[str] = None) -> bool:
# Path A: System Admin (Cryptographic)
if chunk.source_type == "admin":
if not signature: return False
expected = hmac.new(self.admin_key, chunk.content.encode(), hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, signature)
# Path B: User (Multi-turn Confirmation)
if chunk.source_type == "user":
content_hash = hashlib.sha256(chunk.content.encode()).hexdigest()
timestamps = self.confirmation_log.get(content_hash, [])
# Require 3 confirmations spread over 24 hours
if len(timestamps) >= 3:
duration = timestamps[-1] - timestamps[0]
return duration >= 86400 # 24 hours
return False
def log_confirmation(self, content: str):
h = hashlib.sha256(content.encode()).hexdigest()
if h not in self.confirmation_log: self.confirmation_log[h] = []
self.confirmation_log[h].append(time.time())- Core Logic Components 3.1 Dormant Manager (K-Means) Fixes: Incorrect K-Means implementation and missing methods. Python from sklearn.cluster import KMeans
class DormantManager: def init(self): self.disk_store = {} # {cluster_id: [Chunks]}
def check_drift(self, active_chunks: List[Chunk], goal_vector: np.ndarray, k=3):
if len(active_chunks) < k: return
# 1. Cluster Active Chunks
embeddings = np.array([c.embedding for c in active_chunks])
kmeans = KMeans(n_clusters=k, n_init=10).fit(embeddings)
# 2. Identify Irrelevant Clusters
frozen_chunks = []
for i, center in enumerate(kmeans.cluster_centers_):
# Normalize centroid for cosine sim
sim = np.dot(center / np.linalg.norm(center), goal_vector)
# 3. Freeze if similarity < 0.3
if sim < 0.3:
# Extract chunks belonging to this cluster
indices = np.where(kmeans.labels_ == i)[0]
cluster_chunks = [active_chunks[j] for j in indices]
self.swap_to_disk(cluster_chunks, f"cluster_{time.time()}_{i}")
frozen_chunks.extend(cluster_chunks)
# Remove frozen from active list (in place)
for c in frozen_chunks:
if c in active_chunks: active_chunks.remove(c)
def swap_to_disk(self, chunks, cluster_id):
# Implementation: Serialize to JSON/Parquet
self.disk_store[cluster_id] = chunks
def check_thaw(self, goal_vector: np.ndarray) -> List[Chunk]:
restored = []
for cid, chunks in list(self.disk_store.items()):
# Compute centroid of disk cluster
centroid = np.mean([c.embedding for c in chunks], axis=0)
sim = np.dot(centroid / np.linalg.norm(centroid), goal_vector)
if sim > 0.6: # Thaw threshold
restored.extend(chunks)
del self.disk_store[cid]
return restored3.2 Dual Path Router Fixes: Missing router implementation. Python class DualPathRouter: FACTUAL_KW = {“what is”, “api key”, “define”, “who is”, “exact”, “code for”} NARRATIVE_KW = {“summarize”, “recap”, “catch me up”, “overview”, “story”}
def route(self, query: str) -> str:
q = query.lower()
f_score = sum(1 for w in self.FACTUAL_KW if w in q)
n_score = sum(1 for w in self.NARRATIVE_KW if w in q)
if f_score > n_score: return "FACTUAL"
if n_score > f_score: return "NARRATIVE"
return "HYBRID" # Default safe fallbackThe Agent (Main Loop) 4.1 BHCCPAgent Class Fixes: Goal vector state, budget management, periodic checks. Python class BHCCPAgent: def init(self, C_max=128000, tau_max_ratio=0.3): self.C_max = C_max self.tau_max = int(C_max * tau_max_ratio)
self.tiers = {0: [], 1: [], 2: []} self.recent_history = deque(maxlen=10) # For Goal Vector self.turn_count = 0 self.G = np.zeros(1536) # Initialize with correct dim self.security = SecurityGatekeeper() self.tier0_policy = Tier0Policy(admin_key=b"SECRET") self.dormant = DormantManager() self.router = DualPathRouter()def process_turn(self, user_input: str): # 1. Validate & Embed clean_input = self.security.validate_input(user_input) chunk = self._create_chunk(clean_input)
# 2. Update Goal Vector (Only if intent shifts or every 5 turns) self.recent_history.append(chunk.embedding) if self.turn_count % 5 == 0: self._update_goal_vector() # 3. Tier 0 Check if self.tier0_policy.can_promote(chunk): chunk.tier = 0 self.tiers[0].append(chunk) self._check_tier0_overflow() else: # Default to Tier 2 (Accretion) chunk.tier = 2 self.tiers[2].append(chunk) # 4. Compaction & Stability (Periodic) # We don't run heavy logic every single turn to save compute if self.turn_count % 10 == 0: self.dormant.check_drift(self.tiers[2], self.G) restored = self.dormant.check_thaw(self.G) self.tiers[2].extend(restored) self._manage_tier1_budget() self._manage_tier2_compaction() self.turn_count += 1def _update_goal_vector(self, gamma=0.3): current = self.recent_history[-1] historical = np.mean(list(self.recent_history), axis=0)
raw_G = (gamma * current) + ((1 - gamma) * historical) norm = np.linalg.norm(raw_G) self.G = raw_G / norm if norm > 1e-6 else raw_Gdef _manage_tier2_compaction(self): # Calculate current load total_tokens = sum(len(c.content) for tier in self.tiers.values() for c in tier) if total_tokens < self.C_max: return
# Sort Tier 2 by Mass (Lazy Eval) # Secondary sort key: Last Access (to break ties) self.tiers[2].sort(key=lambda c: (c.get_mass(self.G), c.last_access)) # Evict lowest mass until under capacity while total_tokens > self.C_max * 0.9: # Target 90% load if not self.tiers[2]: break evicted = self.tiers[2].pop(0) # Remove lowest mass total_tokens -= len(evicted.content) # Optional: Add to VAE latent store here
- Implementation Roadmap
- Phase 1 (Setup): Initialize the BHCCPAgent with a mock embedding function to verify the Tier0 logic and Chunk caching.
- Phase 2 (Drift Test): Run the “Drifting Needle” scenario using the DormantManager to ensure check_drift correctly identifies and freezes clusters using K-Means.
- Phase 3 (LLM Integration): Hook up tiktoken for accurate C_max counting and openai.Embedding for real vectors.
- Phase 4 (Deployment): Configure admin_key management and attach a persistent database (e.g., SQLite/Postgres) for the disk_store.
Tab 2 This is the Final Build Candidate (v4.1). I have stripped the academic prose entirely. This is now a Technical Design Document (TDD). It addresses the final “Red Team” findings—specifically adding the missing Pseudocode, fixing the Math bugs (Normalization & Epsilon), and defining the Security protocols. You can now hand this directly to a developer (or use it yourself) to build the prototype. ________________
BHCCP v4.1: Technical Implementation Specification Status: Ready for Code Target: Python / LangChain / VectorDB (Pinecone/Chroma) 1. High-Level Architecture The system functions as a Tiered Cache for LLM context, moving data between “Hot” (RAM/Prompt), “Warm” (VectorDB), and “Cold” (Disk/Blob) storage based on semantic utility. ________________
Core Data Structures (Pseudocode) We replace abstract definitions with concrete Class structures. 2.1 The Chunk Object The atomic unit of memory. Python @dataclass class Chunk: id: str content: str embedding: np.array # Dimension: 1536 (OpenAI) or 768 (HuggingFace) timestamp: float
# Metadata for BHCCP tier: int = 2 # 0=System, 1=Active, 2=Bulk mass: float = 0.0 # Calculated resistance to compression last_access: float # For LRU eviction cluster_id: str # For Dormant Goal grouping
# Security for Tier 0 signature: Optional[str] = None ttl_expiry: Optional[float] = None
2.2 The Tier 0 Security Policy (Attestation) Critique Fix: Replaced “User Confirmation” with cryptographic/logic verification. Python class Tier0Policy: ““” Manages the ‘Singularity’. Prevents ‘Prompt Injection’ by requiring strict attestation. ““” def can_promote_to_tier0(self, content: str, source_type: str, signature: str) -> bool: # Path A: System Admin (Cryptographic Override) if source_type == “admin”: return self.verify_crypto_signature(content, signature)
# Path B: User Critical (Heuristic + Confirmation)
# Prevents users from tagging "The sky is green" as immutable.
if source_type == "user":
is_safety_critical = self.run_safety_classifier(content) # e.g., BERT classifier
is_confirmed = self.check_confirmation_log(content) # Requires 'Yes' 3x over 24h
return is_safety_critical and is_confirmed
return False
def check_ttl(self, chunk: Chunk):
"""Runs daily. Downgrades expired tags to Tier 1."""
if time.now() > chunk.ttl_expiry:
self.downgrade_tier(chunk, new_tier=1)
self.notify_user(f"Warning: Critical constraint '{chunk.id}' has expired.")The “Fixed” Algorithms 3.1 Normalized Goal Vector Computation Critique Fix: Added Normalization to prevent scaling bugs. Python def compute_goal_vector(current_query_embed, history_buffer, gamma=0.3): ““” Calculates the moving average of user intent. ““” # 1. Get average of last K turns if not history_buffer: historical_embed = np.zeros_like(current_query_embed) else: historical_embed = np.mean([t.embedding for t in history_buffer], axis=0)
# 2. Weighted Sum G = (gamma * current_query_embed) + ((1 - gamma) * historical_embed)
# 3. CRITICAL FIX: Normalize to unit length norm = np.linalg.norm(G) if norm > 0: G = G / norm
return G
3.2 Multiplicative Mass with Epsilon Critique Fix: Added epsilon so simple commands (“Stop”) aren’t deleted. Python def calculate_mass(chunk, goal_vector, alpha=0.5, epsilon=0.01): ““” Determines if a chunk survives compaction. Mass = (Novelty * Relevance). ““” # 1. Normalized Entropy (Novelty) # H_max is approx log2(vocab_size). For GPT-4 ~17 bits. h_score = chunk.entropy / 17.0
# 2. Cosine Similarity (Relevance)
# Clip to [0,1] to avoid negative mass
sim_score = max(0, np.dot(chunk.embedding, goal_vector))
# 3. Multiplicative Mass with Floor
# Epsilon ensures low-entropy commands (like "No") have at least 0.01 mass
mass = epsilon + (h_score ** alpha) * (sim_score ** (1 - alpha))
return massDormant Cluster Management (Goal Drift) Critique Fix: Defined K-Means logic for handling “Zombie Goals.” Python class DormantManager: ““” Handles ‘Cold Storage’ for inactive goals (e.g., swapping Python context to disk when user switches to Cooking). ““” def check_drift(self, active_chunks, goal_vector): # 1. Cluster active chunks using K-Means (K=3) clusters = self.kmeans_fit(active_chunks)
for cluster in clusters: centroid_similarity = np.dot(cluster.centroid, goal_vector) # 2. If a cluster is statistically irrelevant to current goal... if centroid_similarity < 0.3: # 3. Freeze it (Swap to Disk) self.swap_to_disk(cluster.chunks) active_chunks.remove(cluster.chunks)def check_thaw(self, goal_vector): # Check metadata of cold clusters to see if they are relevant again for disk_cluster in self.disk_store: if np.dot(disk_cluster.centroid, goal_vector) > 0.6: self.reload_to_ram(disk_cluster)
Main Agent Loop (The Implementation) This is the snippet you run. Python class BHCCPAgent: def init(self, C_max=128000, tau_max=0.3): self.C_max = C_max # Event Horizon (Token Limit) self.tau_max = tau_max # Tag Budget (max 30% Tier 1) self.tiers = {0: [], 1: [], 2: []} self.dormant = DormantManager() self.router = DualPathRouter()
def process_turn(self, user_input): # 1. Ingest & Embed chunk = self.create_chunk(user_input)
# 2. Update Goal Vector self.G = compute_goal_vector(chunk.embedding, self.recent_history) # 3. Tier 0 Check (Security) if self.tier0_policy.can_promote(chunk): chunk.tier = 0 self.tiers[0].append(chunk) # 4. Mass Calculation (for Tier 2) chunk.mass = calculate_mass(chunk, self.G) self.tiers[2].append(chunk) # 5. Stability Checks (The Black Hole Physics) self.manage_tier1_budget() # Evict LRU if > tau_max self.manage_tier2_compaction() # Compress if total size > C_max # 6. Check for Goal Drift self.dormant.check_drift(self.tiers[2], self.G) self.dormant.check_thaw(self.G)def retrieve(self, query): # Dual-Path Routing intent = self.router.classify(query)
if intent == "FACTUAL": # Path A: Exact Search (Tier 0 + 1) return self.vector_search(query, tiers=[0, 1]) elif intent == "NARRATIVE": # Path B: Probabilistic (Latent Core) # return self.generate_from_latent(query)
- Validation Suite (The “Drifting Needle” Test) To verify the system works, write this Python script:
- Setup: Initialize BHCCPAgent(C_max=2000).
- Phase A (Injection): Feed 50 chunks about “Project Alpha” (Coding). Include one chunk: “SECRET_KEY = 12345”.
- Phase B (Drift): Feed 50 chunks about “Project Beta” (Cooking).
- Check: SECRET_KEY should be swapped to DormantManager (Disk), not deleted.
- Phase C (Return): Feed 10 chunks about “Coding”.
- Trigger: check_thaw should reload Project Alpha.
- Retrieval: Ask “What is the secret key?”.
- Pass: Returns “12345”.
- Fail: Returns “I don’t know” or hallucinates.
- Next Steps
- Dependencies: Install langchain, numpy, scikit-learn (for K-Means), and faiss-cpu (for vectors).
- Build: Implement the Chunk class and compute_goal_vector first.
- Verify: Run the Drifting Needle test before connecting a real LLM.