feat: replace REST API with gRPC service

Complete migration from FastAPI/HTTP to gRPC: Proto definition (proto/daena/v1/daena.proto): - Daena service with 9 RPCs: GetStatus, GetHealth, GetConfig, ListPlugins, GetQueueState, ListPendingRecords, ListDeadRecords, ListSinks, ReloadConfig - Messages for all request/response types with proper field types gRPC server (src/daena/api/grpc_server.py): - Async gRPC aio server with DaenaServicer implementation - DaenaGRPCServer with start/stop/serve_forever lifecycle - All endpoints delegate to DaenaRuntime gRPC client (src/daena/cli/main.py): - CLI now uses grpc.insecure_channel + DaenaStub - Same commands: status, health, queue, sinks, plugins, config, version Config changes: - Remove FastAPI/uvicorn/http config, replace with grpc config - GRPCConfig with host/port/max_workers Dependency changes: - Remove fastapi, uvicorn, starlette, anyio - Add grpcio, grpcio-tools, protobuf - Keep httpx for HTTP sink forwarding (separate from management API) Tests: - 8 new gRPC integration tests (test_grpc.py) - All 31 tests passing, ruff clean
2026-07-15 10:48:18 +03:30
parent b3243427b7
commit 4da9accabf
21 changed files with 1019 additions and 324 deletions
@@ -10,10 +10,10 @@ logging:
  format: json
 api:
-  http:
+  grpc:
    host: 127.0.0.1
    port: 8642
-    workers: 1
+    max_workers: 10
 pipeline:
  batch_size: 100
@@ -14,7 +14,6 @@ ExecReload=/bin/kill -HUP $MAINPID
 Restart=on-failure
 RestartSec=5
 AmbientCapabilities=CAP_SYS_PTRACE
 NoNewPrivileges=true
 ProtectSystem=full
 ProtectHome=true
@@ -0,0 +1,96 @@
 syntax = "proto3";
 package daena.v1;
 service Daena {
  rpc GetStatus(Empty) returns (StatusResponse);
  rpc GetHealth(Empty) returns (HealthResponse);
  rpc GetConfig(Empty) returns (ConfigResponse);
  rpc ListPlugins(Empty) returns (PluginListResponse);
  rpc GetQueueState(Empty) returns (QueueStateResponse);
  rpc ListPendingRecords(Pagination) returns (RecordList);
  rpc ListDeadRecords(Pagination) returns (RecordList);
  rpc ListSinks(Empty) returns (SinkList);
  rpc ReloadConfig(Empty) returns (ReloadResponse);
 }
 message Empty {}
 message StatusResponse {
  string service = 1;
  string version = 2;
  bool running = 3;
 }
 message HealthResponse {
  bool healthy = 1;
  bool running = 2;
  bool backend = 3;
  bool plugins_loaded = 4;
 }
 message ConfigResponse {
  string config_json = 1;
 }
 message PluginInfo {
  string name = 1;
  string class_name = 2;
 }
 message PluginCategory {
  repeated PluginInfo items = 1;
 }
 message PluginListResponse {
  PluginCategory sources = 1;
  PluginCategory processors = 2;
  PluginCategory sinks = 3;
 }
 message QueueStateResponse {
  int64 pending = 1;
  int64 delivered = 2;
  int64 failed = 3;
  int64 dead = 4;
  int64 total = 5;
 }
 message Pagination {
  int32 limit = 1;
  int32 offset = 2;
 }
 message RecordData {
  string record_id = 1;
  string source = 2;
  string record_type = 3;
  string body = 4;
  string metadata = 5;
  string state = 6;
  string created_at = 7;
  string updated_at = 8;
  int32 retry_count = 9;
  string next_retry_at = 10;
  string last_error = 11;
  string destination = 12;
 }
 message RecordList {
  repeated RecordData records = 1;
  int32 count = 2;
 }
 message SinkInfo {
  string name = 1;
  string type = 2;
 }
 message SinkList {
  repeated SinkInfo sinks = 1;
 }
 message ReloadResponse {
  bool success = 1;
  string message = 2;
 }
@@ -21,8 +21,9 @@ classifiers = [
 ]
 dependencies = [
-    "fastapi>=0.115",
+    "grpcio>=1.70",
-    "uvicorn[standard]>=0.30",
+    "grpcio-tools>=1.70",
    "protobuf>=5.28",
    "pydantic>=2.0",
    "pydantic-settings>=2.0",
    "PyYAML>=6.0",
@@ -0,0 +1 @@
 """Daena gRPC API."""
@@ -0,0 +1,165 @@
 from __future__ import annotations
 import json
 import grpc
 import grpc.aio
 from daena.api.pb2 import daena_pb2 as pb2
 from daena.api.pb2 import daena_pb2_grpc as pb2_grpc
 from daena.core.runtime import DaenaRuntime
 from daena.logutil import get_logger
 from daena.persistence.models import RecordState
 log = get_logger("daena.grpc")
 def _record_to_proto(record) -> pb2.RecordData:
    return pb2.RecordData(
        record_id=record.record_id,
        source=record.source,
        record_type=record.record_type,
        body=json.dumps(record.body),
        metadata=json.dumps(record.metadata),
        state=record.state.value if hasattr(record.state, "value") else record.state,
        created_at=record.created_at or "",
        updated_at=record.updated_at or "",
        retry_count=record.retry_count,
        next_retry_at=record.next_retry_at or "",
        last_error=record.last_error or "",
        destination=record.destination or "",
    )
 # ruff: noqa: N802, ARG002
 class DaenaServicer(pb2_grpc.DaenaServicer):
    def __init__(self, runtime: DaenaRuntime) -> None:
        self._runtime = runtime
    async def GetStatus(self, request, context):
        return pb2.StatusResponse(
            service="daena",
            version="0.1.0",
            running=self._runtime._running,
        )
    async def GetHealth(self, request, context):
        return pb2.HealthResponse(
            healthy=self._runtime._running,
            running=self._runtime._running,
            backend=self._runtime.backend is not None,
            plugins_loaded=bool(self._runtime.registry.list_sources()),
        )
    async def GetConfig(self, request, context):
        cfg = self._runtime.config
        raw = {
            "data_dir": cfg.data_dir,
            "log_dir": cfg.log_dir,
            "api": {"grpc": {"host": cfg.api.grpc.host, "port": cfg.api.grpc.port}},
            "pipeline": cfg.pipeline.model_dump(),
            "persistence": cfg.persistence.model_dump(),
            "sources": [s.model_dump() for s in cfg.sources],
            "processors": [p.model_dump() for p in cfg.processors],
            "sinks": [s.model_dump() for s in cfg.sinks],
        }
        return pb2.ConfigResponse(config_json=json.dumps(raw))
    async def ListPlugins(self, request, context):
        registry = self._runtime.registry
        all_plugins = registry.all_plugins()
        def _cat(key: str):
            items = all_plugins.get(key, {})
            return pb2.PluginCategory(
                items=[pb2.PluginInfo(name=n, class_name=cls.__name__) for n, cls in items.items()]
            )
        return pb2.PluginListResponse(
            sources=_cat("sources"),
            processors=_cat("processors"),
            sinks=_cat("sinks"),
        )
    async def GetQueueState(self, request, context):
        backend = self._runtime.backend
        if backend is None:
            return pb2.QueueStateResponse()
        return pb2.QueueStateResponse(
            pending=backend.count(RecordState.pending),
            delivered=backend.count(RecordState.delivered),
            failed=backend.count(RecordState.failed),
            dead=backend.count(RecordState.dead),
            total=backend.count(),
        )
    async def ListPendingRecords(self, request, context):
        backend = self._runtime.backend
        if backend is None:
            return pb2.RecordList(records=[], count=0)
        records = backend.list_by_state(
            RecordState.pending,
            limit=request.limit or 100,
            offset=request.offset or 0,
        )
        return pb2.RecordList(
            records=[_record_to_proto(r) for r in records],
            count=len(records),
        )
    async def ListDeadRecords(self, request, context):
        backend = self._runtime.backend
        if backend is None:
            return pb2.RecordList(records=[], count=0)
        records = backend.list_by_state(
            RecordState.dead,
            limit=request.limit or 100,
            offset=request.offset or 0,
        )
        return pb2.RecordList(
            records=[_record_to_proto(r) for r in records],
            count=len(records),
        )
    async def ListSinks(self, request, context):
        pipeline = self._runtime.pipeline
        if pipeline is None:
            return pb2.SinkList(sinks=[])
        return pb2.SinkList(
            sinks=[pb2.SinkInfo(name=s.name, type=type(s).__name__) for s in pipeline.sinks]
        )
    async def ReloadConfig(self, request, context):
        try:
            await self._runtime.reload()
            return pb2.ReloadResponse(success=True, message="Configuration reloaded")
        except Exception as exc:
            return pb2.ReloadResponse(success=False, message=str(exc))
 class DaenaGRPCServer:
    """Manages the gRPC server lifecycle."""
    def __init__(self, runtime: DaenaRuntime, host: str = "127.0.0.1", port: int = 8642) -> None:
        self._runtime = runtime
        self._host = host
        self._port = port
        self._server: grpc.aio.Server | None = None
    async def start(self) -> None:
        self._server = grpc.aio.server()
        servicer = DaenaServicer(self._runtime)
        pb2_grpc.add_DaenaServicer_to_server(servicer, self._server)
        address = f"{self._host}:{self._port}"
        self._server.add_insecure_port(address)
        await self._server.start()
        log.info("grpc_server_started", address=address)
    async def stop(self) -> None:
        if self._server:
            await self._server.stop(grace=5)
            log.info("grpc_server_stopped")
    async def serve_forever(self) -> None:
        if self._server:
            await self._server.wait_for_termination()
@@ -0,0 +1 @@
 """gRPC protocol buffer stubs for Daena v1."""
@@ -0,0 +1,62 @@
 # -*- coding: utf-8 -*-
 # ruff: noqa: I001, E501, UP009
 # Generated by the protocol buffer compiler.  DO NOT EDIT!
 # NO CHECKED-IN PROTOBUF GENCODE
 # source: daena/v1/daena.proto
 # Protobuf Python Version: 7.35.0
 """Generated protocol buffer code."""
 from google.protobuf import descriptor as _descriptor
 from google.protobuf import descriptor_pool as _descriptor_pool
 from google.protobuf import runtime_version as _runtime_version
 from google.protobuf import symbol_database as _symbol_database
 from google.protobuf.internal import builder as _builder
 _runtime_version.ValidateProtobufRuntimeVersion(
    _runtime_version.Domain.PUBLIC, 7, 35, 0, "", "daena/v1/daena.proto"
 )
 # @@protoc_insertion_point(imports)
 _sym_db = _symbol_database.Default()
 DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(
    b'\n\x14\x64\x61\x65na/v1/daena.proto\x12\x08\x64\x61\x65na.v1"\x07\n\x05\x45mpty"C\n\x0eStatusResponse\x12\x0f\n\x07service\x18\x01 \x01(\t\x12\x0f\n\x07version\x18\x02 \x01(\t\x12\x0f\n\x07running\x18\x03 \x01(\x08"[\n\x0eHealthResponse\x12\x0f\n\x07healthy\x18\x01 \x01(\x08\x12\x0f\n\x07running\x18\x02 \x01(\x08\x12\x0f\n\x07\x62\x61\x63kend\x18\x03 \x01(\x08\x12\x16\n\x0eplugins_loaded\x18\x04 \x01(\x08"%\n\x0e\x43onfigResponse\x12\x13\n\x0b\x63onfig_json\x18\x01 \x01(\t".\n\nPluginInfo\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x12\n\nclass_name\x18\x02 \x01(\t"5\n\x0ePluginCategory\x12#\n\x05items\x18\x01 \x03(\x0b\x32\x14.daena.v1.PluginInfo"\x96\x01\n\x12PluginListResponse\x12)\n\x07sources\x18\x01 \x01(\x0b\x32\x18.daena.v1.PluginCategory\x12,\n\nprocessors\x18\x02 \x01(\x0b\x32\x18.daena.v1.PluginCategory\x12\'\n\x05sinks\x18\x03 \x01(\x0b\x32\x18.daena.v1.PluginCategory"e\n\x12QueueStateResponse\x12\x0f\n\x07pending\x18\x01 \x01(\x03\x12\x11\n\tdelivered\x18\x02 \x01(\x03\x12\x0e\n\x06\x66\x61iled\x18\x03 \x01(\x03\x12\x0c\n\x04\x64\x65\x61\x64\x18\x04 \x01(\x03\x12\r\n\x05total\x18\x05 \x01(\x03"+\n\nPagination\x12\r\n\x05limit\x18\x01 \x01(\x05\x12\x0e\n\x06offset\x18\x02 \x01(\x05"\xf0\x01\n\nRecordData\x12\x11\n\trecord_id\x18\x01 \x01(\t\x12\x0e\n\x06source\x18\x02 \x01(\t\x12\x13\n\x0brecord_type\x18\x03 \x01(\t\x12\x0c\n\x04\x62ody\x18\x04 \x01(\t\x12\x10\n\x08metadata\x18\x05 \x01(\t\x12\r\n\x05state\x18\x06 \x01(\t\x12\x12\n\ncreated_at\x18\x07 \x01(\t\x12\x12\n\nupdated_at\x18\x08 \x01(\t\x12\x13\n\x0bretry_count\x18\t \x01(\x05\x12\x15\n\rnext_retry_at\x18\n \x01(\t\x12\x12\n\nlast_error\x18\x0b \x01(\t\x12\x13\n\x0b\x64\x65stination\x18\x0c \x01(\t"B\n\nRecordList\x12%\n\x07records\x18\x01 \x03(\x0b\x32\x14.daena.v1.RecordData\x12\r\n\x05\x63ount\x18\x02 \x01(\x05"&\n\x08SinkInfo\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x0c\n\x04type\x18\x02 \x01(\t"-\n\x08SinkList\x12!\n\x05sinks\x18\x01 \x03(\x0b\x32\x12.daena.v1.SinkInfo"2\n\x0eReloadResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t2\x9b\x04\n\x05\x44\x61\x65na\x12\x36\n\tGetStatus\x12\x0f.daena.v1.Empty\x1a\x18.daena.v1.StatusResponse\x12\x36\n\tGetHealth\x12\x0f.daena.v1.Empty\x1a\x18.daena.v1.HealthResponse\x12\x36\n\tGetConfig\x12\x0f.daena.v1.Empty\x1a\x18.daena.v1.ConfigResponse\x12<\n\x0bListPlugins\x12\x0f.daena.v1.Empty\x1a\x1c.daena.v1.PluginListResponse\x12>\n\rGetQueueState\x12\x0f.daena.v1.Empty\x1a\x1c.daena.v1.QueueStateResponse\x12@\n\x12ListPendingRecords\x12\x14.daena.v1.Pagination\x1a\x14.daena.v1.RecordList\x12=\n\x0fListDeadRecords\x12\x14.daena.v1.Pagination\x1a\x14.daena.v1.RecordList\x12\x30\n\tListSinks\x12\x0f.daena.v1.Empty\x1a\x12.daena.v1.SinkList\x12\x39\n\x0cReloadConfig\x12\x0f.daena.v1.Empty\x1a\x18.daena.v1.ReloadResponseb\x06proto3'
 )
 _globals = globals()
 _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
 _builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, "daena.v1.daena_pb2", _globals)
 if not _descriptor._USE_C_DESCRIPTORS:
    DESCRIPTOR._loaded_options = None
    _globals["_EMPTY"]._serialized_start = 34
    _globals["_EMPTY"]._serialized_end = 41
    _globals["_STATUSRESPONSE"]._serialized_start = 43
    _globals["_STATUSRESPONSE"]._serialized_end = 110
    _globals["_HEALTHRESPONSE"]._serialized_start = 112
    _globals["_HEALTHRESPONSE"]._serialized_end = 203
    _globals["_CONFIGRESPONSE"]._serialized_start = 205
    _globals["_CONFIGRESPONSE"]._serialized_end = 242
    _globals["_PLUGININFO"]._serialized_start = 244
    _globals["_PLUGININFO"]._serialized_end = 290
    _globals["_PLUGINCATEGORY"]._serialized_start = 292
    _globals["_PLUGINCATEGORY"]._serialized_end = 345
    _globals["_PLUGINLISTRESPONSE"]._serialized_start = 348
    _globals["_PLUGINLISTRESPONSE"]._serialized_end = 498
    _globals["_QUEUESTATERESPONSE"]._serialized_start = 500
    _globals["_QUEUESTATERESPONSE"]._serialized_end = 601
    _globals["_PAGINATION"]._serialized_start = 603
    _globals["_PAGINATION"]._serialized_end = 646
    _globals["_RECORDDATA"]._serialized_start = 649
    _globals["_RECORDDATA"]._serialized_end = 889
    _globals["_RECORDLIST"]._serialized_start = 891
    _globals["_RECORDLIST"]._serialized_end = 957
    _globals["_SINKINFO"]._serialized_start = 959
    _globals["_SINKINFO"]._serialized_end = 997
    _globals["_SINKLIST"]._serialized_start = 999
    _globals["_SINKLIST"]._serialized_end = 1044
    _globals["_RELOADRESPONSE"]._serialized_start = 1046
    _globals["_RELOADRESPONSE"]._serialized_end = 1096
    _globals["_DAENA"]._serialized_start = 1099
    _globals["_DAENA"]._serialized_end = 1638
 # @@protoc_insertion_point(module_scope)
@@ -0,0 +1,479 @@
 # ruff: noqa: N802, ARG002, I001, F401, E501
 # Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
 """Client and server classes corresponding to protobuf-defined services."""
 import grpc
 import warnings
 from daena.api.pb2 import daena_pb2 as daena_dot_v1_dot_daena__pb2
 GRPC_GENERATED_VERSION = "1.82.1"
 GRPC_VERSION = grpc.__version__
 _version_not_supported = False
 try:
    from grpc._utilities import first_version_is_lower
    _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
 except ImportError:
    _version_not_supported = True
 if _version_not_supported:
    raise RuntimeError(
        f"The grpc package installed is at version {GRPC_VERSION},"
        + " but the generated code in daena/v1/daena_pb2_grpc.py depends on"
        + f" grpcio>={GRPC_GENERATED_VERSION}."
        + f" Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}"
        + f" or downgrade your generated code using grpcio-tools<={GRPC_VERSION}."
    )
 class DaenaStub:
    """Missing associated documentation comment in .proto file."""
    def __init__(self, channel):
        """Constructor.
        Args:
            channel: A grpc.Channel.
        """
        self.GetStatus = channel.unary_unary(
            "/daena.v1.Daena/GetStatus",
            request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            response_deserializer=daena_dot_v1_dot_daena__pb2.StatusResponse.FromString,
            _registered_method=True,
        )
        self.GetHealth = channel.unary_unary(
            "/daena.v1.Daena/GetHealth",
            request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            response_deserializer=daena_dot_v1_dot_daena__pb2.HealthResponse.FromString,
            _registered_method=True,
        )
        self.GetConfig = channel.unary_unary(
            "/daena.v1.Daena/GetConfig",
            request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            response_deserializer=daena_dot_v1_dot_daena__pb2.ConfigResponse.FromString,
            _registered_method=True,
        )
        self.ListPlugins = channel.unary_unary(
            "/daena.v1.Daena/ListPlugins",
            request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            response_deserializer=daena_dot_v1_dot_daena__pb2.PluginListResponse.FromString,
            _registered_method=True,
        )
        self.GetQueueState = channel.unary_unary(
            "/daena.v1.Daena/GetQueueState",
            request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            response_deserializer=daena_dot_v1_dot_daena__pb2.QueueStateResponse.FromString,
            _registered_method=True,
        )
        self.ListPendingRecords = channel.unary_unary(
            "/daena.v1.Daena/ListPendingRecords",
            request_serializer=daena_dot_v1_dot_daena__pb2.Pagination.SerializeToString,
            response_deserializer=daena_dot_v1_dot_daena__pb2.RecordList.FromString,
            _registered_method=True,
        )
        self.ListDeadRecords = channel.unary_unary(
            "/daena.v1.Daena/ListDeadRecords",
            request_serializer=daena_dot_v1_dot_daena__pb2.Pagination.SerializeToString,
            response_deserializer=daena_dot_v1_dot_daena__pb2.RecordList.FromString,
            _registered_method=True,
        )
        self.ListSinks = channel.unary_unary(
            "/daena.v1.Daena/ListSinks",
            request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            response_deserializer=daena_dot_v1_dot_daena__pb2.SinkList.FromString,
            _registered_method=True,
        )
        self.ReloadConfig = channel.unary_unary(
            "/daena.v1.Daena/ReloadConfig",
            request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            response_deserializer=daena_dot_v1_dot_daena__pb2.ReloadResponse.FromString,
            _registered_method=True,
        )
 class DaenaServicer:
    """Missing associated documentation comment in .proto file."""
    def GetStatus(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")
    def GetHealth(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")
    def GetConfig(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")
    def ListPlugins(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")
    def GetQueueState(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")
    def ListPendingRecords(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")
    def ListDeadRecords(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")
    def ListSinks(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")
    def ReloadConfig(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details("Method not implemented!")
        raise NotImplementedError("Method not implemented!")
 def add_DaenaServicer_to_server(servicer, server):
    rpc_method_handlers = {
        "GetStatus": grpc.unary_unary_rpc_method_handler(
            servicer.GetStatus,
            request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString,
            response_serializer=daena_dot_v1_dot_daena__pb2.StatusResponse.SerializeToString,
        ),
        "GetHealth": grpc.unary_unary_rpc_method_handler(
            servicer.GetHealth,
            request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString,
            response_serializer=daena_dot_v1_dot_daena__pb2.HealthResponse.SerializeToString,
        ),
        "GetConfig": grpc.unary_unary_rpc_method_handler(
            servicer.GetConfig,
            request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString,
            response_serializer=daena_dot_v1_dot_daena__pb2.ConfigResponse.SerializeToString,
        ),
        "ListPlugins": grpc.unary_unary_rpc_method_handler(
            servicer.ListPlugins,
            request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString,
            response_serializer=daena_dot_v1_dot_daena__pb2.PluginListResponse.SerializeToString,
        ),
        "GetQueueState": grpc.unary_unary_rpc_method_handler(
            servicer.GetQueueState,
            request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString,
            response_serializer=daena_dot_v1_dot_daena__pb2.QueueStateResponse.SerializeToString,
        ),
        "ListPendingRecords": grpc.unary_unary_rpc_method_handler(
            servicer.ListPendingRecords,
            request_deserializer=daena_dot_v1_dot_daena__pb2.Pagination.FromString,
            response_serializer=daena_dot_v1_dot_daena__pb2.RecordList.SerializeToString,
        ),
        "ListDeadRecords": grpc.unary_unary_rpc_method_handler(
            servicer.ListDeadRecords,
            request_deserializer=daena_dot_v1_dot_daena__pb2.Pagination.FromString,
            response_serializer=daena_dot_v1_dot_daena__pb2.RecordList.SerializeToString,
        ),
        "ListSinks": grpc.unary_unary_rpc_method_handler(
            servicer.ListSinks,
            request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString,
            response_serializer=daena_dot_v1_dot_daena__pb2.SinkList.SerializeToString,
        ),
        "ReloadConfig": grpc.unary_unary_rpc_method_handler(
            servicer.ReloadConfig,
            request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString,
            response_serializer=daena_dot_v1_dot_daena__pb2.ReloadResponse.SerializeToString,
        ),
    }
    generic_handler = grpc.method_handlers_generic_handler("daena.v1.Daena", rpc_method_handlers)
    server.add_generic_rpc_handlers((generic_handler,))
    server.add_registered_method_handlers("daena.v1.Daena", rpc_method_handlers)
 # This class is part of an EXPERIMENTAL API.
 class Daena:
    """Missing associated documentation comment in .proto file."""
    @staticmethod
    def GetStatus(
        request,
        target,
        options=(),
        channel_credentials=None,
        call_credentials=None,
        insecure=False,
        compression=None,
        wait_for_ready=None,
        timeout=None,
        metadata=None,
    ):
        return grpc.experimental.unary_unary(
            request,
            target,
            "/daena.v1.Daena/GetStatus",
            daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            daena_dot_v1_dot_daena__pb2.StatusResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True,
        )
    @staticmethod
    def GetHealth(
        request,
        target,
        options=(),
        channel_credentials=None,
        call_credentials=None,
        insecure=False,
        compression=None,
        wait_for_ready=None,
        timeout=None,
        metadata=None,
    ):
        return grpc.experimental.unary_unary(
            request,
            target,
            "/daena.v1.Daena/GetHealth",
            daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            daena_dot_v1_dot_daena__pb2.HealthResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True,
        )
    @staticmethod
    def GetConfig(
        request,
        target,
        options=(),
        channel_credentials=None,
        call_credentials=None,
        insecure=False,
        compression=None,
        wait_for_ready=None,
        timeout=None,
        metadata=None,
    ):
        return grpc.experimental.unary_unary(
            request,
            target,
            "/daena.v1.Daena/GetConfig",
            daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            daena_dot_v1_dot_daena__pb2.ConfigResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True,
        )
    @staticmethod
    def ListPlugins(
        request,
        target,
        options=(),
        channel_credentials=None,
        call_credentials=None,
        insecure=False,
        compression=None,
        wait_for_ready=None,
        timeout=None,
        metadata=None,
    ):
        return grpc.experimental.unary_unary(
            request,
            target,
            "/daena.v1.Daena/ListPlugins",
            daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            daena_dot_v1_dot_daena__pb2.PluginListResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True,
        )
    @staticmethod
    def GetQueueState(
        request,
        target,
        options=(),
        channel_credentials=None,
        call_credentials=None,
        insecure=False,
        compression=None,
        wait_for_ready=None,
        timeout=None,
        metadata=None,
    ):
        return grpc.experimental.unary_unary(
            request,
            target,
            "/daena.v1.Daena/GetQueueState",
            daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            daena_dot_v1_dot_daena__pb2.QueueStateResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True,
        )
    @staticmethod
    def ListPendingRecords(
        request,
        target,
        options=(),
        channel_credentials=None,
        call_credentials=None,
        insecure=False,
        compression=None,
        wait_for_ready=None,
        timeout=None,
        metadata=None,
    ):
        return grpc.experimental.unary_unary(
            request,
            target,
            "/daena.v1.Daena/ListPendingRecords",
            daena_dot_v1_dot_daena__pb2.Pagination.SerializeToString,
            daena_dot_v1_dot_daena__pb2.RecordList.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True,
        )
    @staticmethod
    def ListDeadRecords(
        request,
        target,
        options=(),
        channel_credentials=None,
        call_credentials=None,
        insecure=False,
        compression=None,
        wait_for_ready=None,
        timeout=None,
        metadata=None,
    ):
        return grpc.experimental.unary_unary(
            request,
            target,
            "/daena.v1.Daena/ListDeadRecords",
            daena_dot_v1_dot_daena__pb2.Pagination.SerializeToString,
            daena_dot_v1_dot_daena__pb2.RecordList.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True,
        )
    @staticmethod
    def ListSinks(
        request,
        target,
        options=(),
        channel_credentials=None,
        call_credentials=None,
        insecure=False,
        compression=None,
        wait_for_ready=None,
        timeout=None,
        metadata=None,
    ):
        return grpc.experimental.unary_unary(
            request,
            target,
            "/daena.v1.Daena/ListSinks",
            daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            daena_dot_v1_dot_daena__pb2.SinkList.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True,
        )
    @staticmethod
    def ReloadConfig(
        request,
        target,
        options=(),
        channel_credentials=None,
        call_credentials=None,
        insecure=False,
        compression=None,
        wait_for_ready=None,
        timeout=None,
        metadata=None,
    ):
        return grpc.experimental.unary_unary(
            request,
            target,
            "/daena.v1.Daena/ReloadConfig",
            daena_dot_v1_dot_daena__pb2.Empty.SerializeToString,
            daena_dot_v1_dot_daena__pb2.ReloadResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True,
        )
@@ -1,15 +0,0 @@
 from __future__ import annotations
 from fastapi import APIRouter
 from daena.api.routes.plugins import router as plugins_router
 from daena.api.routes.queue import router as queue_router
 from daena.api.routes.sinks import router as sinks_router
 from daena.api.routes.system import router as system_router
 api_router = APIRouter()
 api_router.include_router(system_router, prefix="/system", tags=["system"])
 api_router.include_router(plugins_router, prefix="/plugins", tags=["plugins"])
 api_router.include_router(queue_router, prefix="/queue", tags=["queue"])
 api_router.include_router(sinks_router, prefix="/sinks", tags=["sinks"])
@@ -1,16 +0,0 @@
 from __future__ import annotations
 from fastapi import APIRouter, Request
 router = APIRouter()
@router.get("")
 async def list_plugins(request: Request) -> dict:
    runtime = request.app.state.runtime
    registry = runtime.registry
    plugins = registry.all_plugins()
    result: dict[str, list[dict]] = {}
    for category, entries in plugins.items():
        result[category] = [{"name": name, "class": cls.__name__} for name, cls in entries.items()]
    return {"plugins": result}
@@ -1,50 +0,0 @@
 from __future__ import annotations
 from fastapi import APIRouter, Query, Request
 from daena.persistence.models import RecordState
 router = APIRouter()
@router.get("")
 async def get_queue_state(request: Request) -> dict:
    runtime = request.app.state.runtime
    backend = runtime.backend
    if backend is None:
        return {"error": "backend not initialized"}
    return {
        "pending": backend.count(RecordState.pending),
        "delivered": backend.count(RecordState.delivered),
        "failed": backend.count(RecordState.failed),
        "dead": backend.count(RecordState.dead),
        "total": backend.count(),
    }
@router.get("/pending")
 async def get_pending(
    request: Request,
    limit: int = Query(default=100, le=1000),
    offset: int = Query(default=0, ge=0),
 ) -> dict:
    runtime = request.app.state.runtime
    backend = runtime.backend
    if backend is None:
        return {"error": "backend not initialized", "records": []}
    records = backend.list_by_state(RecordState.pending, limit=limit, offset=offset)
    return {"records": [r.to_dict() for r in records], "count": len(records)}
@router.get("/dead")
 async def get_dead(
    request: Request,
    limit: int = Query(default=100, le=1000),
    offset: int = Query(default=0, ge=0),
 ) -> dict:
    runtime = request.app.state.runtime
    backend = runtime.backend
    if backend is None:
        return {"error": "backend not initialized", "records": []}
    records = backend.list_by_state(RecordState.dead, limit=limit, offset=offset)
    return {"records": [r.to_dict() for r in records], "count": len(records)}
@@ -1,17 +0,0 @@
 from __future__ import annotations
 from fastapi import APIRouter, Request
 router = APIRouter()
@router.get("")
 async def list_sinks(request: Request) -> dict:
    runtime = request.app.state.runtime
    pipeline = runtime.pipeline
    if pipeline is None:
        return {"sinks": []}
    sinks = pipeline.sinks
    return {
        "sinks": [{"name": s.name, "type": type(s).__name__, "config": s.config} for s in sinks]
    }
@@ -1,50 +0,0 @@
 from __future__ import annotations
 from fastapi import APIRouter, Request
 from daena.core.runtime import DaenaRuntime
 router = APIRouter()
 def _resolve_runtime(request: Request) -> DaenaRuntime:
    return request.app.state.runtime  # type: ignore[no-any-return]
@router.get("/status")
 async def get_status(request: Request) -> dict:
    runtime = _resolve_runtime(request)
    return {
        "service": "daena",
        "version": "0.1.0",
        "running": runtime._running,
        "uptime": None,
    }
@router.get("/health")
 async def get_health(request: Request) -> dict:
    runtime = _resolve_runtime(request)
    healthy = runtime._running
    return {
        "healthy": healthy,
        "running": runtime._running,
        "backend": runtime.backend is not None,
        "plugins_loaded": bool(runtime.registry.list_sources()),
    }
@router.get("/config")
 async def get_config(request: Request) -> dict:
    runtime = _resolve_runtime(request)
    cfg = runtime.config
    return {
        "data_dir": cfg.data_dir,
        "log_dir": cfg.log_dir,
        "api": cfg.api.model_dump(),
        "pipeline": cfg.pipeline.model_dump(),
        "persistence": cfg.persistence.model_dump(),
        "sources": [s.model_dump() for s in cfg.sources],
        "processors": [p.model_dump() for p in cfg.processors],
        "sinks": [s.model_dump() for s in cfg.sinks],
    }
@@ -1,50 +0,0 @@
 from __future__ import annotations
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from fastapi import FastAPI
 from daena.api.router import api_router
 from daena.core.runtime import DaenaRuntime
 from daena.logutil import get_logger
 log = get_logger("daena.api")
 class DaenaAPI:
    """Wraps the FastAPI application with daena runtime integration."""
    def __init__(self, runtime: DaenaRuntime) -> None:
        self._runtime = runtime
        self._app: FastAPI | None = None
    def build(self) -> FastAPI:
        runtime = self._runtime
        @asynccontextmanager
        async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
            log.info("api_starting")
            await runtime.start()
            yield
            log.info("api_stopping")
            await runtime.stop()
        app = FastAPI(
            title="Daena",
            description="Daena service management API",
            version="0.1.0",
            lifespan=lifespan,
        )
        app.state.runtime = runtime
        app.include_router(api_router, prefix="/api/v1")
        self._app = app
        return app
    @property
    def app(self) -> FastAPI:
        if self._app is None:
            return self.build()
        return self._app
@@ -1,44 +1,16 @@
 from __future__ import annotations
 import asyncio
 from pathlib import Path
-import uvicorn
+from daena.api.grpc_server import DaenaGRPCServer
 from daena.api.server import DaenaAPI
 from daena.config import load_config
 from daena.core.runtime import DaenaRuntime
 from daena.logutil import get_logger, setup_logging
 log = get_logger("daena.app")
 def serve() -> None:
    """Entry point for `daena-serve` (systemd/uvicorn mode)."""
    config_path = _resolve_config_path()
    config = load_config(config_path)
    setup_logging(
        level=config.logging.level,
        fmt=config.logging.format,
        log_file=config.logging.file,
    )
    runtime = _build_runtime(config)
    api = DaenaAPI(runtime)
    app = api.build()
    log.info(
        "starting_server",
        host=config.api.http.host,
        port=config.api.http.port,
    )
    uvicorn.run(
        app,
        host=config.api.http.host,
        port=config.api.http.port,
        workers=config.api.http.workers,
        log_level=config.logging.level,
    )
 def _resolve_config_path() -> str | None:
    for candidate in (
        "/etc/daena/daena.yaml",
@@ -51,11 +23,40 @@ def _resolve_config_path() -> str | None:
    return None
-def _build_runtime(config):
+def serve() -> None:
-    """Lazy import to prevent premature backend init."""
+    """Entry point for `daena-serve` (systemd/gRPC mode)."""
-    from daena.core.runtime import DaenaRuntime
+    config_path = _resolve_config_path()
    config = load_config(config_path)
    setup_logging(
        level=config.logging.level,
        fmt=config.logging.format,
        log_file=config.logging.file,
    )
-    return DaenaRuntime(config)
+    runtime = DaenaRuntime(config)
    server = DaenaGRPCServer(
        runtime,
        host=config.api.grpc.host,
        port=config.api.grpc.port,
    )
    async def run() -> None:
        await runtime.start()
        await server.start()
        log.info(
            "grpc_server_serving",
            address=f"{config.api.grpc.host}:{config.api.grpc.port}",
        )
        try:
            await server.serve_forever()
        except asyncio.CancelledError:
            pass
        finally:
            await server.stop()
            await runtime.stop()
    asyncio.run(run())
 if __name__ == "__main__":
@@ -1,11 +1,15 @@
 from __future__ import annotations
-import httpx
+import json
 import grpc
 import typer
 from rich.console import Console
 from rich.table import Table
 from daena._version import __version__
 from daena.api.pb2 import daena_pb2 as pb2
 from daena.api.pb2 import daena_pb2_grpc as pb2_grpc
 app = typer.Typer(
    name="daena",
@@ -13,121 +17,100 @@ app = typer.Typer(
 )
 console = Console()
-DEFAULT_API_URL = "http://127.0.0.1:8642/api/v1"
+DEFAULT_TARGET = "127.0.0.1:8642"
-def _client(timeout: float = 10.0) -> httpx.Client:
+def _stub(target: str | None = None) -> pb2_grpc.DaenaStub:
-    return httpx.Client(timeout=httpx.Timeout(timeout))
+    tgt = target or DEFAULT_TARGET
    channel = grpc.insecure_channel(tgt)
    return pb2_grpc.DaenaStub(channel)
-def _api_url() -> str:
+def _connect(target: str | None = None) -> pb2_grpc.DaenaStub:
-    return DEFAULT_API_URL  # TODO: make configurable via env/flag
+    try:
        stub = _stub(target)
        stub.GetStatus(pb2.Empty(), timeout=3.0)
        return stub
    except grpc.RpcError:
        console.print("[red]Cannot connect to daena service[/red]")
        raise typer.Exit(1) from None
    except Exception:
        console.print("[red]Cannot connect to daena service[/red]")
        raise typer.Exit(1) from None
@app.command()
 def status() -> None:
    """Show daemon status."""
-    try:
+    stub = _connect()
-        with _client(5.0) as c:
+    resp = stub.GetStatus(pb2.Empty(), timeout=5.0)
-            resp = c.get(f"{_api_url()}/system/status")
+    console.print(f"[bold]Daena[/bold] v{resp.version}")
-            resp.raise_for_status()
+    console.print(f"Running: {resp.running}")
            data = resp.json()
            console.print(f"[bold]Daena[/bold] v{data.get('version', '?')}")
            console.print(f"Running: {data.get('running', '?')}")
    except httpx.ConnectError:
        console.print("[red]Cannot connect to daena service[/red]")
        raise typer.Exit(1) from None
@app.command()
 def health() -> None:
    """Check service health."""
-    try:
+    stub = _connect()
-        with _client(5.0) as c:
+    resp = stub.GetHealth(pb2.Empty(), timeout=5.0)
-            resp = c.get(f"{_api_url()}/system/health")
+    if resp.healthy:
-            resp.raise_for_status()
+        console.print("[green]Healthy[/green]")
-            data = resp.json()
+    else:
-            if data.get("healthy"):
+        console.print("[red]Unhealthy[/red]")
-                console.print("[green]Healthy[/green]")
+        raise typer.Exit(1)
            else:
                console.print("[red]Unhealthy[/red]")
                raise typer.Exit(1)
    except httpx.ConnectError:
        console.print("[red]Cannot connect to daena service[/red]")
        raise typer.Exit(1) from None
@app.command()
 def queue() -> None:
    """Inspect queue state."""
-    try:
+    stub = _connect()
-        with _client() as c:
+    resp = stub.GetQueueState(pb2.Empty(), timeout=5.0)
-            resp = c.get(f"{_api_url()}/queue")
+    table = Table("State", "Count")
-            resp.raise_for_status()
+    for key in ("pending", "delivered", "failed", "dead", "total"):
-            data = resp.json()
+        table.add_row(key, str(getattr(resp, key, 0)))
-            table = Table("State", "Count")
+    console.print(table)
            for key in ("pending", "delivered", "failed", "dead", "total"):
                table.add_row(key, str(data.get(key, 0)))
            console.print(table)
    except httpx.ConnectError:
        console.print("[red]Cannot connect to daena service[/red]")
        raise typer.Exit(1) from None
@app.command()
 def sinks() -> None:
    """List configured sinks."""
-    try:
+    stub = _connect()
-        with _client() as c:
+    resp = stub.ListSinks(pb2.Empty(), timeout=5.0)
-            resp = c.get(f"{_api_url()}/sinks")
+    items = resp.sinks
-            resp.raise_for_status()
+    if not items:
-            data = resp.json()
+        console.print("No sinks configured")
-            sinks_list = data.get("sinks", [])
+        return
-            if not sinks_list:
+    table = Table("Name", "Type")
-                console.print("No sinks configured")
+    for s in items:
-                return
+        table.add_row(s.name, s.type)
-            table = Table("Name", "Type")
+    console.print(table)
            for s in sinks_list:
                table.add_row(s["name"], s["type"])
            console.print(table)
    except httpx.ConnectError:
        console.print("[red]Cannot connect to daena service[/red]")
        raise typer.Exit(1) from None
@app.command()
 def plugins() -> None:
    """List loaded plugins."""
-    try:
+    stub = _connect()
-        with _client() as c:
+    resp = stub.ListPlugins(pb2.Empty(), timeout=5.0)
-            resp = c.get(f"{_api_url()}/plugins")
+    for label, cat in (
-            resp.raise_for_status()
+        ("Sources", resp.sources),
-            data = resp.json()
+        ("Processors", resp.processors),
-            plugins_data = data.get("plugins", {})
+        ("Sinks", resp.sinks),
-            for category, items in plugins_data.items():
+    ):
-                if items:
+        if cat.items:
-                    table = Table(f"{category.title()}")
+            table = Table(label)
-                    for item in items:
+            for item in cat.items:
-                        table.add_row(f"{item['name']} ({item['class']})")
+                table.add_row(f"{item.name} ({item.class_name})")
-                    console.print(table)
+            console.print(table)
-                    console.print()
+            console.print()
    except httpx.ConnectError:
        console.print("[red]Cannot connect to daena service[/red]")
        raise typer.Exit(1) from None
@app.command()
 def config() -> None:
    """Show loaded service configuration."""
-    try:
+    stub = _connect()
-        with _client() as c:
+    resp = stub.GetConfig(pb2.Empty(), timeout=5.0)
-            resp = c.get(f"{_api_url()}/system/config")
+    parsed = json.loads(resp.config_json)
-            resp.raise_for_status()
+    console.print(parsed)
            data = resp.json()
            console.print(data)
    except httpx.ConnectError:
        console.print("[red]Cannot connect to daena service[/red]")
        raise typer.Exit(1) from None
@app.command()
@@ -14,14 +14,14 @@ class LoggingConfig(BaseSettings):
    file: str | None = None
-class APIHTTPConfig(BaseSettings):
+class GRPCConfig(BaseSettings):
    host: str = "127.0.0.1"
    port: int = 8642
-    workers: int = 1
+    max_workers: int = 10
 class APIConfig(BaseSettings):
-    http: APIHTTPConfig = APIHTTPConfig()
+    grpc: GRPCConfig = GRPCConfig()
 class SinkRetryConfig(BaseSettings):
@@ -9,7 +9,7 @@ def test_minimal_config_has_defaults() -> None:
    config = DaenaConfig()
    assert config.data_dir == "/var/lib/daena"
    assert config.logging.level == "info"
-    assert config.api.http.port == 8642
+    assert config.api.grpc.port == 8642
    assert config.pipeline.batch_size == 100
@@ -0,0 +1,105 @@
 from __future__ import annotations
 import asyncio
 import grpc  # noqa: I001
 import pytest
 from daena.api.grpc_server import DaenaGRPCServer
 from daena.api.pb2 import daena_pb2 as pb2
 from daena.api.pb2 import daena_pb2_grpc as pb2_grpc
 from daena.config import DaenaConfig, SinkConfig, SourceConfig
 from daena.core.runtime import DaenaRuntime
@pytest.fixture
 async def grpc_env():
    config = DaenaConfig()
    config.sources = [
        SourceConfig(name="dummy", type="dummy", options={"interval": 0.2, "message": "test"})
    ]
    config.sinks = [SinkConfig(name="log", type="dummy")]
    config.persistence.lmdb.path = "/tmp/daena_grpc_test"
    config.api.grpc.port = 18642
    runtime = DaenaRuntime(config)
    await runtime.start()
    server = DaenaGRPCServer(runtime, host="127.0.0.1", port=18642)
    await server.start()
    channel = grpc.aio.insecure_channel("127.0.0.1:18642")
    stub = pb2_grpc.DaenaStub(channel)
    yield runtime, stub, server
    await channel.close()
    await server.stop()
    await runtime.stop()
@pytest.mark.asyncio
 async def test_get_status(grpc_env) -> None:
    _, stub, _ = grpc_env
    resp = await stub.GetStatus(pb2.Empty(), timeout=5)
    assert resp.service == "daena"
    assert resp.version == "0.1.0"
    assert resp.running is True
@pytest.mark.asyncio
 async def test_get_health(grpc_env) -> None:
    _, stub, _ = grpc_env
    resp = await stub.GetHealth(pb2.Empty(), timeout=5)
    assert resp.healthy is True
    assert resp.running is True
@pytest.mark.asyncio
 async def test_get_config(grpc_env) -> None:
    _, stub, _ = grpc_env
    resp = await stub.GetConfig(pb2.Empty(), timeout=5)
    assert resp.config_json != ""
@pytest.mark.asyncio
 async def test_list_plugins(grpc_env) -> None:
    _, stub, _ = grpc_env
    resp = await stub.ListPlugins(pb2.Empty(), timeout=5)
    assert len(resp.sources.items) > 0
    assert len(resp.sinks.items) > 0
@pytest.mark.asyncio
 async def test_get_queue_state(grpc_env) -> None:
    _, stub, _ = grpc_env
    await asyncio.sleep(1)
    resp = await stub.GetQueueState(pb2.Empty(), timeout=5)
    assert resp.total > 0
    assert resp.pending > 0
@pytest.mark.asyncio
 async def test_list_pending_records(grpc_env) -> None:
    runtime, stub, _ = grpc_env
    await asyncio.sleep(1)
    req = pb2.Pagination(limit=10, offset=0)
    resp = await stub.ListPendingRecords(req, timeout=5)
    assert resp.count > 0
    assert len(resp.records) > 0
    assert resp.records[0].record_id != ""
@pytest.mark.asyncio
 async def test_list_sinks(grpc_env) -> None:
    _, stub, _ = grpc_env
    resp = await stub.ListSinks(pb2.Empty(), timeout=5)
    assert len(resp.sinks) >= 1
    assert resp.sinks[0].name == "log"
@pytest.mark.asyncio
 async def test_reload_config(grpc_env) -> None:
    _, stub, _ = grpc_env
    resp = await stub.ReloadConfig(pb2.Empty(), timeout=10)
    assert resp.success is True
		`@@ -0,0 +1 @@`
							`"""gRPC protocol buffer stubs for Daena v1."""`