feat(observability): RPC + workflow + DB-pool metrics for the agent JSON-RPC endpoint (#379)

## Summary

Brings the agent JSON-RPC endpoint (`POST /agents/{id}/rpc` —
`task/create` / `message/send` / `event/send`) up to the metrics
observability contract, and adds high-signal Postgres connection-pool
saturation metrics. Everything here is OTel-only (no StatsD dual-emit —
these are new series with no existing consumer), and a cheap no-op when
no OTLP endpoint is configured.

### RPC RED metrics (`agentex.rpc.*`)
- `agentex.rpc.request.duration` (timed dispatch → final byte, so
streaming responses are covered end-to-end), `agentex.rpc.requests`,
`agentex.rpc.errors`.
- Attributes follow the OTel RPC semantic conventions: `rpc.system`,
`rpc.method`, `rpc.jsonrpc.error_code`, `error.type`, `streaming`. No
high-cardinality identifiers (no task/agent/request ids). JSON-RPC
failures return HTTP 200, so the error code rides on
`rpc.jsonrpc.error_code` (set only on failure), not an HTTP status.
- Never-raise emission; a single terminal emit per request covering the
whole stream.

### Workflow-level GenAI metric
- `gen_ai.workflow.duration` for `task/create`, labeled
`gen_ai.operation.name=invoke_workflow`, so workflow-level duration is
directly queryable instead of only inferable from RPC duration.
`invoke_agent` stays reserved for the agent loop's own duration, so the
two operation names never mix gateway and in-pod durations.

### DB connection-pool metrics (OTel DB semconv)
- `db.client.connection.wait_time`,
`db.client.connection.pending_requests`, `db.client.connection.timeouts`
— the pre-outage early-warning signals for pool exhaustion.
- Sourced from an `InstrumentedAsyncAdaptedQueuePool` that wraps
`Pool.connect()` (the single, non-recursive acquisition entry point, run
inside the acquiring greenlet). The existing checkout/checkin listeners
can't produce these: by the time `checkout` fires the wait is over,
there's no waiter count, and a pool timeout raises before any connection
is handed out.
- `wait_time` uses explicit second-scale histogram buckets so quantiles
stay usable (the SDK default boundaries assume a millisecond unit).

## Testing
- Unit tests in `tests/unit/utils/test_rpc_metrics.py` and
`test_db_metrics.py`: unconfigured no-op, never-raise on backend faults,
metric name/attribute assertions, and the pool success / timeout /
non-timeout-error paths. All green; `ruff` clean.

## Notes / follow-ups
- Requires an OTLP endpoint configured on the pods to export.
- HTTP auto-instrumentation (route-templated
`http_server_request_duration_seconds`) lands on a separate branch.
- Dashboard/Mimir verification and matrix backfill to follow once the
target environment is available.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- greptile_comment -->

<h3>Greptile Summary</h3>

This PR adds OTel-only RED metrics for the agent JSON-RPC endpoint
(`agentex.rpc.*`), a GenAI workflow-duration histogram for
`task/create`, and Postgres connection-pool acquisition metrics
(`db.client.connection.wait_time/pending_requests/timeouts/failures`)
sourced from a new `InstrumentedAsyncAdaptedQueuePool` subclass. It also
removes the `namespace: agentex` from the local OTel collector's
Prometheus exporter to fix a double-prefix bug that would have appeared
for new `agentex.*`-prefixed instrument names.

- **RPC metrics**: A synchronous `@contextmanager`
(`rpc_request_timing`) wraps both sync and streaming handlers; a
`RpcCallOutcome` dataclass lets handlers classify JSON-RPC error codes,
with `BaseException` catching client disconnects (e.g., `GeneratorExit`)
separately from structured RPC errors. Never-raise contract is enforced
throughout.
- **DB pool metrics**: `InstrumentedAsyncAdaptedQueuePool.connect()` is
the correct interception seam (checkout/checkin events can\u2019t see
the wait or a timeout), instruments are attached post-construction and
carried across `recreate()`, and the `wait_time` histogram uses explicit
second-scale buckets to keep quantiles usable.
- **Collector config**: Removing `namespace: agentex` aligns local
PromQL with the production Mimir pipeline but renames existing
`agentex_db_*` local series to `db_*`; teams with local Grafana
dashboards querying the old prefixed names will need to update their
queries.

<details><summary><h3>Confidence Score: 5/5</h3></summary>

Safe to merge; all changes are additive instrumentation with a
never-raise emission contract and no mutations to the core RPC execution
path.

The RPC and DB pool metric additions are purely observational. The
rpc_request_timing contextmanager correctly handles BaseException
including GeneratorExit, the pool override degrades to a passthrough
when OTel is unconfigured, and recreate() carries instrumentation
forward. Unit tests cover the no-op path, never-raise contract,
double-count regression, and the streaming GeneratorExit scenario.

**Files Needing Attention:** No files require special attention. The
otel-collector-config.yaml namespace removal warrants a local dashboard
audit if Prometheus queries exist against the old agentex_db_*-prefixed
series.
</details>

<details><summary><h3>Important Files Changed</h3></summary>

| Filename | Overview |
|----------|----------|
| agentex/src/utils/rpc_metrics.py | New module: RPC RED metrics and
workflow-duration histogram with correct OTel semconv, never-raise
emission, and single-record-per-request guarantee via contextmanager. |
| agentex/src/utils/db_metrics.py | Adds
InstrumentedAsyncAdaptedQueuePool and three new OTel
connection-acquisition metrics; pending_requests increment occurs
outside the try block, so an OTel SDK raise there leaves the counter
unbalanced without a compensating decrement. |
| agentex/src/api/routes/agents.py | Wraps sync and streaming RPC
handlers with rpc_request_timing; streaming handler correctly carries
instrumentation across yields and GeneratorExit. |
| agentex/src/config/dependencies.py | Adds
poolclass=InstrumentedAsyncAdaptedQueuePool to all three engine
constructors; straightforward wiring change. |
| agentex/otel/otel-collector-config.yaml | Removes namespace: agentex
from Prometheus exporter to prevent double-prefixing for agentex.*
metrics; renames existing local db.* metric series as a side effect. |
| agentex/tests/unit/utils/test_rpc_metrics.py | Comprehensive unit
tests covering no-op, never-raise, attribute correctness, double-count
regression, GeneratorExit, and handler-classification precedence. |
| agentex/tests/unit/utils/test_db_metrics.py | Unit tests for the pool
instrumentation covering passthrough, success, timeout, non-timeout
error, recreate propagation, and end-to-end wiring. |

</details>

<details><summary><h3>Sequence Diagram</h3></summary>

```mermaid
sequenceDiagram
    participant Client
    participant agents_py as agents.py
    participant rpc_timing as rpc_request_timing
    participant use_case as agents_acp_use_case
    participant otel as OTel SDK

    Client->>agents_py: "POST /agents/{id}/rpc"
    agents_py->>rpc_timing: __enter__ (start timer)
    rpc_timing-->>agents_py: RpcCallOutcome

    alt Sync request
        agents_py->>use_case: handle_rpc_request()
        use_case-->>agents_py: result_entity
        agents_py->>rpc_timing: __exit__ (normal)
        rpc_timing->>otel: record duration/requests
        agents_py-->>Client: AgentRPCResponse (200)
    else Streaming request
        agents_py->>use_case: handle_rpc_request()
        use_case-->>agents_py: AsyncIterator
        loop Each chunk
            agents_py-->>Client: NDJSON chunk (yield)
        end
        alt Stream error
            agents_py->>rpc_timing: rpc_call.fail(-32603, e)
            agents_py-->>Client: error frame (yield)
        end
        agents_py->>rpc_timing: __exit__ (GeneratorExit or normal)
        rpc_timing->>otel: record duration/requests/errors
    end

    Note over agents_py,otel: task/create also emits gen_ai.workflow.duration
```
</details>

<sub>Reviews (6): Last reviewed commit: ["Merge branch &#39;main&#39;
into
stephen-wang/rp..."](0e754eb)
| [Re-trigger
Greptile](https://app.greptile.com/api/retrigger?id=48001757)</sub>

<!-- /greptile_comment -->

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: stephen-wang24

agentex/otel/otel-collector-config.yaml

@@ -21,10 +21,14 @@ exporters:
     sampling_initial: 5
     sampling_thereafter: 200
 
-  # Expose Prometheus endpoint for scraping
+  # Expose Prometheus endpoint for scraping.
+  # No `namespace:` on purpose — the cluster's otel-operator -> daemonset
+  # pipeline adds no prefix, so metric names there are exactly their instrument
+  # names (e.g. agentex_rpc_requests_total, db_client_connection_wait_time_seconds).
+  # A namespace here would double-prefix agentex.* metrics (agentex_agentex_...)
+  # and make locally-tested PromQL diverge from Mimir.
   prometheus:
     endpoint: 0.0.0.0:8889
-    namespace: agentex
     send_timestamps: true
     metric_expiration: 5m
 

agentex/src/api/routes/agents.py

@@ -43,6 +43,9 @@
     DAuthorizedResourceIds,
 )
 from src.utils.logging import make_logger
+from src.utils.rpc_metrics import (
+    rpc_request_timing,
+)
 from src.utils.task_authorization import check_task_or_collapse_to_404
 
 logger = make_logger(__name__)
@@ -549,53 +552,62 @@ async def _handle_sync_rpc(
     request_headers: dict[str, str] | None = None,
 ) -> AgentRPCResponse:
     """Handle synchronous JSON-RPC requests."""
-    try:
-        result_entity = await agents_acp_use_case.handle_rpc_request(
-            agent_id=agent_id,
-            agent_name=agent_name,
-            method=request.method,
-            params=request.params,
-            request_headers=request_headers,
-        )
+    with rpc_request_timing(request.method.value, streaming=False) as rpc_call:
+        try:
+            result_entity = await agents_acp_use_case.handle_rpc_request(
+                agent_id=agent_id,
+                agent_name=agent_name,
+                method=request.method,
+                params=request.params,
+                request_headers=request_headers,
+            )
 
-        if isinstance(result_entity, AsyncIterator):
-            raise ValueError(f"Expected non-async iterator, got {type(result_entity)}")
+            if isinstance(result_entity, AsyncIterator):
+                raise ValueError(
+                    f"Expected non-async iterator, got {type(result_entity)}"
+                )
 
-        if isinstance(result_entity, list):
-            serialized_result = [item.model_dump() for item in result_entity]
-        else:
-            serialized_result = result_entity.model_dump()
-
-        # if request.method == AgentRPCMethod.MESSAGE_SEND:
-        #     if isinstance(result_entity, list):
-        #         result = [TaskMessage.model_validate(task_message_entity) for task_message_entity in result_entity]
-        #     else:
-        #         raise ValueError(f"Expected list of TaskMessage entities, got {type(result_entity)}")
-        # elif request.method == AgentRPCMethod.TASK_CREATE:
-        #     result = Task.model_validate(result_entity)
-        # elif request.method == AgentRPCMethod.TASK_CANCEL:
-        #     result = Task.model_validate(result_entity)
-        # elif request.method == AgentRPCMethod.EVENT_SEND:
-        #     result = Event.model_validate(result_entity)
-        # else:
-        #     raise ValueError(f"Unsupported method: {request.method}")
-        # logger.info(f"AgentRPCResponse Result: {result}")
-        return AgentRPCResponse.model_validate(
-            {
-                "id": request.id,
-                "result": serialized_result,
-                "error": None,
-            }
-        )
+            if isinstance(result_entity, list):
+                serialized_result = [item.model_dump() for item in result_entity]
+            else:
+                serialized_result = result_entity.model_dump()
+
+            # if request.method == AgentRPCMethod.MESSAGE_SEND:
+            #     if isinstance(result_entity, list):
+            #         result = [TaskMessage.model_validate(task_message_entity) for task_message_entity in result_entity]
+            #     else:
+            #         raise ValueError(f"Expected list of TaskMessage entities, got {type(result_entity)}")
+            # elif request.method == AgentRPCMethod.TASK_CREATE:
+            #     result = Task.model_validate(result_entity)
+            # elif request.method == AgentRPCMethod.TASK_CANCEL:
+            #     result = Task.model_validate(result_entity)
+            # elif request.method == AgentRPCMethod.EVENT_SEND:
+            #     result = Event.model_validate(result_entity)
+            # else:
+            #     raise ValueError(f"Unsupported method: {request.method}")
+            # logger.info(f"AgentRPCResponse Result: {result}")
+            return AgentRPCResponse.model_validate(
+                {
+                    "id": request.id,
+                    "result": serialized_result,
+                    "error": None,
+                }
+            )
 
-    except ValidationError as e:
-        logger.error(f"Validation error in RPC request: {e}", exc_info=True)
-        error = JSONRPCError(code=-32602, message=f"Invalid parameters: {e}")
-        return AgentRPCResponse(id=request.id, error=error.model_dump(), result=None)
-    except Exception as e:
-        logger.error(f"Error handling JSON-RPC request: {e}", exc_info=True)
-        error = JSONRPCError(code=-32603, message=str(e))
-        return AgentRPCResponse(id=request.id, error=error.model_dump(), result=None)
+        except ValidationError as e:
+            logger.error(f"Validation error in RPC request: {e}", exc_info=True)
+            error = JSONRPCError(code=-32602, message=f"Invalid parameters: {e}")
+            rpc_call.fail(error.code, e)
+            return AgentRPCResponse(
+                id=request.id, error=error.model_dump(), result=None
+            )
+        except Exception as e:
+            logger.error(f"Error handling JSON-RPC request: {e}", exc_info=True)
+            error = JSONRPCError(code=-32603, message=str(e))
+            rpc_call.fail(error.code, e)
+            return AgentRPCResponse(
+                id=request.id, error=error.model_dump(), result=None
+            )
 
 
 async def _handle_streaming_rpc(
@@ -608,56 +620,60 @@ async def _handle_streaming_rpc(
     """Handle streaming JSON-RPC requests."""
 
     async def rpc_response_generator():
-        result_entity_async_iterator = None
-        try:
-            result_entity_async_iterator = await agents_acp_use_case.handle_rpc_request(
-                agent_id=agent_id,
-                agent_name=agent_name,
-                method=request.method,
-                params=request.params,
-                request_headers=request_headers,
-            )
-
-            if not isinstance(result_entity_async_iterator, AsyncIterator):
-                raise ValueError(
-                    f"Expected AsyncIterator, got {type(result_entity_async_iterator)}"
+        with rpc_request_timing(request.method.value, streaming=True) as rpc_call:
+            result_entity_async_iterator = None
+            try:
+                result_entity_async_iterator = (
+                    await agents_acp_use_case.handle_rpc_request(
+                        agent_id=agent_id,
+                        agent_name=agent_name,
+                        method=request.method,
+                        params=request.params,
+                        request_headers=request_headers,
+                    )
                 )
 
-            # At this point we know it's an AsyncIterator[TaskMessage]
-            async for task_message_update_entity in result_entity_async_iterator:
-                logger.debug(
-                    f"Streaming message chunk type: {type(task_message_update_entity).__name__}"
-                )
-                rpc_response = AgentRPCResponse.model_validate(
-                    {
-                        "id": request.id,
-                        "result": task_message_update_entity.model_dump(),
-                        "error": None,
-                    }
-                )
-                # Yield JSON bytes with newline for NDJSON format
-                yield rpc_response.model_dump_json().encode() + b"\n"
+                if not isinstance(result_entity_async_iterator, AsyncIterator):
+                    raise ValueError(
+                        f"Expected AsyncIterator, got {type(result_entity_async_iterator)}"
+                    )
 
-        except Exception as e:
-            logger.error(f"Error in streaming RPC response: {e}", exc_info=True)
-            # Yield error response
-            error_response = AgentRPCResponse(
-                id=request.id,
-                result=None,
-                error=JSONRPCError(code=-32603, message=str(e)).model_dump(),
-            )
-            yield error_response.model_dump_json().encode() + b"\n"
-        finally:
-            # CRITICAL: Ensure the async iterator is properly closed
-            # This ensures HTTP connections are released back to the pool
-            if result_entity_async_iterator is not None and hasattr(
-                result_entity_async_iterator, "aclose"
-            ):
-                try:
-                    await result_entity_async_iterator.aclose()
-                    logger.debug("Closed streaming iterator properly")
-                except Exception as e:
-                    logger.warning(f"Error closing streaming iterator: {e}")
+                # At this point we know it's an AsyncIterator[TaskMessage]
+                async for task_message_update_entity in result_entity_async_iterator:
+                    logger.debug(
+                        f"Streaming message chunk type: {type(task_message_update_entity).__name__}"
+                    )
+                    rpc_response = AgentRPCResponse.model_validate(
+                        {
+                            "id": request.id,
+                            "result": task_message_update_entity.model_dump(),
+                            "error": None,
+                        }
+                    )
+                    # Yield JSON bytes with newline for NDJSON format
+                    yield rpc_response.model_dump_json().encode() + b"\n"
+
+            except Exception as e:
+                logger.error(f"Error in streaming RPC response: {e}", exc_info=True)
+                rpc_call.fail(-32603, e)
+                # Yield error response
+                error_response = AgentRPCResponse(
+                    id=request.id,
+                    result=None,
+                    error=JSONRPCError(code=-32603, message=str(e)).model_dump(),
+                )
+                yield error_response.model_dump_json().encode() + b"\n"
+            finally:
+                # CRITICAL: Ensure the async iterator is properly closed
+                # This ensures HTTP connections are released back to the pool
+                if result_entity_async_iterator is not None and hasattr(
+                    result_entity_async_iterator, "aclose"
+                ):
+                    try:
+                        await result_entity_async_iterator.aclose()
+                        logger.debug("Closed streaming iterator properly")
+                    except Exception as e:
+                        logger.warning(f"Error closing streaming iterator: {e}")
 
     return StreamingResponse(
         rpc_response_generator(),

agentex/src/config/dependencies.py

@@ -19,7 +19,10 @@
 
 from src.config.environment_variables import Environment, EnvironmentVariables
 from src.utils.database import async_db_engine_creator
-from src.utils.db_metrics import PostgresMetricsCollector
+from src.utils.db_metrics import (
+    InstrumentedAsyncAdaptedQueuePool,
+    PostgresMetricsCollector,
+)
 from src.utils.logging import make_logger
 
 logger = make_logger(__name__)
@@ -97,6 +100,7 @@ async def load(self):
                 self.environment_variables.DATABASE_URL,
             ),
             echo=echo_db_engine,
+            poolclass=InstrumentedAsyncAdaptedQueuePool,  # emits pool wait_time/pending_requests/timeouts
             pool_size=async_db_pool_size,
             max_overflow=20,  # Allow 20 additional connections beyond pool_size when needed
             pool_pre_ping=True,
@@ -109,6 +113,7 @@ async def load(self):
                 self.environment_variables.DATABASE_URL,
             ),
             echo=echo_db_engine,
+            poolclass=InstrumentedAsyncAdaptedQueuePool,  # emits pool wait_time/pending_requests/timeouts
             pool_size=middleware_db_pool_size,
             max_overflow=10,  # Allow 10 additional connections for middleware
             pool_pre_ping=True,
@@ -188,6 +193,7 @@ async def load(self):
                 "postgresql+asyncpg://",
                 async_creator=async_db_engine_creator(read_only_db_url),
                 echo=echo_db_engine,
+                poolclass=InstrumentedAsyncAdaptedQueuePool,  # emits pool wait_time/pending_requests/timeouts
                 pool_size=async_db_pool_size,
                 max_overflow=20,
                 pool_pre_ping=True,