feat(observability): add LLMTrace model and operational metrics #533

Merged
freemo merged 1 commits from feature/m6-llm-trace-metrics into master 2026-03-03 22:12:14 +00:00
18 changed files with 2741 additions and 5 deletions
+6
View File
@@ -32,6 +32,12 @@
and DecisionService integration for strategy_choice + subplan_spawn recording.
Configurable via `planner_max_depth`, `planner_max_files_per_subplan`,
`planner_max_tokens_per_subplan`, `planner_min_files_per_subplan` settings. (#205)
- Added `LLMTrace` Pydantic v2 domain model and `llm_traces` database table with
`LLMTraceRepository` for persisting LLM call telemetry (tokens, cost, latency,
tool calls, context hash, streaming flag, retry count, error). Defined 14
`OperationalMetricKey` values with `MetricEntry` / `MetricCollector` for plan-level
metrics. `TraceService` provides recording, querying, metric computation, and
optional LangSmith forwarding when `LANGCHAIN_TRACING_V2=true`. (#500)
- Added `SafetyProfile` domain model with configurable safety constraints (allowed skill
categories, sandbox/checkpoint requirements, human-approval flag, cost/retry limits) and
integrated it into the `Action` model via `from_config`/`as_cli_dict`. Persistence backed
+126
View File
@@ -0,0 +1,126 @@
"""ASV benchmarks for LLM trace observability overhead.
Measures the time to create LLMTrace models, record traces via
TraceService, compute metrics, and exercise lifecycle hooks.
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.application.services.trace_service import TraceService # noqa: E402
from cleveragents.domain.models.observability.llm_trace import LLMTrace # noqa: E402
from cleveragents.domain.models.observability.metrics import ( # noqa: E402
MetricCollector,
OperationalMetricKey,
)
VALID_PLAN_ID = "01HX0000000000PPPPPPPPPPPP"
class InMemoryRepo:
"""Minimal in-memory repo for benchmarking."""
def __init__(self) -> None:
self._store: dict[str, LLMTrace] = {}
def save(self, trace: LLMTrace) -> None:
self._store[trace.trace_id] = trace
def get(self, trace_id: str) -> LLMTrace | None:
return self._store.get(trace_id)
def list_by_plan(self, plan_id: str) -> list[LLMTrace]:
return [t for t in self._store.values() if t.plan_id == plan_id]
def list_by_decision(self, decision_id: str) -> list[LLMTrace]:
return [t for t in self._store.values() if t.decision_id == decision_id]
def _make_trace(idx: int) -> LLMTrace:
idx_str = f"{idx:08d}"
ulid = f"01HX{'A' * (22 - len(idx_str))}{idx_str}"
return LLMTrace(
trace_id=ulid,
plan_id=VALID_PLAN_ID,
actor="planner",
provider="openai",
model="gpt-4o",
prompt_tokens=500 + idx,
completion_tokens=200 + idx,
cost_usd=0.01 + idx * 0.001,
latency_ms=300.0 + idx,
)
class LLMTraceModelCreationSuite:
"""Benchmark LLMTrace model creation."""
def time_create_single_trace(self) -> None:
"""Create a single LLMTrace model."""
_make_trace(0)
def time_create_100_traces(self) -> None:
"""Create 100 LLMTrace models."""
for i in range(100):
_make_trace(i)
class LLMTraceServiceSuite:
"""Benchmark TraceService operations."""
def setup(self) -> None:
self._svc = TraceService(
settings=MagicMock(),
repository=InMemoryRepo(), # type: ignore[arg-type]
)
self._traces = [_make_trace(i) for i in range(100)]
def time_record_100_traces(self) -> None:
"""Record 100 traces."""
for t in self._traces:
self._svc.record_trace(t)
def time_compute_metrics_100(self) -> None:
"""Compute metrics over 100 traces."""
for t in self._traces:
self._svc.record_trace(t)
self._svc.compute_metrics(VALID_PLAN_ID)
class MetricCollectorSuite:
"""Benchmark MetricCollector operations."""
def time_record_metric(self) -> None:
"""Create a single metric entry."""
MetricCollector.record(
OperationalMetricKey.PLAN_DURATION_MS,
1500.0,
VALID_PLAN_ID,
)
def time_plan_duration_convenience(self) -> None:
"""Use plan_duration convenience method."""
MetricCollector.plan_duration(VALID_PLAN_ID, 1500.0)
def time_lifecycle_hooks(self) -> None:
"""Exercise all lifecycle hooks."""
svc = TraceService(
settings=MagicMock(),
repository=InMemoryRepo(), # type: ignore[arg-type]
)
svc.on_plan_start(VALID_PLAN_ID)
svc.on_actor_invocation(VALID_PLAN_ID, "planner", 250.0)
svc.on_tool_execution(VALID_PLAN_ID, "file_write", errored=False)
+89
View File
@@ -0,0 +1,89 @@
# Observability — LLM Trace & Operational Metrics
## Overview
The observability subsystem records telemetry for every LLM provider
call and exposes aggregated operational metrics per plan. Trace data
drives cost analysis, latency profiling, and optional forwarding to
LangSmith for external analysis.
## LLMTrace Model
`cleveragents.domain.models.observability.llm_trace.LLMTrace`
Each trace captures one LLM invocation:
| Field | Type | Description |
|:-------------------|:------------------|:-----------------------------------------|
| `trace_id` | `str` (ULID) | Unique identifier for the trace |
| `plan_id` | `str` (ULID) | Plan that owns this trace |
| `decision_id` | `str \| None` | Decision that triggered the call |
| `actor` | `str` | Actor name |
| `provider` | `str` | Provider identifier (e.g. `openai`) |
| `model` | `str` | Model name (e.g. `gpt-4o`) |
| `prompt_tokens` | `int` | Prompt token count |
| `completion_tokens`| `int` | Completion token count |
| `cost_usd` | `float` | Estimated USD cost |
| `latency_ms` | `float` | Wall-clock latency (ms) |
| `tool_calls` | `list[dict]` | Tool call descriptors from the model |
| `context_hash` | `str \| None` | SHA-256 of the context window |
| `streaming` | `bool` | Whether streaming was used |
| `retry_count` | `int` | Retries before success |
| `error` | `str \| None` | Error message on failure |
| `timestamp` | `datetime` | When the trace was recorded |
The model is **frozen** (immutable) once created.
## Operational Metrics
`cleveragents.domain.models.observability.metrics.OperationalMetricKey`
14 metric keys grouped by subsystem:
| Key | Subsystem | Description |
|:---------------------------|:----------|:-----------------------------------|
| `PLAN_DURATION_MS` | Plan | Total plan execution time |
| `PLAN_TOTAL_COST_USD` | Plan | Aggregate cost across all LLM calls|
| `PLAN_DECISION_COUNT` | Plan | Number of decisions made |
| `SUBPLAN_COUNT` | Plan | Number of subplans spawned |
| `ACTOR_INVOCATION_COUNT` | Actor | Actor call count |
| `ACTOR_LATENCY_MS` | Actor | Actor invocation latency |
| `TOOL_INVOCATION_COUNT` | Tool | Tool execution count |
| `TOOL_ERROR_RATE` | Tool | Tool error occurrences |
| `CONTEXT_BUILD_TIME_MS` | Context | Context assembly time |
| `CONTEXT_TOKEN_COUNT` | Context | Context token count |
| `LLM_CALL_COUNT` | LLM | Total LLM calls |
| `LLM_TOTAL_TOKENS` | LLM | Sum of all tokens |
| `LLM_TOTAL_COST_USD` | LLM | Sum of all costs |
| `LLM_AVG_LATENCY_MS` | LLM | Average call latency |
## TraceService
`cleveragents.application.services.trace_service.TraceService`
Registered in the DI container as `trace_service`.
### Methods
| Method | Description |
|:------------------------|:------------------------------------------------|
| `record_trace(trace)` | Persist trace and optionally forward to LangSmith|
| `get_traces(plan_id)` | List traces for a plan |
| `get_trace(trace_id)` | Retrieve a single trace |
| `compute_metrics(plan_id)` | Compute LLM-level metrics from traces |
| `on_plan_start(plan_id)` | Lifecycle hook: plan start |
| `on_actor_invocation(...)` | Lifecycle hook: actor invocation |
| `on_tool_execution(...)` | Lifecycle hook: tool execution |
### LangSmith Forwarding
When `LANGCHAIN_TRACING_V2=true` is set in the environment,
`record_trace` automatically forwards each trace to LangSmith
via the `langsmith` SDK. Forwarding is best-effort: failures
are logged but do not raise exceptions.
## Database
Table `llm_traces` with indexes on `plan_id`, `decision_id`,
`actor`, and `provider`. Repository:
`cleveragents.infrastructure.database.llm_trace_repository.LLMTraceRepository`.
+276
View File
@@ -0,0 +1,276 @@
Feature: LLM trace observability
As a platform operator
I want to record LLM call telemetry and compute operational metrics
So that I can analyse cost, latency, and usage patterns
Background:
Given a trace service with an in-memory repository
# --- model validation --------------------------------------------------
Scenario: Create a valid LLM trace
Given a valid LLM trace payload
When I create the LLM trace model
Then the trace should have all required fields populated
And the trace should be frozen
Scenario: Reject trace with invalid trace_id
When I create a trace with an invalid trace_id
Then a validation error should be raised
Scenario: Reject trace with negative prompt tokens
When I create a trace with negative prompt_tokens
Then a validation error should be raised
Scenario: Reject trace with negative cost
When I create a trace with negative cost_usd
Then a validation error should be raised
Scenario: Trace with optional fields omitted
Given a minimal LLM trace payload
When I create the LLM trace model
Then the trace decision_id should be None
And the trace tool_calls should be empty
And the trace context_hash should be None
And the trace streaming should be False
And the trace retry_count should be 0
And the trace error should be None
# --- query model -------------------------------------------------------
Scenario: Create a valid trace query
When I create a trace query with plan_id filter
Then the query should have default limit 100
And the query should have default offset 0
Scenario: Reject query with limit below 1
When I create a trace query with limit 0
Then a validation error should be raised
# --- metric keys -------------------------------------------------------
Scenario: All 14 operational metric keys are defined
Then the OperationalMetricKey enum should have exactly 14 members
Scenario: Metric keys have correct string values
Then PLAN_DURATION_MS should equal "plan_duration_ms"
And LLM_AVG_LATENCY_MS should equal "llm_avg_latency_ms"
And SUBPLAN_COUNT should equal "subplan_count"
# --- metric entry ------------------------------------------------------
Scenario: Create a metric entry via collector
Given a plan_id for metrics
When I record a PLAN_DURATION_MS metric with value 1500.0
Then the metric entry key should be PLAN_DURATION_MS
And the metric entry value should be 1500.0
And the metric entry should have a timestamp
Scenario: Metric entry with labels
Given a plan_id for metrics
When I record an ACTOR_LATENCY_MS metric with actor label
Then the metric entry labels should contain actor
# --- trace recording ---------------------------------------------------
Scenario: Record and retrieve a trace
Given a valid LLM trace
When I record the trace via the service
Then I should be able to retrieve it by trace_id
And I should be able to list it by plan_id
Scenario: Record multiple traces for a plan
Given three valid LLM traces for the same plan
When I record all traces via the service
Then listing by plan_id should return 3 traces
Scenario: List traces by decision
Given two traces with the same decision_id
When I record all traces via the service
Then listing by decision_id should return 2 traces
Scenario: Get non-existent trace returns None
When I query a non-existent trace_id
Then the trace query result should be None
# --- metric computation ------------------------------------------------
Scenario: Compute metrics for a plan with traces
Given three valid LLM traces for the same plan
When I record all traces via the service
And I compute metrics for the plan
Then I should get LLM_CALL_COUNT equal to 3
And I should get LLM_TOTAL_TOKENS greater than 0
And I should get LLM_TOTAL_COST_USD greater than 0
And I should get LLM_AVG_LATENCY_MS greater than 0
Scenario: Compute metrics for a plan with no traces
When I compute metrics for a plan with no traces
Then the metrics list should be empty
# --- lifecycle hooks ---------------------------------------------------
Scenario: Plan start hook returns decision count metric
Given a plan_id for metrics
When I call on_plan_start
Then I should get a PLAN_DECISION_COUNT metric with value 0
Scenario: Actor invocation hook returns metrics
Given a plan_id for metrics
When I call on_actor_invocation with actor "planner" and latency 250.0
Then I should get ACTOR_INVOCATION_COUNT and ACTOR_LATENCY_MS metrics
Scenario: Tool execution hook returns metrics
Given a plan_id for metrics
When I call on_tool_execution with tool "file_write" and no error
Then I should get a TOOL_INVOCATION_COUNT metric
And I should not get a TOOL_ERROR_RATE metric
Scenario: Tool execution hook with error returns error rate
Given a plan_id for metrics
When I call on_tool_execution with tool "file_write" and error
Then I should get both TOOL_INVOCATION_COUNT and TOOL_ERROR_RATE metrics
# --- LangSmith forwarding ----------------------------------------------
Scenario: LangSmith forwarding when env var is set
Given LANGCHAIN_TRACING_V2 is set to "true"
And a valid LLM trace
When I record the trace via the service
Then the LangSmith forwarder should have been called
Scenario: LangSmith forwarding skipped when env var is not set
Given LANGCHAIN_TRACING_V2 is not set
And a valid LLM trace
When I record the trace via the service
Then the LangSmith forwarder should not have been called
Scenario: LangSmith forwarding failure is logged not raised
Given LANGCHAIN_TRACING_V2 is set to "true"
And a valid LLM trace
And the LangSmith forwarder will raise an exception
When I record the trace via the service
Then no exception should propagate
And the trace should still be persisted
# --- convenience collectors --------------------------------------------
Scenario: MetricCollector plan_duration convenience
Given a plan_id for metrics
When I call MetricCollector.plan_duration with 2000.0
Then the metric key should be PLAN_DURATION_MS
And the metric value should be 2000.0
Scenario: MetricCollector plan_cost convenience
Given a plan_id for metrics
When I call MetricCollector.plan_cost with 0.05
Then the metric key should be PLAN_TOTAL_COST_USD
And the metric value should be 0.05
Scenario: MetricCollector llm_call_count convenience
Given a plan_id for metrics
When I call MetricCollector.llm_call_count with 10
Then the metric key should be LLM_CALL_COUNT
And the metric value should be 10.0
# --- database model ----------------------------------------------------
Scenario: LLMTraceModel table exists
Then the LLMTraceModel should have tablename "llm_traces"
# --- repository integration (SQLAlchemy in-memory) ---------------------
Scenario: Repository rejects None session_factory
When I create a repository with None session_factory
Then a ValueError should be raised for the repository
Scenario: Repository save and get roundtrip
Given a SQLAlchemy in-memory repository
And a valid LLM trace
When I save the trace via the repository
Then I should be able to get it by trace_id from the repository
Scenario: Repository save trace with tool calls
Given a SQLAlchemy in-memory repository
And a valid LLM trace with tool calls
When I save the trace via the repository
Then I should be able to get it by trace_id from the repository
And the retrieved trace should have tool calls
Scenario: Repository get non-existent trace returns None
Given a SQLAlchemy in-memory repository
When I get a non-existent trace from the repository
Then the repository result should be None
Scenario: Repository list by plan returns matching traces
Given a SQLAlchemy in-memory repository
And two traces for the same plan in the repository
When I save both traces via the repository
Then listing by plan from the repository should return 2 traces
Scenario: Repository list by plan returns empty for unknown plan
Given a SQLAlchemy in-memory repository
When I list traces for a non-existent plan from the repository
Then the repository list should be empty
Scenario: Repository list by decision returns matching traces
Given a SQLAlchemy in-memory repository
And two traces for the same decision in the repository
When I save both traces via the repository
Then listing by decision from the repository should return 2 traces
Scenario: Repository list by decision returns empty for unknown decision
Given a SQLAlchemy in-memory repository
When I list traces for a non-existent decision from the repository
Then the repository list should be empty
Scenario: Repository save raises DatabaseError on failure
Given a SQLAlchemy repository with a broken session
And a valid LLM trace
When I save the trace via the broken repository
Then a DatabaseError should be raised with message containing "save"
Scenario: Repository get raises DatabaseError on failure
Given a SQLAlchemy repository with a broken session
When I get a trace from the broken repository
Then a DatabaseError should be raised with message containing "get"
Scenario: Repository list_by_plan raises DatabaseError on failure
Given a SQLAlchemy repository with a broken session
When I list by plan from the broken repository
Then a DatabaseError should be raised with message containing "list"
Scenario: Repository list_by_decision raises DatabaseError on failure
Given a SQLAlchemy repository with a broken session
When I list by decision from the broken repository
Then a DatabaseError should be raised with message containing "list"
# --- LangSmith internal forwarder detail --------------------------------
Scenario: LangSmith forwarder sends trace data via SDK
Given a valid LLM trace
And the langsmith SDK is mocked as available
When I call the internal LangSmith forwarder
Then the langsmith Client create_run should have been called
Scenario: LangSmith forwarder includes error field when present
Given a valid LLM trace with error "connection timeout"
And the langsmith SDK is mocked as available
When I call the internal LangSmith forwarder
Then the langsmith Client create_run should have been called with error
Scenario: LangSmith forwarder skips when SDK not installed
Given a valid LLM trace
And the langsmith SDK is not importable
When I call the internal LangSmith forwarder
Then no langsmith Client should have been created
Scenario: LangSmith enabled check is case insensitive
Given LANGCHAIN_TRACING_V2 is set to "TRUE"
Then the langsmith_enabled check should return True
Scenario: Repository save trace with no tool calls stores null
Given a SQLAlchemy in-memory repository
And a valid LLM trace
When I save the trace via the repository
Then the raw tool_calls_json in the database should be null
+14 -3
View File
@@ -7,6 +7,7 @@ support ``--format`` (rich/plain/json/yaml) and ``--check`` for diagnostics.
from __future__ import annotations
from typing import Any
from unittest.mock import patch as _patch
from behave import then, when
from behave.runner import Context
@@ -18,6 +19,16 @@ from cleveragents.cli.commands.system import (
)
from cleveragents.cli.formatting import format_output
def _build_info_data_stable() -> dict[str, Any]:
"""Build info data with deterministic server_mode for tests."""
with _patch(
"cleveragents.cli.commands.server.resolve_server_mode",
return_value="disabled",
):
return build_info_data()
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@@ -113,8 +124,8 @@ def step_system_version_json_nested_key(context: Context, key: str) -> None:
@when("I run the system info command")
def step_run_system_info(context: Context) -> None:
"""Run info with default (rich) format."""
data = build_info_data()
"""Run info in default (rich) format."""
data = _build_info_data_stable()
context.sys_info_data = data
context.sys_info_output = _format_data(data, "rich")
@@ -122,7 +133,7 @@ def step_run_system_info(context: Context) -> None:
@when('I run the system info command with format "{fmt}"')
def step_run_system_info_fmt(context: Context, fmt: str) -> None:
"""Run info with a specific output format."""
data = build_info_data()
data = _build_info_data_stable()
context.sys_info_data = data
context.sys_info_output = _format_data(data, fmt)
+948
View File
@@ -0,0 +1,948 @@
"""Step definitions for the LLM trace observability feature."""
from __future__ import annotations
import builtins
import os
from datetime import datetime
from typing import Any
from unittest.mock import MagicMock, patch
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from sqlalchemy import create_engine
from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
from sqlalchemy.orm import sessionmaker
from cleveragents.application.services.trace_service import (
TraceService,
_forward_trace_to_langsmith,
)
from cleveragents.core.exceptions import DatabaseError as AppDatabaseError
from cleveragents.domain.models.observability.llm_trace import LLMTrace, LLMTraceQuery
from cleveragents.domain.models.observability.metrics import (
MetricCollector,
MetricEntry,
OperationalMetricKey,
)
from cleveragents.infrastructure.database.llm_trace_repository import LLMTraceRepository
from cleveragents.infrastructure.database.models import Base, LLMTraceModel
# ---------------------------------------------------------------------------
# Test constants
# ---------------------------------------------------------------------------
VALID_ULID_1 = "01HXAAAAAAAAAAAAAAAAAAAAAA"
VALID_ULID_2 = "01HXBBBBBBBBBBBBBBBBBBBBBB"
VALID_ULID_3 = "01HXCCCCCCCCCCCCCCCCCCCCCC"
VALID_PLAN_ID = "01HX0000000000PPPPPPPPPPPP"
VALID_DECISION_ID = "01HX0000000000DDDDDDDDDDDD"
def _make_trace(
trace_id: str = VALID_ULID_1,
plan_id: str = VALID_PLAN_ID,
decision_id: str | None = None,
actor: str = "planner",
provider: str = "openai",
model: str = "gpt-4o",
prompt_tokens: int = 500,
completion_tokens: int = 200,
cost_usd: float = 0.01,
latency_ms: float = 350.0,
**kwargs: Any,
) -> LLMTrace:
"""Build a test LLMTrace with sensible defaults."""
return LLMTrace(
trace_id=trace_id,
plan_id=plan_id,
decision_id=decision_id,
actor=actor,
provider=provider,
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
cost_usd=cost_usd,
latency_ms=latency_ms,
**kwargs,
)
# ---------------------------------------------------------------------------
# In-memory repository stub
# ---------------------------------------------------------------------------
class InMemoryLLMTraceRepository:
"""Minimal in-memory repository for test isolation."""
def __init__(self) -> None:
self._store: dict[str, LLMTrace] = {}
def save(self, trace: LLMTrace) -> None:
self._store[trace.trace_id] = trace
def get(self, trace_id: str) -> LLMTrace | None:
return self._store.get(trace_id)
def list_by_plan(self, plan_id: str) -> list[LLMTrace]:
return sorted(
(t for t in self._store.values() if t.plan_id == plan_id),
key=lambda t: t.timestamp.isoformat(),
)
def list_by_decision(self, decision_id: str) -> list[LLMTrace]:
return sorted(
(t for t in self._store.values() if t.decision_id == decision_id),
key=lambda t: t.timestamp.isoformat(),
)
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a trace service with an in-memory repository")
def step_create_service(context: Context) -> None:
context.repo = InMemoryLLMTraceRepository()
context.settings = MagicMock()
context.service = TraceService(
settings=context.settings,
repository=context.repo, # type: ignore[arg-type]
)
context.error = None
context.traces_list = [] # list[LLMTrace]
context.metrics_list = [] # list[MetricEntry]
context.langsmith_called = False
# Ensure LANGCHAIN_TRACING_V2 is unset by default
os.environ.pop("LANGCHAIN_TRACING_V2", None)
# ---------------------------------------------------------------------------
# Model validation
# ---------------------------------------------------------------------------
@given("a valid LLM trace payload")
def step_valid_payload(context: Context) -> None:
context.trace_kwargs = {
"trace_id": VALID_ULID_1,
"plan_id": VALID_PLAN_ID,
"actor": "planner",
"provider": "openai",
"model": "gpt-4o",
"prompt_tokens": 500,
"completion_tokens": 200,
"cost_usd": 0.01,
"latency_ms": 350.0,
}
@given("a minimal LLM trace payload")
def step_minimal_payload(context: Context) -> None:
context.trace_kwargs = {
"trace_id": VALID_ULID_1,
"plan_id": VALID_PLAN_ID,
"actor": "planner",
"provider": "openai",
"model": "gpt-4o",
"prompt_tokens": 0,
"completion_tokens": 0,
"cost_usd": 0.0,
"latency_ms": 0.0,
}
@when("I create the LLM trace model")
def step_create_model(context: Context) -> None:
try:
context.trace = LLMTrace(**context.trace_kwargs)
except ValidationError as exc:
context.error = exc
@when("I create a trace with an invalid trace_id")
def step_invalid_trace_id(context: Context) -> None:
try:
_make_trace(trace_id="INVALID")
except ValidationError as exc:
context.error = exc
@when("I create a trace with negative prompt_tokens")
def step_negative_prompt(context: Context) -> None:
try:
_make_trace(prompt_tokens=-1)
except ValidationError as exc:
context.error = exc
@when("I create a trace with negative cost_usd")
def step_negative_cost(context: Context) -> None:
try:
_make_trace(cost_usd=-0.5)
except ValidationError as exc:
context.error = exc
@then("the trace should have all required fields populated")
def step_all_fields(context: Context) -> None:
t = context.trace
assert t.trace_id == VALID_ULID_1
assert t.plan_id == VALID_PLAN_ID
assert t.actor == "planner"
assert t.provider == "openai"
assert t.model == "gpt-4o"
assert t.prompt_tokens == 500
assert t.completion_tokens == 200
@then("the trace should be frozen")
def step_frozen(context: Context) -> None:
try:
context.trace.actor = "modified" # type: ignore[misc]
raise AssertionError("Expected frozen model to reject assignment")
except ValidationError:
pass
@then("the trace decision_id should be None")
def step_decision_none(context: Context) -> None:
assert context.trace.decision_id is None
@then("the trace tool_calls should be empty")
def step_tool_calls_empty(context: Context) -> None:
assert context.trace.tool_calls == []
@then("the trace context_hash should be None")
def step_context_hash_none(context: Context) -> None:
assert context.trace.context_hash is None
@then("the trace streaming should be False")
def step_streaming_false(context: Context) -> None:
assert context.trace.streaming is False
@then("the trace retry_count should be 0")
def step_retry_zero(context: Context) -> None:
assert context.trace.retry_count == 0
@then("the trace error should be None")
def step_error_none(context: Context) -> None:
assert context.trace.error is None
# ---------------------------------------------------------------------------
# Query model
# ---------------------------------------------------------------------------
@when("I create a trace query with plan_id filter")
def step_create_query(context: Context) -> None:
context.query = LLMTraceQuery(plan_id=VALID_PLAN_ID)
@when("I create a trace query with limit {limit:d}")
def step_query_bad_limit(context: Context, limit: int) -> None:
try:
LLMTraceQuery(limit=limit)
except ValidationError as exc:
context.error = exc
@then("the query should have default limit {limit:d}")
def step_query_limit(context: Context, limit: int) -> None:
assert context.query.limit == limit
@then("the query should have default offset {offset:d}")
def step_query_offset(context: Context, offset: int) -> None:
assert context.query.offset == offset
# ---------------------------------------------------------------------------
# Metric keys
# ---------------------------------------------------------------------------
@then("the OperationalMetricKey enum should have exactly {count:d} members")
def step_metric_key_count(context: Context, count: int) -> None:
assert len(OperationalMetricKey) == count
@then('PLAN_DURATION_MS should equal "{value}"')
def step_plan_duration_value(context: Context, value: str) -> None:
assert OperationalMetricKey.PLAN_DURATION_MS.value == value
@then('LLM_AVG_LATENCY_MS should equal "{value}"')
def step_llm_avg_latency_value(context: Context, value: str) -> None:
assert OperationalMetricKey.LLM_AVG_LATENCY_MS.value == value
@then('SUBPLAN_COUNT should equal "{value}"')
def step_subplan_count_value(context: Context, value: str) -> None:
assert OperationalMetricKey.SUBPLAN_COUNT.value == value
# ---------------------------------------------------------------------------
# Metric entry
# ---------------------------------------------------------------------------
@given("a plan_id for metrics")
def step_plan_id_for_metrics(context: Context) -> None:
context.metric_plan_id = VALID_PLAN_ID
@when("I record a PLAN_DURATION_MS metric with value {value:g}")
def step_record_metric(context: Context, value: float) -> None:
context.metric_entry = MetricCollector.record(
OperationalMetricKey.PLAN_DURATION_MS,
value,
context.metric_plan_id,
)
@when("I record an ACTOR_LATENCY_MS metric with actor label")
def step_record_metric_with_label(context: Context) -> None:
context.metric_entry = MetricCollector.record(
OperationalMetricKey.ACTOR_LATENCY_MS,
150.0,
context.metric_plan_id,
labels={"actor": "planner"},
)
@then("the metric entry key should be PLAN_DURATION_MS")
def step_metric_key_plan_duration(context: Context) -> None:
assert context.metric_entry.key == OperationalMetricKey.PLAN_DURATION_MS
@then("the metric entry value should be {value:g}")
def step_metric_value(context: Context, value: float) -> None:
assert context.metric_entry.value == value
@then("the metric entry should have a timestamp")
def step_metric_timestamp(context: Context) -> None:
assert isinstance(context.metric_entry.timestamp, datetime)
@then("the metric entry labels should contain actor")
def step_metric_labels(context: Context) -> None:
assert "actor" in context.metric_entry.labels
# ---------------------------------------------------------------------------
# Trace recording
# ---------------------------------------------------------------------------
@given("a valid LLM trace")
def step_valid_trace(context: Context) -> None:
context.trace = _make_trace()
@given("three valid LLM traces for the same plan")
def step_three_traces(context: Context) -> None:
context.traces_list = [
_make_trace(
trace_id=VALID_ULID_1, prompt_tokens=500, cost_usd=0.01, latency_ms=300.0
),
_make_trace(
trace_id=VALID_ULID_2, prompt_tokens=300, cost_usd=0.005, latency_ms=200.0
),
_make_trace(
trace_id=VALID_ULID_3, prompt_tokens=700, cost_usd=0.02, latency_ms=500.0
),
]
@given("two traces with the same decision_id")
def step_two_traces_same_decision(context: Context) -> None:
context.traces_list = [
_make_trace(trace_id=VALID_ULID_1, decision_id=VALID_DECISION_ID),
_make_trace(trace_id=VALID_ULID_2, decision_id=VALID_DECISION_ID),
]
@when("I record the trace via the service")
def step_record_trace(context: Context) -> None:
def mock_internal_forward(trace: LLMTrace) -> None:
context.langsmith_called = True
if getattr(context, "langsmith_will_fail", False):
raise RuntimeError("LangSmith connection failed")
try:
with patch(
"cleveragents.application.services.trace_service._forward_trace_to_langsmith",
side_effect=mock_internal_forward,
):
context.service.record_trace(context.trace)
except Exception as exc:
context.error = exc
@when("I record all traces via the service")
def step_record_all_traces(context: Context) -> None:
for t in context.traces_list:
context.service.record_trace(t)
@then("I should be able to retrieve it by trace_id")
def step_retrieve_by_id(context: Context) -> None:
result = context.service.get_trace(context.trace.trace_id)
assert result is not None
assert result.trace_id == context.trace.trace_id
@then("I should be able to list it by plan_id")
def step_list_by_plan(context: Context) -> None:
results = context.service.get_traces(context.trace.plan_id)
assert len(results) >= 1
assert any(t.trace_id == context.trace.trace_id for t in results)
@then("listing by plan_id should return {count:d} traces")
def step_list_count(context: Context, count: int) -> None:
plan_id = context.traces_list[0].plan_id
results = context.service.get_traces(plan_id)
assert len(results) == count
@then("listing by decision_id should return {count:d} traces")
def step_list_by_decision_count(context: Context, count: int) -> None:
decision_id = context.traces_list[0].decision_id
assert decision_id is not None
results = context.service.get_traces_by_decision(decision_id)
assert len(results) == count
@when("I query a non-existent trace_id")
def step_query_missing(context: Context) -> None:
context.result = context.service.get_trace("01HXZZZZZZZZZZZZZZZZZZZZZZ")
@then("the trace query result should be None")
def step_result_none(context: Context) -> None:
assert context.result is None
# ---------------------------------------------------------------------------
# Metric computation
# ---------------------------------------------------------------------------
@when("I compute metrics for the plan")
def step_compute_metrics(context: Context) -> None:
plan_id = context.traces_list[0].plan_id
context.metrics_list = context.service.compute_metrics(plan_id)
@when("I compute metrics for a plan with no traces")
def step_compute_empty(context: Context) -> None:
context.metrics_list = context.service.compute_metrics("01HXEEEEEEEEEEEEEEEEEEEEEE")
@then("I should get LLM_CALL_COUNT equal to {count:d}")
def step_llm_call_count(context: Context, count: int) -> None:
entry = _find_metric(context.metrics_list, OperationalMetricKey.LLM_CALL_COUNT)
assert entry is not None
assert entry.value == float(count)
@then("I should get LLM_TOTAL_TOKENS greater than {value:d}")
def step_llm_total_tokens(context: Context, value: int) -> None:
entry = _find_metric(context.metrics_list, OperationalMetricKey.LLM_TOTAL_TOKENS)
assert entry is not None
assert entry.value > value
@then("I should get LLM_TOTAL_COST_USD greater than {value:d}")
def step_llm_total_cost(context: Context, value: int) -> None:
entry = _find_metric(context.metrics_list, OperationalMetricKey.LLM_TOTAL_COST_USD)
assert entry is not None
assert entry.value > value
@then("I should get LLM_AVG_LATENCY_MS greater than {value:d}")
def step_llm_avg_latency(context: Context, value: int) -> None:
entry = _find_metric(context.metrics_list, OperationalMetricKey.LLM_AVG_LATENCY_MS)
assert entry is not None
assert entry.value > value
@then("the metrics list should be empty")
def step_metrics_empty(context: Context) -> None:
assert len(context.metrics_list) == 0
def _find_metric(
metrics: list[MetricEntry], key: OperationalMetricKey
) -> MetricEntry | None:
for m in metrics:
if m.key == key:
return m
return None
# ---------------------------------------------------------------------------
# Lifecycle hooks
# ---------------------------------------------------------------------------
@when("I call on_plan_start")
def step_on_plan_start(context: Context) -> None:
context.metric_entry = context.service.on_plan_start(context.metric_plan_id)
@then("I should get a PLAN_DECISION_COUNT metric with value {value:d}")
def step_plan_decision_count(context: Context, value: int) -> None:
assert context.metric_entry is not None
assert context.metric_entry.key == OperationalMetricKey.PLAN_DECISION_COUNT
assert context.metric_entry.value == float(value)
@when('I call on_actor_invocation with actor "{actor}" and latency {latency:g}')
def step_on_actor_invocation(context: Context, actor: str, latency: float) -> None:
context.hook_metrics = context.service.on_actor_invocation(
context.metric_plan_id, actor, latency
)
@then("I should get ACTOR_INVOCATION_COUNT and ACTOR_LATENCY_MS metrics")
def step_actor_metrics(context: Context) -> None:
keys = {m.key for m in context.hook_metrics}
assert OperationalMetricKey.ACTOR_INVOCATION_COUNT in keys
assert OperationalMetricKey.ACTOR_LATENCY_MS in keys
@when('I call on_tool_execution with tool "{tool}" and no error')
def step_on_tool_no_error(context: Context, tool: str) -> None:
context.hook_metrics = context.service.on_tool_execution(
context.metric_plan_id, tool, errored=False
)
@when('I call on_tool_execution with tool "{tool}" and error')
def step_on_tool_error(context: Context, tool: str) -> None:
context.hook_metrics = context.service.on_tool_execution(
context.metric_plan_id, tool, errored=True
)
@then("I should get a TOOL_INVOCATION_COUNT metric")
def step_tool_invocation(context: Context) -> None:
keys = {m.key for m in context.hook_metrics}
assert OperationalMetricKey.TOOL_INVOCATION_COUNT in keys
@then("I should not get a TOOL_ERROR_RATE metric")
def step_no_tool_error_rate(context: Context) -> None:
keys = {m.key for m in context.hook_metrics}
assert OperationalMetricKey.TOOL_ERROR_RATE not in keys
@then("I should get both TOOL_INVOCATION_COUNT and TOOL_ERROR_RATE metrics")
def step_tool_both_metrics(context: Context) -> None:
keys = {m.key for m in context.hook_metrics}
assert OperationalMetricKey.TOOL_INVOCATION_COUNT in keys
assert OperationalMetricKey.TOOL_ERROR_RATE in keys
# ---------------------------------------------------------------------------
# LangSmith forwarding
# ---------------------------------------------------------------------------
@given('LANGCHAIN_TRACING_V2 is set to "{value}"')
def step_langsmith_enabled(context: Context, value: str) -> None:
os.environ["LANGCHAIN_TRACING_V2"] = value
@given("LANGCHAIN_TRACING_V2 is not set")
def step_langsmith_disabled(context: Context) -> None:
os.environ.pop("LANGCHAIN_TRACING_V2", None)
@given("the LangSmith forwarder will raise an exception")
def step_langsmith_error(context: Context) -> None:
context.langsmith_will_fail = True
@then("the LangSmith forwarder should have been called")
def step_langsmith_called(context: Context) -> None:
assert context.langsmith_called is True
@then("the LangSmith forwarder should not have been called")
def step_langsmith_not_called(context: Context) -> None:
assert context.langsmith_called is False
@then("no exception should propagate")
def step_no_exception(context: Context) -> None:
assert context.error is None
@then("the trace should still be persisted")
def step_trace_persisted(context: Context) -> None:
result = context.service.get_trace(context.trace.trace_id)
assert result is not None
# ---------------------------------------------------------------------------
# Convenience collectors
# ---------------------------------------------------------------------------
@when("I call MetricCollector.plan_duration with {value:g}")
def step_collector_plan_duration(context: Context, value: float) -> None:
context.metric_entry = MetricCollector.plan_duration(context.metric_plan_id, value)
@when("I call MetricCollector.plan_cost with {value:g}")
def step_collector_plan_cost(context: Context, value: float) -> None:
context.metric_entry = MetricCollector.plan_cost(context.metric_plan_id, value)
@when("I call MetricCollector.llm_call_count with {value:d}")
def step_collector_llm_call_count(context: Context, value: int) -> None:
context.metric_entry = MetricCollector.llm_call_count(
context.metric_plan_id, float(value)
)
@then("the metric key should be PLAN_DURATION_MS")
def step_key_plan_duration(context: Context) -> None:
assert context.metric_entry.key == OperationalMetricKey.PLAN_DURATION_MS
@then("the metric key should be PLAN_TOTAL_COST_USD")
def step_key_plan_cost(context: Context) -> None:
assert context.metric_entry.key == OperationalMetricKey.PLAN_TOTAL_COST_USD
@then("the metric key should be LLM_CALL_COUNT")
def step_key_llm_call(context: Context) -> None:
assert context.metric_entry.key == OperationalMetricKey.LLM_CALL_COUNT
@then("the metric value should be {value:g}")
def step_metric_exact_value(context: Context, value: float) -> None:
assert context.metric_entry.value == value
# ---------------------------------------------------------------------------
# Database model
# ---------------------------------------------------------------------------
@then('the LLMTraceModel should have tablename "{name}"')
def step_table_name(context: Context, name: str) -> None:
assert LLMTraceModel.__tablename__ == name
# ---------------------------------------------------------------------------
# Repository integration (SQLAlchemy in-memory)
# ---------------------------------------------------------------------------
def _create_in_memory_engine_and_session():
"""Create an in-memory SQLite engine and session factory for testing."""
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
factory = sessionmaker(bind=engine)
return engine, factory
@when("I create a repository with None session_factory")
def step_repo_none_factory(context: Context) -> None:
try:
LLMTraceRepository(session_factory=None) # type: ignore[arg-type]
except ValueError as exc:
context.error = exc
@then("a ValueError should be raised for the repository")
def step_repo_valueerror_raised(context: Context) -> None:
assert context.error is not None, "Expected a ValueError"
assert isinstance(context.error, ValueError)
@given("a SQLAlchemy in-memory repository")
def step_sqla_repo(context: Context) -> None:
engine, factory = _create_in_memory_engine_and_session()
context.sqla_engine = engine
context.sqla_factory = factory
context.sqla_repo = LLMTraceRepository(session_factory=factory)
@given("a valid LLM trace with tool calls")
def step_trace_with_tool_calls(context: Context) -> None:
context.trace = _make_trace(
tool_calls=[{"name": "search", "arguments": {"q": "test"}}],
)
@when("I save the trace via the repository")
def step_save_via_repo(context: Context) -> None:
context.sqla_repo.save(context.trace)
@then("I should be able to get it by trace_id from the repository")
def step_get_from_repo(context: Context) -> None:
result = context.sqla_repo.get(context.trace.trace_id)
assert result is not None
assert result.trace_id == context.trace.trace_id
assert result.plan_id == context.trace.plan_id
assert result.actor == context.trace.actor
assert result.provider == context.trace.provider
assert result.model == context.trace.model
assert result.prompt_tokens == context.trace.prompt_tokens
assert result.completion_tokens == context.trace.completion_tokens
context.repo_result = result
@then("the retrieved trace should have tool calls")
def step_retrieved_has_tool_calls(context: Context) -> None:
result = context.sqla_repo.get(context.trace.trace_id)
assert result is not None
assert len(result.tool_calls) > 0
assert result.tool_calls[0]["name"] == "search"
@when("I get a non-existent trace from the repository")
def step_get_nonexistent_from_repo(context: Context) -> None:
context.repo_result = context.sqla_repo.get("01HXZZZZZZZZZZZZZZZZZZZZZZ")
@then("the repository result should be None")
def step_repo_result_none(context: Context) -> None:
assert context.repo_result is None
@given("two traces for the same plan in the repository")
def step_two_traces_same_plan_repo(context: Context) -> None:
context.repo_traces = [
_make_trace(trace_id=VALID_ULID_1, plan_id=VALID_PLAN_ID),
_make_trace(trace_id=VALID_ULID_2, plan_id=VALID_PLAN_ID),
]
@when("I save both traces via the repository")
def step_save_both_repo(context: Context) -> None:
for t in context.repo_traces:
context.sqla_repo.save(t)
@then("listing by plan from the repository should return {count:d} traces")
def step_list_by_plan_repo(context: Context, count: int) -> None:
results = context.sqla_repo.list_by_plan(VALID_PLAN_ID)
assert len(results) == count
@when("I list traces for a non-existent plan from the repository")
def step_list_nonexistent_plan_repo(context: Context) -> None:
context.repo_list_result = context.sqla_repo.list_by_plan(
"01HXZZZZZZZZZZZZZZZZZZZZZZ"
)
@then("the repository list should be empty")
def step_repo_list_empty(context: Context) -> None:
assert len(context.repo_list_result) == 0
@given("two traces for the same decision in the repository")
def step_two_traces_same_decision_repo(context: Context) -> None:
context.repo_traces = [
_make_trace(
trace_id=VALID_ULID_1,
plan_id=VALID_PLAN_ID,
decision_id=VALID_DECISION_ID,
),
_make_trace(
trace_id=VALID_ULID_2,
plan_id=VALID_PLAN_ID,
decision_id=VALID_DECISION_ID,
),
]
@then("listing by decision from the repository should return {count:d} traces")
def step_list_by_decision_repo(context: Context, count: int) -> None:
results = context.sqla_repo.list_by_decision(VALID_DECISION_ID)
assert len(results) == count
@when("I list traces for a non-existent decision from the repository")
def step_list_nonexistent_decision_repo(context: Context) -> None:
context.repo_list_result = context.sqla_repo.list_by_decision(
"01HXZZZZZZZZZZZZZZZZZZZZZZ"
)
# --- Repository error paths -----------------------------------------------
class _BrokenSession:
"""A mock session that raises SQLAlchemyDatabaseError on all ops."""
def add(self, _obj: Any) -> None:
raise SQLAlchemyDatabaseError("mock", {}, Exception("broken"))
def commit(self) -> None:
raise SQLAlchemyDatabaseError("mock", {}, Exception("broken"))
def rollback(self) -> None:
pass
def query(self, *_args: Any, **_kwargs: Any) -> Any:
raise SQLAlchemyDatabaseError("mock", {}, Exception("broken"))
@given("a SQLAlchemy repository with a broken session")
def step_broken_repo(context: Context) -> None:
context.broken_repo = LLMTraceRepository(
session_factory=lambda: _BrokenSession() # type: ignore[return-value]
)
@when("I save the trace via the broken repository")
def step_save_broken(context: Context) -> None:
try:
context.broken_repo.save(context.trace)
except AppDatabaseError as exc:
context.error = exc
except Exception as exc:
# tenacity may wrap in RetryError; unwrap to the root cause
cause = getattr(exc, "__cause__", None) or exc
context.error = cause
@when("I get a trace from the broken repository")
def step_get_broken(context: Context) -> None:
try:
context.broken_repo.get(VALID_ULID_1)
except AppDatabaseError as exc:
context.error = exc
except Exception as exc:
cause = getattr(exc, "__cause__", None) or exc
context.error = cause
@when("I list by plan from the broken repository")
def step_list_plan_broken(context: Context) -> None:
try:
context.broken_repo.list_by_plan(VALID_PLAN_ID)
except AppDatabaseError as exc:
context.error = exc
except Exception as exc:
cause = getattr(exc, "__cause__", None) or exc
context.error = cause
@when("I list by decision from the broken repository")
def step_list_decision_broken(context: Context) -> None:
try:
context.broken_repo.list_by_decision(VALID_DECISION_ID)
except AppDatabaseError as exc:
context.error = exc
except Exception as exc:
cause = getattr(exc, "__cause__", None) or exc
context.error = cause
# ---------------------------------------------------------------------------
# LangSmith internal forwarder detail tests
# ---------------------------------------------------------------------------
@given("the langsmith SDK is mocked as available")
def step_langsmith_sdk_available(context: Context) -> None:
context.mock_langsmith_client = MagicMock()
context.mock_langsmith_module = MagicMock()
context.mock_langsmith_module.Client.return_value = context.mock_langsmith_client
@given('a valid LLM trace with error "{error_msg}"')
def step_trace_with_error(context: Context, error_msg: str) -> None:
context.trace = _make_trace(error=error_msg)
@when("I call the internal LangSmith forwarder")
def step_call_internal_forwarder(context: Context) -> None:
if hasattr(context, "mock_langsmith_module"):
# Mock the langsmith import inside _forward_trace_to_langsmith
real_import = builtins.__import__
def mock_import(name: str, *args: Any, **kwargs: Any) -> Any:
if name == "langsmith":
return context.mock_langsmith_module
return real_import(name, *args, **kwargs)
with patch.object(builtins, "__import__", side_effect=mock_import):
_forward_trace_to_langsmith(context.trace)
elif getattr(context, "langsmith_not_importable", False):
# Block the langsmith import to simulate SDK not installed
real_import = builtins.__import__
def block_langsmith(name: str, *args: Any, **kwargs: Any) -> Any:
if name == "langsmith":
raise ImportError("No module named 'langsmith'")
return real_import(name, *args, **kwargs)
with patch.object(builtins, "__import__", side_effect=block_langsmith):
_forward_trace_to_langsmith(context.trace)
else:
_forward_trace_to_langsmith(context.trace)
@then("the langsmith Client create_run should have been called")
def step_langsmith_create_run_called(context: Context) -> None:
context.mock_langsmith_client.create_run.assert_called_once()
@then("the langsmith Client create_run should have been called with error")
def step_langsmith_create_run_with_error(context: Context) -> None:
context.mock_langsmith_client.create_run.assert_called_once()
call_kwargs = context.mock_langsmith_client.create_run.call_args
# The call uses **run_data, so check kwargs
all_args = call_kwargs.kwargs if call_kwargs.kwargs else {}
assert "error" in all_args, f"Expected 'error' in call kwargs: {all_args}"
@given("the langsmith SDK is not importable")
def step_langsmith_not_importable(context: Context) -> None:
context.langsmith_not_importable = True
context.langsmith_client_created = False
@then("no langsmith Client should have been created")
def step_no_langsmith_client(context: Context) -> None:
# Verified by the fact that no error was raised and the function
# returned cleanly when the import was blocked.
assert not getattr(context, "langsmith_client_created", False)
@then("the langsmith_enabled check should return True")
def step_langsmith_enabled_true(context: Context) -> None:
assert TraceService._langsmith_enabled() is True
@then("the raw tool_calls_json in the database should be null")
def step_raw_tool_calls_null(context: Context) -> None:
session = context.sqla_factory()
row = (
session.query(LLMTraceModel).filter_by(trace_id=context.trace.trace_id).first()
)
assert row is not None
assert row.tool_calls_json is None
+10 -2
View File
@@ -11,6 +11,14 @@ Suite Teardown Cleanup Test Environment
*** Variables ***
${PYTHON} python
*** Keywords ***
Run CLI With Clean Home
[Documentation] Run a CLI command with a temp HOME to avoid stale config
[Arguments] @{cmd}
${tmpdir}= Evaluate __import__('tempfile').mkdtemp(prefix='cli_core_')
${result}= Run Process @{cmd} timeout=60s env:HOME=${tmpdir}
RETURN ${result}
*** Test Cases ***
Version Command Default Rich Format
[Documentation] Version command with default (rich) format shows version string
@@ -54,7 +62,7 @@ Info Command Default Rich Format
Info Command JSON Format
[Documentation] Info command with --format json returns structured data
${result}= Run Process ${PYTHON} -m cleveragents info --format json timeout=60s
${result}= Run CLI With Clean Home ${PYTHON} -m cleveragents info --format json
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} "version": "1.0.0"
Should Contain ${result.stdout} "data_dir"
@@ -63,7 +71,7 @@ Info Command JSON Format
Info Command Plain Format
[Documentation] Info command with --format plain returns key-value pairs
${result}= Run Process ${PYTHON} -m cleveragents info --format plain timeout=60s
${result}= Run CLI With Clean Home ${PYTHON} -m cleveragents info --format plain
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} version: 1.0.0
Should Contain ${result.stdout} server_mode:
+214
View File
@@ -0,0 +1,214 @@
"""Helper script for LLM trace Robot Framework tests.
Usage:
python helper_llm_trace.py record-and-retrieve
python helper_llm_trace.py list-by-plan
python helper_llm_trace.py compute-metrics
python helper_llm_trace.py metric-key-count
python helper_llm_trace.py validation
python helper_llm_trace.py lifecycle-hooks
"""
from __future__ import annotations
import sys
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock
# Ensure source tree is importable — force the local checkout to win
# over any inherited PYTHONPATH (e.g. /app/src in CI containers).
_SRC = str(Path(__file__).resolve().parents[1] / "src")
sys.path.insert(0, _SRC)
from cleveragents.application.services.trace_service import TraceService # noqa: E402
from cleveragents.domain.models.observability.llm_trace import LLMTrace # noqa: E402
from cleveragents.domain.models.observability.metrics import ( # noqa: E402
OperationalMetricKey,
)
VALID_ULID_1 = "01HXAAAAAAAAAAAAAAAAAAAAAA"
VALID_ULID_2 = "01HXBBBBBBBBBBBBBBBBBBBBBB"
VALID_ULID_3 = "01HXCCCCCCCCCCCCCCCCCCCCCC"
VALID_PLAN_ID = "01HX0000000000PPPPPPPPPPPP"
class InMemoryRepo:
"""Minimal in-memory repo for testing."""
def __init__(self) -> None:
self._store: dict[str, LLMTrace] = {}
def save(self, trace: LLMTrace) -> None:
self._store[trace.trace_id] = trace
def get(self, trace_id: str) -> LLMTrace | None:
return self._store.get(trace_id)
def list_by_plan(self, plan_id: str) -> list[LLMTrace]:
return [t for t in self._store.values() if t.plan_id == plan_id]
def list_by_decision(self, decision_id: str) -> list[LLMTrace]:
return [t for t in self._store.values() if t.decision_id == decision_id]
def _make_trace(trace_id: str = VALID_ULID_1, **kwargs: Any) -> LLMTrace:
defaults: dict[str, Any] = {
"trace_id": trace_id,
"plan_id": VALID_PLAN_ID,
"actor": "planner",
"provider": "openai",
"model": "gpt-4o",
"prompt_tokens": 500,
"completion_tokens": 200,
"cost_usd": 0.01,
"latency_ms": 350.0,
}
defaults.update(kwargs)
return LLMTrace(**defaults)
def _make_service() -> TraceService:
repo = InMemoryRepo()
return TraceService(settings=MagicMock(), repository=repo) # type: ignore[arg-type]
def cmd_record_and_retrieve() -> None:
"""Record a trace and retrieve by ID."""
svc = _make_service()
trace = _make_trace()
svc.record_trace(trace)
result = svc.get_trace(trace.trace_id)
if result is None:
print("FAIL: trace not found after recording")
sys.exit(1)
if result.trace_id != trace.trace_id:
print(f"FAIL: trace_id mismatch {result.trace_id} != {trace.trace_id}")
sys.exit(1)
print("llm-trace-record-ok")
def cmd_list_by_plan() -> None:
"""Record multiple traces and list by plan."""
svc = _make_service()
for uid in (VALID_ULID_1, VALID_ULID_2, VALID_ULID_3):
svc.record_trace(_make_trace(trace_id=uid))
results = svc.get_traces(VALID_PLAN_ID)
if len(results) != 3:
print(f"FAIL: expected 3 traces, got {len(results)}")
sys.exit(1)
print("llm-trace-list-plan-ok")
def cmd_compute_metrics() -> None:
"""Compute metrics from traces."""
svc = _make_service()
for uid in (VALID_ULID_1, VALID_ULID_2, VALID_ULID_3):
svc.record_trace(_make_trace(trace_id=uid))
metrics = svc.compute_metrics(VALID_PLAN_ID)
if len(metrics) != 4:
print(f"FAIL: expected 4 metrics, got {len(metrics)}")
sys.exit(1)
keys = {m.key for m in metrics}
expected = {
OperationalMetricKey.LLM_CALL_COUNT,
OperationalMetricKey.LLM_TOTAL_TOKENS,
OperationalMetricKey.LLM_TOTAL_COST_USD,
OperationalMetricKey.LLM_AVG_LATENCY_MS,
}
if keys != expected:
print(f"FAIL: unexpected metric keys {keys}")
sys.exit(1)
print("llm-trace-metrics-ok")
def cmd_metric_key_count() -> None:
"""Verify all 14 metric keys exist."""
if len(OperationalMetricKey) != 14:
print(f"FAIL: expected 14 keys, got {len(OperationalMetricKey)}")
sys.exit(1)
print("llm-trace-metric-keys-ok")
def cmd_validation() -> None:
"""Verify model validation rejects bad inputs."""
from pydantic import ValidationError
# Invalid trace_id
try:
_make_trace(trace_id="INVALID")
print("FAIL: invalid trace_id accepted")
sys.exit(1)
except ValidationError:
pass
# Negative prompt tokens
try:
_make_trace(prompt_tokens=-1)
print("FAIL: negative prompt_tokens accepted")
sys.exit(1)
except ValidationError:
pass
# Negative cost
try:
_make_trace(cost_usd=-0.5)
print("FAIL: negative cost_usd accepted")
sys.exit(1)
except ValidationError:
pass
print("llm-trace-validation-ok")
def cmd_lifecycle_hooks() -> None:
"""Verify lifecycle hooks produce correct metrics."""
svc = _make_service()
# on_plan_start
entry = svc.on_plan_start(VALID_PLAN_ID)
if entry is None or entry.key != OperationalMetricKey.PLAN_DECISION_COUNT:
print("FAIL: on_plan_start incorrect")
sys.exit(1)
# on_actor_invocation
entries = svc.on_actor_invocation(VALID_PLAN_ID, "planner", 250.0)
keys = {m.key for m in entries}
if OperationalMetricKey.ACTOR_INVOCATION_COUNT not in keys:
print("FAIL: on_actor_invocation missing ACTOR_INVOCATION_COUNT")
sys.exit(1)
# on_tool_execution
entries = svc.on_tool_execution(VALID_PLAN_ID, "file_write", errored=True)
keys = {m.key for m in entries}
if OperationalMetricKey.TOOL_ERROR_RATE not in keys:
print("FAIL: on_tool_execution missing TOOL_ERROR_RATE")
sys.exit(1)
print("llm-trace-hooks-ok")
def main() -> None:
"""Dispatch subcommand."""
if len(sys.argv) < 2:
print("Usage: helper_llm_trace.py <command>")
sys.exit(1)
cmd = sys.argv[1]
commands = {
"record-and-retrieve": cmd_record_and_retrieve,
"list-by-plan": cmd_list_by_plan,
"compute-metrics": cmd_compute_metrics,
"metric-key-count": cmd_metric_key_count,
"validation": cmd_validation,
"lifecycle-hooks": cmd_lifecycle_hooks,
}
fn = commands.get(cmd)
if fn is None:
print(f"Unknown command: {cmd}")
sys.exit(1)
fn()
if __name__ == "__main__":
main()
+47
View File
@@ -0,0 +1,47 @@
*** Settings ***
Documentation Smoke tests for LLM trace observability
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_llm_trace.py
*** Test Cases ***
Record And Retrieve LLM Trace
[Documentation] Record a trace and verify retrieval by ID
${result}= Run Process ${PYTHON} ${HELPER} record-and-retrieve cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} llm-trace-record-ok
List Traces By Plan
[Documentation] Record multiple traces and list by plan ID
${result}= Run Process ${PYTHON} ${HELPER} list-by-plan cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} llm-trace-list-plan-ok
Compute Metrics For Plan
[Documentation] Compute operational metrics from recorded traces
${result}= Run Process ${PYTHON} ${HELPER} compute-metrics cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} llm-trace-metrics-ok
Verify Metric Key Count
[Documentation] All 14 operational metric keys should be defined
${result}= Run Process ${PYTHON} ${HELPER} metric-key-count cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} llm-trace-metric-keys-ok
Verify LLM Trace Model Validation
[Documentation] Model should reject invalid inputs
${result}= Run Process ${PYTHON} ${HELPER} validation cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} llm-trace-validation-ok
Verify Lifecycle Hooks
[Documentation] Lifecycle hooks should produce correct metric entries
${result}= Run Process ${PYTHON} ${HELPER} lifecycle-hooks cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} llm-trace-hooks-ok
+26
View File
@@ -41,6 +41,7 @@ from cleveragents.application.services.skeleton_compressor import (
SkeletonCompressorService,
)
from cleveragents.application.services.subplan_service import SubplanService
from cleveragents.application.services.trace_service import TraceService
from cleveragents.application.services.vector_store_service import VectorStoreService
from cleveragents.config.settings import Settings, get_settings
from cleveragents.domain.models.acms.stubs import (
@@ -49,6 +50,9 @@ from cleveragents.domain.models.acms.stubs import (
InMemoryVectorBackend,
)
from cleveragents.domain.providers.ai_provider import AIProviderInterface
from cleveragents.infrastructure.database.llm_trace_repository import (
LLMTraceRepository,
)
from cleveragents.infrastructure.database.repositories import (
CheckpointRepository,
NamespacedProjectRepository,
@@ -188,6 +192,21 @@ def _build_checkpoint_service(
)
def _build_trace_service(
database_url: str,
settings: Settings | None = None,
) -> TraceService:
"""Build a TraceService backed by a database LLMTraceRepository."""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine(database_url, echo=False)
factory = sessionmaker(bind=engine, expire_on_commit=False)
repository = LLMTraceRepository(session_factory=factory)
resolved_settings = settings or get_settings()
return TraceService(settings=resolved_settings, repository=repository)
class Container(containers.DeclarativeContainer):
"""Dependency injection container using dependency-injector.
@@ -335,6 +354,13 @@ class Container(containers.DeclarativeContainer):
SkeletonCompressorService,
)
# Trace Service - LLM observability (Forgejo #500)
trace_service = providers.Factory(
_build_trace_service,
database_url=database_url,
settings=settings,
)
# Autonomy Guardrail Service - Singleton so all callers share state
autonomy_guardrail_service = providers.Singleton(
AutonomyGuardrailService,
@@ -116,6 +116,9 @@ from cleveragents.application.services.subplan_service import (
from cleveragents.application.services.tool_registry_service import (
ToolRegistryService,
)
from cleveragents.application.services.trace_service import (
TraceService,
)
from cleveragents.application.services.validation_apply import (
ApplyValidationGate,
ApplyValidationResult,
@@ -201,6 +204,7 @@ __all__ = [
"SubplanService",
"SyntaxCheckRule",
"ToolRegistryService",
"TraceService",
"ValidationAttachment",
"ValidationCommand",
"ValidationPipeline",
@@ -0,0 +1,314 @@
"""Trace service for LLM observability.
Coordinates the recording, querying, and analysis of ``LLMTrace`` entries.
Optionally forwards traces to LangSmith when the ``LANGCHAIN_TRACING_V2``
environment variable is set to ``"true"``.
Metric collection hooks produce ``MetricEntry`` objects from aggregated
trace data for a given plan.
Design decisions:
- Dependency injection: ``Settings`` and ``LLMTraceRepository`` are
injected via the constructor (consistent with existing service patterns).
- Stateless: All trace state lives in the repository.
- LangSmith forwarding is best-effort failures are logged but do not
raise.
Based on Forgejo issue #500.
"""
from __future__ import annotations
import logging
import os
from typing import TYPE_CHECKING, Any
from cleveragents.domain.models.observability.llm_trace import LLMTrace
from cleveragents.domain.models.observability.metrics import (
MetricCollector,
MetricEntry,
OperationalMetricKey,
)
if TYPE_CHECKING:
from cleveragents.config.settings import Settings
from cleveragents.infrastructure.database.llm_trace_repository import (
LLMTraceRepository,
)
logger = logging.getLogger(__name__)
class TraceService:
"""Orchestrates LLM trace persistence, querying, and metric computation.
Attributes:
_settings: Application settings (injected).
_repository: Persistence layer for ``LLMTrace`` records.
"""
def __init__(
self,
settings: Settings,
repository: LLMTraceRepository,
) -> None:
"""Initialise with injected dependencies.
Args:
settings: Application settings.
repository: LLM trace repository for persistence.
"""
self._settings = settings
self._repository = repository
# ---- write path -------------------------------------------------------
def record_trace(self, trace: LLMTrace) -> None:
"""Persist an ``LLMTrace`` entry and optionally forward to LangSmith.
Args:
trace: The trace to record.
"""
self._repository.save(trace)
logger.debug("Recorded LLM trace %s for plan %s", trace.trace_id, trace.plan_id)
# Best-effort LangSmith forwarding
if self._langsmith_enabled():
self.forward_to_langsmith(trace)
# ---- read path --------------------------------------------------------
def get_traces(self, plan_id: str) -> list[LLMTrace]:
"""Query all traces for a plan.
Args:
plan_id: ULID of the plan.
Returns:
List of ``LLMTrace`` objects ordered by timestamp.
"""
return self._repository.list_by_plan(plan_id)
def get_traces_by_decision(self, decision_id: str) -> list[LLMTrace]:
"""Query all traces for a decision.
Args:
decision_id: ULID of the decision.
Returns:
List of ``LLMTrace`` objects ordered by timestamp.
"""
return self._repository.list_by_decision(decision_id)
def get_trace(self, trace_id: str) -> LLMTrace | None:
"""Retrieve a single trace by ID.
Args:
trace_id: ULID of the trace.
Returns:
The ``LLMTrace`` or ``None``.
"""
return self._repository.get(trace_id)
# ---- metric computation -----------------------------------------------
def compute_metrics(self, plan_id: str) -> list[MetricEntry]:
"""Compute operational metrics from all traces for a plan.
Aggregates LLM-level metrics (call count, total tokens, total
cost, average latency) from the trace entries.
Args:
plan_id: ULID of the plan.
Returns:
List of ``MetricEntry`` objects for the computed metrics.
"""
traces = self._repository.list_by_plan(plan_id)
metrics: list[MetricEntry] = []
if not traces:
return metrics
total_tokens = sum(t.prompt_tokens + t.completion_tokens for t in traces)
total_cost = sum(t.cost_usd for t in traces)
total_latency = sum(t.latency_ms for t in traces)
call_count = len(traces)
avg_latency = total_latency / call_count if call_count > 0 else 0.0
metrics.append(
MetricCollector.record(
OperationalMetricKey.LLM_CALL_COUNT,
float(call_count),
plan_id,
)
)
metrics.append(
MetricCollector.record(
OperationalMetricKey.LLM_TOTAL_TOKENS,
float(total_tokens),
plan_id,
)
)
metrics.append(
MetricCollector.record(
OperationalMetricKey.LLM_TOTAL_COST_USD,
total_cost,
plan_id,
)
)
metrics.append(
MetricCollector.record(
OperationalMetricKey.LLM_AVG_LATENCY_MS,
avg_latency,
plan_id,
)
)
return metrics
# ---- plan lifecycle hooks (stubs) -------------------------------------
def on_plan_start(self, plan_id: str) -> MetricEntry | None:
"""Hook: called when a plan begins execution.
Returns a stub ``MetricEntry`` for plan decision count
initialisation. Real timing starts here.
Args:
plan_id: ULID of the plan.
Returns:
A ``MetricEntry`` for ``PLAN_DECISION_COUNT`` initialised to 0.
"""
return MetricCollector.record(
OperationalMetricKey.PLAN_DECISION_COUNT,
0.0,
plan_id,
)
def on_actor_invocation(
self, plan_id: str, actor: str, latency_ms: float
) -> list[MetricEntry]:
"""Hook: called after an actor invocation completes.
Args:
plan_id: ULID of the plan.
actor: Name of the invoked actor.
latency_ms: Actor invocation latency in milliseconds.
Returns:
List of ``MetricEntry`` objects for actor metrics.
"""
return [
MetricCollector.record(
OperationalMetricKey.ACTOR_INVOCATION_COUNT,
1.0,
plan_id,
labels={"actor": actor},
),
MetricCollector.record(
OperationalMetricKey.ACTOR_LATENCY_MS,
latency_ms,
plan_id,
labels={"actor": actor},
),
]
def on_tool_execution(
self, plan_id: str, tool_name: str, *, errored: bool = False
) -> list[MetricEntry]:
"""Hook: called after a tool execution completes.
Args:
plan_id: ULID of the plan.
tool_name: Name of the tool executed.
errored: Whether the execution errored.
Returns:
List of ``MetricEntry`` objects for tool metrics.
"""
entries = [
MetricCollector.record(
OperationalMetricKey.TOOL_INVOCATION_COUNT,
1.0,
plan_id,
labels={"tool": tool_name},
),
]
if errored:
entries.append(
MetricCollector.record(
OperationalMetricKey.TOOL_ERROR_RATE,
1.0,
plan_id,
labels={"tool": tool_name},
)
)
return entries
# ---- LangSmith forwarding ---------------------------------------------
@staticmethod
def _langsmith_enabled() -> bool:
"""Check whether LangSmith forwarding is enabled."""
return os.environ.get("LANGCHAIN_TRACING_V2", "").lower() == "true"
def forward_to_langsmith(self, trace: LLMTrace) -> None:
"""Forward a trace to LangSmith (best-effort).
This is a best-effort operation: failures are logged at
warning level but do not propagate exceptions.
Args:
trace: The trace to forward.
"""
try:
_forward_trace_to_langsmith(trace)
except Exception:
logger.warning(
"Failed to forward trace %s to LangSmith",
trace.trace_id,
exc_info=True,
)
def _forward_trace_to_langsmith(trace: LLMTrace) -> None:
"""Internal: forward a trace to LangSmith via the langsmith SDK.
Separated for testability. Imports ``langsmith`` lazily to avoid
hard dependency.
Args:
trace: The trace to forward.
"""
try:
from langsmith import Client # type: ignore[import-untyped]
except ImportError:
logger.debug("langsmith SDK not installed; skipping trace forwarding")
return
client = Client()
run_data: dict[str, Any] = {
"name": f"llm-trace-{trace.trace_id}",
"run_type": "llm",
"inputs": {
"actor": trace.actor,
"provider": trace.provider,
"model": trace.model,
"prompt_tokens": trace.prompt_tokens,
"streaming": trace.streaming,
},
"outputs": {
"completion_tokens": trace.completion_tokens,
"cost_usd": trace.cost_usd,
"latency_ms": trace.latency_ms,
"tool_calls": trace.tool_calls,
},
}
if trace.error:
run_data["error"] = trace.error
client.create_run(**run_data)
logger.debug("Forwarded trace %s to LangSmith", trace.trace_id)
@@ -0,0 +1,28 @@
"""Observability domain models for CleverAgents.
Provides the ``LLMTrace`` model for recording LLM call telemetry and
the ``OperationalMetricKey`` enum plus ``MetricEntry`` / ``MetricCollector``
for aggregated plan-level operational metrics.
Based on Forgejo issue #500.
"""
from __future__ import annotations
from cleveragents.domain.models.observability.llm_trace import (
LLMTrace,
LLMTraceQuery,
)
from cleveragents.domain.models.observability.metrics import (
MetricCollector,
MetricEntry,
OperationalMetricKey,
)
__all__ = [
"LLMTrace",
"LLMTraceQuery",
"MetricCollector",
"MetricEntry",
"OperationalMetricKey",
]
@@ -0,0 +1,207 @@
"""LLM trace domain model for observability.
An ``LLMTrace`` records a single LLM provider call with token usage,
cost, latency, and optional tool-call metadata. Traces are keyed by
a ULID ``trace_id`` and linked to a ``plan_id`` and optional
``decision_id`` for drill-down analysis.
``LLMTraceQuery`` provides query/filter parameters for trace retrieval
from the repository layer.
Based on ``docs/specification.md`` Observability section and
Forgejo issue #500.
"""
from __future__ import annotations
import re
from datetime import datetime
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
# ULID is 26 characters, Crockford's base32
ULID_PATTERN = r"^[0-9A-HJKMNP-TV-Z]{26}$"
class LLMTrace(BaseModel):
"""Immutable record of a single LLM provider call.
Each trace captures the full observability surface for one
LLM invocation: identity (trace/plan/decision), actor and
provider metadata, token usage, cost, latency, tool calls,
context fingerprint, streaming flag, retry count, and any
error message.
Attributes:
trace_id: ULID uniquely identifying this trace entry.
plan_id: ULID of the plan that owns this trace.
decision_id: Optional ULID of the decision that triggered
this call.
actor: Name of the actor that made the call.
provider: LLM provider identifier (e.g. ``openai``,
``anthropic``).
model: Model name (e.g. ``gpt-4o``, ``claude-3-opus``).
prompt_tokens: Number of tokens in the prompt.
completion_tokens: Number of tokens in the completion.
cost_usd: Estimated cost in USD for this call.
latency_ms: Wall-clock latency in milliseconds.
tool_calls: List of tool call descriptors returned by the
model, if any.
context_hash: SHA-256 fingerprint of the context window
sent to the model, for deduplication analysis.
streaming: Whether the call used streaming mode.
retry_count: Number of retries before success (0 = first
attempt succeeded).
error: Error message if the call failed, ``None`` on success.
"""
trace_id: str = Field(
...,
min_length=26,
max_length=26,
description="ULID uniquely identifying this trace entry",
)
plan_id: str = Field(
...,
min_length=26,
max_length=26,
description="ULID of the plan that owns this trace",
)
decision_id: str | None = Field(
default=None,
description="Optional ULID of the decision that triggered this call",
)
actor: str = Field(
...,
min_length=1,
max_length=255,
description="Name of the actor that made the call",
)
provider: str = Field(
...,
min_length=1,
max_length=128,
description="LLM provider identifier (e.g. openai, anthropic)",
)
model: str = Field(
...,
min_length=1,
max_length=255,
description="Model name (e.g. gpt-4o, claude-3-opus)",
)
prompt_tokens: int = Field(
...,
ge=0,
description="Number of tokens in the prompt",
)
completion_tokens: int = Field(
...,
ge=0,
description="Number of tokens in the completion",
)
cost_usd: float = Field(
...,
ge=0.0,
description="Estimated cost in USD for this call",
)
latency_ms: float = Field(
...,
ge=0.0,
description="Wall-clock latency in milliseconds",
)
tool_calls: list[dict[str, Any]] = Field(
default_factory=list,
description="Tool call descriptors returned by the model",
)
context_hash: str | None = Field(
default=None,
max_length=64,
description="SHA-256 fingerprint of the context window",
)
streaming: bool = Field(
default=False,
description="Whether the call used streaming mode",
)
retry_count: int = Field(
default=0,
ge=0,
description="Number of retries before success",
)
error: str | None = Field(
default=None,
description="Error message if the call failed",
)
timestamp: datetime = Field(
default_factory=datetime.utcnow,
description="When the trace was recorded",
)
@field_validator("trace_id", "plan_id")
@classmethod
def _validate_ulid(cls: type[LLMTrace], v: str) -> str:
"""Ensure ULID fields match Crockford's base32 pattern."""
if not re.match(ULID_PATTERN, v):
raise ValueError(f"Invalid ULID: {v!r}")
return v
@field_validator("decision_id")
@classmethod
def _validate_optional_ulid(cls: type[LLMTrace], v: str | None) -> str | None:
"""Validate optional ULID field."""
if v is not None and not re.match(ULID_PATTERN, v):
raise ValueError(f"Invalid ULID: {v!r}")
return v
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
)
class LLMTraceQuery(BaseModel):
"""Query parameters for filtering LLM traces.
All filter fields are optional; omitted fields match everything.
Attributes:
plan_id: Filter by plan ULID.
decision_id: Filter by decision ULID.
actor: Filter by actor name.
provider: Filter by provider identifier.
model: Filter by model name.
min_cost_usd: Minimum cost threshold.
max_cost_usd: Maximum cost threshold.
min_latency_ms: Minimum latency threshold.
max_latency_ms: Maximum latency threshold.
has_error: If set, filter by error presence.
limit: Maximum number of results (default 100).
offset: Pagination offset (default 0).
"""
plan_id: str | None = Field(default=None, description="Filter by plan ULID")
decision_id: str | None = Field(default=None, description="Filter by decision ULID")
actor: str | None = Field(default=None, description="Filter by actor name")
provider: str | None = Field(
default=None, description="Filter by provider identifier"
)
model: str | None = Field(default=None, description="Filter by model name")
min_cost_usd: float | None = Field(
default=None, ge=0.0, description="Minimum cost threshold"
)
max_cost_usd: float | None = Field(
default=None, ge=0.0, description="Maximum cost threshold"
)
min_latency_ms: float | None = Field(
default=None, ge=0.0, description="Minimum latency threshold"
)
max_latency_ms: float | None = Field(
default=None, ge=0.0, description="Maximum latency threshold"
)
has_error: bool | None = Field(default=None, description="Filter by error presence")
limit: int = Field(default=100, ge=1, le=10000, description="Max results")
offset: int = Field(default=0, ge=0, description="Pagination offset")
model_config = ConfigDict(
str_strip_whitespace=True,
)
@@ -0,0 +1,174 @@
"""Operational metric definitions for observability.
Defines the 14 operational metric keys tracked across the plan
lifecycle and a lightweight ``MetricEntry`` model for recording
individual metric observations.
``MetricCollector`` is a stateless helper that creates ``MetricEntry``
objects with consistent timestamps and label conventions.
Based on ``docs/specification.md`` Observability section and
Forgejo issue #500.
"""
from __future__ import annotations
from datetime import datetime
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
class OperationalMetricKey(StrEnum):
"""Enumeration of the 14 operational metric keys.
Metrics are grouped by subsystem:
**Plan-level**
``PLAN_DURATION_MS``, ``PLAN_TOTAL_COST_USD``,
``PLAN_DECISION_COUNT``, ``SUBPLAN_COUNT``
**Actor-level**
``ACTOR_INVOCATION_COUNT``, ``ACTOR_LATENCY_MS``
**Tool-level**
``TOOL_INVOCATION_COUNT``, ``TOOL_ERROR_RATE``
**Context-level**
``CONTEXT_BUILD_TIME_MS``, ``CONTEXT_TOKEN_COUNT``
**LLM-level**
``LLM_CALL_COUNT``, ``LLM_TOTAL_TOKENS``,
``LLM_TOTAL_COST_USD``, ``LLM_AVG_LATENCY_MS``
"""
PLAN_DURATION_MS = "plan_duration_ms"
PLAN_TOTAL_COST_USD = "plan_total_cost_usd"
PLAN_DECISION_COUNT = "plan_decision_count"
ACTOR_INVOCATION_COUNT = "actor_invocation_count"
ACTOR_LATENCY_MS = "actor_latency_ms"
TOOL_INVOCATION_COUNT = "tool_invocation_count"
TOOL_ERROR_RATE = "tool_error_rate"
CONTEXT_BUILD_TIME_MS = "context_build_time_ms"
CONTEXT_TOKEN_COUNT = "context_token_count"
LLM_CALL_COUNT = "llm_call_count"
LLM_TOTAL_TOKENS = "llm_total_tokens"
LLM_TOTAL_COST_USD = "llm_total_cost_usd"
LLM_AVG_LATENCY_MS = "llm_avg_latency_ms"
SUBPLAN_COUNT = "subplan_count"
class MetricEntry(BaseModel):
"""A single metric observation.
Each entry records the key, numeric value, plan context, timestamp,
and optional labels for dimensional filtering.
Attributes:
key: The operational metric being recorded.
value: Numeric value of the observation.
plan_id: ULID of the plan this metric belongs to.
timestamp: When the observation was recorded.
labels: Arbitrary dimensional labels for drill-down
(e.g. ``{"actor": "planner", "provider": "openai"}``).
"""
key: OperationalMetricKey = Field(
...,
description="The operational metric being recorded",
)
value: float = Field(
...,
description="Numeric value of the observation",
)
plan_id: str = Field(
...,
min_length=26,
max_length=26,
description="ULID of the plan this metric belongs to",
)
timestamp: datetime = Field(
default_factory=datetime.utcnow,
description="When the observation was recorded",
)
labels: dict[str, Any] = Field(
default_factory=dict,
description="Dimensional labels for drill-down filtering",
)
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
)
class MetricCollector:
"""Stateless factory for creating ``MetricEntry`` objects.
Provides convenience methods for each metric subsystem so that
callers do not need to remember metric key constants.
"""
@staticmethod
def record(
key: OperationalMetricKey,
value: float,
plan_id: str,
*,
labels: dict[str, Any] | None = None,
timestamp: datetime | None = None,
) -> MetricEntry:
"""Create a ``MetricEntry`` for the given key and value.
Args:
key: The operational metric key.
value: Numeric observation value.
plan_id: ULID of the owning plan.
labels: Optional dimensional labels.
timestamp: Override timestamp (defaults to ``utcnow``).
Returns:
A new ``MetricEntry`` instance.
"""
kwargs: dict[str, Any] = {
"key": key,
"value": value,
"plan_id": plan_id,
"labels": labels or {},
}
if timestamp is not None:
kwargs["timestamp"] = timestamp
return MetricEntry(**kwargs)
# ----- convenience helpers -----
@staticmethod
def plan_duration(plan_id: str, duration_ms: float, **labels: Any) -> MetricEntry:
"""Record ``PLAN_DURATION_MS``."""
return MetricCollector.record(
OperationalMetricKey.PLAN_DURATION_MS,
duration_ms,
plan_id,
labels=dict(labels),
)
@staticmethod
def plan_cost(plan_id: str, cost_usd: float, **labels: Any) -> MetricEntry:
"""Record ``PLAN_TOTAL_COST_USD``."""
return MetricCollector.record(
OperationalMetricKey.PLAN_TOTAL_COST_USD,
cost_usd,
plan_id,
labels=dict(labels),
)
@staticmethod
def llm_call_count(plan_id: str, count: float, **labels: Any) -> MetricEntry:
"""Record ``LLM_CALL_COUNT``."""
return MetricCollector.record(
OperationalMetricKey.LLM_CALL_COUNT,
count,
plan_id,
labels=dict(labels),
)
@@ -0,0 +1,173 @@
"""LLM trace persistence repository.
Provides ``LLMTraceRepository`` for CRUD operations on the
``llm_traces`` table. Follows the session-factory pattern (ADR-007)
with retry logic (ADR-033).
Based on Forgejo issue #500.
"""
from __future__ import annotations
import json
from collections.abc import Callable
from datetime import datetime
from typing import Any
from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import Session
from cleveragents.core.exceptions import DatabaseError
from cleveragents.core.retry_patterns import (
retry_database_operation as database_retry,
)
from cleveragents.domain.models.observability.llm_trace import LLMTrace
from cleveragents.infrastructure.database.models import LLMTraceModel
class LLMTraceRepository:
"""Repository for persisting and querying ``LLMTrace`` records.
Uses the session-factory pattern: each public method obtains a
session from the factory. Callers are responsible for commit.
"""
def __init__(
self,
session_factory: Callable[[], Session],
) -> None:
"""Initialise with a callable returning a SQLAlchemy Session."""
if session_factory is None:
raise ValueError("session_factory must not be None")
self._sf = session_factory
def _session(self) -> Session:
return self._sf()
@database_retry
def save(self, trace: LLMTrace) -> None:
"""Persist a single ``LLMTrace`` row.
Args:
trace: The trace to persist.
Raises:
DatabaseError: On unrecoverable persistence failure.
"""
session = self._session()
try:
model = LLMTraceModel(
trace_id=trace.trace_id,
plan_id=trace.plan_id,
decision_id=trace.decision_id,
actor=trace.actor,
provider=trace.provider,
model=trace.model,
prompt_tokens=trace.prompt_tokens,
completion_tokens=trace.completion_tokens,
cost_usd=trace.cost_usd,
latency_ms=trace.latency_ms,
tool_calls_json=json.dumps(trace.tool_calls)
if trace.tool_calls
else None,
context_hash=trace.context_hash,
streaming=trace.streaming,
retry_count=trace.retry_count,
error=trace.error,
timestamp=trace.timestamp.isoformat(),
)
session.add(model)
session.commit()
except (SQLAlchemyDatabaseError, OperationalError) as exc:
session.rollback()
raise DatabaseError(f"Failed to save LLM trace: {exc}") from exc
@database_retry
def get(self, trace_id: str) -> LLMTrace | None:
"""Retrieve a single trace by ID.
Args:
trace_id: ULID of the trace to retrieve.
Returns:
The ``LLMTrace`` or ``None`` if not found.
"""
session = self._session()
try:
row = session.query(LLMTraceModel).filter_by(trace_id=trace_id).first()
if row is None:
return None
return self._to_domain(row)
except (SQLAlchemyDatabaseError, OperationalError) as exc:
raise DatabaseError(f"Failed to get LLM trace: {exc}") from exc
@database_retry
def list_by_plan(self, plan_id: str) -> list[LLMTrace]:
"""List all traces for a given plan.
Args:
plan_id: ULID of the plan.
Returns:
List of ``LLMTrace`` objects ordered by timestamp.
"""
session = self._session()
try:
rows = (
session.query(LLMTraceModel)
.filter_by(plan_id=plan_id)
.order_by(LLMTraceModel.timestamp)
.all()
)
return [self._to_domain(r) for r in rows]
except (SQLAlchemyDatabaseError, OperationalError) as exc:
raise DatabaseError(f"Failed to list traces by plan: {exc}") from exc
@database_retry
def list_by_decision(self, decision_id: str) -> list[LLMTrace]:
"""List all traces for a given decision.
Args:
decision_id: ULID of the decision.
Returns:
List of ``LLMTrace`` objects ordered by timestamp.
"""
session = self._session()
try:
rows = (
session.query(LLMTraceModel)
.filter_by(decision_id=decision_id)
.order_by(LLMTraceModel.timestamp)
.all()
)
return [self._to_domain(r) for r in rows]
except (SQLAlchemyDatabaseError, OperationalError) as exc:
raise DatabaseError(f"Failed to list traces by decision: {exc}") from exc
@staticmethod
def _to_domain(row: Any) -> LLMTrace:
"""Convert a database row to a domain ``LLMTrace``."""
tool_calls: list[dict[str, Any]] = []
if row.tool_calls_json:
tool_calls = json.loads(row.tool_calls_json)
return LLMTrace(
trace_id=row.trace_id,
plan_id=row.plan_id,
decision_id=row.decision_id,
actor=row.actor,
provider=row.provider,
model=row.model,
prompt_tokens=row.prompt_tokens,
completion_tokens=row.completion_tokens,
cost_usd=row.cost_usd,
latency_ms=row.latency_ms,
tool_calls=tool_calls,
context_hash=row.context_hash,
streaming=row.streaming,
retry_count=row.retry_count,
error=row.error,
timestamp=datetime.fromisoformat(row.timestamp),
)
@@ -2984,3 +2984,47 @@ class ToolInvocationModel(Base): # type: ignore[misc]
Index("ix_tool_invocations_plan_id", "plan_id"),
Index("ix_tool_invocations_tool_name", "tool_name"),
)
# ---------------------------------------------------------------------------
# LLM Trace Models (Observability — Forgejo #500)
# ---------------------------------------------------------------------------
class LLMTraceModel(Base): # type: ignore[misc]
"""Database model for persisted LLMTrace records.
Each row captures a single LLM provider call with full
observability metadata: token usage, cost, latency, tool
calls, context fingerprint, streaming flag, retry count,
and error information.
Table: ``llm_traces``
"""
__allow_unmapped__ = True
__tablename__ = "llm_traces"
trace_id = Column(String(26), primary_key=True)
plan_id = Column(String(26), nullable=False)
decision_id = Column(String(26), nullable=True)
actor = Column(String(255), nullable=False)
provider = Column(String(128), nullable=False)
model = Column(String(255), nullable=False)
prompt_tokens = Column(Integer, nullable=False, default=0)
completion_tokens = Column(Integer, nullable=False, default=0)
cost_usd = Column(Float, nullable=False, default=0.0)
latency_ms = Column(Float, nullable=False, default=0.0)
tool_calls_json = Column(Text, nullable=True)
context_hash = Column(String(64), nullable=True)
streaming = Column(Boolean, nullable=False, default=False)
retry_count = Column(Integer, nullable=False, default=0)
error = Column(Text, nullable=True)
timestamp = Column(String(30), nullable=False)
__table_args__ = (
Index("ix_llm_traces_plan_id", "plan_id"),
Index("ix_llm_traces_decision_id", "decision_id"),
Index("ix_llm_traces_actor", "actor"),
Index("ix_llm_traces_provider", "provider"),
)
+41
View File
@@ -667,3 +667,44 @@ _DEFAULT_MAX_TOKENS_HOT # noqa: B018, F821
_DEFAULT_MAX_DECISIONS_WARM # noqa: B018, F821
_DEFAULT_MAX_DECISIONS_COLD # noqa: B018, F821
_utc_now # noqa: B018, F821
# LLM Trace Observability — public API (issue #500)
LLMTrace # noqa: B018, F821
LLMTraceQuery # noqa: B018, F821
LLMTraceModel # noqa: B018, F821
LLMTraceRepository # noqa: B018, F821
TraceService # noqa: B018, F821
OperationalMetricKey # noqa: B018, F821
MetricEntry # noqa: B018, F821
MetricCollector # noqa: B018, F821
trace_service # noqa: B018, F821
forward_to_langsmith # noqa: B018, F821
_forward_trace_to_langsmith # noqa: B018, F821
compute_metrics # noqa: B018, F821
record_trace # noqa: B018, F821
get_traces # noqa: B018, F821
get_traces_by_decision # noqa: B018, F821
get_trace # noqa: B018, F821
on_plan_start # noqa: B018, F821
on_actor_invocation # noqa: B018, F821
on_tool_execution # noqa: B018, F821
list_by_plan # noqa: B018, F821
list_by_decision # noqa: B018, F821
plan_duration # noqa: B018, F821
plan_cost # noqa: B018, F821
llm_call_count # noqa: B018, F821
PLAN_DURATION_MS # noqa: B018, F821
PLAN_TOTAL_COST_USD # noqa: B018, F821
PLAN_DECISION_COUNT # noqa: B018, F821
ACTOR_INVOCATION_COUNT # noqa: B018, F821
ACTOR_LATENCY_MS # noqa: B018, F821
TOOL_INVOCATION_COUNT # noqa: B018, F821
TOOL_ERROR_RATE # noqa: B018, F821
CONTEXT_BUILD_TIME_MS # noqa: B018, F821
CONTEXT_TOKEN_COUNT # noqa: B018, F821
LLM_CALL_COUNT # noqa: B018, F821
LLM_TOTAL_TOKENS # noqa: B018, F821
LLM_TOTAL_COST_USD # noqa: B018, F821
LLM_AVG_LATENCY_MS # noqa: B018, F821
SUBPLAN_COUNT # noqa: B018, F821
_build_trace_service # noqa: B018, F821