Release/v2.3 -> master

trustgraph-base/trustgraph/base/__init__.py

@@ -5,7 +5,7 @@ from . consumer import Consumer
 from . producer import Producer
 from . publisher import Publisher
 from . subscriber import Subscriber
-from . metrics import ProcessorMetrics, ConsumerMetrics, ProducerMetrics
+from . metrics import ProcessorMetrics, ConsumerMetrics, ProducerMetrics, SubscriberMetrics
 from . logging import add_logging_args, setup_logging
 from . flow_processor import FlowProcessor
 from . consumer_spec import ConsumerSpec
@@ -22,6 +22,7 @@ from . text_completion_client import (
     TextCompletionClientSpec, TextCompletionClient, TextCompletionResult,
 )
 from . prompt_client import PromptClientSpec, PromptClient, PromptResult
+from . config_client import ConfigClientSpec, ConfigClient
 from . triples_store_service import TriplesStoreService
 from . graph_embeddings_store_service import GraphEmbeddingsStoreService
 from . document_embeddings_store_service import DocumentEmbeddingsStoreService

trustgraph-base/trustgraph/base/backend.py

@@ -159,6 +159,62 @@ class PubSubBackend(Protocol):
         """
         ...
 
+    async def create_queue(self, topic: str, subscription: str) -> None:
+        """
+        Pre-create a queue so it exists before any consumer connects.
+
+        The topic and subscription together identify the queue, mirroring
+        create_consumer where the queue name is derived from both.
+
+        Idempotent — creating an already-existing queue succeeds silently.
+
+        Args:
+            topic: Queue identifier in class:topicspace:topic format
+            subscription: Subscription/consumer group name
+        """
+        ...
+
+    async def delete_queue(self, topic: str, subscription: str) -> None:
+        """
+        Delete a queue and any messages it contains.
+
+        The topic and subscription together identify the queue, mirroring
+        create_consumer where the queue name is derived from both.
+
+        Idempotent — deleting a non-existent queue succeeds silently.
+
+        Args:
+            topic: Queue identifier in class:topicspace:topic format
+            subscription: Subscription/consumer group name
+        """
+        ...
+
+    async def queue_exists(self, topic: str, subscription: str) -> bool:
+        """
+        Check whether a queue exists.
+
+        Args:
+            topic: Queue identifier in class:topicspace:topic format
+            subscription: Subscription/consumer group name
+
+        Returns:
+            True if the queue exists, False otherwise.
+        """
+        ...
+
+    async def ensure_queue(self, topic: str, subscription: str) -> None:
+        """
+        Ensure a queue exists, creating it if necessary.
+
+        Convenience wrapper — checks existence, creates if missing.
+        Used by system services on startup.
+
+        Args:
+            topic: Queue identifier in class:topicspace:topic format
+            subscription: Subscription/consumer group name
+        """
+        ...
+
     def close(self) -> None:
         """Close the backend connection."""
         ...

trustgraph-base/trustgraph/base/config_client.py

@@ -0,0 +1,92 @@
+
+from . request_response_spec import RequestResponse, RequestResponseSpec
+from .. schema import ConfigRequest, ConfigResponse, ConfigKey, ConfigValue
+
+CONFIG_TIMEOUT = 10
+
+
+class ConfigClient(RequestResponse):
+
+    async def _request(self, timeout=CONFIG_TIMEOUT, **kwargs):
+        resp = await self.request(
+            ConfigRequest(**kwargs),
+            timeout=timeout,
+        )
+        if resp.error:
+            raise RuntimeError(
+                f"{resp.error.type}: {resp.error.message}"
+            )
+        return resp
+
+    async def get(self, type, key, timeout=CONFIG_TIMEOUT):
+        """Get a single config value. Returns the value string or None."""
+        resp = await self._request(
+            operation="get",
+            keys=[ConfigKey(type=type, key=key)],
+            timeout=timeout,
+        )
+        if resp.values and len(resp.values) > 0:
+            return resp.values[0].value
+        return None
+
+    async def put(self, type, key, value, timeout=CONFIG_TIMEOUT):
+        """Put a single config value."""
+        await self._request(
+            operation="put",
+            values=[ConfigValue(type=type, key=key, value=value)],
+            timeout=timeout,
+        )
+
+    async def put_many(self, values, timeout=CONFIG_TIMEOUT):
+        """Put multiple config values in a single request.
+        values is a list of (type, key, value) tuples."""
+        await self._request(
+            operation="put",
+            values=[
+                ConfigValue(type=t, key=k, value=v)
+                for t, k, v in values
+            ],
+            timeout=timeout,
+        )
+
+    async def delete(self, type, key, timeout=CONFIG_TIMEOUT):
+        """Delete a single config key."""
+        await self._request(
+            operation="delete",
+            keys=[ConfigKey(type=type, key=key)],
+            timeout=timeout,
+        )
+
+    async def delete_many(self, keys, timeout=CONFIG_TIMEOUT):
+        """Delete multiple config keys in a single request.
+        keys is a list of (type, key) tuples."""
+        await self._request(
+            operation="delete",
+            keys=[
+                ConfigKey(type=t, key=k)
+                for t, k in keys
+            ],
+            timeout=timeout,
+        )
+
+    async def keys(self, type, timeout=CONFIG_TIMEOUT):
+        """List all keys for a config type."""
+        resp = await self._request(
+            operation="list",
+            type=type,
+            timeout=timeout,
+        )
+        return resp.directory
+
+
+class ConfigClientSpec(RequestResponseSpec):
+    def __init__(
+            self, request_name, response_name,
+    ):
+        super(ConfigClientSpec, self).__init__(
+            request_name=request_name,
+            request_schema=ConfigRequest,
+            response_name=response_name,
+            response_schema=ConfigResponse,
+            impl=ConfigClient,
+        )

trustgraph-base/trustgraph/base/consumer_spec.py

@@ -23,7 +23,7 @@ class ConsumerSpec(Spec):
             taskgroup = processor.taskgroup,
             flow = flow,
             backend = processor.pubsub,
-            topic = definition[self.name],
+            topic = definition["topics"][self.name],
             subscriber = processor.id + "--" + flow.name + "--" + self.name,
             schema = self.schema,
             handler = self.handler,

trustgraph-base/trustgraph/base/flow_processor.py

@@ -29,9 +29,9 @@ class FlowProcessor(AsyncProcessor):
         # Initialise base class
         super(FlowProcessor, self).__init__(**params)
 
-        # Register configuration handler
+        # Register configuration handler for this processor's config type
         self.register_config_handler(
-            self.on_configure_flows, types=["active-flow"]
+            self.on_configure_flows, types=[f"processor:{self.id}"]
         )
 
         # Initialise flow information state
@@ -66,17 +66,16 @@ class FlowProcessor(AsyncProcessor):
 
         logger.info(f"Got config version {version}")
 
-        # Skip over invalid data
-        if "active-flow" not in config: return
-
-        # Check there's configuration information for me
-        if self.id in config["active-flow"]:
-
-            # Get my flow config
-            flow_config = json.loads(config["active-flow"][self.id])
+        config_type = f"processor:{self.id}"
 
+        # Get my flow config — each key is a variant, each value is
+        # the JSON config for that flow variant
+        if config_type in config:
+            flow_config = {
+                k: json.loads(v)
+                for k, v in config[config_type].items()
+            }
         else:
-
             logger.debug("No configuration settings for me.")
             flow_config = {}
 

trustgraph-base/trustgraph/base/parameter_spec.py

@@ -18,6 +18,6 @@ class ParameterSpec(Spec):
 
     def add(self, flow: Any, processor: Any, definition: dict[str, Any]) -> None:
 
-        value = definition.get(self.name, None)
+        value = definition.get("parameters", {}).get(self.name, None)
 
         flow.parameter[self.name] = Parameter(value)

trustgraph-base/trustgraph/base/producer_spec.py

@@ -19,7 +19,7 @@ class ProducerSpec(Spec):
 
         producer = Producer(
             backend = processor.pubsub,
-            topic = definition[self.name],
+            topic = definition["topics"][self.name],
             schema = self.schema,
             metrics = producer_metrics,
         )

trustgraph-base/trustgraph/base/pulsar_backend.py

@@ -266,6 +266,26 @@ class PulsarBackend:
 
         return PulsarBackendConsumer(pulsar_consumer, schema)
 
+    async def create_queue(self, topic: str, subscription: str) -> None:
+        """No-op — Pulsar auto-creates topics on first use.
+        TODO: Use admin REST API for explicit persistent topic creation."""
+        pass
+
+    async def delete_queue(self, topic: str, subscription: str) -> None:
+        """No-op — to be replaced with admin REST API calls.
+        TODO: Delete subscription and persistent topic via admin API."""
+        pass
+
+    async def queue_exists(self, topic: str, subscription: str) -> bool:
+        """Returns True — Pulsar auto-creates on subscribe.
+        TODO: Use admin REST API for actual existence check."""
+        return True
+
+    async def ensure_queue(self, topic: str, subscription: str) -> None:
+        """No-op — Pulsar auto-creates topics on first use.
+        TODO: Use admin REST API for explicit creation."""
+        pass
+
     def close(self) -> None:
         """Close the Pulsar client."""
         self.client.close()

trustgraph-base/trustgraph/base/rabbitmq_backend.py

@@ -19,6 +19,7 @@ Uses basic_consume (push) instead of basic_get (polling) for
 efficient message delivery.
 """
 
+import asyncio
 import json
 import time
 import logging
@@ -170,28 +171,37 @@ class RabbitMQBackendConsumer:
         self._connection = pika.BlockingConnection(self._connection_params)
         self._channel = self._connection.channel()
 
-        # Declare the topic exchange
+        # Declare the topic exchange (idempotent, also done by producers)
         self._channel.exchange_declare(
             exchange=self._exchange_name,
             exchange_type='topic',
             durable=True,
         )
 
-        # Declare the queue — anonymous if exclusive
-        result = self._channel.queue_declare(
-            queue=self._queue_name,
-            durable=self._durable,
-            exclusive=self._exclusive,
-            auto_delete=self._auto_delete,
-        )
-        # Capture actual name (important for anonymous queues where name='')
-        self._queue_name = result.method.queue
+        if self._exclusive:
+            # Anonymous ephemeral queue (response/notify class).
+            # These are per-consumer and must be created here — the
+            # broker assigns the name.
+            result = self._channel.queue_declare(
+                queue='',
+                durable=False,
+                exclusive=True,
+                auto_delete=True,
+            )
+            self._queue_name = result.method.queue
 
-        self._channel.queue_bind(
-            queue=self._queue_name,
-            exchange=self._exchange_name,
-            routing_key=self._routing_key,
-        )
+            self._channel.queue_bind(
+                queue=self._queue_name,
+                exchange=self._exchange_name,
+                routing_key=self._routing_key,
+            )
+        else:
+            # Named queue (flow/request class). Queue must already
+            # exist — created by the flow service or ensure_queue.
+            # We just verify it exists and bind to consume.
+            self._channel.queue_declare(
+                queue=self._queue_name, passive=True,
+            )
 
         self._channel.basic_qos(prefetch_count=1)
 
@@ -409,5 +419,124 @@ class RabbitMQBackend:
             queue_name, schema, queue_durable, exclusive, auto_delete,
         )
 
+    def _create_queue_sync(self, exchange, routing_key, queue_name, durable):
+        """Blocking queue creation — run via asyncio.to_thread."""
+        connection = None
+        try:
+            connection = pika.BlockingConnection(self._connection_params)
+            channel = connection.channel()
+            channel.exchange_declare(
+                exchange=exchange,
+                exchange_type='topic',
+                durable=True,
+            )
+            channel.queue_declare(
+                queue=queue_name,
+                durable=durable,
+                exclusive=False,
+                auto_delete=False,
+            )
+            channel.queue_bind(
+                queue=queue_name,
+                exchange=exchange,
+                routing_key=routing_key,
+            )
+            logger.info(f"Created queue: {queue_name}")
+        finally:
+            if connection and connection.is_open:
+                try:
+                    connection.close()
+                except Exception:
+                    pass
+
+    async def create_queue(self, topic: str, subscription: str) -> None:
+        """Pre-create a named queue bound to the topic exchange.
+
+        Only applies to shared queues (flow/request class). Response and
+        notify queues are anonymous/auto-delete and created by consumers.
+        """
+        exchange, routing_key, cls, durable = self._parse_queue_id(topic)
+
+        if cls in ('response', 'notify'):
+            return
+
+        queue_name = f"{exchange}.{routing_key}.{subscription}"
+        await asyncio.to_thread(
+            self._create_queue_sync, exchange, routing_key,
+            queue_name, durable,
+        )
+
+    def _delete_queue_sync(self, queue_name):
+        """Blocking queue deletion — run via asyncio.to_thread."""
+        connection = None
+        try:
+            connection = pika.BlockingConnection(self._connection_params)
+            channel = connection.channel()
+            channel.queue_delete(queue=queue_name)
+            logger.info(f"Deleted queue: {queue_name}")
+        except Exception as e:
+            # Idempotent — queue may already be gone
+            logger.debug(f"Queue delete for {queue_name}: {e}")
+        finally:
+            if connection and connection.is_open:
+                try:
+                    connection.close()
+                except Exception:
+                    pass
+
+    async def delete_queue(self, topic: str, subscription: str) -> None:
+        """Delete a named queue and any messages it contains.
+
+        Only applies to shared queues (flow/request class). Response and
+        notify queues are anonymous/auto-delete and managed by the broker.
+        """
+        exchange, routing_key, cls, durable = self._parse_queue_id(topic)
+
+        if cls in ('response', 'notify'):
+            return
+
+        queue_name = f"{exchange}.{routing_key}.{subscription}"
+        await asyncio.to_thread(self._delete_queue_sync, queue_name)
+
+    def _queue_exists_sync(self, queue_name):
+        """Blocking queue existence check — run via asyncio.to_thread.
+        Uses passive=True which checks without creating."""
+        connection = None
+        try:
+            connection = pika.BlockingConnection(self._connection_params)
+            channel = connection.channel()
+            channel.queue_declare(queue=queue_name, passive=True)
+            return True
+        except pika.exceptions.ChannelClosedByBroker:
+            # 404 NOT_FOUND — queue does not exist
+            return False
+        finally:
+            if connection and connection.is_open:
+                try:
+                    connection.close()
+                except Exception:
+                    pass
+
+    async def queue_exists(self, topic: str, subscription: str) -> bool:
+        """Check whether a named queue exists.
+
+        Only applies to shared queues (flow/request class). Response and
+        notify queues are anonymous/ephemeral — always returns False.
+        """
+        exchange, routing_key, cls, durable = self._parse_queue_id(topic)
+
+        if cls in ('response', 'notify'):
+            return False
+
+        queue_name = f"{exchange}.{routing_key}.{subscription}"
+        return await asyncio.to_thread(
+            self._queue_exists_sync, queue_name
+        )
+
+    async def ensure_queue(self, topic: str, subscription: str) -> None:
+        """Ensure a queue exists, creating it if necessary."""
+        if not await self.queue_exists(topic, subscription):
+            await self.create_queue(topic, subscription)
+
     def close(self) -> None:
         pass

trustgraph-base/trustgraph/base/request_response_spec.py

@@ -137,10 +137,10 @@ class RequestResponseSpec(Spec):
                 "--" + str(uuid.uuid4())
             ),
             consumer_name = flow.id,
-            request_topic = definition[self.request_name],
+            request_topic = definition["topics"][self.request_name],
             request_schema = self.request_schema,
             request_metrics = request_metrics,
-            response_topic = definition[self.response_name],
+            response_topic = definition["topics"][self.response_name],
             response_schema = self.response_schema,
             response_metrics = response_metrics,
         )

trustgraph-base/trustgraph/base/subscriber_spec.py

@@ -20,7 +20,7 @@ class SubscriberSpec(Spec):
 
         subscriber = Subscriber(
             backend = processor.pubsub,
-            topic = definition[self.name],
+            topic = definition["topics"][self.name],
             subscription = flow.id,
             consumer_name = flow.id,
             schema = self.schema,