From 693d952214f8dda1a31df29b535b12e49b1934c6 Mon Sep 17 00:00:00 2001
From: Cyber MacGeddon <cybermaggedon@gmail.com>
Date: Wed, 17 Dec 2025 14:16:53 +0000
Subject: [PATCH] Fixing schema issues

---
 .../trustgraph/base/pulsar_backend.py         | 36 +++++++++++--------
 .../trustgraph/config/service/flow.py         |  3 --
 .../trustgraph/config/service/service.py      |  3 +-
 .../gateway/dispatch/text_completion.py       |  4 +--
 4 files changed, 24 insertions(+), 22 deletions(-)

diff --git a/trustgraph-base/trustgraph/base/pulsar_backend.py b/trustgraph-base/trustgraph/base/pulsar_backend.py
index 144a48ba..ff0caa4e 100644
--- a/trustgraph-base/trustgraph/base/pulsar_backend.py
+++ b/trustgraph-base/trustgraph/base/pulsar_backend.py
@@ -62,20 +62,11 @@ def dict_to_dataclass(data: dict, cls: type) -> Any:
         if key in field_types:
             field_type = field_types[key]
 
-            # Handle Optional types (type | None)
-            if hasattr(field_type, '__args__'):
-                # Get the non-None type from Union
-                actual_type = next((t for t in field_type.__args__ if t is not type(None)), field_type)
-            else:
-                actual_type = field_type
-
-            # Recursively convert nested dataclasses
-            if is_dataclass(actual_type) and isinstance(value, dict):
-                kwargs[key] = dict_to_dataclass(value, actual_type)
-            elif hasattr(actual_type, '__origin__'):
-                # Handle generic types like list[T] or dict[K, V]
-                if actual_type.__origin__ == list:
-                    item_type = actual_type.__args__[0] if actual_type.__args__ else None
+            # Check if this is a generic type (list, dict, etc.)
+            if hasattr(field_type, '__origin__'):
+                # Handle list[T]
+                if field_type.__origin__ == list:
+                    item_type = field_type.__args__[0] if field_type.__args__ else None
                     if item_type and is_dataclass(item_type) and isinstance(value, list):
                         kwargs[key] = [
                             dict_to_dataclass(item, item_type) if isinstance(item, dict) else item
@@ -83,8 +74,19 @@ def dict_to_dataclass(data: dict, cls: type) -> Any:
                         ]
                     else:
                         kwargs[key] = value
+                # Handle Optional[T] (which is Union[T, None])
+                elif hasattr(field_type, '__args__') and type(None) in field_type.__args__:
+                    # Get the non-None type from Union
+                    actual_type = next((t for t in field_type.__args__ if t is not type(None)), None)
+                    if actual_type and is_dataclass(actual_type) and isinstance(value, dict):
+                        kwargs[key] = dict_to_dataclass(value, actual_type)
+                    else:
+                        kwargs[key] = value
                 else:
                     kwargs[key] = value
+            # Handle direct dataclass fields
+            elif is_dataclass(field_type) and isinstance(value, dict):
+                kwargs[key] = dict_to_dataclass(value, field_type)
             else:
                 kwargs[key] = value
 
@@ -211,11 +213,15 @@ class PulsarBackend:
         Example: q1/tg/flow/my-queue -> persistent://tg/flow/my-queue
 
         Args:
-            generic_topic: Generic topic string
+            generic_topic: Generic topic string or already-formatted Pulsar URI
 
         Returns:
             Pulsar topic URI
         """
+        # If already a Pulsar URI, return as-is
+        if '://' in generic_topic:
+            return generic_topic
+
         parts = generic_topic.split('/', 3)
         if len(parts) != 4:
             raise ValueError(f"Invalid topic format: {generic_topic}, expected qos/tenant/namespace/queue")
diff --git a/trustgraph-flow/trustgraph/config/service/flow.py b/trustgraph-flow/trustgraph/config/service/flow.py
index b99b7d0a..42696c31 100644
--- a/trustgraph-flow/trustgraph/config/service/flow.py
+++ b/trustgraph-flow/trustgraph/config/service/flow.py
@@ -361,9 +361,6 @@ class FlowConfig:
         else:
 
             resp = FlowResponse(
-                value=None,
-                directory=None,
-                values=None,
                 error=Error(
                     type = "bad-operation",
                     message = "Bad operation"
diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py
index b84660a1..42b256df 100644
--- a/trustgraph-flow/trustgraph/config/service/service.py
+++ b/trustgraph-flow/trustgraph/config/service/service.py
@@ -235,13 +235,12 @@ class Processor(AsyncProcessor):
             )
 
         except Exception as e:
-            
+
             resp = FlowResponse(
                 error=Error(
                     type = "flow-error",
                     message = str(e),
                 ),
-                text=None,
             )
 
             await self.flow_response_producer.send(
diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/text_completion.py b/trustgraph-flow/trustgraph/gateway/dispatch/text_completion.py
index d29d1918..0e77584e 100644
--- a/trustgraph-flow/trustgraph/gateway/dispatch/text_completion.py
+++ b/trustgraph-flow/trustgraph/gateway/dispatch/text_completion.py
@@ -6,12 +6,12 @@ from . requestor import ServiceRequestor
 
 class TextCompletionRequestor(ServiceRequestor):
     def __init__(
-            self, pulsar_client, request_queue, response_queue, timeout,
+            self, backend, request_queue, response_queue, timeout,
             consumer, subscriber,
     ):
 
         super(TextCompletionRequestor, self).__init__(
-            pulsar_client=pulsar_client,
+            backend=backend,
             request_queue=request_queue,
             response_queue=response_queue,
             request_schema=TextCompletionRequest,