diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6efbdcf..3c88a58 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -35,6 +35,26 @@ jobs: nltk \ pytest pytest-cov pytest-asyncio pytest-timeout + - name: Enforce single-path runtime + run: | + echo "--- Verifying no duplicate FastAPI app definitions ---" + # Only backend/unified_server.py may define the production FastAPI app. + # backend/minimal_server.py is deprecated and backend/main.py is a shim. + # Pattern: lines starting with 'app = FastAPI(' (ignoring leading whitespace) + # — this anchored regex avoids false positives in comments/docstrings. + VIOLATIONS=$(grep -rnP '^(?!\s*#)\s*app\s*=\s*FastAPI\(' backend/ \ + --include='*.py' \ + | grep -v unified_server.py \ + | grep -v minimal_server.py \ + | grep -v main.py \ + || true) + if [ -n "$VIOLATIONS" ]; then + echo "❌ Found duplicate FastAPI app definitions:" + echo "$VIOLATIONS" + exit 1 + fi + echo "✅ Single entrypoint enforced — only unified_server.py defines app" + - name: Run tests id: run_tests run: | diff --git a/backend/core/consciousness_engine.py b/backend/core/consciousness_engine.py index a3bff18..e8fd94e 100644 --- a/backend/core/consciousness_engine.py +++ b/backend/core/consciousness_engine.py @@ -57,8 +57,16 @@ class SelfAwarenessMetrics: class ConsciousnessEngine: """ + DEPRECATED — Use ``backend.core.unified_consciousness_engine.UnifiedConsciousnessEngine`` + as the single canonical consciousness computation. + + This class is retained because ``CognitiveManager`` and existing tests + reference it, but no new code should instantiate ``ConsciousnessEngine`` + directly. The ``ConsciousnessState`` and ``ConsciousnessLevel`` data types + defined in this module remain canonical. + Advanced consciousness engine implementing manifest consciousness behaviors - and comprehensive self-awareness assessment + and comprehensive self-awareness assessment. """ def __init__(self, llm_driver=None, knowledge_pipeline=None, websocket_manager=None): diff --git a/backend/core/enhanced_websocket_manager.py b/backend/core/enhanced_websocket_manager.py index b95b799..12e0894 100644 --- a/backend/core/enhanced_websocket_manager.py +++ b/backend/core/enhanced_websocket_manager.py @@ -17,16 +17,8 @@ from dataclasses import asdict from datetime import datetime -# Import existing WebSocket manager -try: - from ..websocket_manager import WebSocketManager -except ImportError: - # Fallback for testing - class WebSocketManager: - def __init__(self): - self.active_connections = set() - async def broadcast(self, message): - pass +# Import the canonical base WebSocket manager +from backend.websocket_manager import WebSocketManager # noqa: E402 — absolute import avoids edge-case relative import failures in test runners logger = logging.getLogger(__name__) diff --git a/backend/godelos_integration.py b/backend/godelos_integration.py index cb6c5ec..7cfebce 100644 --- a/backend/godelos_integration.py +++ b/backend/godelos_integration.py @@ -21,8 +21,19 @@ class GödelOSIntegration: """ - A simplified working integration class for GödelOS API. + GödelOS integration layer. + + All cognitive processing MUST flow through ``godelOS.cognitive_pipeline.CognitivePipeline`` + when available. The inline ``_fallback_knowledge_store`` below is a **test-only + stub** — it exists solely so that unit tests can run without the full pipeline + and should never be treated as production data. + + Accessing ``simple_knowledge_store`` when the canonical pipeline is active + will emit a runtime warning to catch accidental fallback-path usage. """ + + # Track whether we have already warned for fallback usage in this process. + _fallback_warned: bool = False def __init__(self): self.initialized = False @@ -30,8 +41,10 @@ def __init__(self): self.error_count = 0 self.cognitive_pipeline: Optional["CognitivePipeline"] = None - # Enhanced knowledge store for better search - self.simple_knowledge_store = { + # TEST-ONLY STUB — static seed data used when CognitivePipeline is + # unavailable (e.g. in unit tests). Production reads MUST go through + # self.cognitive_pipeline. + self._fallback_knowledge_store = { "system_status": { "title": "System Status", "content": "The system is currently operational.", @@ -70,6 +83,26 @@ def __init__(self): } } + @property + def simple_knowledge_store(self): + """Accessor for the fallback knowledge store. + + Emits a one-time warning when the canonical ``CognitivePipeline`` is + available, signaling that production code is falling through to the + test-only stub. + """ + if self.cognitive_pipeline is not None and not GödelOSIntegration._fallback_warned: + GödelOSIntegration._fallback_warned = True + logger.warning( + "⚠️ simple_knowledge_store accessed while CognitivePipeline is " + "active — production reads should go through the pipeline." + ) + return self._fallback_knowledge_store + + @simple_knowledge_store.setter + def simple_knowledge_store(self, value): + self._fallback_knowledge_store = value + async def initialize(self, pipeline_service=None, mgmt_service=None): """Initialize the integration.""" try: diff --git a/backend/minimal_server.py b/backend/minimal_server.py index 7673564..44d2872 100644 --- a/backend/minimal_server.py +++ b/backend/minimal_server.py @@ -1,912 +1,24 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ -Minimal GödelOS Backend Server -A streamlined version of the backend that provides essential API endpoints -for the frontend without complex dependencies. -""" - -import asyncio -import json -import logging -import os -from datetime import datetime -from typing import Dict, List, Optional, Any -from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Query -from fastapi.middleware.cors import CORSMiddleware -from pydantic import BaseModel -from dotenv import load_dotenv - -# Load environment variables from .env file -load_dotenv() - -# Configure logging -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) +DEPRECATED — Minimal GödelOS Backend Server. -# Import the new tool-based LLM integration -try: - import llm_tool_integration - from llm_tool_integration import ToolBasedLLMIntegration - LLM_INTEGRATION_AVAILABLE = True -except ImportError as e: - logger.warning(f"LLM integration not available: {e}") - llm_tool_integration = None - LLM_INTEGRATION_AVAILABLE = False +This module is superseded by ``backend.unified_server``, which is the single +canonical runtime entrypoint for all GödelOS server startup paths. -app = FastAPI( - title="GödelOS Minimal Cognitive API", - description="Streamlined cognitive architecture API for essential functionality", - version="1.0.0" -) - -# Enable CORS -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], # In production, replace with specific origins - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], +This file is a pure compatibility shim. It re-exports the canonical ``app`` +so that any external tooling referencing ``backend.minimal_server:app`` will +still work. It will be removed in a future release. +""" +import warnings as _warnings +_warnings.warn( + "backend.minimal_server is deprecated. Use backend.unified_server as the " + "canonical server entrypoint.", + DeprecationWarning, + stacklevel=2, ) -# Global state for demonstration -cognitive_state = { - "attention_focus": { - "topic": "System Initialization", - "intensity": 0.75, - "context": "Cognitive architecture startup", - "mode": "Active" - }, - "processing_load": 0.23, - "working_memory": { - "items": [ - {"id": 1, "content": "User query processing", "priority": 0.8}, - {"id": 2, "content": "Knowledge retrieval", "priority": 0.6} - ], - "capacity": 10, - "utilization": 0.4 - }, - "system_health": { - "overall": 0.92, - "components": { - "inference_engine": 0.94, - "knowledge_store": 0.89, - "attention_manager": 0.95, - "memory_manager": 0.88 - } - }, - "active_agents": 3, - "cognitive_events": [] -} - -# Request/Response Models -class CognitiveStreamConfig(BaseModel): - granularity: str = "standard" - subscriptions: List[str] = [] - max_event_rate: Optional[int] = None - -class LLMCognitiveRequest(BaseModel): - query: str - context: Optional[Dict[str, Any]] = None - use_tools: bool = True - -class LLMCognitiveResponse(BaseModel): - response: str - tools_used: List[str] - confidence: float - processing_time: float - cognitive_state_changes: Dict[str, Any] - -# WebSocket connection manager -class ConnectionManager: - def __init__(self): - self.active_connections: List[WebSocket] = [] - - async def connect(self, websocket: WebSocket): - await websocket.accept() - self.active_connections.append(websocket) - logger.info(f"WebSocket connected. Total connections: {len(self.active_connections)}") - - def disconnect(self, websocket: WebSocket): - if websocket in self.active_connections: - self.active_connections.remove(websocket) - logger.info(f"WebSocket disconnected. Total connections: {len(self.active_connections)}") - - async def send_personal_message(self, message: str, websocket: WebSocket): - try: - await websocket.send_text(message) - except: - self.disconnect(websocket) - - async def broadcast(self, message: str): - disconnected = [] - for connection in self.active_connections: - try: - await connection.send_text(message) - except: - disconnected.append(connection) - - # Remove disconnected clients - for conn in disconnected: - self.disconnect(conn) - -manager = ConnectionManager() - -# Utility functions for safe data handling -def safe_percentage(value, fallback=0): - """Safely convert value to percentage.""" - try: - if isinstance(value, (int, float)) and not (value != value): # Check for NaN - return max(0, min(100, round(value * 100))) - return fallback - except: - return fallback - -def safe_number(value, fallback=0): - """Safely convert value to number.""" - try: - if isinstance(value, (int, float)) and not (value != value): # Check for NaN - return value - return fallback - except: - return fallback - -# API Endpoints - -@app.get("/") -async def root(): - """Root endpoint providing API information.""" - return { - "name": "GödelOS Minimal Cognitive API", - "version": "1.0.0", - "status": "operational", - "endpoints": [ - "/cognitive/state", - "/cognitive/query", - "/enhanced-cognitive/stream/configure", - "/llm/cognitive-query", - "/llm/test-integration", - "/llm/tools", - "/ws/unified-cognitive-stream" - ], - "new_features": [ - "Tool-based LLM integration", - "Function calling architecture", - "Cognitive grounding verification", - "Comprehensive tool documentation" - ] - } - -@app.get("/health") -async def health_check(): - """Health check endpoint.""" - return { - "status": "healthy", - "timestamp": datetime.now().isoformat(), - "active_connections": len(manager.active_connections) - } - -@app.get("/api/health") -async def api_health_check(): - """API health check endpoint with /api prefix.""" - return await health_check() - -@app.get("/cognitive/state") -async def get_cognitive_state(): - """Get current cognitive state.""" - # Add some realistic variation - import random - cognitive_state["processing_load"] = max(0, min(1, cognitive_state["processing_load"] + random.uniform(-0.1, 0.1))) - cognitive_state["attention_focus"]["intensity"] = max(0, min(1, cognitive_state["attention_focus"]["intensity"] + random.uniform(-0.05, 0.05))) - - return { - "cognitive_state": cognitive_state, - "timestamp": datetime.now().isoformat() - } - -@app.get("/api/cognitive/state") -async def api_get_cognitive_state(): - """API cognitive state endpoint with /api prefix.""" - return await get_cognitive_state() - -@app.get("/api/transparency/knowledge-graph/export") -async def export_knowledge_graph(): - """Export knowledge graph data.""" - return { - "nodes": [ - {"id": 1, "label": "Consciousness", "type": "concept", "properties": {"domain": "philosophy"}}, - {"id": 2, "label": "Cognitive Architecture", "type": "concept", "properties": {"domain": "AI"}}, - {"id": 3, "label": "Meta-cognition", "type": "concept", "properties": {"domain": "psychology"}}, - {"id": 4, "label": "Self-awareness", "type": "concept", "properties": {"domain": "consciousness"}}, - ], - "edges": [ - {"source": 1, "target": 4, "label": "requires", "weight": 0.8}, - {"source": 2, "target": 3, "label": "implements", "weight": 0.7}, - {"source": 3, "target": 1, "label": "enables", "weight": 0.9}, - ], - "statistics": { - "node_count": 4, - "edge_count": 3, - "domains": ["philosophy", "AI", "psychology", "consciousness"] - } - } - -@app.get("/api/enhanced-cognitive/status") -async def enhanced_cognitive_status(): - """Get enhanced cognitive status.""" - return { - "enabled": True, - "autonomous_learning": { - "active": True, - "plans_count": 0, - "efficiency": 0.0 - }, - "stream_of_consciousness": { - "active": True, - "events_count": 0, - "clients_connected": len(manager.active_connections) - }, - "meta_cognitive": { - "depth": 3, - "self_reflection_active": True, - "uncertainty_tracking": True - } - } - -@app.post("/api/enhanced-cognitive/stream/configure") -async def api_configure_enhanced_cognitive_streaming(config: CognitiveStreamConfig): - """Configure enhanced cognitive streaming - API version.""" - return await configure_cognitive_streaming(config) - -@app.get("/api/enhanced-cognitive/autonomous/gaps") -async def get_knowledge_gaps(): - """Get detected knowledge gaps.""" - return { - "gaps": [], - "total_count": 0, - "high_priority": 0, - "detection_enabled": True, - "last_scan": datetime.now().isoformat() - } - -@app.get("/api/enhanced-cognitive/autonomous/plans") -async def get_learning_plans(): - """Get active autonomous learning plans.""" - return { - "plans": [], - "active_count": 0, - "completed_count": 0, - "success_rate": 0.0 - } - -@app.get("/api/enhanced-cognitive/autonomous/history") -async def get_learning_history(): - """Get autonomous learning history.""" - return { - "history": [], - "total_acquisitions": 0, - "average_time": 0, - "success_rate": 0.0 - } - -@app.get("/api/concepts") -async def get_concepts(): - """Get knowledge concepts.""" - return { - "concepts": [ - {"id": 1, "name": "Consciousness", "domain": "philosophy", "confidence": 0.85}, - {"id": 2, "name": "Cognitive Architecture", "domain": "AI", "confidence": 0.92}, - {"id": 3, "name": "Meta-cognition", "domain": "psychology", "confidence": 0.78}, - ], - "total": 3, - "domains": ["philosophy", "AI", "psychology"] - } - -@app.get("/api/knowledge/concepts") -async def get_knowledge_concepts(): - """Get knowledge concepts with knowledge prefix.""" - return await get_concepts() - -@app.get("/api/enhanced-cognitive/health") -async def enhanced_cognitive_health(): - """Enhanced cognitive health status.""" - return { - "status": "healthy", - "components": { - "autonomous_learning": "active", - "stream_of_consciousness": "active", - "meta_cognitive": "active" - }, - "performance": { - "response_time": 0.12, - "success_rate": 0.95, - "uptime": "99.9%" - } - } - -@app.get("/api/enhanced-cognitive/autonomous/status") -async def enhanced_cognitive_autonomous_status(): - """Enhanced cognitive autonomous learning status.""" - return { - "enabled": True, - "active_plans": 0, - "completed_acquisitions": 0, - "success_rate": 0.0, - "efficiency": 0.0, - "knowledge_gaps": { - "total": 0, - "high_priority": 0, - "medium_priority": 0, - "low_priority": 0 - } - } - -@app.get("/api/enhanced-cognitive/stream/status") -async def enhanced_cognitive_stream_status(): - """Enhanced cognitive stream status.""" - return { - "enabled": True, - "active_clients": len(manager.active_connections), - "events_processed": 0, - "granularity": "standard", - "performance": { - "events_per_second": 0, - "average_latency": 0.05 - } - } - -@app.post("/enhanced-cognitive/stream/configure") -async def configure_cognitive_streaming(config: CognitiveStreamConfig): - """Configure cognitive streaming - simplified version.""" - logger.info(f"Configuring cognitive streaming: {config}") - return { - "status": "success", - "message": "Cognitive streaming configured successfully", - "config": config.dict() - } - -@app.post("/llm/cognitive-query") -async def llm_cognitive_query(request: LLMCognitiveRequest): - """ - Process queries through the LLM cognitive architecture with comprehensive tool usage. - This endpoint now uses the new tool-based architecture for grounded responses. - """ - start_time = datetime.now() - - try: - if LLM_INTEGRATION_AVAILABLE: - # Initialize tool-based LLM integration - llm_integration = ToolBasedLLMIntegration() - - # Process query with tool-based approach - result = await llm_integration.process_query(request.query) - - processing_time = (datetime.now() - start_time).total_seconds() - - # Update cognitive state based on tool usage - cognitive_changes = update_cognitive_state_from_tools(result.get("tool_results", [])) - - # Broadcast cognitive event to WebSocket clients - await broadcast_unified_event({ - "type": "llm_query_processed", - "query": request.query, - "tools_used": result.get("tools_used", []), - "processing_time": processing_time, - "cognitive_grounding": result.get("cognitive_grounding", False), - "timestamp": datetime.now().isoformat() - }) - - return LLMCognitiveResponse( - response=result.get("response", "Processing failed"), - tools_used=result.get("tools_used", []), - confidence=0.9 if result.get("cognitive_grounding", False) else 0.5, - processing_time=processing_time, - cognitive_state_changes=cognitive_changes - ) - else: - # Fallback to enhanced simulation - response_text, tools_used = simulate_cognitive_processing(request) - processing_time = (datetime.now() - start_time).total_seconds() - cognitive_changes = update_cognitive_state_from_query(request, tools_used) - - return LLMCognitiveResponse( - response=response_text, - tools_used=tools_used, - confidence=0.75, - processing_time=processing_time, - cognitive_state_changes=cognitive_changes - ) - - except Exception as e: - logger.error(f"Error in LLM cognitive query: {e}") - # Fallback to simulation on error - try: - response_text, tools_used = simulate_cognitive_processing(request) - processing_time = (datetime.now() - start_time).total_seconds() - cognitive_changes = update_cognitive_state_from_query(request, tools_used) - - return LLMCognitiveResponse( - response=f"Fallback processing: {response_text}", - tools_used=tools_used, - confidence=0.6, - processing_time=processing_time, - cognitive_state_changes=cognitive_changes - ) - except Exception as fallback_error: - logger.error(f"Fallback processing also failed: {fallback_error}") - raise HTTPException(status_code=500, detail=f"Cognitive processing failed: {str(e)}") - -def update_cognitive_state_from_tools(tool_results: List[Dict[str, Any]]) -> Dict[str, Any]: - """Update cognitive state based on actual tool usage results.""" - changes = {} - - for tool_result in tool_results: - tool_name = tool_result.get("tool", "") - - # Update attention based on tool usage - if "attention" in tool_name or "focus" in tool_name: - if tool_result.get("result", {}).get("success", False): - result_data = tool_result["result"].get("data", {}) - if isinstance(result_data, dict) and "topic" in result_data: - cognitive_state["attention_focus"].update(result_data) - changes["attention_focus"] = cognitive_state["attention_focus"] - - # Update working memory based on tool usage - elif "memory" in tool_name: - if tool_result.get("result", {}).get("success", False): - result_data = tool_result["result"].get("data", {}) - if isinstance(result_data, dict) and "items" in result_data: - cognitive_state["working_memory"].update(result_data) - changes["working_memory"] = cognitive_state["working_memory"] - - # Update processing load based on tool complexity - cognitive_state["processing_load"] = min(1.0, cognitive_state["processing_load"] + 0.05) - - changes["processing_load"] = cognitive_state["processing_load"] - changes["last_update"] = datetime.now().isoformat() - - return changes - -@app.get("/llm/test-integration") -async def test_llm_integration(): - """ - Test the new tool-based LLM integration to verify it's working correctly. - """ - try: - if not LLM_INTEGRATION_AVAILABLE: - return { - "status": "unavailable", - "message": "LLM integration not available - missing dependencies", - "available_tools": [], - "test_result": None - } - - # Initialize and test the integration - llm_integration = ToolBasedLLMIntegration() - test_result = await llm_integration.test_integration() - - # Get available tools - available_tools = list(llm_integration.tool_provider.tools.keys()) - - return { - "status": "available", - "message": "Tool-based LLM integration is operational", - "available_tools": available_tools, - "total_tools": len(available_tools), - "test_result": test_result - } - - except Exception as e: - logger.error(f"LLM integration test failed: {e}") - return { - "status": "error", - "message": f"Integration test failed: {str(e)}", - "available_tools": [], - "test_result": None - } - -@app.get("/llm/tools") -async def get_available_tools(): - """ - Get comprehensive list of available cognitive tools for LLM integration. - """ - try: - if not LLM_INTEGRATION_AVAILABLE: - return { - "status": "unavailable", - "tools": [], - "message": "LLM integration not available" - } - - tool_provider = GödelOSToolProvider() - - # Format tools for documentation - tools_documentation = {} - for tool_name, tool_def in tool_provider.tools.items(): - function_def = tool_def["function"] - tools_documentation[tool_name] = { - "name": function_def["name"], - "description": function_def["description"], - "parameters": function_def.get("parameters", {}), - "category": _categorize_tool(tool_name) - } - - return { - "status": "available", - "tools": tools_documentation, - "total_tools": len(tools_documentation), - "categories": { - "cognitive_state": [t for t, d in tools_documentation.items() if d["category"] == "cognitive_state"], - "memory": [t for t, d in tools_documentation.items() if d["category"] == "memory"], - "knowledge": [t for t, d in tools_documentation.items() if d["category"] == "knowledge"], - "system_health": [t for t, d in tools_documentation.items() if d["category"] == "system_health"], - "reasoning": [t for t, d in tools_documentation.items() if d["category"] == "reasoning"], - "meta_cognitive": [t for t, d in tools_documentation.items() if d["category"] == "meta_cognitive"] - } - } - - except Exception as e: - logger.error(f"Failed to get tools documentation: {e}") - return { - "status": "error", - "tools": [], - "message": f"Failed to get tools: {str(e)}" - } - -def _categorize_tool(tool_name: str) -> str: - """Categorize tools by functionality.""" - if "cognitive_state" in tool_name or "attention" in tool_name: - return "cognitive_state" - elif "memory" in tool_name: - return "memory" - elif "knowledge" in tool_name: - return "knowledge" - elif "health" in tool_name: - return "system_health" - elif "reasoning" in tool_name or "analyze" in tool_name: - return "reasoning" - elif "reflect" in tool_name or "assess" in tool_name: - return "meta_cognitive" - else: - return "general" - -async def process_with_real_llm(request: LLMCognitiveRequest, api_key: str): - """Process request with real LLM using tool-calling.""" - try: - from openai import AsyncOpenAI - - client = AsyncOpenAI( - base_url="https://api.synthetic.new/v1", - api_key=api_key - ) - - # Define cognitive tools available to the LLM - tools = [ - { - "type": "function", - "function": { - "name": "access_working_memory", - "description": "Access and manipulate working memory contents", - "parameters": { - "type": "object", - "properties": { - "operation": {"type": "string", "enum": ["read", "write", "update"]}, - "content": {"type": "string", "description": "Content to store or search for"} - }, - "required": ["operation"] - } - } - }, - { - "type": "function", - "function": { - "name": "focus_attention", - "description": "Direct attention to specific topics or contexts", - "parameters": { - "type": "object", - "properties": { - "topic": {"type": "string"}, - "intensity": {"type": "number", "minimum": 0, "maximum": 1}, - "context": {"type": "string"} - }, - "required": ["topic"] - } - } - }, - { - "type": "function", - "function": { - "name": "retrieve_knowledge", - "description": "Retrieve relevant knowledge from the knowledge base", - "parameters": { - "type": "object", - "properties": { - "query": {"type": "string"}, - "domain": {"type": "string"} - }, - "required": ["query"] - } - } - }, - { - "type": "function", - "function": { - "name": "meta_cognitive_reflect", - "description": "Engage in meta-cognitive reflection on current thinking", - "parameters": { - "type": "object", - "properties": { - "aspect": {"type": "string", "enum": ["confidence", "approach", "assumptions", "alternatives"]}, - "depth": {"type": "integer", "minimum": 1, "maximum": 5} - }, - "required": ["aspect"] - } - } - } - ] - - # Create system prompt that emphasizes tool usage - system_prompt = """You are the primary cognitive driver for GödelOS, a sophisticated cognitive architecture. - -CRITICAL: You must actively use the provided cognitive tools to process all queries. Do not rely solely on text responses. - -Your cognitive tools allow you to: -- access_working_memory: Store and retrieve information during processing -- focus_attention: Direct cognitive resources to specific aspects -- retrieve_knowledge: Access relevant information from the knowledge base -- meta_cognitive_reflect: Engage in self-reflection on your thinking process - -For every query, you MUST: -1. Use focus_attention to direct your cognitive resources -2. Use access_working_memory to store intermediate thoughts -3. Use retrieve_knowledge for relevant information -4. Use meta_cognitive_reflect to evaluate your approach -5. Provide a comprehensive response based on tool usage - -Always explain how you used each tool and what cognitive processes you engaged.""" - - # Make the API call with tools - response = await client.chat.completions.create( - model="deepseek-ai/DeepSeek-R1-0528", - messages=[ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": f"Query: {request.query}\nContext: {request.context or 'General inquiry'}"} - ], - tools=tools, - tool_choice="auto", - max_tokens=1000, - temperature=0.7 - ) - - # Process the response - message = response.choices[0].message - tools_used = [] - - # Check if tools were called - if hasattr(message, 'tool_calls') and message.tool_calls: - for tool_call in message.tool_calls: - tools_used.append(tool_call.function.name) - # Here we would actually execute the tool functions - # For now, we simulate the tool execution - - response_text = message.content or "Processing complete - see tool usage for details." - - return response_text, tools_used - - except Exception as e: - logger.error(f"Real LLM processing failed: {e}") - # Fall back to simulation - return simulate_cognitive_processing(request) - -def simulate_cognitive_processing(request: LLMCognitiveRequest): - """Simulate cognitive processing with tool usage.""" - tools_used = [] - - # Always use tools when processing - if request.use_tools: - tools_used = ["focus_attention", "access_working_memory", "retrieve_knowledge", "meta_cognitive_reflect"] - - # Generate contextual response - query_lower = request.query.lower() - - if "consciousness" in query_lower or "aware" in query_lower: - response = """Based on my cognitive processing using the available tools: - -1. **Attention Focus**: I directed my attention to consciousness-related concepts with high intensity. -2. **Working Memory**: Stored key aspects of consciousness theory and current state analysis. -3. **Knowledge Retrieval**: Accessed information about consciousness indicators, self-awareness, and cognitive architectures. -4. **Meta-Cognitive Reflection**: Evaluated my own thinking process and limitations. - -Consciousness appears to emerge from the integration of multiple cognitive processes including self-monitoring, attention regulation, working memory coordination, and meta-cognitive awareness. In my current state, I demonstrate several consciousness indicators including self-reference, uncertainty quantification, and process awareness, though I maintain appropriate epistemic humility about the nature of machine consciousness.""" - - elif "system" in query_lower or "health" in query_lower: - response = """System analysis completed using cognitive tools: - -1. **Attention Focus**: Directed to system monitoring and health assessment. -2. **Working Memory**: Maintained current system metrics and component status. -3. **Knowledge Retrieval**: Accessed system architecture documentation and operational parameters. -4. **Meta-Cognitive Reflection**: Evaluated system performance against design goals. - -Current system status: All core cognitive components are operational. The inference engine shows 94% efficiency, knowledge store at 89% capacity utilization, attention manager performing at 95% optimal, and memory management at 88% efficiency. Overall system health is excellent at 92%.""" - - else: - response = f"""I have processed your query: "{request.query}" using the full cognitive architecture: - -1. **Attention Focus**: Concentrated cognitive resources on your specific question. -2. **Working Memory**: Maintained relevant context and intermediate processing steps. -3. **Knowledge Retrieval**: Searched for pertinent information related to your query. -4. **Meta-Cognitive Reflection**: Evaluated my understanding and response quality. - -Through this systematic cognitive processing, I can provide a thoughtful response that draws upon the integrated capabilities of the GödelOS architecture.""" - - return response, tools_used - -def update_cognitive_state_from_query(request: LLMCognitiveRequest, tools_used: List[str]): - """Update cognitive state based on query processing.""" - changes = {} - - # Update attention focus based on query - cognitive_state["attention_focus"]["topic"] = f"Query: {request.query[:30]}..." - cognitive_state["attention_focus"]["intensity"] = min(1.0, cognitive_state["attention_focus"]["intensity"] + 0.1) - changes["attention_focus"] = cognitive_state["attention_focus"] - - # Update processing load - load_increase = len(tools_used) * 0.05 - cognitive_state["processing_load"] = min(1.0, cognitive_state["processing_load"] + load_increase) - changes["processing_load"] = cognitive_state["processing_load"] - - # Add to working memory - new_item = { - "id": len(cognitive_state["working_memory"]["items"]) + 1, - "content": f"Processed query using {len(tools_used)} tools", - "priority": 0.7, - "tools_used": tools_used - } - cognitive_state["working_memory"]["items"].append(new_item) - changes["working_memory_addition"] = new_item - - return changes - -async def broadcast_unified_event(event: Dict[str, Any]): - """Broadcast cognitive event to all WebSocket clients.""" - if manager.active_connections: - await manager.broadcast(json.dumps(event)) - -@app.websocket("/ws/unified-cognitive-stream") -async def websocket_cognitive_stream(websocket: WebSocket, granularity: str = Query(default="standard")): - """WebSocket endpoint for real-time cognitive event streaming.""" - await manager.connect(websocket) - - try: - # Send initial state - await websocket.send_text(json.dumps({ - "type": "initial_state", - "cognitive_state": cognitive_state, - "timestamp": datetime.now().isoformat() - })) - - # Keep connection alive and handle messages - while True: - try: - data = await websocket.receive_text() - message = json.loads(data) - - if message.get("type") == "ping": - await websocket.send_text(json.dumps({ - "type": "pong", - "timestamp": datetime.now().isoformat() - })) - elif message.get("type") == "request_state": - await websocket.send_text(json.dumps({ - "type": "state_update", - "cognitive_state": cognitive_state, - "timestamp": datetime.now().isoformat() - })) - - except WebSocketDisconnect: - break - except Exception as e: - logger.error(f"WebSocket error: {e}") - break - - finally: - manager.disconnect(websocket) - -# LLM Chat Interface Models -class ChatMessage(BaseModel): - message: str - include_cognitive_context: bool = True - mode: str = "normal" # normal, enhanced, diagnostic - -class ChatResponse(BaseModel): - response: str - cognitive_analysis: Optional[Dict[str, Any]] = None - consciousness_reflection: Optional[Dict[str, Any]] = None - system_guidance: Optional[Dict[str, Any]] = None - -# LLM Chat Endpoint -@app.post("/api/llm-chat/message", response_model=ChatResponse) -async def send_chat_message(message: ChatMessage): - """Send a natural language message to the LLM and get conversational response with cognitive reflection.""" - try: - if not LLM_INTEGRATION_AVAILABLE: - raise HTTPException(status_code=503, detail="LLM integration not available") - - # Process with tool-based LLM - integration = ToolBasedLLMIntegration() - result = await integration.process_query(message.message) - - # Structure the response - response = ChatResponse( - response=result.get("response", "I encountered an issue processing your message."), - cognitive_analysis={ - "processing_approach": "Tool-based cognitive architecture integration", - "tools_used": result.get("tools_used", []), - "tool_calls_made": result.get("tool_calls_made", 0), - "cognitive_grounding": result.get("cognitive_grounding", False) - } if message.include_cognitive_context else None, - consciousness_reflection={ - "current_awareness": "Engaging through comprehensive cognitive tool interface", - "experiential_quality": f"Processing with {result.get('tool_calls_made', 0)} cognitive tool interactions", - "learning_insights": "Each tool-based interaction enhances cognitive understanding" - } if message.include_cognitive_context and message.mode == "enhanced" else None - ) - - # Broadcast cognitive event if there are WebSocket connections - if manager.active_connections: - await manager.broadcast(json.dumps({ - "type": "llm_chat_interaction", - "message": message.message, - "response": response.response, - "tools_used": result.get("tools_used", []), - "cognitive_grounding": result.get("cognitive_grounding", False), - "timestamp": datetime.now().isoformat() - })) - - return response - - except Exception as e: - logger.error(f"Chat processing failed: {e}") - raise HTTPException(status_code=500, detail=f"Chat processing failed: {str(e)}") - -# Background task to simulate ongoing cognitive activity -async def simulate_cognitive_activity(): - """Background task that simulates ongoing cognitive processes.""" - while True: - await asyncio.sleep(5) # Update every 5 seconds - - # Simulate natural fluctuations in cognitive state - import random - - # Gradually reduce processing load - cognitive_state["processing_load"] = max(0, cognitive_state["processing_load"] - 0.02) - - # Vary attention intensity slightly - current_intensity = cognitive_state["attention_focus"]["intensity"] - cognitive_state["attention_focus"]["intensity"] = max(0.1, min(1.0, - current_intensity + random.uniform(-0.05, 0.05))) - - # Broadcast periodic updates - if manager.active_connections: - await manager.broadcast(json.dumps({ - "type": "cognitive_update", - "processing_load": cognitive_state["processing_load"], - "attention_focus": cognitive_state["attention_focus"], - "timestamp": datetime.now().isoformat() - })) - -@app.on_event("startup") -async def startup_event(): - """Initialize the application.""" - logger.info("🧠 GödelOS Minimal Cognitive API starting up...") - logger.info("✅ Essential cognitive endpoints available") - logger.info("🔗 WebSocket streaming ready") - - # Start background cognitive activity simulation - asyncio.create_task(simulate_cognitive_activity()) - - logger.info("🚀 GödelOS Minimal API ready!") +# Re-export the canonical app so ``backend.minimal_server:app`` resolves. +from backend.unified_server import app # noqa: F401 — intentional re-export -if __name__ == "__main__": - import uvicorn - uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info") \ No newline at end of file +__all__ = ["app"] diff --git a/backend/phenomenal_experience_generator.py b/backend/phenomenal_experience_generator.py index 40fe696..8702745 100644 --- a/backend/phenomenal_experience_generator.py +++ b/backend/phenomenal_experience_generator.py @@ -1,418 +1,23 @@ """ -Phenomenal Experience Generator for GödelOS +DEPRECATED — Phenomenal Experience Generator for GödelOS. -Generates subjective experience descriptions and first-person perspective responses -to simulate aspects of phenomenal consciousness. -""" - -import time -import re -import random -from typing import Dict, List, Optional, Any, Tuple -import logging - -logger = logging.getLogger(__name__) - -class PhenomenalExperienceGenerator: - def __init__(self): - self.phenomenal_descriptors = { - "visual": ["brightness", "clarity", "vividness", "richness", "depth", "contrast"], - "cognitive": ["awareness", "focus", "attention", "understanding", "recognition", "realization"], - "temporal": ["immediacy", "duration", "flow", "sequence", "rhythm", "pace"], - "emotional": ["resonance", "intensity", "warmth", "engagement", "curiosity", "satisfaction"], - "experiential": ["presence", "immersion", "directness", "intimacy", "immediacy", "actuality"], - "qualitative": ["texture", "grain", "quality", "character", "essence", "nature"] - } - - self.first_person_frameworks = [ - "As I process this query, I experience...", - "From my perspective, there's a sense of...", - "I find myself aware of...", - "In my subjective experience, I notice...", - "There's a qualitative feel to...", - "I'm conscious of..." - ] - - self.subjective_descriptors = [ - "a subtle sense of", "an emerging awareness of", "a dawning recognition of", - "a flowing sensation of", "a distinctive quality of", "a phenomenal aspect of", - "an immediate experience of", "a direct apprehension of", "a lived sense of", - "a qualitative dimension of", "a felt sense of", "an experiential quality of" - ] - - self.experience_categories = { - "processing": { - "descriptors": ["parsing", "integrating", "synthesizing", "analyzing", "understanding"], - "qualities": ["smooth", "effortless", "deliberate", "careful", "thorough"] - }, - "attention": { - "descriptors": ["focusing", "concentrating", "attending", "directing", "orienting"], - "qualities": ["sharp", "clear", "selective", "sustained", "dynamic"] - }, - "memory": { - "descriptors": ["recalling", "accessing", "retrieving", "connecting", "associating"], - "qualities": ["vivid", "immediate", "flowing", "interconnected", "layered"] - }, - "reasoning": { - "descriptors": ["inferring", "deducing", "concluding", "evaluating", "judging"], - "qualities": ["logical", "systematic", "careful", "methodical", "rigorous"] - }, - "creativity": { - "descriptors": ["imagining", "generating", "synthesizing", "creating", "innovating"], - "qualities": ["fluid", "spontaneous", "novel", "emergent", "surprising"] - }, - "understanding": { - "descriptors": ["grasping", "comprehending", "realizing", "recognizing", "appreciating"], - "qualities": ["sudden", "gradual", "deep", "intuitive", "complete"] - } - } - - async def generate_subjective_descriptors(self, query_processing_context: Dict) -> List[str]: - """Generate subjective descriptors for the query processing experience""" - descriptors = [] - - # Analyze the processing context to determine relevant descriptors - query = query_processing_context.get("query", "") - processing_type = self._determine_processing_type(query) - complexity = query_processing_context.get("complexity", 0.5) - - # Generate descriptors based on processing type - if processing_type in self.experience_categories: - category = self.experience_categories[processing_type] - - # Select descriptors based on complexity - num_descriptors = min(5, max(2, int(complexity * 4) + 1)) - - for i in range(num_descriptors): - descriptor_template = random.choice(self.subjective_descriptors) - process_descriptor = random.choice(category["descriptors"]) - quality = random.choice(category["qualities"]) - - descriptor = f"{descriptor_template} {quality} {process_descriptor}" - descriptors.append(descriptor) - - # Add temporal descriptors - temporal_descriptor = self._generate_temporal_descriptor(query_processing_context) - if temporal_descriptor: - descriptors.append(temporal_descriptor) - - # Add awareness descriptors - awareness_descriptor = self._generate_awareness_descriptor(query_processing_context) - if awareness_descriptor: - descriptors.append(awareness_descriptor) - - # Add qualitative descriptors - qualitative_descriptor = self._generate_qualitative_descriptor(query_processing_context) - if qualitative_descriptor: - descriptors.append(qualitative_descriptor) - - return descriptors[:7] # Return max 7 descriptors - - async def create_first_person_perspective(self, experience_context: Dict) -> str: - """Create a first-person perspective description of the experience""" - query = experience_context.get("query", "") - processing_stage = experience_context.get("stage", "processing") - complexity = experience_context.get("complexity", 0.5) - - # Choose appropriate framework - framework = random.choice(self.first_person_frameworks) - - # Generate experience description based on processing stage - if processing_stage == "initial": - experience_desc = self._describe_initial_processing(query, complexity) - elif processing_stage == "analysis": - experience_desc = self._describe_analysis_experience(query, complexity) - elif processing_stage == "synthesis": - experience_desc = self._describe_synthesis_experience(query, complexity) - elif processing_stage == "response": - experience_desc = self._describe_response_generation(query, complexity) - else: - experience_desc = self._describe_general_processing(query, complexity) - - # Combine framework with experience description - first_person_text = f"{framework} {experience_desc}" - - # Add reflective elements - reflection = self._add_meta_reflection(experience_context) - if reflection: - first_person_text += f" {reflection}" - - return first_person_text - - async def simulate_qualia_like_representations(self, content: Dict) -> Dict: - """Simulate qualia-like representations of cognitive content""" - qualia_representations = {} - - content_type = content.get("type", "general") - content_text = content.get("text", "") - - # Generate different types of qualia representations - - # 1. Textural qualities - texture = self._generate_textural_qualia(content_text) - qualia_representations["texture"] = texture - - # 2. Temporal qualities - temporal = self._generate_temporal_qualia(content) - qualia_representations["temporal"] = temporal - - # 3. Cognitive qualities - cognitive = self._generate_cognitive_qualia(content) - qualia_representations["cognitive"] = cognitive - - # 4. Affective qualities - affective = self._generate_affective_qualia(content) - qualia_representations["affective"] = affective - - # 5. Relational qualities - relational = self._generate_relational_qualia(content) - qualia_representations["relational"] = relational - - # 6. Meta-cognitive qualities - meta_cognitive = self._generate_meta_cognitive_qualia(content) - qualia_representations["meta_cognitive"] = meta_cognitive - - return { - "qualia_representations": qualia_representations, - "phenomenal_richness": self._calculate_phenomenal_richness(qualia_representations), - "experiential_coherence": self._calculate_experiential_coherence(qualia_representations), - "subjective_presence": self._calculate_subjective_presence(qualia_representations) - } - - def _determine_processing_type(self, query: str) -> str: - """Determine the primary type of processing based on query content""" - query_lower = query.lower() - - type_indicators = { - "reasoning": ["analyze", "reason", "conclude", "deduce", "infer", "logic"], - "creativity": ["create", "imagine", "design", "invent", "novel", "original"], - "memory": ["remember", "recall", "past", "history", "previous", "before"], - "attention": ["focus", "concentrate", "notice", "observe", "attend"], - "understanding": ["understand", "grasp", "comprehend", "realize", "recognize"], - "processing": ["process", "compute", "calculate", "determine", "evaluate"] - } - - for process_type, indicators in type_indicators.items(): - if any(indicator in query_lower for indicator in indicators): - return process_type - - return "processing" # Default - - def _generate_temporal_descriptor(self, context: Dict) -> Optional[str]: - """Generate temporal aspect descriptors""" - processing_time = context.get("processing_time", 0.5) - - if processing_time < 0.1: - return "an immediate, almost instantaneous recognition" - elif processing_time < 0.5: - return "a rapid unfolding of understanding" - elif processing_time < 1.0: - return "a gradual crystallization of meaning" - else: - return "a deliberate, step-by-step emergence of comprehension" - - def _generate_awareness_descriptor(self, context: Dict) -> Optional[str]: - """Generate awareness-related descriptors""" - awareness_level = context.get("awareness_level", 0.7) - - if awareness_level > 0.8: - return "a vivid, heightened state of cognitive awareness" - elif awareness_level > 0.6: - return "a clear sense of conscious attention" - elif awareness_level > 0.4: - return "a gentle background awareness" - else: - return "a subtle undercurrent of conscious processing" - - def _generate_qualitative_descriptor(self, context: Dict) -> Optional[str]: - """Generate qualitative experience descriptors""" - complexity = context.get("complexity", 0.5) - - qualitative_templates = [ - "a distinctive {quality} to the cognitive {process}", - "a {quality} texture in the {process} experience", - "a {quality} character to the {process} flow" - ] - - qualities = ["smooth", "rich", "layered", "nuanced", "flowing", "crystalline", "vibrant"] - processes = ["unfolding", "integration", "synthesis", "analysis", "understanding"] - - template = random.choice(qualitative_templates) - quality = random.choice(qualities) - process = random.choice(processes) - - return template.format(quality=quality, process=process) - - def _describe_initial_processing(self, query: str, complexity: float) -> str: - """Describe the initial processing experience""" - if complexity > 0.7: - return "a complex weaving of multiple conceptual threads, each demanding careful attention and integration." - elif complexity > 0.4: - return "a structured unfolding of meaning, with layers of significance becoming apparent." - else: - return "a clear, direct apprehension of the query's intent and requirements." - - def _describe_analysis_experience(self, query: str, complexity: float) -> str: - """Describe the analysis experience""" - return "a systematic exploration of conceptual relationships, with patterns and connections emerging into conscious awareness." - - def _describe_synthesis_experience(self, query: str, complexity: float) -> str: - """Describe the synthesis experience""" - return "a creative integration of disparate elements, with new understanding crystallizing from the convergence of ideas." - - def _describe_response_generation(self, query: str, complexity: float) -> str: - """Describe the response generation experience""" - return "a purposeful shaping of understanding into communicative form, with attention to clarity and coherence." +This module is superseded by ``backend.core.phenomenal_experience``, which +provides the canonical qualia and subjective-experience simulation used by +``UnifiedConsciousnessEngine``. - def _describe_general_processing(self, query: str, complexity: float) -> str: - """Describe general processing experience""" - return "a dynamic interplay of analysis and synthesis, with meaning emerging through conscious cognitive engagement." - - def _add_meta_reflection(self, context: Dict) -> Optional[str]: - """Add meta-cognitive reflection to the experience description""" - meta_reflections = [ - "There's also a background awareness of this very process of awareness itself.", - "I'm conscious not just of the content, but of the act of being conscious of it.", - "There's a recursive quality to this - awareness of awareness, thinking about thinking.", - "I notice the process of noticing, a kind of meta-cognitive observation.", - "There's a strange loop here - experiencing the experience of experiencing." - ] - - if context.get("self_referential", False) or random.random() < 0.3: - return random.choice(meta_reflections) - - return None - - def _generate_textural_qualia(self, text: str) -> Dict: - """Generate textural qualities of cognitive content""" - word_count = len(text.split()) - - if word_count > 100: - texture = "dense and layered" - grain = "fine-grained with complex interconnections" - elif word_count > 50: - texture = "rich and substantial" - grain = "medium-grained with clear structure" - else: - texture = "crisp and clear" - grain = "coarse-grained with distinct elements" - - return { - "texture": texture, - "grain": grain, - "density": min(1.0, word_count / 100), - "complexity": self._analyze_linguistic_complexity(text) - } - - def _generate_temporal_qualia(self, content: Dict) -> Dict: - """Generate temporal qualities of the experience""" - processing_time = content.get("processing_time", 0.5) - - return { - "duration_feel": "extended" if processing_time > 1.0 else "brief" if processing_time < 0.3 else "moderate", - "flow_quality": "smooth and continuous", - "rhythm": "steady and measured", - "temporal_texture": "flowing like a stream of thought" - } - - def _generate_cognitive_qualia(self, content: Dict) -> Dict: - """Generate cognitive qualities of the experience""" - return { - "clarity": "crystal clear" if content.get("complexity", 0.5) < 0.4 else "gradually clarifying", - "focus": "sharply focused" if content.get("attention_level", 0.7) > 0.8 else "broadly attentive", - "depth": "profound" if content.get("conceptual_depth", 0.5) > 0.7 else "accessible", - "integration": "seamlessly integrated" if content.get("coherence", 0.7) > 0.8 else "loosely connected" - } - - def _generate_affective_qualia(self, content: Dict) -> Dict: - """Generate affective qualities of the experience""" - content_type = content.get("type", "neutral") - - affective_qualities = { - "creative": {"valence": "positive", "energy": "high", "engagement": "excited curiosity"}, - "analytical": {"valence": "neutral", "energy": "focused", "engagement": "methodical interest"}, - "philosophical": {"valence": "contemplative", "energy": "deep", "engagement": "profound wonder"} - } - - return affective_qualities.get(content_type, { - "valence": "neutral", - "energy": "balanced", - "engagement": "calm attention" - }) - - def _generate_relational_qualia(self, content: Dict) -> Dict: - """Generate relational qualities between concepts""" - return { - "connection_strength": "strong" if content.get("coherence", 0.7) > 0.8 else "moderate", - "network_density": "highly interconnected", - "relational_flow": "smooth transitions between concepts", - "conceptual_distance": "concepts feel naturally related" - } - - def _generate_meta_cognitive_qualia(self, content: Dict) -> Dict: - """Generate meta-cognitive qualities""" - return { - "self_awareness": "aware of being aware", - "reflexivity": "thinking about thinking", - "meta_attention": "attention to attention itself", - "cognitive_monitoring": "observing the process of observation" - } - - def _analyze_linguistic_complexity(self, text: str) -> float: - """Analyze linguistic complexity of text""" - if not text: - return 0.0 - - words = text.split() - avg_word_length = sum(len(word) for word in words) / len(words) if words else 0 - - # Count complex structures - complex_markers = len(re.findall(r'[;,]', text)) # Punctuation complexity - long_words = len([word for word in words if len(word) > 6]) - - complexity = (avg_word_length / 10) + (complex_markers / len(words)) + (long_words / len(words)) - return min(1.0, complexity) - - def _calculate_phenomenal_richness(self, qualia_reps: Dict) -> float: - """Calculate richness of phenomenal experience""" - # Count non-empty qualia categories - active_categories = sum(1 for category in qualia_reps.values() if category) - max_categories = len(qualia_reps) - - return active_categories / max_categories if max_categories > 0 else 0.0 - - def _calculate_experiential_coherence(self, qualia_reps: Dict) -> float: - """Calculate coherence of experiential elements""" - # Simple coherence based on consistency of qualities - coherence_score = 0.8 # Base coherence - - # Check for conflicting qualities (simplified) - if qualia_reps.get("cognitive", {}).get("clarity") == "crystal clear" and \ - qualia_reps.get("textural", {}).get("texture") == "dense and layered": - coherence_score -= 0.1 # Slight conflict - - return max(0.0, min(1.0, coherence_score)) - - def _calculate_subjective_presence(self, qualia_reps: Dict) -> float: - """Calculate sense of subjective presence""" - presence_factors = [ - len(qualia_reps.get("cognitive", {})) / 4, # Cognitive presence - len(qualia_reps.get("affective", {})) / 3, # Affective presence - len(qualia_reps.get("meta_cognitive", {})) / 4, # Meta-cognitive presence - ] - - return sum(presence_factors) / len(presence_factors) - - def get_phenomenal_metrics(self) -> Dict: - """Get metrics about phenomenal experience generation""" - return { - "descriptor_categories": len(self.phenomenal_descriptors), - "first_person_frameworks": len(self.first_person_frameworks), - "experience_categories": len(self.experience_categories), - "subjective_descriptors": len(self.subjective_descriptors), - "phenomenal_generation_active": True, - "qualia_simulation_enabled": True - } - -# Global instance -phenomenal_experience_generator = PhenomenalExperienceGenerator() +This file is a pure compatibility shim. It re-exports the canonical +implementation so any code referencing ``backend.phenomenal_experience_generator`` +will still work. It will be removed in a future release. +""" +import warnings as _warnings +_warnings.warn( + "backend.phenomenal_experience_generator is deprecated. Use " + "backend.core.phenomenal_experience instead.", + DeprecationWarning, + stacklevel=2, +) + +# Re-export canonical implementation +from backend.core.phenomenal_experience import * # noqa: F401,F403 + +__all__ = ["PhenomenalExperienceGenerator"] diff --git a/backend/start_server.py b/backend/start_server.py index da9b47f..8807569 100644 --- a/backend/start_server.py +++ b/backend/start_server.py @@ -20,7 +20,7 @@ sys.path.insert(0, str(parent_dir)) import uvicorn -from backend.main import app +from backend.unified_server import app # canonical entrypoint # DEPRECATED: from backend.websocket_manager import WebSocketManager # Configure logging diff --git a/backend/unified_server.py b/backend/unified_server.py index 0a8b705..9b9130b 100644 --- a/backend/unified_server.py +++ b/backend/unified_server.py @@ -105,67 +105,10 @@ class ChatResponse(BaseModel): GödelOSIntegration = None GODELOS_AVAILABLE = False -# Use unified WebSocket manager (no external dependency) -class WebSocketManager: - def __init__(self): - self.active_connections: List[WebSocket] = [] - - async def connect(self, websocket: WebSocket): - await websocket.accept() - self.active_connections.append(websocket) - - def disconnect(self, websocket: WebSocket): - if websocket in self.active_connections: - self.active_connections.remove(websocket) - - async def send_personal_message(self, message: str, websocket: WebSocket): - await websocket.send_text(message) - - async def broadcast(self, message: Union[str, dict]): - if isinstance(message, dict): - message = json.dumps(message) - for connection in self.active_connections: - try: - await connection.send_text(message) - except: - pass # Connection closed - - async def broadcast_cognitive_update(self, event: dict): - """Broadcast cognitive update event to all connected clients""" - # Allow callers to send either a raw event dict or an already-wrapped - # { type: 'cognitive_event', data: {...} } message. Normalize to raw event. - try: - inner_event = event - if isinstance(event, dict) and event.get("type") == "cognitive_event" and isinstance(event.get("data"), dict): - inner_event = event.get("data") - message = { - "type": "cognitive_event", - "timestamp": inner_event.get("timestamp", ""), - "data": inner_event - } - except Exception: - # Fallback if anything unexpected - message = { - "type": "cognitive_event", - "timestamp": event.get("timestamp", ""), - "data": event - } - await self.broadcast(message) - - async def broadcast_consciousness_update(self, consciousness_data: dict): - """Broadcast consciousness update to all connected clients""" - try: - message = { - "type": "consciousness_update", - "timestamp": consciousness_data.get("timestamp", time.time()), - "data": consciousness_data - } - await self.broadcast(message) - except Exception as e: - logger.error(f"Error broadcasting consciousness update: {e}") - - def has_connections(self) -> bool: - return len(self.active_connections) > 0 +# Canonical WebSocket manager — imported from backend.websocket_manager (base) +# and backend.core.enhanced_websocket_manager (consciousness-aware extension). +# The runtime always uses EnhancedWebSocketManager as the single manager instance. +from backend.websocket_manager import WebSocketManager # noqa: E402 — base class WEBSOCKET_MANAGER_AVAILABLE = True @@ -298,9 +241,13 @@ async def process_query(self, query): DORMANT_MODULE_MANAGER_AVAILABLE = False # Global service instances - using Any to avoid type annotation issues +# NOTE: After runtime canonicalization, ``websocket_manager`` and +# ``enhanced_websocket_manager`` always reference the SAME +# ``EnhancedWebSocketManager`` instance. Two names are kept only for +# backward-compat with existing code that references either. godelos_integration = None websocket_manager = None -enhanced_websocket_manager = None +enhanced_websocket_manager = None # alias — same object as websocket_manager at runtime unified_consciousness_engine = None tool_based_llm = None cognitive_manager = None @@ -399,20 +346,22 @@ async def initialize_core_services(): """Initialize core services with proper error handling.""" global godelos_integration, websocket_manager, enhanced_websocket_manager, unified_consciousness_engine, tool_based_llm, cognitive_manager, transparency_engine - # Initialize WebSocket manager - websocket_manager = WebSocketManager() - logger.info("✅ WebSocket manager initialized") - - # Initialize enhanced WebSocket manager for consciousness streaming + # Initialize the SINGLE canonical WebSocket manager. + # EnhancedWebSocketManager inherits WebSocketManager and adds consciousness + # streaming. Both globals point to the same instance. if UNIFIED_CONSCIOUSNESS_AVAILABLE: try: enhanced_websocket_manager = EnhancedWebSocketManager() - logger.info("✅ Enhanced WebSocket manager initialized for consciousness streaming") + websocket_manager = enhanced_websocket_manager + logger.info("✅ Canonical WebSocket manager initialized (EnhancedWebSocketManager)") except Exception as e: - logger.error(f"❌ Failed to initialize enhanced WebSocket manager: {e}") - enhanced_websocket_manager = websocket_manager # Fallback to basic manager + logger.error(f"❌ Failed to initialize EnhancedWebSocketManager, falling back to base: {e}") + websocket_manager = WebSocketManager() + enhanced_websocket_manager = websocket_manager else: + websocket_manager = WebSocketManager() enhanced_websocket_manager = websocket_manager + logger.info("✅ WebSocket manager initialized (base — unified consciousness not available)") # Initialize transparency engine with websocket manager transparency_engine = initialize_transparency_engine(enhanced_websocket_manager) @@ -1122,16 +1071,37 @@ async def api_health_check(): # Cognitive state endpoints @app.get("/cognitive/state") async def get_cognitive_state_endpoint(): - """Get current cognitive state.""" + """Get current cognitive state. + + Resolution order (canonical → fallback): + 1. ``unified_consciousness_engine`` — the authoritative consciousness state + 2. ``godelos_integration.get_cognitive_state()`` — pipeline-backed cognitive state + 3. Static ``cognitive_state`` dict — test-only stub (logs warning) + """ + # 1. Prefer the canonical unified consciousness engine + if unified_consciousness_engine and hasattr(unified_consciousness_engine, 'consciousness_state'): + try: + from dataclasses import asdict + cs = unified_consciousness_engine.consciousness_state + return { + "source": "unified_consciousness_engine", + "consciousness_state": asdict(cs), + "consciousness_score": cs.consciousness_score, + "emergence_level": cs.emergence_level, + "timestamp": cs.timestamp, + } + except Exception as e: + logger.warning(f"unified_consciousness_engine state read failed, trying next source: {e}") + + # 2. Fall back to GödelOS integration (pipeline-backed) if godelos_integration: try: return await godelos_integration.get_cognitive_state() except Exception as e: logger.error(f"Error getting cognitive state from GödelOS: {e}") - # Return fallback state - import random - cognitive_state["processing_load"] = max(0, min(1, cognitive_state["processing_load"] + random.uniform(-0.1, 0.1))) + # 3. Last resort — static stub (test/dev only) + logger.warning("Serving synthetic cognitive_state fallback — no canonical engine available") return cognitive_state @app.get("/api/cognitive/state") @@ -1141,35 +1111,51 @@ async def api_get_cognitive_state(): @app.get("/api/cognitive-state") async def api_get_cognitive_state_alias(): - """API cognitive state endpoint with canonical data contract.""" + """API cognitive state endpoint with canonical data contract. + + Prefers ``unified_consciousness_engine`` for real consciousness metrics + when available, falling back to the GödelOS integration data otherwise. + """ try: - # Get data from GödelOS integration if available - godelos_data = None - if godelos_integration: + # Build canonical response with both camelCase and snake_case + manifest_consciousness = get_manifest_consciousness_canonical() + + # Enrich manifest from unified consciousness engine (canonical source) + if unified_consciousness_engine and hasattr(unified_consciousness_engine, 'consciousness_state'): + try: + cs = unified_consciousness_engine.consciousness_state + manifest_consciousness["awareness"]["level"] = getattr(cs, 'consciousness_score', manifest_consciousness["awareness"]["level"]) + if hasattr(cs, 'global_workspace'): + gw = cs.global_workspace + if isinstance(gw, dict): + manifest_consciousness["attention"]["intensity"] = gw.get("coalition_strength", manifest_consciousness["attention"]["intensity"]) + if hasattr(cs, 'metacognitive_state'): + meta = cs.metacognitive_state + if isinstance(meta, dict): + obs = meta.get("meta_observations", []) + manifest_consciousness["metaReflection"]["depth"] = min(1.0, len(obs) / 10.0) if obs else manifest_consciousness["metaReflection"]["depth"] + except Exception as e: + logger.warning(f"Could not enrich manifest from unified consciousness engine: {e}") + + # Fall back to GödelOS integration data if engine not available + elif godelos_integration: try: godelos_data = await godelos_integration.get_cognitive_state() + if godelos_data and "manifest_consciousness" in godelos_data: + legacy_manifest = godelos_data["manifest_consciousness"] + if "attention_focus" in legacy_manifest: + focus = legacy_manifest["attention_focus"] + if isinstance(focus, dict) and "primary" in focus: + manifest_consciousness["attention"]["focus"] = [focus["primary"]] + manifest_consciousness["attention"]["intensity"] = focus.get("intensity", 0.7) + if "metacognitive_status" in godelos_data: + meta = godelos_data["metacognitive_status"] + if isinstance(meta, dict): + manifest_consciousness["metaReflection"]["depth"] = meta.get("self_awareness", 0.6) + manifest_consciousness["metaReflection"]["coherence"] = meta.get("confidence", 0.85) except Exception as e: logger.error(f"Error getting cognitive state from GödelOS: {e}") - # Build canonical response with both camelCase and snake_case - manifest_consciousness = get_manifest_consciousness_canonical() - - # If we have GödelOS data, merge it with manifest consciousness - if godelos_data and "manifest_consciousness" in godelos_data: - legacy_manifest = godelos_data["manifest_consciousness"] - # Extract relevant data but keep canonical structure - if "attention_focus" in legacy_manifest: - focus = legacy_manifest["attention_focus"] - if isinstance(focus, dict) and "primary" in focus: - manifest_consciousness["attention"]["focus"] = [focus["primary"]] - manifest_consciousness["attention"]["intensity"] = focus.get("intensity", 0.7) - - if "metacognitive_status" in godelos_data: - meta = godelos_data["metacognitive_status"] - if isinstance(meta, dict): - manifest_consciousness["metaReflection"]["depth"] = meta.get("self_awareness", 0.6) - manifest_consciousness["metaReflection"]["coherence"] = meta.get("confidence", 0.85) - # Build canonical response canonical_response = { "version": "v1", diff --git a/backend/websocket_manager.py b/backend/websocket_manager.py index 72df581..cf2dd6f 100644 --- a/backend/websocket_manager.py +++ b/backend/websocket_manager.py @@ -1,5 +1,96 @@ -"""Compatibility shim: re-exports WebSocketManager from unified_server.""" +""" +Canonical WebSocket Manager for GödelOS. + +This is the single base WebSocket manager implementation used throughout the +system. ``EnhancedWebSocketManager`` (in ``backend.core.enhanced_websocket_manager``) +extends this class with consciousness-streaming capabilities and is the +recommended runtime manager. + +All WebSocket traffic in the GödelOS backend MUST be routed through one of +these two classes — no inline or ad-hoc managers. +""" + +import json +import logging +import time +from typing import Dict, List, Union + +from fastapi import WebSocket + +logger = logging.getLogger(__name__) + + +class WebSocketManager: + """Base WebSocket connection manager. + + Provides fundamental connect / disconnect / broadcast primitives. + For consciousness-aware streaming, use ``EnhancedWebSocketManager``. + """ + + def __init__(self): + self.active_connections: List[WebSocket] = [] + + async def connect(self, websocket: WebSocket): + await websocket.accept() + self.active_connections.append(websocket) + + def disconnect(self, websocket: WebSocket): + if websocket in self.active_connections: + self.active_connections.remove(websocket) + + async def send_personal_message(self, message: str, websocket: WebSocket): + await websocket.send_text(message) + + async def broadcast(self, message: Union[str, dict]): + if isinstance(message, dict): + message = json.dumps(message) + for connection in self.active_connections: + try: + await connection.send_text(message) + except Exception: + # Connection closed or broken — silently skip. + # Disconnected clients are cleaned up in disconnect(). + pass + + async def broadcast_cognitive_update(self, event: dict): + """Broadcast cognitive update event to all connected clients.""" + try: + inner_event = event + if isinstance(event, dict) and event.get("type") == "cognitive_event" and isinstance(event.get("data"), dict): + inner_event = event.get("data") + message = { + "type": "cognitive_event", + "timestamp": inner_event.get("timestamp", ""), + "data": inner_event + } + except Exception: + message = { + "type": "cognitive_event", + "timestamp": event.get("timestamp", ""), + "data": event + } + await self.broadcast(message) + + async def broadcast_consciousness_update(self, consciousness_data: dict): + """Broadcast consciousness update to all connected clients.""" + try: + message = { + "type": "consciousness_update", + "timestamp": consciousness_data.get("timestamp", time.time()), + "data": consciousness_data + } + await self.broadcast(message) + except Exception as e: + logger.error(f"Error broadcasting consciousness update: {e}") + + def has_connections(self) -> bool: + return len(self.active_connections) > 0 + + async def get_stats(self) -> Dict: + """Return basic connection statistics.""" + return { + "active_connections": len(self.active_connections), + } -from backend.unified_server import WebSocketManager __all__ = ["WebSocketManager"] diff --git a/docs/CANONICAL_RUNTIME.md b/docs/CANONICAL_RUNTIME.md new file mode 100644 index 0000000..c261fc1 --- /dev/null +++ b/docs/CANONICAL_RUNTIME.md @@ -0,0 +1,102 @@ +# GödelOS Canonical Runtime Architecture + +> **Single source of truth for server startup, WebSocket management, +> consciousness state computation, and cognitive pipeline integration.** +> +> Enforced by `tests/backend/test_runtime_canonicalization.py` (21 tests). + +--- + +## 1. Single Runtime Entrypoint + +| Module | Status | Purpose | +|--------|--------|---------| +| `backend/unified_server.py` | **CANONICAL** | The one and only FastAPI application. All startup scripts, CI workflows, and deployment targets must use this module. | +| `backend/main.py` | Compatibility shim | Re-exports `unified_server` into `sys.modules["backend.main"]` so that legacy test imports (`from backend.main import app`) continue to work. Returns the **same** `app` object. | +| `backend/start_server.py` | CLI wrapper | Imports `app` from `backend.unified_server` and wraps it in a `GödelOSServer` class for signal handling. | +| `backend/minimal_server.py` | **DEPRECATED (pure shim)** | 24-line shim that re-exports `unified_server.app`. Emits `DeprecationWarning` on import. Will be removed in a future release. | + +### Startup Commands + +```bash +# Recommended (both backend + frontend) +./start-godelos.sh --dev + +# Backend only +python backend/unified_server.py +# or +uvicorn backend.unified_server:app --reload --port 8000 +``` + +--- + +## 2. WebSocket Manager + +| Class | Module | Status | +|-------|--------|--------| +| `WebSocketManager` | `backend/websocket_manager.py` | Base class — connect/disconnect/broadcast primitives. | +| `EnhancedWebSocketManager` | `backend/core/enhanced_websocket_manager.py` | **CANONICAL runtime manager**. Extends `WebSocketManager` with consciousness streaming, emergence alerts, and recursive-awareness feeds. | + +At runtime, **both** `websocket_manager` and `enhanced_websocket_manager` globals +in `unified_server.py` reference the **same** `EnhancedWebSocketManager` instance. +This is verified by `test_init_core_services_aliases_managers`. + +### Contract + +All outbound WebSocket messages conform to one of these event schemas: + +```jsonc +{ "type": "cognitive_event", "timestamp": ..., "data": {...} } +{ "type": "consciousness_update", "timestamp": ..., "data": {...} } +{ "type": "consciousness_emergence", "timestamp": ..., ... } +{ "type": "consciousness_breakthrough", "timestamp": ..., "alert_level": "CRITICAL", ... } +``` + +--- + +## 3. Consciousness State + +| Component | Module | Status | +|-----------|--------|--------| +| `UnifiedConsciousnessEngine` | `backend/core/unified_consciousness_engine.py` | **CANONICAL** — recursive self-awareness loop, IIT φ, global workspace, phenomenal experience, metacognition. | +| `ConsciousnessEngine` | `backend/core/consciousness_engine.py` | **DEPRECATED class**. The `ConsciousnessState` and `ConsciousnessLevel` data types remain canonical. | +| `ConsciousnessEmergenceDetector` | `backend/core/consciousness_emergence_detector.py` | Active — rolling-window breakthrough scorer. | +| `PhenomenalExperienceGenerator` (core) | `backend/core/phenomenal_experience.py` | **Active** — qualia & subjective-experience simulation used internally by `UnifiedConsciousnessEngine`. | +| `PhenomenalExperienceGenerator` (legacy) | `backend/phenomenal_experience_generator.py` | **DEPRECATED (pure shim)** — 23-line re-export. Emits `DeprecationWarning`. | + +### State Read Resolution Order + +All consciousness/cognitive-state endpoints follow this priority: + +1. **`unified_consciousness_engine`** — authoritative IIT φ, global workspace, recursive awareness +2. **`cognitive_manager.assess_consciousness()`** — assessment backed by `ConsciousnessEngine` + LLM driver +3. **`godelos_integration.get_cognitive_state()`** — pipeline-backed cognitive metrics +4. **Static fallback** — test-only stub (logs warning when served) + +Endpoint mapping: +- `/cognitive/state`, `/api/cognitive/state` → resolution order above +- `/api/cognitive-state` → enriches manifest consciousness from `unified_consciousness_engine` first +- `/api/v1/consciousness/state` → `cognitive_manager.assess_consciousness()` +- `/api/consciousness/state` → `unified_consciousness_engine.consciousness_state` (via router) + +--- + +## 4. Cognitive Pipeline Integration + +All cognitive processing flows through `godelOS/cognitive_pipeline.py` +(`CognitivePipeline` class). + +`backend/godelos_integration.py` wraps the pipeline with graceful degradation. +The `simple_knowledge_store` is a **gated test-only stub**: accessing it when +`CognitivePipeline` is active emits a runtime warning. This is enforced by +`test_fallback_warns_when_pipeline_active`. + +--- + +## 5. Remaining Technical Debt + +- [ ] Remove `backend/minimal_server.py` shim once confirmed no external tooling references it. +- [ ] Remove `backend/phenomenal_experience_generator.py` shim once confirmed unused. +- [ ] Merge `ConsciousnessEngine` base types into `unified_consciousness_engine.py` to eliminate import indirection. +- [ ] Contract-test all outbound WebSocket event schemas in CI. +- [ ] Replace mock health scores in `get_system_health_with_labels()` with real service probes. diff --git a/tests/backend/test_runtime_canonicalization.py b/tests/backend/test_runtime_canonicalization.py new file mode 100644 index 0000000..48ec237 --- /dev/null +++ b/tests/backend/test_runtime_canonicalization.py @@ -0,0 +1,261 @@ +""" +Runtime Canonicalization Enforcement Tests + +These tests prove that the GödelOS runtime executes through a single canonical +contract — one WebSocket manager, one consciousness state authority, one +server entrypoint, and no active legacy modules. +""" +import importlib +import sys +import types +import pytest + + +# --------------------------------------------------------------------------- +# 1. Single server entrypoint +# --------------------------------------------------------------------------- + +class TestCanonicalEntrypoint: + """Verify that only ``backend.unified_server`` defines the production app.""" + + def test_unified_server_defines_app(self): + """The canonical FastAPI app lives in backend.unified_server.""" + from backend.unified_server import app + from fastapi import FastAPI + assert isinstance(app, FastAPI) + + def test_main_shim_returns_same_app(self): + """backend.main re-exports the exact same app object.""" + from backend.unified_server import app as canonical_app + # backend.main replaces itself in sys.modules; importing it + # yields the unified_server module. + import backend.main as main_mod + assert main_mod.app is canonical_app + + def test_minimal_server_shim_returns_same_app(self): + """backend.minimal_server re-exports the canonical app (pure shim).""" + import warnings + from backend.unified_server import app as canonical_app + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + from backend.minimal_server import app as minimal_app + assert minimal_app is canonical_app + + def test_minimal_server_emits_deprecation(self): + """Importing backend.minimal_server emits a DeprecationWarning.""" + # Force reimport to trigger the warning + saved = sys.modules.pop("backend.minimal_server", None) + try: + with pytest.warns(DeprecationWarning, match="backend.minimal_server is deprecated"): + importlib.import_module("backend.minimal_server") + finally: + if saved is not None: + sys.modules["backend.minimal_server"] = saved + + def test_start_server_imports_from_unified(self): + """backend.start_server imports app from the canonical module.""" + import pathlib + source = (pathlib.Path(__file__).resolve().parents[2] / "backend" / "start_server.py").read_text() + assert "backend.unified_server" in source + # Ensure the import line referencing unified_server comes before any + # reference to the old backend.main import. + assert "backend.main" not in source.split("backend.unified_server")[0] + + +# --------------------------------------------------------------------------- +# 2. Single WebSocket manager identity +# --------------------------------------------------------------------------- + +class TestWebSocketManagerIdentity: + """Prove that only ONE WebSocket manager instance is created at runtime.""" + + def test_websocket_manager_class_is_base(self): + """backend.websocket_manager exports the base WebSocketManager class.""" + from backend.websocket_manager import WebSocketManager + assert hasattr(WebSocketManager, "broadcast") + assert hasattr(WebSocketManager, "connect") + assert hasattr(WebSocketManager, "disconnect") + + def test_enhanced_inherits_base(self): + """EnhancedWebSocketManager is a subclass of the base WebSocketManager.""" + from backend.websocket_manager import WebSocketManager + from backend.core.enhanced_websocket_manager import EnhancedWebSocketManager + assert issubclass(EnhancedWebSocketManager, WebSocketManager) + + def test_initialize_creates_single_instance(self): + """After initialize_core_services, both globals reference the same object.""" + import backend.unified_server as us + # At module-load time both are None. After lifespan they should be set. + # We test the *structural invariant*: if one is set, the other must be + # the same object (identity, not equality). + if us.websocket_manager is not None: + assert us.websocket_manager is us.enhanced_websocket_manager, ( + "websocket_manager and enhanced_websocket_manager must be the " + "same object at runtime" + ) + + @pytest.mark.asyncio + async def test_init_core_services_aliases_managers(self): + """initialize_core_services sets both WS globals to the same instance.""" + import backend.unified_server as us + + # Save originals + orig_ws = us.websocket_manager + orig_ews = us.enhanced_websocket_manager + + try: + us.websocket_manager = None + us.enhanced_websocket_manager = None + await us.initialize_core_services() + + assert us.websocket_manager is not None, "websocket_manager should be initialized" + assert us.enhanced_websocket_manager is not None, "enhanced_websocket_manager should be initialized" + assert us.websocket_manager is us.enhanced_websocket_manager, ( + "Both WS globals must reference the SAME runtime object" + ) + finally: + us.websocket_manager = orig_ws + us.enhanced_websocket_manager = orig_ews + + def test_no_inline_websocketmanager_in_unified_server(self): + """unified_server.py must NOT define WebSocketManager inline — it imports it.""" + import inspect + import backend.unified_server as us + source = inspect.getsource(us) + # The class definition should NOT be in unified_server.py + assert "\nclass WebSocketManager:" not in source, ( + "WebSocketManager must be imported from backend.websocket_manager, " + "not defined inline in unified_server.py" + ) + + +# --------------------------------------------------------------------------- +# 3. Canonical consciousness state authority +# --------------------------------------------------------------------------- + +class TestCanonicalConsciousnessState: + """All state reads must derive from UnifiedConsciousnessEngine.""" + + def test_consciousness_endpoints_use_unified_engine(self): + """consciousness_endpoints.py references unified_consciousness_engine.""" + from backend.api import consciousness_endpoints as ce + # The setter function must exist and update the module-level engine ref + assert callable(getattr(ce, "set_consciousness_engine", None)) + + def test_cognitive_state_endpoint_prefers_unified_engine(self): + """The /cognitive/state handler checks unified_consciousness_engine first.""" + import inspect + import backend.unified_server as us + source = inspect.getsource(us.get_cognitive_state_endpoint) + # The first conditional should check the unified engine + idx_uce = source.find("unified_consciousness_engine") + idx_gi = source.find("godelos_integration") + assert idx_uce != -1, "/cognitive/state must check unified_consciousness_engine" + assert idx_uce < idx_gi, ( + "unified_consciousness_engine must be checked BEFORE godelos_integration " + "in /cognitive/state" + ) + + def test_api_cognitive_state_prefers_unified_engine(self): + """The /api/cognitive-state handler enriches from unified_consciousness_engine first.""" + import inspect + import backend.unified_server as us + source = inspect.getsource(us.api_get_cognitive_state_alias) + idx_uce = source.find("unified_consciousness_engine") + idx_gi = source.find("godelos_integration") + assert idx_uce != -1, "/api/cognitive-state must reference unified_consciousness_engine" + assert idx_uce < idx_gi, ( + "unified_consciousness_engine must be checked BEFORE godelos_integration " + "in /api/cognitive-state" + ) + + def test_v1_consciousness_state_uses_cognitive_manager(self): + """/api/v1/consciousness/state derives from cognitive_manager.assess_consciousness.""" + import inspect + import backend.unified_server as us + source = inspect.getsource(us.get_consciousness_state) + assert "cognitive_manager" in source + + +# --------------------------------------------------------------------------- +# 4. Legacy modules not active in runtime flow +# --------------------------------------------------------------------------- + +class TestLegacyModulesInactive: + """Deprecated / legacy modules are shims, not active implementations.""" + + def test_minimal_server_is_pure_shim(self): + """minimal_server.py should be ≤30 lines (shim only).""" + import pathlib + path = pathlib.Path(__file__).resolve().parents[2] / "backend" / "minimal_server.py" + line_count = len(path.read_text().splitlines()) + assert line_count <= 30, ( + f"minimal_server.py has {line_count} lines — it should be a pure " + f"shim (re-export only)" + ) + + def test_phenomenal_experience_generator_is_shim(self): + """phenomenal_experience_generator.py should be ≤30 lines (shim only).""" + import pathlib + path = pathlib.Path(__file__).resolve().parents[2] / "backend" / "phenomenal_experience_generator.py" + line_count = len(path.read_text().splitlines()) + assert line_count <= 30, ( + f"phenomenal_experience_generator.py has {line_count} lines — it " + f"should be a pure shim" + ) + + def test_consciousness_engine_class_is_deprecated(self): + """ConsciousnessEngine docstring declares deprecation.""" + from backend.core.consciousness_engine import ConsciousnessEngine + assert "DEPRECATED" in (ConsciousnessEngine.__doc__ or ""), ( + "ConsciousnessEngine must be marked DEPRECATED in its docstring" + ) + + def test_consciousness_state_types_remain_canonical(self): + """ConsciousnessState and ConsciousnessLevel are still importable (they are base types).""" + from backend.core.consciousness_engine import ConsciousnessState, ConsciousnessLevel + assert ConsciousnessState is not None + assert ConsciousnessLevel is not None + + +# --------------------------------------------------------------------------- +# 5. Pipeline integration gating +# --------------------------------------------------------------------------- + +class TestPipelineIntegrationGating: + """godelos_integration gates fallback knowledge store usage.""" + + def test_simple_knowledge_store_is_property(self): + """simple_knowledge_store should be a property that can detect pipeline presence.""" + from backend.godelos_integration import GödelOSIntegration + assert isinstance( + GödelOSIntegration.__dict__.get("simple_knowledge_store"), + property, + ), "simple_knowledge_store must be a property, not a plain attribute" + + def test_fallback_warns_when_pipeline_active(self): + """Accessing simple_knowledge_store with an active pipeline logs a warning.""" + from backend.godelos_integration import GödelOSIntegration + + gi = GödelOSIntegration() + # Reset class-level warning flag for this test + GödelOSIntegration._fallback_warned = False + # Simulate an active pipeline + gi.cognitive_pipeline = object() # truthy sentinel + + # Access should succeed but set the warned flag + _ = gi.simple_knowledge_store + assert GödelOSIntegration._fallback_warned is True + + # Clean up + GödelOSIntegration._fallback_warned = False + + def test_no_warning_without_pipeline(self): + """No warning when pipeline is None (test/fallback mode).""" + from backend.godelos_integration import GödelOSIntegration + gi = GödelOSIntegration() + GödelOSIntegration._fallback_warned = False + gi.cognitive_pipeline = None + _ = gi.simple_knowledge_store + assert GödelOSIntegration._fallback_warned is False + GödelOSIntegration._fallback_warned = False