Refactored schemas for pub/sub plugin architecture

trustgraph-base/trustgraph/schema/core/metadata.py

@@ -1,16 +1,14 @@
-
+from dataclasses import dataclass, field
-from pulsar.schema import Record, String, Array
 from .primitives import Triple
 
-class Metadata(Record):
+@dataclass
-
+class Metadata:
     # Source identifier
-    id = String()
+    id: str = ""
 
     # Subgraph
-    metadata = Array(Triple())
+    metadata: list[Triple] = field(default_factory=list)
 
     # Collection management
-    user = String()
+    user: str = ""
-    collection = String()
+    collection: str = ""
-

trustgraph-base/trustgraph/schema/core/primitives.py

@@ -1,34 +1,39 @@
 
-from pulsar.schema import Record, String, Boolean, Array, Integer
+from dataclasses import dataclass, field
 
-class Error(Record):
+@dataclass
-    type = String()
+class Error:
-    message = String()
+    type: str = ""
+    message: str = ""
 
-class Value(Record):
+@dataclass
-    value = String()
+class Value:
-    is_uri = Boolean()
+    value: str = ""
-    type = String()
+    is_uri: bool = False
+    type: str = ""
 
-class Triple(Record):
+@dataclass
-    s = Value()
+class Triple:
-    p = Value()
+    s: Value | None = None
-    o = Value()
+    p: Value | None = None
+    o: Value | None = None
 
-class Field(Record):
+@dataclass
-    name = String()
+class Field:
+    name: str = ""
     # int, string, long, bool, float, double, timestamp
-    type = String()
+    type: str = ""
-    size = Integer()
+    size: int = 0
-    primary = Boolean()
+    primary: bool = False
-    description = String()
+    description: str = ""
     # NEW FIELDS for structured data:
-    required = Boolean()  # Whether field is required
+    required: bool = False  # Whether field is required
-    enum_values = Array(String())  # For enum type fields
+    enum_values: list[str] = field(default_factory=list)  # For enum type fields
-    indexed = Boolean()  # Whether field should be indexed
+    indexed: bool = False  # Whether field should be indexed
 
-class RowSchema(Record):
+@dataclass
-    name = String()
+class RowSchema:
-    description = String()
+    name: str = ""
-    fields = Array(Field())
+    description: str = ""
+    fields: list[Field] = field(default_factory=list)
 

trustgraph-base/trustgraph/schema/core/topic.py

@@ -1,4 +1,23 @@
 
-def topic(topic, kind='persistent', tenant='tg', namespace='flow'):
+def topic(queue_name, qos='q1', tenant='tg', namespace='flow'):
-    return f"{kind}://{tenant}/{namespace}/{topic}"
+    """
+    Create a generic topic identifier that can be mapped by backends.
+
+    Args:
+        queue_name: The queue/topic name
+        qos: Quality of service
+             - 'q0' = best-effort (no ack)
+             - 'q1' = at-least-once (ack required)
+             - 'q2' = exactly-once (two-phase ack)
+        tenant: Tenant identifier for multi-tenancy
+        namespace: Namespace within tenant
+
+    Returns:
+        Generic topic string: qos/tenant/namespace/queue_name
+
+    Examples:
+        topic('my-queue')  # q1/tg/flow/my-queue
+        topic('config', qos='q2', namespace='config')  # q2/tg/config/config
+    """
+    return f"{qos}/{tenant}/{namespace}/{queue_name}"
 

trustgraph-base/trustgraph/schema/knowledge/document.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, Bytes
+from dataclasses import dataclass
 
 from ..core.metadata import Metadata
 from ..core.topic import topic
@@ -6,24 +6,27 @@ from ..core.topic import topic
 ############################################################################
 
 # PDF docs etc.
-class Document(Record):
+@dataclass
-    metadata = Metadata()
+class Document:
-    data = Bytes()
+    metadata: Metadata | None = None
+    data: bytes = b""
 
 ############################################################################
 
 # Text documents / text from PDF
 
-class TextDocument(Record):
+@dataclass
-    metadata = Metadata()
+class TextDocument:
-    text = Bytes()
+    metadata: Metadata | None = None
+    text: bytes = b""
 
 ############################################################################
 
 # Chunks of text
 
-class Chunk(Record):
+@dataclass
-    metadata = Metadata()
+class Chunk:
-    chunk = Bytes()
+    metadata: Metadata | None = None
+    chunk: bytes = b""
 
 ############################################################################

trustgraph-base/trustgraph/schema/knowledge/embeddings.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, Bytes, String, Boolean, Integer, Array, Double, Map
+from dataclasses import dataclass, field
 
 from ..core.metadata import Metadata
 from ..core.primitives import Value, RowSchema
@@ -8,49 +8,55 @@ from ..core.topic import topic
 
 # Graph embeddings are embeddings associated with a graph entity
 
-class EntityEmbeddings(Record):
+@dataclass
-    entity = Value()
+class EntityEmbeddings:
-    vectors = Array(Array(Double()))
+    entity: Value | None = None
+    vectors: list[list[float]] = field(default_factory=list)
 
 # This is a 'batching' mechanism for the above data
-class GraphEmbeddings(Record):
+@dataclass
-    metadata = Metadata()
+class GraphEmbeddings:
-    entities = Array(EntityEmbeddings())
+    metadata: Metadata | None = None
+    entities: list[EntityEmbeddings] = field(default_factory=list)
 
 ############################################################################
 
 # Document embeddings are embeddings associated with a chunk
 
-class ChunkEmbeddings(Record):
+@dataclass
-    chunk = Bytes()
+class ChunkEmbeddings:
-    vectors = Array(Array(Double()))
+    chunk: bytes = b""
+    vectors: list[list[float]] = field(default_factory=list)
 
 # This is a 'batching' mechanism for the above data
-class DocumentEmbeddings(Record):
+@dataclass
-    metadata = Metadata()
+class DocumentEmbeddings:
-    chunks = Array(ChunkEmbeddings())
+    metadata: Metadata | None = None
+    chunks: list[ChunkEmbeddings] = field(default_factory=list)
 
 ############################################################################
 
 # Object embeddings are embeddings associated with the primary key of an
 # object
 
-class ObjectEmbeddings(Record):
+@dataclass
-    metadata = Metadata()
+class ObjectEmbeddings:
-    vectors = Array(Array(Double()))
+    metadata: Metadata | None = None
-    name = String()
+    vectors: list[list[float]] = field(default_factory=list)
-    key_name = String()
+    name: str = ""
-    id = String()
+    key_name: str = ""
+    id: str = ""
 
 ############################################################################
 
 # Structured object embeddings with enhanced capabilities
 
-class StructuredObjectEmbedding(Record):
+@dataclass
-    metadata = Metadata()
+class StructuredObjectEmbedding:
-    vectors = Array(Array(Double()))
+    metadata: Metadata | None = None
-    schema_name = String()
+    vectors: list[list[float]] = field(default_factory=list)
-    object_id = String()  # Primary key value
+    schema_name: str = ""
-    field_embeddings = Map(Array(Double()))  # Per-field embeddings
+    object_id: str = ""  # Primary key value
+    field_embeddings: dict[str, list[float]] = field(default_factory=dict)  # Per-field embeddings
 
 ############################################################################

trustgraph-base/trustgraph/schema/knowledge/graph.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String, Array
+from dataclasses import dataclass, field
 
 from ..core.primitives import Value, Triple
 from ..core.metadata import Metadata
@@ -8,21 +8,24 @@ from ..core.topic import topic
 
 # Entity context are an entity associated with textual context
 
-class EntityContext(Record):
+@dataclass
-    entity = Value()
+class EntityContext:
-    context = String()
+    entity: Value | None = None
+    context: str = ""
 
 # This is a 'batching' mechanism for the above data
-class EntityContexts(Record):
+@dataclass
-    metadata = Metadata()
+class EntityContexts:
-    entities = Array(EntityContext())
+    metadata: Metadata | None = None
+    entities: list[EntityContext] = field(default_factory=list)
 
 ############################################################################
 
 # Graph triples
 
-class Triples(Record):
+@dataclass
-    metadata = Metadata()
+class Triples:
-    triples = Array(Triple())
+    metadata: Metadata | None = None
+    triples: list[Triple] = field(default_factory=list)
 
 ############################################################################

trustgraph-base/trustgraph/schema/knowledge/knowledge.py

@@ -1,5 +1,4 @@
-
+from dataclasses import dataclass, field
-from pulsar.schema import Record, Bytes, String, Array, Long, Boolean
 from ..core.primitives import Triple, Error
 from ..core.topic import topic
 from ..core.metadata import Metadata
@@ -22,40 +21,40 @@ from .embeddings import GraphEmbeddings
 #   <- ()
 #   <- (error)
 
-class KnowledgeRequest(Record):
+@dataclass
-
+class KnowledgeRequest:
     # get-kg-core, delete-kg-core, list-kg-cores, put-kg-core
     # load-kg-core, unload-kg-core
-    operation = String()
+    operation: str = ""
 
     # list-kg-cores, delete-kg-core, put-kg-core
-    user = String()
+    user: str = ""
 
     # get-kg-core, list-kg-cores, delete-kg-core, put-kg-core,
     # load-kg-core, unload-kg-core
-    id = String()
+    id: str = ""
 
     # load-kg-core
-    flow = String()
+    flow: str = ""
 
     # load-kg-core
-    collection = String()
+    collection: str = ""
 
     # put-kg-core
-    triples = Triples()
+    triples: Triples | None = None
-    graph_embeddings = GraphEmbeddings()
+    graph_embeddings: GraphEmbeddings | None = None
 
-class KnowledgeResponse(Record):
+@dataclass
-    error = Error()
+class KnowledgeResponse:
-    ids = Array(String())
+    error: Error | None = None
-    eos = Boolean()     # Indicates end of knowledge core stream
+    ids: list[str] = field(default_factory=list)
-    triples = Triples()
+    eos: bool = False     # Indicates end of knowledge core stream
-    graph_embeddings = GraphEmbeddings()
+    triples: Triples | None = None
+    graph_embeddings: GraphEmbeddings | None = None
 
 knowledge_request_queue = topic(
-    'knowledge', kind='non-persistent', namespace='request'
+    'knowledge', qos='q0', namespace='request'
 )
 knowledge_response_queue = topic(
-    'knowledge', kind='non-persistent', namespace='response',
+    'knowledge', qos='q0', namespace='response',
 )
-

trustgraph-base/trustgraph/schema/knowledge/nlp.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String, Boolean
+from dataclasses import dataclass
 
 from ..core.topic import topic
 
@@ -6,21 +6,25 @@ from ..core.topic import topic
 
 # NLP extraction data types
 
-class Definition(Record):
+@dataclass
-    name = String()
+class Definition:
-    definition = String()
+    name: str = ""
+    definition: str = ""
 
-class Topic(Record):
+@dataclass
-    name = String()
+class Topic:
-    definition = String()
+    name: str = ""
+    definition: str = ""
 
-class Relationship(Record):
+@dataclass
-    s = String()
+class Relationship:
-    p = String()
+    s: str = ""
-    o = String()
+    p: str = ""
-    o_entity = Boolean()
+    o: str = ""
+    o_entity: bool = False
 
-class Fact(Record):
+@dataclass
-    s = String()
+class Fact:
-    p = String()
+    s: str = ""
-    o = String()
+    p: str = ""
+    o: str = ""

trustgraph-base/trustgraph/schema/knowledge/object.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String, Map, Double, Array
+from dataclasses import dataclass, field
 
 from ..core.metadata import Metadata
 from ..core.topic import topic
@@ -7,11 +7,13 @@ from ..core.topic import topic
 
 # Extracted object from text processing
 
-class ExtractedObject(Record):
+@dataclass
-    metadata = Metadata()
+class ExtractedObject:
-    schema_name = String()  # Which schema this object belongs to
+    metadata: Metadata | None = None
-    values = Array(Map(String()))  # Array of objects, each object is field name -> value
+    schema_name: str = ""  # Which schema this object belongs to
-    confidence = Double()
+    values: list[dict[str, str]] = field(default_factory=list)  # Array of objects, each object is field name -> value
-    source_span = String()  # Text span where object was found
+    confidence: float = 0.0
+    source_span: str = ""  # Text span where object was found
 
 ############################################################################
+

trustgraph-base/trustgraph/schema/knowledge/rows.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, Array, Map, String
+from dataclasses import dataclass, field
 
 from ..core.metadata import Metadata
 from ..core.primitives import RowSchema
@@ -8,9 +8,10 @@ from ..core.topic import topic
 
 # Stores rows of information
 
-class Rows(Record):
+@dataclass
-    metadata = Metadata()
+class Rows:
-    row_schema = RowSchema()
+    metadata: Metadata | None = None
-    rows = Array(Map(String()))
+    row_schema: RowSchema | None = None
+    rows: list[dict[str, str]] = field(default_factory=list)
 
 ############################################################################

trustgraph-base/trustgraph/schema/knowledge/structured.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String, Bytes, Map
+from dataclasses import dataclass, field
 
 from ..core.metadata import Metadata
 from ..core.topic import topic
@@ -7,11 +7,13 @@ from ..core.topic import topic
 
 # Structured data submission for fire-and-forget processing
 
-class StructuredDataSubmission(Record):
+@dataclass
-    metadata = Metadata()
+class StructuredDataSubmission:
-    format = String()  # "json", "csv", "xml"
+    metadata: Metadata | None = None
-    schema_name = String()  # Reference to schema in config
+    format: str = ""  # "json", "csv", "xml"
-    data = Bytes()  # Raw data to ingest
+    schema_name: str = ""  # Reference to schema in config
-    options = Map(String())  # Format-specific options
+    data: bytes = b""  # Raw data to ingest
+    options: dict[str, str] = field(default_factory=dict)  # Format-specific options
 
 ############################################################################
+

trustgraph-base/trustgraph/schema/services/agent.py

@@ -1,5 +1,5 @@
 
-from pulsar.schema import Record, String, Array, Map, Boolean
+from dataclasses import dataclass, field
 
 from ..core.topic import topic
 from ..core.primitives import Error
@@ -8,33 +8,36 @@ from ..core.primitives import Error
 
 # Prompt services, abstract the prompt generation
 
-class AgentStep(Record):
+@dataclass
-    thought = String()
+class AgentStep:
-    action = String()
+    thought: str = ""
-    arguments = Map(String())
+    action: str = ""
-    observation = String()
+    arguments: dict[str, str] = field(default_factory=dict)
-    user = String()              # User context for the step
+    observation: str = ""
+    user: str = ""              # User context for the step
 
-class AgentRequest(Record):
+@dataclass
-    question = String()
+class AgentRequest:
-    state = String()
+    question: str = ""
-    group = Array(String())
+    state: str = ""
-    history = Array(AgentStep())
+    group: list[str] = field(default_factory=list)
-    user = String()              # User context for multi-tenancy
+    history: list['AgentStep'] = field(default_factory=list)
-    streaming = Boolean()        # NEW: Enable streaming response delivery (default false)
+    user: str = ""              # User context for multi-tenancy
+    streaming: bool = False     # NEW: Enable streaming response delivery (default false)
 
-class AgentResponse(Record):
+@dataclass
+class AgentResponse:
     # Streaming-first design
-    chunk_type = String()        # "thought", "action", "observation", "answer", "error"
+    chunk_type: str = ""        # "thought", "action", "observation", "answer", "error"
-    content = String()           # The actual content (interpretation depends on chunk_type)
+    content: str = ""           # The actual content (interpretation depends on chunk_type)
-    end_of_message = Boolean()   # Current chunk type (thought/action/etc.) is complete
+    end_of_message: bool = False   # Current chunk type (thought/action/etc.) is complete
-    end_of_dialog = Boolean()    # Entire agent dialog is complete
+    end_of_dialog: bool = False    # Entire agent dialog is complete
 
     # Legacy fields (deprecated but kept for backward compatibility)
-    answer = String()
+    answer: str = ""
-    error = Error()
+    error: Error | None = None
-    thought = String()
+    thought: str = ""
-    observation = String()
+    observation: str = ""
 
 ############################################################################
 

trustgraph-base/trustgraph/schema/services/collection.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String, Integer, Array
+from dataclasses import dataclass, field
 from datetime import datetime
 
 from ..core.primitives import Error
@@ -10,37 +10,40 @@ from ..core.topic import topic
 
 # Collection metadata operations (for librarian service)
 
-class CollectionMetadata(Record):
+@dataclass
+class CollectionMetadata:
     """Collection metadata record"""
-    user = String()
+    user: str = ""
-    collection = String()
+    collection: str = ""
-    name = String()
+    name: str = ""
-    description = String()
+    description: str = ""
-    tags = Array(String())
+    tags: list[str] = field(default_factory=list)
 
 ############################################################################
 
-class CollectionManagementRequest(Record):
+@dataclass
+class CollectionManagementRequest:
     """Request for collection management operations"""
-    operation = String()  # e.g., "delete-collection"
+    operation: str = ""  # e.g., "delete-collection"
 
     # For 'list-collections'
-    user = String()
+    user: str = ""
-    collection = String()
+    collection: str = ""
-    timestamp = String()  # ISO timestamp
+    timestamp: str = ""  # ISO timestamp
-    name = String()
+    name: str = ""
-    description = String()
+    description: str = ""
-    tags = Array(String())
+    tags: list[str] = field(default_factory=list)
 
     # For list
-    tag_filter = Array(String())  # Optional filter by tags
+    tag_filter: list[str] = field(default_factory=list)  # Optional filter by tags
-    limit = Integer()
+    limit: int = 0
 
-class CollectionManagementResponse(Record):
+@dataclass
+class CollectionManagementResponse:
     """Response for collection management operations"""
-    error = Error()  # Only populated if there's an error
+    error: Error | None = None  # Only populated if there's an error
-    timestamp = String()  # ISO timestamp
+    timestamp: str = ""  # ISO timestamp
-    collections = Array(CollectionMetadata())
+    collections: list[CollectionMetadata] = field(default_factory=list)
 
 
 ############################################################################
@@ -48,8 +51,9 @@ class CollectionManagementResponse(Record):
 # Topics
 
 collection_request_queue = topic(
-    'collection', kind='non-persistent', namespace='request'
+    'collection', qos='q0', namespace='request'
 )
 collection_response_queue = topic(
-    'collection', kind='non-persistent', namespace='response'
+    'collection', qos='q0', namespace='response'
 )
+

trustgraph-base/trustgraph/schema/services/config.py

@@ -1,5 +1,5 @@
 
-from pulsar.schema import Record, Bytes, String, Boolean, Array, Map, Integer
+from dataclasses import dataclass, field
 
 from ..core.topic import topic
 from ..core.primitives import Error
@@ -13,58 +13,61 @@ from ..core.primitives import Error
 #   put(values) -> ()
 #   delete(keys) -> ()
 #   config() -> (version, config)
-class ConfigKey(Record):
+@dataclass
-    type = String()
+class ConfigKey:
-    key = String()
+    type: str = ""
+    key: str = ""
 
-class ConfigValue(Record):
+@dataclass
-    type = String()
+class ConfigValue:
-    key = String()
+    type: str = ""
-    value = String()
+    key: str = ""
+    value: str = ""
 
 # Prompt services, abstract the prompt generation
-class ConfigRequest(Record):
+@dataclass
-
+class ConfigRequest:
-    operation = String() # get, list, getvalues, delete, put, config
+    operation: str = ""  # get, list, getvalues, delete, put, config
 
     # get, delete
-    keys = Array(ConfigKey())
+    keys: list[ConfigKey] = field(default_factory=list)
 
     # list, getvalues
-    type = String()
+    type: str = ""
 
     # put
-    values = Array(ConfigValue())
+    values: list[ConfigValue] = field(default_factory=list)
-
-class ConfigResponse(Record):
 
+@dataclass
+class ConfigResponse:
     # get, list, getvalues, config
-    version = Integer()
+    version: int = 0
 
     # get, getvalues
-    values = Array(ConfigValue())
+    values: list[ConfigValue] = field(default_factory=list)
 
     # list
-    directory = Array(String())
+    directory: list[str] = field(default_factory=list)
 
     # config
-    config = Map(Map(String()))
+    config: dict[str, dict[str, str]] = field(default_factory=dict)
 
     # Everything
-    error = Error()
+    error: Error | None = None
 
-class ConfigPush(Record):
+@dataclass
-    version = Integer()
+class ConfigPush:
-    config = Map(Map(String()))
+    version: int = 0
+    config: dict[str, dict[str, str]] = field(default_factory=dict)
 
 config_request_queue = topic(
-    'config', kind='non-persistent', namespace='request'
+    'config', qos='q0', namespace='request'
 )
 config_response_queue = topic(
-    'config', kind='non-persistent', namespace='response'
+    'config', qos='q0', namespace='response'
 )
 config_push_queue = topic(
-    'config', kind='persistent', namespace='config'
+    'config', qos='q2', namespace='config'
 )
 
 ############################################################################

trustgraph-base/trustgraph/schema/services/diagnosis.py

@@ -1,33 +1,36 @@
-from pulsar.schema import Record, String, Map, Double, Array
+from dataclasses import dataclass, field
 from ..core.primitives import Error
 
 ############################################################################
 
 # Structured data diagnosis services
 
-class StructuredDataDiagnosisRequest(Record):
+@dataclass
-    operation = String()  # "detect-type", "generate-descriptor", "diagnose", or "schema-selection"
+class StructuredDataDiagnosisRequest:
-    sample = String()     # Data sample to analyze (text content)
+    operation: str = ""  # "detect-type", "generate-descriptor", "diagnose", or "schema-selection"
-    type = String()       # Data type (csv, json, xml) - optional, required for generate-descriptor
+    sample: str = ""     # Data sample to analyze (text content)
-    schema_name = String() # Target schema name for descriptor generation - optional
+    type: str = ""       # Data type (csv, json, xml) - optional, required for generate-descriptor
+    schema_name: str = "" # Target schema name for descriptor generation - optional
 
     # JSON encoded options (e.g., delimiter for CSV)
-    options = Map(String())
+    options: dict[str, str] = field(default_factory=dict)
 
-class StructuredDataDiagnosisResponse(Record):
+@dataclass
-    error = Error()
+class StructuredDataDiagnosisResponse:
+    error: Error | None = None
 
-    operation = String()         # The operation that was performed
+    operation: str = ""         # The operation that was performed
-    detected_type = String()     # Detected data type (for detect-type/diagnose) - optional
+    detected_type: str = ""     # Detected data type (for detect-type/diagnose) - optional
-    confidence = Double()        # Confidence score for type detection - optional
+    confidence: float = 0.0     # Confidence score for type detection - optional
 
     # JSON encoded descriptor (for generate-descriptor/diagnose) - optional
-    descriptor = String()
+    descriptor: str = ""
 
     # JSON encoded additional metadata (e.g., field count, sample records)
-    metadata = Map(String())
+    metadata: dict[str, str] = field(default_factory=dict)
 
     # Array of matching schema IDs (for schema-selection operation) - optional
-    schema_matches = Array(String())
+    schema_matches: list[str] = field(default_factory=list)
 
 ############################################################################
+

trustgraph-base/trustgraph/schema/services/flow.py

@@ -1,5 +1,5 @@
 
-from pulsar.schema import Record, Bytes, String, Boolean, Array, Map, Integer
+from dataclasses import dataclass, field
 
 from ..core.topic import topic
 from ..core.primitives import Error
@@ -18,54 +18,54 @@ from ..core.primitives import Error
 #   stop_flow(flowid) -> ()
 
 # Prompt services, abstract the prompt generation
-class FlowRequest(Record):
+@dataclass
-
+class FlowRequest:
-    operation = String() # list-classes, get-class, put-class, delete-class
+    operation: str = ""  # list-classes, get-class, put-class, delete-class
                          # list-flows, get-flow, start-flow, stop-flow
 
     # get_class, put_class, delete_class, start_flow
-    class_name = String()
+    class_name: str = ""
 
     # put_class
-    class_definition = String()
+    class_definition: str = ""
 
     # start_flow
-    description = String()
+    description: str = ""
 
     # get_flow, start_flow, stop_flow
-    flow_id = String()
+    flow_id: str = ""
 
     # start_flow - optional parameters for flow customization
-    parameters = Map(String())
+    parameters: dict[str, str] = field(default_factory=dict)
-
-class FlowResponse(Record):
 
+@dataclass
+class FlowResponse:
     # list_classes
-    class_names = Array(String())
+    class_names: list[str] = field(default_factory=list)
 
     # list_flows
-    flow_ids = Array(String())
+    flow_ids: list[str] = field(default_factory=list)
 
     # get_class
-    class_definition = String()
+    class_definition: str = ""
 
     # get_flow
-    flow = String()
+    flow: str = ""
 
     # get_flow
-    description = String()
+    description: str = ""
 
     # get_flow - parameters used when flow was started
-    parameters = Map(String())
+    parameters: dict[str, str] = field(default_factory=dict)
 
     # Everything
-    error = Error()
+    error: Error | None = None
 
 flow_request_queue = topic(
-    'flow', kind='non-persistent', namespace='request'
+    'flow', qos='q0', namespace='request'
 )
 flow_response_queue = topic(
-    'flow', kind='non-persistent', namespace='response'
+    'flow', qos='q0', namespace='response'
 )
 
 ############################################################################

trustgraph-base/trustgraph/schema/services/library.py

@@ -1,9 +1,8 @@
-
+from dataclasses import dataclass, field
-from pulsar.schema import Record, Bytes, String, Array, Long
 from ..core.primitives import Triple, Error
 from ..core.topic import topic
 from ..core.metadata import Metadata
-from ..knowledge.document import Document, TextDocument
+# Note: Document imports will be updated after knowledge schemas are converted
 
 # add-document
 #   -> (document_id, document_metadata, content)
@@ -50,76 +49,79 @@ from ..knowledge.document import Document, TextDocument
 #   <- (processing_metadata[])
 #   <- (error)
 
-class DocumentMetadata(Record):
+@dataclass
-    id = String()
+class DocumentMetadata:
-    time = Long()
+    id: str = ""
-    kind = String()
+    time: int = 0
-    title = String()
+    kind: str = ""
-    comments = String()
+    title: str = ""
-    metadata = Array(Triple())
+    comments: str = ""
-    user = String()
+    metadata: list[Triple] = field(default_factory=list)
-    tags = Array(String())
+    user: str = ""
+    tags: list[str] = field(default_factory=list)
 
-class ProcessingMetadata(Record):
+@dataclass
-    id = String()
+class ProcessingMetadata:
-    document_id = String()
+    id: str = ""
-    time = Long()
+    document_id: str = ""
-    flow = String()
+    time: int = 0
-    user = String()
+    flow: str = ""
-    collection = String()
+    user: str = ""
-    tags = Array(String())
+    collection: str = ""
+    tags: list[str] = field(default_factory=list)
 
-class Criteria(Record):
+@dataclass
-    key = String()
+class Criteria:
-    value = String()
+    key: str = ""
-    operator = String()
+    value: str = ""
-
+    operator: str = ""
-class LibrarianRequest(Record):
 
+@dataclass
+class LibrarianRequest:
     # add-document, remove-document, update-document, get-document-metadata,
     # get-document-content, add-processing, remove-processing, list-documents,
     # list-processing
-    operation = String()
+    operation: str = ""
 
     # add-document, remove-document, update-document, get-document-metadata,
     # get-document-content
-    document_id = String()
+    document_id: str = ""
 
     # add-processing, remove-processing
-    processing_id = String()
+    processing_id: str = ""
 
     # add-document, update-document
-    document_metadata = DocumentMetadata()
+    document_metadata: DocumentMetadata | None = None
 
     # add-processing
-    processing_metadata = ProcessingMetadata()
+    processing_metadata: ProcessingMetadata | None = None
 
     # add-document
-    content = Bytes()
+    content: bytes = b""
 
     # list-documents, list-processing
-    user = String()
+    user: str = ""
 
     # list-documents?, list-processing?
-    collection = String()
+    collection: str = ""
 
     #
-    criteria = Array(Criteria())
+    criteria: list[Criteria] = field(default_factory=list)
 
-class LibrarianResponse(Record):
+@dataclass
-    error = Error()
+class LibrarianResponse:
-    document_metadata = DocumentMetadata()
+    error: Error | None = None
-    content = Bytes()
+    document_metadata: DocumentMetadata | None = None
-    document_metadatas = Array(DocumentMetadata())
+    content: bytes = b""
-    processing_metadatas = Array(ProcessingMetadata())
+    document_metadatas: list[DocumentMetadata] = field(default_factory=list)
+    processing_metadatas: list[ProcessingMetadata] = field(default_factory=list)
 
 # FIXME: Is this right?  Using persistence on librarian so that
 # message chunking works
 
 librarian_request_queue = topic(
-    'librarian', kind='persistent', namespace='request'
+    'librarian', qos='q1', namespace='request'
 )
 librarian_response_queue = topic(
-    'librarian', kind='persistent', namespace='response',
+    'librarian', qos='q1', namespace='response',
 )
-

trustgraph-base/trustgraph/schema/services/llm.py

@@ -1,5 +1,5 @@
 
-from pulsar.schema import Record, String, Array, Double, Integer, Boolean
+from dataclasses import dataclass, field
 
 from ..core.topic import topic
 from ..core.primitives import Error
@@ -8,46 +8,49 @@ from ..core.primitives import Error
 
 # LLM text completion
 
-class TextCompletionRequest(Record):
+@dataclass
-    system = String()
+class TextCompletionRequest:
-    prompt = String()
+    system: str = ""
-    streaming = Boolean()  # Default false for backward compatibility
+    prompt: str = ""
+    streaming: bool = False  # Default false for backward compatibility
 
-class TextCompletionResponse(Record):
+@dataclass
-    error = Error()
+class TextCompletionResponse:
-    response = String()
+    error: Error | None = None
-    in_token = Integer()
+    response: str = ""
-    out_token = Integer()
+    in_token: int = 0
-    model = String()
+    out_token: int = 0
-    end_of_stream = Boolean()  # Indicates final message in stream
+    model: str = ""
+    end_of_stream: bool = False  # Indicates final message in stream
 
 ############################################################################
 
 # Embeddings
 
-class EmbeddingsRequest(Record):
+@dataclass
-    text = String()
+class EmbeddingsRequest:
+    text: str = ""
 
-class EmbeddingsResponse(Record):
+@dataclass
-    error = Error()
+class EmbeddingsResponse:
-    vectors = Array(Array(Double()))
+    error: Error | None = None
+    vectors: list[list[float]] = field(default_factory=list)
 
 ############################################################################
 
 # Tool request/response
 
-class ToolRequest(Record):
+@dataclass
-    name = String()
+class ToolRequest:
-
+    name: str = ""
     # Parameters are JSON encoded
-    parameters = String()
+    parameters: str = ""
-
-class ToolResponse(Record):
-    error = Error()
 
+@dataclass
+class ToolResponse:
+    error: Error | None = None
     # Plain text aka "unstructured"
-    text = String()
+    text: str = ""
-
     # JSON-encoded object aka "structured"
-    object = String()
+    object: str = ""
 

trustgraph-base/trustgraph/schema/services/lookup.py

@@ -1,5 +1,4 @@
-
+from dataclasses import dataclass
-from pulsar.schema import Record, String
 
 from ..core.primitives import Error, Value, Triple
 from ..core.topic import topic
@@ -9,13 +8,14 @@ from ..core.metadata import Metadata
 
 # Lookups
 
-class LookupRequest(Record):
+@dataclass
-    kind = String()
+class LookupRequest:
-    term = String()
+    kind: str = ""
+    term: str = ""
 
-class LookupResponse(Record):
+@dataclass
-    text = String()
+class LookupResponse:
-    error = Error()
+    text: str = ""
+    error: Error | None = None
 
 ############################################################################
-

trustgraph-base/trustgraph/schema/services/nlp_query.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String, Array, Map, Integer, Double
+from dataclasses import dataclass, field
 
 from ..core.primitives import Error
 from ..core.topic import topic
@@ -7,15 +7,18 @@ from ..core.topic import topic
 
 # NLP to Structured Query Service - converts natural language to GraphQL
 
-class QuestionToStructuredQueryRequest(Record):
+@dataclass
-    question = String()
+class QuestionToStructuredQueryRequest:
-    max_results = Integer()
+    question: str = ""
+    max_results: int = 0
 
-class QuestionToStructuredQueryResponse(Record):
+@dataclass
-    error = Error()
+class QuestionToStructuredQueryResponse:
-    graphql_query = String()  # Generated GraphQL query
+    error: Error | None = None
-    variables = Map(String())  # GraphQL variables if any
+    graphql_query: str = ""  # Generated GraphQL query
-    detected_schemas = Array(String())  # Which schemas the query targets
+    variables: dict[str, str] = field(default_factory=dict)  # GraphQL variables if any
-    confidence = Double()
+    detected_schemas: list[str] = field(default_factory=list)  # Which schemas the query targets
+    confidence: float = 0.0
 
 ############################################################################
+

trustgraph-base/trustgraph/schema/services/objects_query.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String, Map, Array
+from dataclasses import dataclass, field
 
 from ..core.primitives import Error
 from ..core.topic import topic
@@ -7,22 +7,25 @@ from ..core.topic import topic
 
 # Objects Query Service - executes GraphQL queries against structured data
 
-class GraphQLError(Record):
+@dataclass
-    message = String()
+class GraphQLError:
-    path = Array(String())       # Path to the field that caused the error
+    message: str = ""
-    extensions = Map(String())   # Additional error metadata
+    path: list[str] = field(default_factory=list)       # Path to the field that caused the error
+    extensions: dict[str, str] = field(default_factory=dict)   # Additional error metadata
 
-class ObjectsQueryRequest(Record):
+@dataclass
-    user = String()              # Cassandra keyspace (follows pattern from TriplesQueryRequest)
+class ObjectsQueryRequest:
-    collection = String()        # Data collection identifier (required for partition key)
+    user: str = ""              # Cassandra keyspace (follows pattern from TriplesQueryRequest)
-    query = String()             # GraphQL query string
+    collection: str = ""        # Data collection identifier (required for partition key)
-    variables = Map(String())    # GraphQL variables
+    query: str = ""             # GraphQL query string
-    operation_name = String()    # Operation to execute for multi-operation documents
+    variables: dict[str, str] = field(default_factory=dict)    # GraphQL variables
+    operation_name: str = ""    # Operation to execute for multi-operation documents
 
-class ObjectsQueryResponse(Record):
+@dataclass
-    error = Error()              # System-level error (connection, timeout, etc.)
+class ObjectsQueryResponse:
-    data = String()              # JSON-encoded GraphQL response data
+    error: Error | None = None              # System-level error (connection, timeout, etc.)
-    errors = Array(GraphQLError()) # GraphQL field-level errors
+    data: str = ""              # JSON-encoded GraphQL response data
-    extensions = Map(String())   # Query metadata (execution time, etc.)
+    errors: list[GraphQLError] = field(default_factory=list) # GraphQL field-level errors
+    extensions: dict[str, str] = field(default_factory=dict)   # Query metadata (execution time, etc.)
 
 ############################################################################

trustgraph-base/trustgraph/schema/services/prompt.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String, Map, Boolean
+from dataclasses import dataclass, field
 
 from ..core.primitives import Error
 from ..core.topic import topic
@@ -18,27 +18,28 @@ from ..core.topic import topic
 # extract-rows
 #   schema, chunk -> rows
 
-class PromptRequest(Record):
+@dataclass
-    id = String()
+class PromptRequest:
+    id: str = ""
 
     # JSON encoded values
-    terms = Map(String())
+    terms: dict[str, str] = field(default_factory=dict)
 
     # Streaming support (default false for backward compatibility)
-    streaming = Boolean()
+    streaming: bool = False
-
-class PromptResponse(Record):
 
+@dataclass
+class PromptResponse:
     # Error case
-    error = Error()
+    error: Error | None = None
 
     # Just plain text
-    text = String()
+    text: str = ""
 
     # JSON encoded
-    object = String()
+    object: str = ""
 
     # Indicates final message in stream
-    end_of_stream = Boolean()
+    end_of_stream: bool = False
 
 ############################################################################

trustgraph-base/trustgraph/schema/services/query.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String, Integer, Array, Double
+from dataclasses import dataclass, field
 
 from ..core.primitives import Error, Value, Triple
 from ..core.topic import topic
@@ -7,49 +7,55 @@ from ..core.topic import topic
 
 # Graph embeddings query
 
-class GraphEmbeddingsRequest(Record):
+@dataclass
-    vectors = Array(Array(Double()))
+class GraphEmbeddingsRequest:
-    limit = Integer()
+    vectors: list[list[float]] = field(default_factory=list)
-    user = String()
+    limit: int = 0
-    collection = String()
+    user: str = ""
+    collection: str = ""
 
-class GraphEmbeddingsResponse(Record):
+@dataclass
-    error = Error()
+class GraphEmbeddingsResponse:
-    entities = Array(Value())
+    error: Error | None = None
+    entities: list[Value] = field(default_factory=list)
 
 ############################################################################
 
 # Graph triples query
 
-class TriplesQueryRequest(Record):
+@dataclass
-    user = String()
+class TriplesQueryRequest:
-    collection = String()
+    user: str = ""
-    s = Value()
+    collection: str = ""
-    p = Value()
+    s: Value | None = None
-    o = Value()
+    p: Value | None = None
-    limit = Integer()
+    o: Value | None = None
+    limit: int = 0
 
-class TriplesQueryResponse(Record):
+@dataclass
-    error = Error()
+class TriplesQueryResponse:
-    triples = Array(Triple())
+    error: Error | None = None
+    triples: list[Triple] = field(default_factory=list)
 
 ############################################################################
 
 # Doc embeddings query
 
-class DocumentEmbeddingsRequest(Record):
+@dataclass
-    vectors = Array(Array(Double()))
+class DocumentEmbeddingsRequest:
-    limit = Integer()
+    vectors: list[list[float]] = field(default_factory=list)
-    user = String()
+    limit: int = 0
-    collection = String()
+    user: str = ""
+    collection: str = ""
 
-class DocumentEmbeddingsResponse(Record):
+@dataclass
-    error = Error()
+class DocumentEmbeddingsResponse:
-    chunks = Array(String())
+    error: Error | None = None
+    chunks: list[str] = field(default_factory=list)
 
 document_embeddings_request_queue = topic(
-    "non-persistent://trustgraph/document-embeddings-request"
+    "document-embeddings-request", qos='q0', tenant='trustgraph', namespace='flow'
 )
 document_embeddings_response_queue = topic(
-    "non-persistent://trustgraph/document-embeddings-response"
+    "document-embeddings-response", qos='q0', tenant='trustgraph', namespace='flow'
 )

trustgraph-base/trustgraph/schema/services/retrieval.py

@@ -1,5 +1,4 @@
-
+from dataclasses import dataclass
-from pulsar.schema import Record, Bytes, String, Boolean, Integer, Array, Double
 from ..core.topic import topic
 from ..core.primitives import Error, Value
 
@@ -7,36 +6,39 @@ from ..core.primitives import Error, Value
 
 # Graph RAG text retrieval
 
-class GraphRagQuery(Record):
+@dataclass
-    query = String()
+class GraphRagQuery:
-    user = String()
+    query: str = ""
-    collection = String()
+    user: str = ""
-    entity_limit = Integer()
+    collection: str = ""
-    triple_limit = Integer()
+    entity_limit: int = 0
-    max_subgraph_size = Integer()
+    triple_limit: int = 0
-    max_path_length = Integer()
+    max_subgraph_size: int = 0
-    streaming = Boolean()
+    max_path_length: int = 0
+    streaming: bool = False
 
-class GraphRagResponse(Record):
+@dataclass
-    error = Error()
+class GraphRagResponse:
-    response = String()
+    error: Error | None = None
-    chunk = String()
+    response: str = ""
-    end_of_stream = Boolean()
+    chunk: str = ""
+    end_of_stream: bool = False
 
 ############################################################################
 
 # Document RAG text retrieval
 
-class DocumentRagQuery(Record):
+@dataclass
-    query = String()
+class DocumentRagQuery:
-    user = String()
+    query: str = ""
-    collection = String()
+    user: str = ""
-    doc_limit = Integer()
+    collection: str = ""
-    streaming = Boolean()
+    doc_limit: int = 0
-
+    streaming: bool = False
-class DocumentRagResponse(Record):
-    error = Error()
-    response = String()
-    chunk = String()
-    end_of_stream = Boolean()
 
+@dataclass
+class DocumentRagResponse:
+    error: Error | None = None
+    response: str = ""
+    chunk: str = ""
+    end_of_stream: bool = False

trustgraph-base/trustgraph/schema/services/storage.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String
+from dataclasses import dataclass
 
 from ..core.primitives import Error
 from ..core.topic import topic
@@ -7,15 +7,17 @@ from ..core.topic import topic
 
 # Storage management operations
 
-class StorageManagementRequest(Record):
+@dataclass
+class StorageManagementRequest:
     """Request for storage management operations sent to store processors"""
-    operation = String()  # e.g., "delete-collection"
+    operation: str = ""  # e.g., "delete-collection"
-    user = String()
+    user: str = ""
-    collection = String()
+    collection: str = ""
 
-class StorageManagementResponse(Record):
+@dataclass
+class StorageManagementResponse:
     """Response from storage processors for management operations"""
-    error = Error()  # Only populated if there's an error, if null success
+    error: Error | None = None  # Only populated if there's an error, if null success
 
 ############################################################################
 
@@ -23,20 +25,21 @@ class StorageManagementResponse(Record):
 
 # Topics for sending collection management requests to different storage types
 vector_storage_management_topic = topic(
-    'vector-storage-management', kind='non-persistent', namespace='request'
+    'vector-storage-management', qos='q0', namespace='request'
 )
 
 object_storage_management_topic = topic(
-    'object-storage-management', kind='non-persistent', namespace='request'
+    'object-storage-management', qos='q0', namespace='request'
 )
 
 triples_storage_management_topic = topic(
-    'triples-storage-management', kind='non-persistent', namespace='request'
+    'triples-storage-management', qos='q0', namespace='request'
 )
 
 # Topic for receiving responses from storage processors
 storage_management_response_topic = topic(
-    'storage-management', kind='non-persistent', namespace='response'
+    'storage-management', qos='q0', namespace='response'
 )
 
 ############################################################################
+

trustgraph-base/trustgraph/schema/services/structured_query.py

@@ -1,4 +1,4 @@
-from pulsar.schema import Record, String, Map, Array
+from dataclasses import dataclass, field
 
 from ..core.primitives import Error
 from ..core.topic import topic
@@ -7,14 +7,17 @@ from ..core.topic import topic
 
 # Structured Query Service - executes GraphQL queries
 
-class StructuredQueryRequest(Record):
+@dataclass
-    question = String()
+class StructuredQueryRequest:
-    user = String()        # Cassandra keyspace identifier
+    question: str = ""
-    collection = String()  # Data collection identifier
+    user: str = ""        # Cassandra keyspace identifier
+    collection: str = ""  # Data collection identifier
 
-class StructuredQueryResponse(Record):
+@dataclass
-    error = Error()
+class StructuredQueryResponse:
-    data = String()  # JSON-encoded GraphQL response data
+    error: Error | None = None
-    errors = Array(String())  # GraphQL errors if any
+    data: str = ""  # JSON-encoded GraphQL response data
+    errors: list[str] = field(default_factory=list)  # GraphQL errors if any
 
 ############################################################################
+