added more logging to comm protocol

Author: bradley-erickson

VERSION

@@ -1 +1 @@
-0.1.0+2025.10.01T21.16.15.146Z.8b205d0c.master
+0.1.0+2025.10.13T19.17.03.311Z.f12616b4.berickson.2025.10.13.comm.protocol

learning_observer/VERSION

@@ -1 +1 @@
-0.1.0+2025.10.01T17.50.57.101Z.d6047986.master
+0.1.0+2025.10.13T19.17.03.311Z.f12616b4.berickson.2025.10.13.comm.protocol

learning_observer/learning_observer/communication_protocol/executor.py

@@ -446,7 +446,7 @@ async def handle_select(keys, fields=learning_observer.communication_protocol.qu
 
         # Determine fields to keep based on the current resulting_value if fields is All
         if fields == learning_observer.communication_protocol.query.SelectFields.All:
-            current_fields_to_keep = {key: key for key in resulting_value.keys() if key != 'provenance'}
+            current_fields_to_keep = {key: key for key in resulting_value.keys() if key != 'provenance'} if resulting_value else {}
         else:
             current_fields_to_keep = fields_to_keep
 
@@ -666,10 +666,38 @@ async def execute_dag(endpoint, parameters, functions, target_exports):
 
     See `learning_observer/communication_protocol/test_cases.py` for usage examples.
     """
-    target_nodes = [endpoint['exports'][key]['returns'] for key in target_exports]
-
+    exports = endpoint.get('exports', {})
+    nodes = endpoint.get('execution_dag', {})
     visited = set()
-    nodes = endpoint['execution_dag']
+
+    # --- Resolve targets and collect any obvious errors early ---
+    target_nodes = []
+    target_errors = {}  # maps target node name -> error dict
+
+    for key in target_exports:
+        if key not in exports:
+            # Unknown export requested
+            target_name = f'__missing_export__:{key}'
+            target_nodes.append(target_name)
+            target_errors[target_name] = DAGExecutionException(
+                f'Export `{key}` not found in endpoint.exports.',
+                inspect.currentframe().f_code.co_name,
+                {'requested_export': key, 'available_exports': list(exports.keys())}
+            ).to_dict()
+            continue
+
+        target_node = exports[key].get('returns')
+        if target_node not in nodes:
+            # Export exists, but its `returns` node is missing from the DAG
+            target_nodes.append(target_node)
+            target_errors[target_node] = DAGExecutionException(
+                f'Target DAG node `{target_node}` not found in execution_dag.',
+                inspect.currentframe().f_code.co_name,
+                {'target_node': target_node, 'available_nodes': list(nodes.keys())}
+            ).to_dict()
+            continue
+
+        target_nodes.append(target_node)
 
     async def dispatch_node(node):
         """
@@ -743,10 +771,17 @@ async def visit(node_name):
         visited.add(node_name)
         return nodes[node_name]
 
+    out = {}
+    for e in target_nodes:
+        if e in target_errors:
+            out[e] = _clean_json_via_generator(target_errors[e])
+        else:
+            out[e] = _clean_json_via_generator(await visit(e))
+    return out
+
     # Include execution history in output if operating in development settings
     if learning_observer.settings.RUN_MODE == learning_observer.settings.RUN_MODES.DEV:
         return {e: _clean_json_via_generator(await visit(e)) for e in target_nodes}
-
     # HACK currently `dashboard.py` relies on the provenance to tell users which
     # items need updating, such as John Doe's history essay. This ought to be
     # handled by the communication protocol during execution. Once that occurs,

learning_observer/learning_observer/dashboard.py

@@ -587,6 +587,53 @@ async def websocket_dashboard_handler(request):
     Returns:
         aiohttp web response.
     '''
+    active_user = await learning_observer.auth.get_active_user(request)
+    user_context = {
+        'user_id': (active_user or {}).get('user_id'),
+        'user_email': (active_user or {}).get('email'),
+        'user_role': (active_user or {}).get('role'),
+    }
+
+    def _log_protocol_event(event, **extra):
+        '''
+        Emit structured debug logs describing websocket activity.
+        '''
+        payload = {
+            'event': event,
+            'user_id': user_context.get('user_id'),
+            'user_email': user_context.get('user_email'),
+            'user_role': user_context.get('user_role'),
+            'remote': request.remote,
+            'forwarded_for': request.headers.get('X-Forwarded-For'),
+            'request_path': str(request.rel_url),
+        }
+        payload.update(extra)
+        try:
+            debug_log('communication_protocol_event', json.dumps(payload, sort_keys=True))
+        except TypeError:
+            # Fall back to logging the raw payload if serialization fails.
+            debug_log('communication_protocol_event', payload)
+
+    def _summarize_query(query):
+        '''
+        Provide a compact description of the query for log aggregation.
+        '''
+        summary = {}
+        for key, value in (query or {}).items():
+            if not isinstance(value, dict):
+                summary[key] = {'non_dict_value': repr(value)}
+                continue
+            execution_dag = value.get('execution_dag')
+            summary[key] = {
+                'target_exports': value.get('target_exports', []),
+                'execution_dag_type': type(execution_dag).__name__,
+                'execution_dag_name': execution_dag if isinstance(execution_dag, str) else None,
+                'kwargs': value.get('kwargs', {}),
+            }
+        return summary
+
+    _log_protocol_event('connection_opened')
+
     ws = aiohttp.web.WebSocketResponse(receive_timeout=0.3)
     await ws.prepare(request)
     client_query = None
@@ -661,11 +708,15 @@ async def _drive_generator(generator, dag_kwargs, target=None):
         try:
             received_params = await ws.receive_json()
             client_query = received_params
+            _log_protocol_event(
+                'query_received',
+                query_summary=_summarize_query(client_query),
+            )
             # TODO we should validate the client_query structure
         except (TypeError, ValueError):
             # these Errors may signal a close
             if (await ws.receive()).type == aiohttp.WSMsgType.CLOSE:
-                debug_log("Socket closed!")
+                _log_protocol_event('connection_closed', reason='client_close_frame')
                 return aiohttp.web.Response()
         except asyncio.exceptions.TimeoutError:
             # this is the normal path of the code
@@ -674,7 +725,7 @@ async def _drive_generator(generator, dag_kwargs, target=None):
                 continue
 
         if ws.closed:
-            debug_log("Socket closed.")
+            _log_protocol_event('connection_closed', reason='websocket_closed_flag')
             return aiohttp.web.Response()
 
         if client_query != previous_client_query:

scripts/generic_websocket_dashboard.js

@@ -1,53 +1,48 @@
 /*
-  This is test code for our new generic dashboard framework.
+  Simple Node.js websocket client for exercising the communication protocol.
 
-  It runs with node.js. We're developing it in node so that we can use
-  this as a starting point for thinking about a front-end framework, and
-  perhaps share test code.
-
-  We don't have a clean plan for where we'll go (e.g. reuse code versus
-  prototypes). It's starting as a clone of the Python code with the same
-  filename.
+  Update the REQUEST payload to point at the execution DAG and exports you want
+  to test.
 */
 
-// var d3 = require('d3');
-// d3.text("https://www.google.com", function(d) {console.log(d);});
+// Remove this line when running within a browser terminal (i.e. non-Node.js environment)
 const WebSocket = require('ws');
 
-server = 'ws://localhost:8888/wsapi/generic_dashboard'
-
-messages = [
-    {
-        "action": "subscribe",
-        "keys": [{
-            "source" : "da_timeline.visualize.handle_event",
-            "KeyField.STUDENT": "guest-225d890e93a6b04c0aefe515b9d2dac9"
-        }],
-        "refresh": [0.5, "seconds"]
-    },
-    {
-        "action": "subscribe",
-        "keys":  [{
-            "source" : "da_timeline.visualize.handle_event",
-            "KeyField.STUDENT": "INVALID-STUDENT"
-        }],
-        "refresh": [2, "seconds"]
-    },
-    {
-        "action": "start"
-    }
-]
-
-socket = new WebSocket(server);
-socket.on('message', msg => console.log(msg.toString()));
+const SERVER = 'ws://localhost:8888/wsapi/communication_protocol';
 
-socket.onopen = function() {
-    console.log("Open");
-    for(var i=0; i<messages.length; i++) {
-	socket.send(JSON.stringify(messages[i]));
+const REQUEST = {
+  docs_request: {
+    execution_dag: 'writing_observer',
+    target_exports: ['docs_with_roster'],
+    kwargs: {
+      course_id: 'COURSE-123',
     }
+  }
 };
-
-// We can comment in/out console.log(`d`) based on whether we want a giant amount of debug data
-socket.onerror = function(d) { console.log("Error"); console.log(d); };
-socket.onclose = function(d) { console.log("Close"); /* console.log(d); */ };
+const socket = new WebSocket(SERVER);
+
+socket.on('open', () => {
+  console.log('Open');
+  socket.send(JSON.stringify(REQUEST));
+});
+
+socket.on('message', (msg) => {
+  try {
+    const parsed = JSON.parse(msg.toString());
+    console.log(parsed);
+  } catch (error) {
+    console.log(msg.toString());
+  }
+});
+
+socket.on('error', (err) => {
+  console.log('Error');
+  console.log(err);
+});
+
+socket.on('close', (event) => {
+  console.log('Close');
+  if (event) {
+    console.log(event);
+  }
+});

scripts/generic_websocket_dashboard.py

@@ -1,48 +1,41 @@
-'''
-This is a test script for a web socket interaction.
-'''
+"""Simple websocket client for manual communication protocol testing."""
 
 import aiohttp
 import asyncio
+import json
 
-messages = [
-    {
-        "action": "subscribe",
-        "keys": [{
-            "source": "da_timeline.visualize.handle_event",
-            "KeyField.STUDENT": "guest-225d890e93a6b04c0aefe515b9d2dac9"
-        }],
-        "refresh": [0.5, "seconds"]
-    },
-    {
-        "action": "subscribe",
-        "keys": [{
-            "source": "da_timeline.visualize.handle_event",
-            "KeyField.STUDENT": "INVALID-STUDENT"
-        }],
-        "refresh": [2, "seconds"]
-    },
-    {
-        "action": "start"
+# Example request payload for the communication protocol websocket. Adjust the
+# execution_dag, target_exports, and kwargs to match the reducers you want to
+# exercise.
+REQUEST = {
+    "docs_request": {
+        "execution_dag": "writing_observer",
+        "target_exports": ["docs_with_roster"],
+        "kwargs": {
+            "course_id": 12345678901
+        },
     }
-]
+}
 
 
 async def main():
     async with aiohttp.ClientSession() as session:
-        async with session.ws_connect('http://localhost:8888/wsapi/generic_dashboard', timeout=0.5) as ws:
-            for message in messages:
-                await ws.send_json(message)
-
+        async with session.ws_connect(
+            "http://localhost:8888/wsapi/communication_protocol",
+            timeout=5,
+        ) as ws:
+            await ws.send_json(REQUEST)
             async for msg in ws:
-                print(msg.type)
                 if msg.type == aiohttp.WSMsgType.TEXT:
-                    print("Message")
-                    print(msg.data)
+                    try:
+                        parsed = json.loads(msg.data)
+                    except json.JSONDecodeError:
+                        parsed = msg.data
+                    print(json.dumps(parsed, indent=2))
                 elif msg.type == aiohttp.WSMsgType.ERROR:
-                    print("Error")
-                    print(msg)
+                    print("Error received from websocket:", msg)
                     break
-    return True
 
-asyncio.run(main())
+
+if __name__ == '__main__':
+    asyncio.run(main())