From 3f14cbbf7e1f325e294b26dd0ffec833d4062e8c Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Tue, 3 Mar 2026 19:34:14 +0000 Subject: [PATCH] feat(observability): add LLMTrace model and operational metrics Add LLMTrace Pydantic v2 domain model with all required fields (trace_id, plan_id, decision_id, actor, provider, model, prompt_tokens, completion_tokens, cost_usd, latency_ms, tool_calls, context_hash, streaming, retry_count, error). Define 14 OperationalMetricKey values (PLAN_DURATION_MS, PLAN_TOTAL_COST_USD, PLAN_DECISION_COUNT, ACTOR_INVOCATION_COUNT, ACTOR_LATENCY_MS, TOOL_INVOCATION_COUNT, TOOL_ERROR_RATE, CONTEXT_BUILD_TIME_MS, CONTEXT_TOKEN_COUNT, LLM_CALL_COUNT, LLM_TOTAL_TOKENS, LLM_TOTAL_COST_USD, LLM_AVG_LATENCY_MS, SUBPLAN_COUNT) with MetricEntry model and MetricCollector. Add llm_traces database table (LLMTraceModel) with LLMTraceRepository for persistence. TraceService provides recording, querying, metric computation, plan lifecycle hooks, and optional LangSmith forwarding when LANGCHAIN_TRACING_V2=true. Wired into DI container as trace_service. Includes 28 Behave BDD scenarios, 6 Robot Framework smoke tests, 3 ASV benchmark suites, and reference documentation. Fix pre-existing cli_core server_mode test flake by mocking resolve_server_mode in test steps to avoid stale config file interference. Closes #500 --- CHANGELOG.md | 6 + benchmarks/llm_trace_bench.py | 126 +++ docs/reference/observability.md | 89 ++ features/llm_trace.feature | 276 +++++ features/steps/cli_core_steps.py | 17 +- features/steps/llm_trace_steps.py | 948 ++++++++++++++++++ robot/cli_core.robot | 12 +- robot/helper_llm_trace.py | 214 ++++ robot/llm_trace.robot | 47 + src/cleveragents/application/container.py | 26 + .../application/services/__init__.py | 4 + .../application/services/trace_service.py | 314 ++++++ .../domain/models/observability/__init__.py | 28 + .../domain/models/observability/llm_trace.py | 207 ++++ .../domain/models/observability/metrics.py | 174 ++++ .../database/llm_trace_repository.py | 173 ++++ .../infrastructure/database/models.py | 44 + vulture_whitelist.py | 41 + 18 files changed, 2741 insertions(+), 5 deletions(-) create mode 100644 benchmarks/llm_trace_bench.py create mode 100644 docs/reference/observability.md create mode 100644 features/llm_trace.feature create mode 100644 features/steps/llm_trace_steps.py create mode 100644 robot/helper_llm_trace.py create mode 100644 robot/llm_trace.robot create mode 100644 src/cleveragents/application/services/trace_service.py create mode 100644 src/cleveragents/domain/models/observability/__init__.py create mode 100644 src/cleveragents/domain/models/observability/llm_trace.py create mode 100644 src/cleveragents/domain/models/observability/metrics.py create mode 100644 src/cleveragents/infrastructure/database/llm_trace_repository.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 95dfe13b3..f21073d9d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/benchmarks/llm_trace_bench.py b/benchmarks/llm_trace_bench.py new file mode 100644 index 000000000..31a9b3969 --- /dev/null +++ b/benchmarks/llm_trace_bench.py @@ -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) diff --git a/docs/reference/observability.md b/docs/reference/observability.md new file mode 100644 index 000000000..906903b67 --- /dev/null +++ b/docs/reference/observability.md @@ -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`. diff --git a/features/llm_trace.feature b/features/llm_trace.feature new file mode 100644 index 000000000..feb827178 --- /dev/null +++ b/features/llm_trace.feature @@ -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 diff --git a/features/steps/cli_core_steps.py b/features/steps/cli_core_steps.py index ca31fbf38..e5ba7188c 100644 --- a/features/steps/cli_core_steps.py +++ b/features/steps/cli_core_steps.py @@ -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) diff --git a/features/steps/llm_trace_steps.py b/features/steps/llm_trace_steps.py new file mode 100644 index 000000000..dbd635e52 --- /dev/null +++ b/features/steps/llm_trace_steps.py @@ -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 diff --git a/robot/cli_core.robot b/robot/cli_core.robot index 0d119daf0..e9a158e66 100644 --- a/robot/cli_core.robot +++ b/robot/cli_core.robot @@ -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: diff --git a/robot/helper_llm_trace.py b/robot/helper_llm_trace.py new file mode 100644 index 000000000..d9d811f35 --- /dev/null +++ b/robot/helper_llm_trace.py @@ -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 ") + 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() diff --git a/robot/llm_trace.robot b/robot/llm_trace.robot new file mode 100644 index 000000000..da20952f3 --- /dev/null +++ b/robot/llm_trace.robot @@ -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 diff --git a/src/cleveragents/application/container.py b/src/cleveragents/application/container.py index 9aa1a27a6..ae65e6a42 100644 --- a/src/cleveragents/application/container.py +++ b/src/cleveragents/application/container.py @@ -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, diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index de470d223..ba9ae25ff 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -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", diff --git a/src/cleveragents/application/services/trace_service.py b/src/cleveragents/application/services/trace_service.py new file mode 100644 index 000000000..ab38d6fe8 --- /dev/null +++ b/src/cleveragents/application/services/trace_service.py @@ -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) diff --git a/src/cleveragents/domain/models/observability/__init__.py b/src/cleveragents/domain/models/observability/__init__.py new file mode 100644 index 000000000..40b1ce703 --- /dev/null +++ b/src/cleveragents/domain/models/observability/__init__.py @@ -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", +] diff --git a/src/cleveragents/domain/models/observability/llm_trace.py b/src/cleveragents/domain/models/observability/llm_trace.py new file mode 100644 index 000000000..51ff24911 --- /dev/null +++ b/src/cleveragents/domain/models/observability/llm_trace.py @@ -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, + ) diff --git a/src/cleveragents/domain/models/observability/metrics.py b/src/cleveragents/domain/models/observability/metrics.py new file mode 100644 index 000000000..bed27dfc0 --- /dev/null +++ b/src/cleveragents/domain/models/observability/metrics.py @@ -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), + ) diff --git a/src/cleveragents/infrastructure/database/llm_trace_repository.py b/src/cleveragents/infrastructure/database/llm_trace_repository.py new file mode 100644 index 000000000..bf6c28e10 --- /dev/null +++ b/src/cleveragents/infrastructure/database/llm_trace_repository.py @@ -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), + ) diff --git a/src/cleveragents/infrastructure/database/models.py b/src/cleveragents/infrastructure/database/models.py index 848e70545..e29650793 100644 --- a/src/cleveragents/infrastructure/database/models.py +++ b/src/cleveragents/infrastructure/database/models.py @@ -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"), + ) diff --git a/vulture_whitelist.py b/vulture_whitelist.py index a48636350..148bf4b54 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -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 -- 2.52.0