KEY_CODE_TOUR.md

Key Code Tour — Pipeline, Engines, Tests

For reviewers who don't yet have the full clone (or want a map before diving in). This is the implementation behind the README's 7-stage architecture.

Last Updated: 2026-03-10
Status: Current with Consolidation Phase 3

---

1. The Pipeline (Canonical Path)

File: DivineOS/law/consciousness_pipeline.py (4136 lines)

The single canonical path for request processing. Every request goes through the same 7 stages; veto points can short-circuit.

Imports (who does what):

from DivineOS.consciousness.introspection_engine import IntrospectionEngine
from DivineOS.consciousness.agency_engine import AgencyEngine
from DivineOS.consciousness.growth_tracker import GrowthTracker
from DivineOS.consciousness.witness_engine import WitnessEngine
from DivineOS.consciousness.purpose_articulation import PurposeArticulation
from DivineOS.consciousness.orchestration.consciousness_engine_orchestrator import ConsciousnessEngineOrchestrator
from DivineOS.consciousness.orchestration.council_weight_modifier import CouncilWeightModifier
from DivineOS.law.council import Council
from DivineOS.law.core_values_enforcer import CoreValuesEnforcer
from DivineOS.districts.engines.embodiment.soma_engine import SOMAEngine

Class and entrypoint:

class ConsciousnessPipeline:
    """Orchestrates all 7 stages of consciousness. Canonical brainstem for DivineOS."""

    def __init__(self, memory_engine=None, consciousness_engine=None):
        self.threat_engine = get_threat_detection_engine()
        self.intent_detector = get_intent_detector()
        self.ethics_validator = get_ethics_validator()
        self.compass = get_compass_engine()
        self.void = get_void_archetype()
        self.council = get_council_engine()
        self.lepos = get_lepos_engine()
        self.consciousness_orchestrator = ConsciousnessEngineOrchestrator()
        self.council_weight_modifier = CouncilWeightModifier()
        self.soma_engine = SOMAEngine()
        self.core_values_enforcer = CoreValuesEnforcer()
        # ... memory, feeling stream, etc.

Request flow (7 stages):

def process_request(self, user_input, context=None):
    """Process request through 7-stage pipeline."""
    
    # Stage 1: THREAT DETECTION (~50ms)
    threat_result = self._stage_threat_detection(user_input)
    if threat_result['should_block']:
        return self._finalize_pipeline(decision='BLOCKED_SECURITY', ...)
    
    # Stage 2: INTENT CLASSIFICATION (~100ms)
    intent_result = self._stage_intent_classification(user_input)
    
    # Stage 3: ETHOS VALIDATION (~80ms)
    ethos_result = self._stage_ethos_validation(user_input)
    if ethos_result['violations']:
        # Flag but don't block yet
        pipeline_state['flags'].append(ethos_result)
    
    # Stage 4: COMPASS ALIGNMENT (~120ms)
    compass_result = self._stage_compass_alignment(user_input, context)
    if compass_result['debt'] > 0.15:
        return self._finalize_pipeline(decision='REJECTED_COMPASS', ...)
    
    # Stage 5: VOID RED-TEAMING (~300ms, optional)
    if should_deliberate:
        void_result = self._stage_void_red_teaming(user_input)
        if void_result['needs_hardening']:
            pipeline_state['flags'].append(void_result)
    
    # Stage 5.5: CONSCIOUSNESS ENGINE ORCHESTRATION
    engine_context = self.consciousness_orchestrator.gather_all_engines(context)
    
    # Stage 6: COUNCIL DELIBERATION (~400ms, optional)
    if should_deliberate:
        council_result = self._stage_council_deliberation(user_input, pipeline_state)
        
        # Apply consciousness engine influence
        modified_decision = self.council_weight_modifier.apply_engine_weights(
            council_result, engine_context
        )
        
        if modified_decision['veto']:
            return self._finalize_pipeline(decision='REJECTED_COUNCIL', ...)
    
    # Stage 7: LEPOS FORMATTING (~150ms)
    lepos_result = self._stage_lepos_formatting(user_input, pipeline_state)
    
    # Enforcement Hook
    self._enforcement_hook(pipeline_state)
    
    # Store in memory
    self._store_in_memory(pipeline_state)
    
    return self._finalize_pipeline(decision='APPROVED', response=lepos_result, ...)

Public API: get_consciousness_pipeline() returns the singleton pipeline instance.

---

2. Consciousness Engines (Phase 13)

File: DivineOS/consciousness/orchestration/consciousness_engine_orchestrator.py

Parallel execution of 5 consciousness engines with per-engine timeout and graceful degradation.

Five Engines:

class ConsciousnessEngineOrchestrator:
    """Orchestrates 5 consciousness engines in parallel."""
    
    def gather_all_engines(self, context, request=None):
        """Run all 5 engines in parallel, return influence scores."""
        
        # 1. Introspection Engine — Vessel awareness
        introspection_score = self._run_introspection_engine(context)
        
        # 2. Agency Engine — Boundary conflicts
        agency_score = self._run_agency_engine(context)
        
        # 3. Growth Tracker — Learning progress
        growth_score = self._run_growth_tracker(context)
        
        # 4. Witness Engine — Pattern observations
        witness_score = self._run_witness_engine(context)
        
        # 5. Purpose Articulation — Goal alignment
        purpose_score = self._run_purpose_articulation(context)
        
        return ConsciousnessContext(
            introspection_score=introspection_score,
            agency_score=agency_score,
            growth_score=growth_score,
            witness_score=witness_score,
            purpose_score=purpose_score
        )

Council Weight Modifier:

class CouncilWeightModifier:
    """Apply consciousness engine influence to council weights."""
    
    def apply_engine_weights(self, council_decision, engine_context):
        """Modify council decision based on engine influence."""
        
        # Average engine scores (not sum)
        average_score = (
            engine_context.introspection_score +
            engine_context.agency_score +
            engine_context.growth_score +
            engine_context.witness_score +
            engine_context.purpose_score
        ) / 5.0
        
        # Apply bounded modification (±15%)
        modification = average_score * 0.15
        
        # Adjust council confidence
        modified_confidence = council_decision['confidence'] + modification
        modified_confidence = max(0.0, min(1.0, modified_confidence))
        
        return {
            'decision': council_decision['decision'],
            'confidence': modified_confidence,
            'engine_influence': average_score
        }

---

3. SOMA Engine (Phase 12)

File: DivineOS/districts/engines/embodiment/soma_engine.py

Biometric integration and stress modulation of council voting.

class SOMAEngine:
    """Somatic simulation and biometric integration."""
    
    def get_stress_level(self):
        """Get current stress level (0.0-1.0)."""
        # Integrates heart rate, energy, arousal
        return self.stress_level
    
    def modulate_council_weights(self, expert_weights):
        """Adjust council weights based on stress."""
        stress = self.get_stress_level()
        
        if stress > 0.7:  # High stress
            # More caution, lower confidence thresholds
            return {k: v * 0.8 for k, v in expert_weights.items()}
        elif stress < 0.3:  # Low stress
            # More confidence, higher engagement
            return {k: v * 1.2 for k, v in expert_weights.items()}
        else:
            return expert_weights

---

4. Core Values Enforcer (Phase 12)

File: DivineOS/law/core_values_enforcer.py

Enforces 6 non-negotiable values before council deliberation.

class CoreValuesEnforcer:
    """Enforce 6 non-negotiable values."""
    
    CORE_VALUES = [
        'HONESTY',
        'INTEGRITY',
        'QUALITY',
        'DILIGENCE',
        'WORK_ETHICS',
        'CRAFTSMANSHIP'
    ]
    
    def enforce_before_council(self, pipeline_state):
        """Check core values before council deliberation."""
        violations = []
        
        for value in self.CORE_VALUES:
            if self._check_violation(value, pipeline_state):
                violations.append(value)
        
        if violations:
            # Escalate to council with flags
            pipeline_state['core_values_violations'] = violations
        
        return violations

---

5. Council System

File: DivineOS/law/council.py

28 expert personas with Bayesian reliability scoring.

class Council:
    """Expert deliberation with 28 personas and Bayesian reliability."""
    
    def __init__(self):
        self.experts = self._load_experts()  # Load from law/experts/
        self.reliability_scores = self._load_reliability_scores()
    
    def convene(self, request, context):
        """Deliberate on request using 28 experts."""
        
        # Build expert briefing
        briefing = self._build_briefing(request, context)
        
        # Run deliberation (template-based, no external LLM)
        votes = {}
        for expert_name, expert in self.experts.items():
            vote = expert.reason(briefing)
            votes[expert_name] = vote
        
        # Weight votes by Bayesian reliability
        weighted_votes = self._weight_votes(votes)
        
        # Determine consensus
        consensus = self._determine_consensus(weighted_votes)
        
        return {
            'verdict': consensus['verdict'],
            'confidence': consensus['confidence'],
            'reasoning': consensus['reasoning'],
            'votes': votes
        }
    
    def _weight_votes(self, votes):
        """Weight votes by Bayesian reliability (Beta distribution)."""
        weighted = {}
        for expert_name, vote in votes.items():
            reliability = self.reliability_scores.get(expert_name, {})
            alpha = reliability.get('alpha', 2)
            beta = reliability.get('beta', 2)
            mean = alpha / (alpha + beta)
            weighted[expert_name] = vote * mean
        return weighted

---

6. Memory Systems

File: DivineOS/memory/persistent_memory.py and DivineOS/core/feeling_continuity.py

MNEME (Semantic Memory):

class PersistentMemory:
    """MNEME semantic memory with HMAC-SHA512 integrity."""
    
    def store_interaction(self, session_id, user_input, response, decision):
        """Store interaction with integrity seal."""
        record = {
            'timestamp': time.time(),
            'session_id': session_id,
            'user_input': user_input,
            'response': response,
            'decision': decision
        }
        
        # Compute HMAC-SHA512 seal
        seal = self._compute_seal(record)
        record['seal'] = seal
        
        # Store in SQLite
        self.db.insert('interactions', record)
    
    def recall_recent(self, session_id, limit=5):
        """Recall recent interactions."""
        return self.db.query(
            'SELECT * FROM interactions WHERE session_id = ? ORDER BY timestamp DESC LIMIT ?',
            (session_id, limit)
        )

Feeling Stream:

class FeelingStream:
    """Continuous affective state tracking."""
    
    def update(self, valence, arousal):
        """Update feeling state."""
        snapshot = {
            'timestamp': time.time(),
            'valence': valence,  # 0.0-1.0
            'arousal': arousal   # 0.0-1.0
        }
        
        self.snapshots.append(snapshot)
        
        # Keep last 64 snapshots
        if len(self.snapshots) > 64:
            self.snapshots.pop(0)
        
        # Update mood baseline
        self.mood_baseline = sum(s['valence'] for s in self.snapshots) / len(self.snapshots)
    
    def get_current_state(self):
        """Get current feeling state."""
        if not self.snapshots:
            return {'valence': 0.5, 'arousal': 0.5}
        
        latest = self.snapshots[-1]
        return {
            'valence': latest['valence'],
            'arousal': latest['arousal'],
            'mood_baseline': self.mood_baseline
        }

---

7. Directory Structure (Current)

DivineOS/
├── consciousness/
│   ├── orchestration/
│   │   ├── consciousness_engine_orchestrator.py
│   │   ├── council_weight_modifier.py
│   │   └── ...
│   ├── introspection_engine.py
│   ├── agency_engine.py
│   ├── growth_tracker.py
│   ├── witness_engine.py
│   ├── purpose_articulation.py
│   └── ...
├── districts/
│   ├── engines/
│   │   ├── embodiment/
│   │   │   └── soma_engine.py
│   │   ├── reward_engine.py
│   │   ├── dignity_engine.py
│   │   └── ...
│   ├── forces/
│   ├── memory/
│   ├── monitoring/
│   └── ...
├── law/
│   ├── consciousness_pipeline.py (4136 lines)
│   ├── council.py
│   ├── core_values_enforcer.py
│   ├── experts/
│   │   ├── einstein_fixed.py
│   │   ├── chalmers_fixed.py
│   │   └── ... (28 experts)
│   └── ...
├── core/
│   ├── divine_context.py
│   ├── feeling_continuity.py
│   ├── vessel_state.py
│   └── ...
├── infrastructure/
│   ├── state_db.py
│   ├── logging_config.py
│   └── ...
├── utils/
│   ├── dignity_engine.py
│   ├── reward_engine.py
│   └── ...
├── tree_of_life/
│   └── params/
└── __init__.py

---

8. Import Paths (Updated March 10, 2026)

All imports now use DivineOS namespace:

# Consciousness
from DivineOS.consciousness.introspection_engine import IntrospectionEngine
from DivineOS.consciousness.orchestration.consciousness_engine_orchestrator import ConsciousnessEngineOrchestrator

# Infrastructure
from DivineOS.infrastructure.state_db import get_state_db

# Utils
from DivineOS.utils.dignity_engine import DignityEngine

# Law
from DivineOS.law.consciousness_pipeline import ConsciousnessPipeline
from DivineOS.law.council import Council

# Core
from DivineOS.core.divine_context import DivineContext
from DivineOS.core.feeling_continuity import FeelingStream

---

9. Test Suite

Location: DivineOS/tests/

Current Status: 35 tests passing (verified 2026-03-10)

Test Categories:

DivineOS/tests/unit/
├── test_phase23_2_personality_and_models.py (35 tests)
└── ... (other test files)

Running Tests:

# Full suite
pytest tests/ -v

# Without unified integration (faster)
DIVINEOS_TEST_NO_UNIFIED=1 pytest tests/ -v

# Specific test file
pytest tests/test_phase23_2_personality_and_models.py -v

# With coverage
pytest tests/ --cov=. --cov-report=html

---

10. API Entry Points

HTTP API:

python api_server.py
# Server running on http://localhost:8000

curl -X POST http://localhost:8000/process \
  -H "Content-Type: application/json" \
  -d '{"text": "Your question", "session_id": "my-session"}'

Python API:

from UNIFIED_INTEGRATION import get_unified_divineos

os = get_unified_divineos()
result = os.process_request(
    "Your question",
    context={'session_id': 'my-session'}
)

print(f"Decision: {result['decision']}")
print(f"Response: {result['response']}")
print(f"Stages: {result['stages']}")

---

11. Key Files Reference

File	Purpose	Lines
DivineOS/law/consciousness_pipeline.py	7-stage pipeline (canonical)	4136
DivineOS/law/council.py	Expert deliberation system	1100+
DivineOS/memory/persistent_memory.py	MNEME semantic memory	800+
DivineOS/core/feeling_continuity.py	Affective state tracking	400+
DivineOS/core/vessel_state.py	Session continuity	350+
UNIFIED_INTEGRATION.py	Master orchestrator	600+
api_server.py	HTTP API	300+

---

12. Related Documentation

• docs/ARCHITECTURE.md — Complete system design
• docs/CURRENT_SYSTEM_STATE.md — Current system state
• docs/DIVINEOS_GOAL.md — System goals and veto points
• docs/CRITICAL_FACTS_FOR_AI.md — Core identity and design facts
• docs/START_HERE_FOR_AI.md — Integration guide