Canonicalize GödelOS Runtime: Unify Server Entrypoint, WebSocket Manager, and Consciousness State

.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: |

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):

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__)
 

backend/godelos_integration.py

@@ -21,17 +21,30 @@
 
 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
         self.start_time = time.time()
         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: