From 62cd7c20fe2308317df4aa655d4caf9e183f4b9c Mon Sep 17 00:00:00 2001
From: Cyber MacGeddon <cybermaggedon@gmail.com>
Date: Wed, 3 Dec 2025 18:54:37 +0000
Subject: [PATCH] Fix structural errors

---
 .../trustgraph/api/async_socket_client.py     | 40 +++++++++++++------
 trustgraph-base/trustgraph/api/bulk_client.py |  4 +-
 2 files changed, 29 insertions(+), 15 deletions(-)

diff --git a/trustgraph-base/trustgraph/api/async_socket_client.py b/trustgraph-base/trustgraph/api/async_socket_client.py
index 2e058505..b68a69d5 100644
--- a/trustgraph-base/trustgraph/api/async_socket_client.py
+++ b/trustgraph-base/trustgraph/api/async_socket_client.py
@@ -189,13 +189,17 @@ class AsyncSocketFlowInstance:
         request.update(kwargs)
 
         if streaming:
-            async for chunk in self.client._send_request_streaming("text-completion", self.flow_id, request):
-                if hasattr(chunk, 'content'):
-                    yield chunk.content
+            return self._text_completion_streaming(request)
         else:
             result = await self.client._send_request("text-completion", self.flow_id, request)
             return result.get("response", "")
 
+    async def _text_completion_streaming(self, request):
+        """Helper for streaming text completion"""
+        async for chunk in self.client._send_request_streaming("text-completion", self.flow_id, request):
+            if hasattr(chunk, 'content'):
+                yield chunk.content
+
     async def graph_rag(self, question: str, user: str, collection: str,
                         max_subgraph_size: int = 1000, max_subgraph_count: int = 5,
                         max_entity_distance: int = 3, streaming: bool = False, **kwargs):
@@ -212,13 +216,17 @@ class AsyncSocketFlowInstance:
         request.update(kwargs)
 
         if streaming:
-            async for chunk in self.client._send_request_streaming("graph-rag", self.flow_id, request):
-                if hasattr(chunk, 'content'):
-                    yield chunk.content
+            return self._graph_rag_streaming(request)
         else:
             result = await self.client._send_request("graph-rag", self.flow_id, request)
             return result.get("response", "")
 
+    async def _graph_rag_streaming(self, request):
+        """Helper for streaming graph RAG"""
+        async for chunk in self.client._send_request_streaming("graph-rag", self.flow_id, request):
+            if hasattr(chunk, 'content'):
+                yield chunk.content
+
     async def document_rag(self, question: str, user: str, collection: str,
                            doc_limit: int = 10, streaming: bool = False, **kwargs):
         """Document RAG with optional streaming"""
@@ -232,13 +240,17 @@ class AsyncSocketFlowInstance:
         request.update(kwargs)
 
         if streaming:
-            async for chunk in self.client._send_request_streaming("document-rag", self.flow_id, request):
-                if hasattr(chunk, 'content'):
-                    yield chunk.content
+            return self._document_rag_streaming(request)
         else:
             result = await self.client._send_request("document-rag", self.flow_id, request)
             return result.get("response", "")
 
+    async def _document_rag_streaming(self, request):
+        """Helper for streaming document RAG"""
+        async for chunk in self.client._send_request_streaming("document-rag", self.flow_id, request):
+            if hasattr(chunk, 'content'):
+                yield chunk.content
+
     async def prompt(self, id: str, variables: Dict[str, str], streaming: bool = False, **kwargs):
         """Execute prompt with optional streaming"""
         request = {
@@ -249,13 +261,17 @@ class AsyncSocketFlowInstance:
         request.update(kwargs)
 
         if streaming:
-            async for chunk in self.client._send_request_streaming("prompt", self.flow_id, request):
-                if hasattr(chunk, 'content'):
-                    yield chunk.content
+            return self._prompt_streaming(request)
         else:
             result = await self.client._send_request("prompt", self.flow_id, request)
             return result.get("response", "")
 
+    async def _prompt_streaming(self, request):
+        """Helper for streaming prompt"""
+        async for chunk in self.client._send_request_streaming("prompt", self.flow_id, request):
+            if hasattr(chunk, 'content'):
+                yield chunk.content
+
     async def graph_embeddings_query(self, text: str, user: str, collection: str, limit: int = 10, **kwargs):
         """Query graph embeddings for semantic search"""
         request = {
diff --git a/trustgraph-base/trustgraph/api/bulk_client.py b/trustgraph-base/trustgraph/api/bulk_client.py
index aa5648da..958df41a 100644
--- a/trustgraph-base/trustgraph/api/bulk_client.py
+++ b/trustgraph-base/trustgraph/api/bulk_client.py
@@ -68,9 +68,7 @@ class BulkClient:
             if loop.is_running():
                 loop = asyncio.new_event_loop()
                 asyncio.set_event_loop(loop)
-        except Runtime
-
-Error:
+        except RuntimeError:
             loop = asyncio.new_event_loop()
             asyncio.set_event_loop(loop)