diff --git a/.gitignore b/.gitignore index 68b627e13..5a54fe3c7 100644 --- a/.gitignore +++ b/.gitignore @@ -172,3 +172,4 @@ src/cleveragents/acp/ # Git worktrees for parallel task branches worktrees/ *.bak +ca-cow-backup-* diff --git a/features/observability/audit_bridge.feature b/features/observability/audit_bridge.feature new file mode 100644 index 000000000..9d78ae927 --- /dev/null +++ b/features/observability/audit_bridge.feature @@ -0,0 +1,156 @@ +Feature: Automatic Audit Bridge + As a security-conscious system + I want domain events with audit semantics to be automatically written to audit.jsonl + So that all security-relevant operations are tracked with structured records + + # --- AuditBridge creation --- + + Scenario: AuditBridge can be instantiated with an event bus + Given a ReactiveEventBus for audit bridge + When I create an AuditBridge with the event bus + Then the audit bridge should be created successfully + + Scenario: AuditBridge subscribes to all audit event types + Given a ReactiveEventBus for audit bridge + When I create an AuditBridge with the event bus + Then the audit bridge should cover at least 10 event types + + # --- Audit record schema --- + + Scenario: AuditRecord has all required fields + When I build an audit record from a plan_applied event + Then the audit record should have a "timestamp" field + And the audit record should have a "event_type" field + And the audit record should have a "actor" field + And the audit record should have a "action" field + And the audit record should have a "resource" field + And the audit record should have a "outcome" field + And the audit record should have a "metadata" field + And the audit record should have a "correlation_id" field + And the audit record should have a "plan_id" field + And the audit record should have a "session_id" field + + Scenario: AuditRecord timestamp is ISO-8601 UTC + When I build an audit record from a plan_applied event + Then the audit record timestamp should be a valid ISO-8601 string + + Scenario: AuditRecord event_type is the dot-separated string + When I build an audit record from a plan_applied event + Then the audit record event_type should equal "plan.applied" + + Scenario: AuditRecord outcome is success for plan_applied + When I build an audit record from a plan_applied event + Then the audit record outcome should equal "success" + + Scenario: AuditRecord outcome is failure for plan_errored + When I build an audit record from a plan_errored event + Then the audit record outcome should equal "failure" + + Scenario: AuditRecord outcome is failure for auth_failure + When I build an audit record from an auth_failure event + Then the audit record outcome should equal "failure" + + Scenario: AuditRecord outcome is failure for invariant_violated + When I build an audit record from an invariant_violated event + Then the audit record outcome should equal "failure" + + Scenario: AuditRecord actor is populated from event actor_name + When I build an audit record from a plan_applied event with actor "admin" + Then the audit record actor should equal "admin" + + Scenario: AuditRecord plan_id is propagated from event + When I build an audit record from a plan_applied event with plan_id "PLAN-001" + Then the audit record plan_id should equal "PLAN-001" + + Scenario: AuditRecord correlation_id is propagated from event + When I build an audit record from a plan_applied event with correlation_id "CORR-XYZ" + Then the audit record correlation_id should equal "CORR-XYZ" + + # --- Event routing --- + + Scenario: plan_applied event is written to audit log + Given an AuditBridge with in-memory capture + When a plan_applied event is emitted on the bus + Then the audit bridge should have 1 record + And the first record event_type should equal "plan.applied" + + Scenario: plan_cancelled event is written to audit log + Given an AuditBridge with in-memory capture + When a plan_cancelled event is emitted on the bus + Then the audit bridge should have 1 record + + Scenario: config_changed event is written to audit log + Given an AuditBridge with in-memory capture + When a config_changed event is emitted on the bus + Then the audit bridge should have 1 record + + Scenario: auth_success event is written to audit log + Given an AuditBridge with in-memory capture + When an auth_success event is emitted on the bus + Then the audit bridge should have 1 record + + Scenario: auth_failure event is written to audit log + Given an AuditBridge with in-memory capture + When an auth_failure event is emitted on the bus + Then the audit bridge should have 1 record + + Scenario: plan_created event is NOT written to audit log + Given an AuditBridge with in-memory capture + When a plan_created event is emitted on the bus + Then the audit bridge should have 0 records + + Scenario: actor_invoked event is NOT written to audit log + Given an AuditBridge with in-memory capture + When an actor_invoked event is emitted on the bus + Then the audit bridge should have 0 records + + Scenario: Multiple audit events are all captured + Given an AuditBridge with in-memory capture + When a plan_applied event is emitted on the bus + And a config_changed event is emitted on the bus + And an auth_failure event is emitted on the bus + Then the audit bridge should have 3 records + + # --- JSONL file output --- + + Scenario: AuditBridge writes records to audit.jsonl file + Given an AuditBridge writing to a temp file + When a plan_applied event is emitted on the bus + Then the audit jsonl file should exist + And the audit jsonl file should contain 1 line + + Scenario: Each line in audit.jsonl is valid JSON + Given an AuditBridge writing to a temp file + When a plan_applied event is emitted on the bus + And a config_changed event is emitted on the bus + Then each line in the audit jsonl file should be valid JSON + + Scenario: audit.jsonl records contain all required fields + Given an AuditBridge writing to a temp file + When a plan_applied event is emitted on the bus + Then the first jsonl record should have field "timestamp" + And the first jsonl record should have field "event_type" + And the first jsonl record should have field "actor" + And the first jsonl record should have field "action" + And the first jsonl record should have field "outcome" + And the first jsonl record should have field "metadata" + And the first jsonl record should have field "correlation_id" + + # --- Error resilience --- + + Scenario: AuditBridge handles file write errors gracefully + Given an AuditBridge with an invalid file path + When a plan_applied event is emitted on the bus + Then no exception should propagate from the audit bridge write + + # --- to_dict and to_json --- + + Scenario: AuditRecord to_dict returns a plain dict + When I build an audit record from a plan_applied event + Then to_dict should return a dict + And the dict should have key "timestamp" + And the dict should have key "event_type" + + Scenario: AuditRecord to_json returns a JSON string + When I build an audit record from a plan_applied event + Then to_json should return a valid JSON string diff --git a/features/observability/prometheus_metrics.feature b/features/observability/prometheus_metrics.feature new file mode 100644 index 000000000..01a5d9f35 --- /dev/null +++ b/features/observability/prometheus_metrics.feature @@ -0,0 +1,105 @@ +Feature: Prometheus Metrics Registry + As a platform operator + I want Prometheus-compatible metrics export + So that I can monitor plan execution, tool calls, LLM usage, and context assembly + + # --- Registry creation --- + + Scenario: PrometheusRegistry can be instantiated + When I create a PrometheusRegistry + Then the registry should be created successfully + + Scenario: PrometheusRegistry registers instruments for all 14 metric keys + When I create a PrometheusRegistry + Then the registry should have 14 instruments registered + + Scenario: PrometheusRegistry registers histogram for PLAN_DURATION_MS + When I create a PrometheusRegistry + Then the instrument for plan_duration_ms should be a Histogram + + Scenario: PrometheusRegistry registers counter for LLM_CALL_COUNT + When I create a PrometheusRegistry + Then the instrument for llm_call_count should be a Counter + + Scenario: PrometheusRegistry registers gauge for PLAN_DECISION_COUNT + When I create a PrometheusRegistry + Then the instrument for plan_decision_count should be a Gauge + + Scenario: PrometheusRegistry registers counter for PLAN_TOTAL_COST_USD + When I create a PrometheusRegistry + Then the instrument for plan_total_cost_usd should be a Counter + + Scenario: PrometheusRegistry registers histogram for ACTOR_LATENCY_MS + When I create a PrometheusRegistry + Then the instrument for actor_latency_ms should be a Histogram + + Scenario: PrometheusRegistry registers counter for TOOL_INVOCATION_COUNT + When I create a PrometheusRegistry + Then the instrument for tool_invocation_count should be a Counter + + Scenario: PrometheusRegistry registers gauge for TOOL_ERROR_RATE + When I create a PrometheusRegistry + Then the instrument for tool_error_rate should be a Gauge + + Scenario: PrometheusRegistry registers histogram for CONTEXT_BUILD_TIME_MS + When I create a PrometheusRegistry + Then the instrument for context_build_time_ms should be a Histogram + + # --- Recording metrics --- + + Scenario: Recording a histogram MetricEntry observes the value + Given a PrometheusRegistry + And a plan_id "01HX0000000000MMMMMMMMMMMM" + When I record a PLAN_DURATION_MS histogram entry with value 1500.0 + Then the Prometheus text output should contain "cleveragents_plan_duration_ms" + + Scenario: Recording a counter MetricEntry increments the counter + Given a PrometheusRegistry + And a plan_id "01HX0000000000MMMMMMMMMMMM" + When I record a LLM_CALL_COUNT counter entry with value 3.0 + Then the Prometheus text output should contain "cleveragents_llm_call_count" + + Scenario: Recording a gauge MetricEntry sets the gauge value + Given a PrometheusRegistry + And a plan_id "01HX0000000000MMMMMMMMMMMM" + When I record a PLAN_DECISION_COUNT gauge entry with value 7.0 + Then the Prometheus text output should contain "cleveragents_plan_decision_count" + + Scenario: generate_text returns bytes + Given a PrometheusRegistry + When I call generate_text + Then the generate_text result should be bytes + + Scenario: generate_text output is non-empty after recording a metric + Given a PrometheusRegistry + And a plan_id "01HX0000000000MMMMMMMMMMMM" + When I record a LLM_TOTAL_TOKENS counter entry with value 1000.0 + And I call generate_text + Then the generate_text result should be non-empty bytes + + # --- File export --- + + Scenario: export_to_file writes metrics to a file + Given a PrometheusRegistry with a temp metrics file + And a plan_id "01HX0000000000MMMMMMMMMMMM" + When I record a ACTOR_INVOCATION_COUNT counter entry with value 5.0 + And I export metrics to file + Then the metrics file should exist + And the metrics file should contain "cleveragents_actor_invocation_count" + + Scenario: export_to_file raises RuntimeError when no path configured + Given a PrometheusRegistry + When I call export_to_file without a path + Then a RuntimeError should be raised + + # --- Settings integration --- + + Scenario: PrometheusRegistry respects metrics_export_prometheus setting + Given settings with metrics_export_prometheus enabled + When I create a MetricsEmitter from settings + Then the emitter prometheus_enabled should be True + + Scenario: PrometheusRegistry is disabled by default in settings + Given default settings for prometheus + When I create a MetricsEmitter from settings + Then the emitter prometheus_enabled should be False diff --git a/features/observability/structlog_config.feature b/features/observability/structlog_config.feature new file mode 100644 index 000000000..4ab4c187e --- /dev/null +++ b/features/observability/structlog_config.feature @@ -0,0 +1,81 @@ +Feature: JSON Structlog Configuration + As a platform operator + I want configurable structured logging with JSON output and correlation IDs + So that I can parse and correlate log records across plan executions + + # --- configure_structlog --- + + Scenario: configure_structlog can be called with json format + When I call configure_structlog with log_format "json" + Then no exception should be raised from configure_structlog + + Scenario: configure_structlog can be called with text format + When I call configure_structlog with log_format "text" + Then no exception should be raised from configure_structlog + + Scenario: configure_structlog can be called with INFO log level + When I call configure_structlog with log_level "INFO" + Then no exception should be raised from configure_structlog + + Scenario: configure_structlog can be called with DEBUG log level + When I call configure_structlog with log_level "DEBUG" + Then no exception should be raised from configure_structlog + + # --- Correlation ID context variable --- + + Scenario: set_correlation_id stores a value retrievable by get_correlation_id + When I set the correlation ID to "CORR-ABC-123" + Then get_correlation_id should return "CORR-ABC-123" + + Scenario: get_correlation_id returns None when not set + Given the correlation ID is cleared + Then get_correlation_id should return None + + Scenario: set_correlation_id can be updated + When I set the correlation ID to "CORR-FIRST" + And I set the correlation ID to "CORR-SECOND" + Then get_correlation_id should return "CORR-SECOND" + + Scenario: set_correlation_id can be cleared with None + When I set the correlation ID to "CORR-TO-CLEAR" + And I set the correlation ID to None + Then get_correlation_id should return None + + # --- inject_correlation_id processor --- + + Scenario: inject_correlation_id adds correlation_id to event dict when set + Given the correlation ID is set to "PROC-CORR-001" + When I run the inject_correlation_id processor on an empty event dict + Then the event dict should contain key "correlation_id" with value "PROC-CORR-001" + + Scenario: inject_correlation_id does not overwrite existing correlation_id + Given the correlation ID is set to "PROC-CORR-002" + When I run the inject_correlation_id processor on an event dict with correlation_id "EXISTING" + Then the event dict should contain key "correlation_id" with value "EXISTING" + + Scenario: inject_correlation_id does nothing when correlation ID is None + Given the correlation ID is cleared + When I run the inject_correlation_id processor on an empty event dict + Then the event dict should not contain key "correlation_id" + + # --- Settings integration --- + + Scenario: Settings has log_format field defaulting to text + When I create default settings for structlog + Then settings.log_format should equal "text" + + Scenario: Settings log_format can be set to json + When I create settings with log_format "json" + Then settings.log_format should equal "json" + + Scenario: Settings has metrics_prometheus_port field + When I create default settings for structlog + Then settings.metrics_prometheus_port should be an integer + + Scenario: Settings has audit_jsonl_path field + When I create default settings for structlog + Then settings.audit_jsonl_path should be a string + + Scenario: Settings has metrics_file_path field + When I create default settings for structlog + Then settings.metrics_file_path should be a string diff --git a/features/steps/audit_bridge_steps.py b/features/steps/audit_bridge_steps.py new file mode 100644 index 000000000..4342b16b5 --- /dev/null +++ b/features/steps/audit_bridge_steps.py @@ -0,0 +1,403 @@ +"""Step definitions for the automatic audit bridge feature (Forgejo #940).""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path +from typing import Any + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.reactive import ReactiveEventBus +from cleveragents.infrastructure.events.types import EventType +from cleveragents.infrastructure.observability.audit_bridge import ( + AUDIT_EVENT_TYPES, + AuditBridge, + _build_audit_record, +) + +__all__: list[str] = [] + +VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_event( + event_type: EventType, + *, + plan_id: str | None = VALID_PLAN_ID, + actor_name: str | None = None, + correlation_id: str | None = None, + session_id: str | None = None, + details: dict[str, Any] | None = None, +) -> DomainEvent: + kwargs: dict[str, Any] = { + "event_type": event_type, + "plan_id": plan_id, + "actor_name": actor_name, + "session_id": session_id, + "details": details or {}, + } + if correlation_id is not None: + kwargs["correlation_id"] = correlation_id + return DomainEvent(**kwargs) + + +# --------------------------------------------------------------------------- +# Given — event bus +# --------------------------------------------------------------------------- + + +@given("a ReactiveEventBus for audit bridge") +def step_given_reactive_event_bus(context: Context) -> None: + context.event_bus = ReactiveEventBus() + + +# --------------------------------------------------------------------------- +# When — AuditBridge creation +# --------------------------------------------------------------------------- + + +@when("I create an AuditBridge with the event bus") +def step_create_audit_bridge(context: Context) -> None: + context.temp_dir = tempfile.mkdtemp() + context.audit_path = Path(context.temp_dir) / "audit.jsonl" + context.audit_bridge = AuditBridge( + event_bus=context.event_bus, + audit_path=context.audit_path, + capture_in_memory=True, + ) + + +# --------------------------------------------------------------------------- +# Then — AuditBridge creation assertions +# --------------------------------------------------------------------------- + + +@then("the audit bridge should be created successfully") +def step_audit_bridge_created(context: Context) -> None: + assert context.audit_bridge is not None + assert isinstance(context.audit_bridge, AuditBridge) + + +@then("the audit bridge should cover at least {count:d} event types") +def step_audit_bridge_covers_event_types(context: Context, count: int) -> None: + assert len(AUDIT_EVENT_TYPES) >= count, ( + f"Expected at least {count} audit event types, got {len(AUDIT_EVENT_TYPES)}" + ) + + +# --------------------------------------------------------------------------- +# When — building audit records +# --------------------------------------------------------------------------- + + +@when("I build an audit record from a plan_applied event") +def step_build_record_plan_applied(context: Context) -> None: + event = _make_event(EventType.PLAN_APPLIED) + context.audit_record = _build_audit_record(event) + + +@when("I build an audit record from a plan_errored event") +def step_build_record_plan_errored(context: Context) -> None: + event = _make_event(EventType.PLAN_ERRORED) + context.audit_record = _build_audit_record(event) + + +@when("I build an audit record from an auth_failure event") +def step_build_record_auth_failure(context: Context) -> None: + event = _make_event(EventType.AUTH_FAILURE) + context.audit_record = _build_audit_record(event) + + +@when("I build an audit record from an invariant_violated event") +def step_build_record_invariant_violated(context: Context) -> None: + event = _make_event(EventType.INVARIANT_VIOLATED) + context.audit_record = _build_audit_record(event) + + +@when('I build an audit record from a plan_applied event with actor "{actor}"') +def step_build_record_with_actor(context: Context, actor: str) -> None: + event = _make_event(EventType.PLAN_APPLIED, actor_name=actor) + context.audit_record = _build_audit_record(event) + + +@when('I build an audit record from a plan_applied event with plan_id "{plan_id}"') +def step_build_record_with_plan_id(context: Context, plan_id: str) -> None: + event = _make_event(EventType.PLAN_APPLIED, plan_id=plan_id) + context.audit_record = _build_audit_record(event) + + +@when('I build an audit record from a plan_applied event with correlation_id "{cid}"') +def step_build_record_with_correlation_id(context: Context, cid: str) -> None: + event = _make_event(EventType.PLAN_APPLIED, correlation_id=cid) + context.audit_record = _build_audit_record(event) + + +# --------------------------------------------------------------------------- +# Then — AuditRecord field assertions +# --------------------------------------------------------------------------- + + +@then('the audit record should have a "{field}" field') +def step_record_has_field(context: Context, field: str) -> None: + assert hasattr(context.audit_record, field), f"AuditRecord missing field '{field}'" + + +@then("the audit record timestamp should be a valid ISO-8601 string") +def step_record_timestamp_iso8601(context: Context) -> None: + ts = context.audit_record.timestamp + assert isinstance(ts, str), f"Expected str timestamp, got {type(ts).__name__}" + assert "T" in ts, f"Timestamp '{ts}' does not look like ISO-8601" + + +@then('the audit record event_type should equal "{expected}"') +def step_record_event_type_equals(context: Context, expected: str) -> None: + assert context.audit_record.event_type == expected, ( + f"Expected event_type '{expected}', got '{context.audit_record.event_type}'" + ) + + +@then('the audit record outcome should equal "{expected}"') +def step_record_outcome_equals(context: Context, expected: str) -> None: + assert context.audit_record.outcome == expected, ( + f"Expected outcome '{expected}', got '{context.audit_record.outcome}'" + ) + + +@then('the audit record actor should equal "{expected}"') +def step_record_actor_equals(context: Context, expected: str) -> None: + assert context.audit_record.actor == expected, ( + f"Expected actor '{expected}', got '{context.audit_record.actor}'" + ) + + +@then('the audit record plan_id should equal "{expected}"') +def step_record_plan_id_equals(context: Context, expected: str) -> None: + assert context.audit_record.plan_id == expected, ( + f"Expected plan_id '{expected}', got '{context.audit_record.plan_id}'" + ) + + +@then('the audit record correlation_id should equal "{expected}"') +def step_record_correlation_id_equals(context: Context, expected: str) -> None: + assert context.audit_record.correlation_id == expected, ( + f"Expected correlation_id '{expected}', got " + f"'{context.audit_record.correlation_id}'" + ) + + +# --------------------------------------------------------------------------- +# Given — AuditBridge with in-memory capture +# --------------------------------------------------------------------------- + + +@given("an AuditBridge with in-memory capture") +def step_given_audit_bridge_in_memory(context: Context) -> None: + context.temp_dir = tempfile.mkdtemp() + context.audit_path = Path(context.temp_dir) / "audit.jsonl" + context.event_bus = ReactiveEventBus() + context.audit_bridge = AuditBridge( + event_bus=context.event_bus, + audit_path=context.audit_path, + capture_in_memory=True, + ) + + +@given("an AuditBridge writing to a temp file") +def step_given_audit_bridge_temp_file(context: Context) -> None: + context.temp_dir = tempfile.mkdtemp() + context.audit_path = Path(context.temp_dir) / "audit.jsonl" + context.event_bus = ReactiveEventBus() + context.audit_bridge = AuditBridge( + event_bus=context.event_bus, + audit_path=context.audit_path, + capture_in_memory=False, + ) + + +@given("an AuditBridge with an invalid file path") +def step_given_audit_bridge_invalid_path(context: Context) -> None: + context.event_bus = ReactiveEventBus() + # Use a path that cannot be written to (root-owned directory) + context.audit_bridge = AuditBridge( + event_bus=context.event_bus, + audit_path=Path("/proc/invalid_audit_test_path/audit.jsonl"), + capture_in_memory=True, + ) + + +# --------------------------------------------------------------------------- +# When — emitting events +# --------------------------------------------------------------------------- + + +@when("a plan_applied event is emitted on the bus") +def step_emit_plan_applied_on_bus(context: Context) -> None: + context.event_bus.emit(_make_event(EventType.PLAN_APPLIED)) + + +@when("a plan_cancelled event is emitted on the bus") +def step_emit_plan_cancelled_on_bus(context: Context) -> None: + context.event_bus.emit(_make_event(EventType.PLAN_CANCELLED)) + + +@when("a config_changed event is emitted on the bus") +def step_emit_config_changed_on_bus(context: Context) -> None: + context.event_bus.emit(_make_event(EventType.CONFIG_CHANGED)) + + +@when("an auth_success event is emitted on the bus") +def step_emit_auth_success_on_bus(context: Context) -> None: + context.event_bus.emit(_make_event(EventType.AUTH_SUCCESS)) + + +@when("an auth_failure event is emitted on the bus") +def step_emit_auth_failure_on_bus(context: Context) -> None: + context.event_bus.emit(_make_event(EventType.AUTH_FAILURE)) + + +@when("a plan_created event is emitted on the bus") +def step_emit_plan_created_on_bus(context: Context) -> None: + context.event_bus.emit(_make_event(EventType.PLAN_CREATED)) + + +@when("an actor_invoked event is emitted on the bus") +def step_emit_actor_invoked_on_bus(context: Context) -> None: + context.event_bus.emit(_make_event(EventType.ACTOR_INVOKED)) + + +# --------------------------------------------------------------------------- +# Then — in-memory record assertions +# --------------------------------------------------------------------------- + + +@then("the audit bridge should have {count:d} record") +def step_audit_bridge_has_n_records_singular(context: Context, count: int) -> None: + actual = len(context.audit_bridge.records) + assert actual == count, f"Expected {count} record(s), got {actual}" + + +@then("the audit bridge should have {count:d} records") +def step_audit_bridge_has_n_records(context: Context, count: int) -> None: + actual = len(context.audit_bridge.records) + assert actual == count, f"Expected {count} record(s), got {actual}" + + +@then('the first record event_type should equal "{expected}"') +def step_first_record_event_type(context: Context, expected: str) -> None: + records = context.audit_bridge.records + assert len(records) >= 1, "No records in audit bridge" + assert records[0].event_type == expected, ( + f"Expected event_type '{expected}', got '{records[0].event_type}'" + ) + + +# --------------------------------------------------------------------------- +# Then — JSONL file assertions +# --------------------------------------------------------------------------- + + +@then("the audit jsonl file should exist") +def step_audit_jsonl_exists(context: Context) -> None: + assert context.audit_path.exists(), ( + f"Audit JSONL file {context.audit_path} does not exist" + ) + + +@then("the audit jsonl file should contain {count:d} line") +def step_audit_jsonl_has_n_lines_singular(context: Context, count: int) -> None: + lines = [ + line + for line in context.audit_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + assert len(lines) == count, f"Expected {count} line(s), got {len(lines)}" + + +@then("the audit jsonl file should contain {count:d} lines") +def step_audit_jsonl_has_n_lines(context: Context, count: int) -> None: + lines = [ + line + for line in context.audit_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + assert len(lines) == count, f"Expected {count} line(s), got {len(lines)}" + + +@then("each line in the audit jsonl file should be valid JSON") +def step_audit_jsonl_valid_json(context: Context) -> None: + lines = [ + line + for line in context.audit_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + assert len(lines) > 0, "No lines in audit JSONL file" + for i, line in enumerate(lines): + try: + json.loads(line) + except json.JSONDecodeError as exc: + raise AssertionError( + f"Line {i + 1} is not valid JSON: {exc}\nLine: {line}" + ) from exc + + +@then('the first jsonl record should have field "{field}"') +def step_first_jsonl_record_has_field(context: Context, field: str) -> None: + lines = [ + line + for line in context.audit_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + assert len(lines) >= 1, "No lines in audit JSONL file" + record = json.loads(lines[0]) + assert field in record, ( + f"Expected field '{field}' in JSONL record, got keys: {list(record.keys())}" + ) + + +# --------------------------------------------------------------------------- +# Then — error resilience +# --------------------------------------------------------------------------- + + +@then("no exception should propagate from the audit bridge write") +def step_no_exception_from_audit_bridge(context: Context) -> None: + # If we got here without an exception, the test passes + pass + + +# --------------------------------------------------------------------------- +# Then — to_dict / to_json +# --------------------------------------------------------------------------- + + +@then("to_dict should return a dict") +def step_to_dict_returns_dict(context: Context) -> None: + result = context.audit_record.to_dict() + assert isinstance(result, dict), f"Expected dict, got {type(result).__name__}" + + +@then('the dict should have key "{key}"') +def step_dict_has_key(context: Context, key: str) -> None: + result = context.audit_record.to_dict() + assert key in result, ( + f"Expected key '{key}' in dict, got keys: {list(result.keys())}" + ) + + +@then("to_json should return a valid JSON string") +def step_to_json_returns_valid_json(context: Context) -> None: + result = context.audit_record.to_json() + assert isinstance(result, str), f"Expected str, got {type(result).__name__}" + try: + json.loads(result) + except json.JSONDecodeError as exc: + raise AssertionError(f"to_json() returned invalid JSON: {exc}") from exc diff --git a/features/steps/prometheus_metrics_steps.py b/features/steps/prometheus_metrics_steps.py new file mode 100644 index 000000000..baff38269 --- /dev/null +++ b/features/steps/prometheus_metrics_steps.py @@ -0,0 +1,261 @@ +"""Step definitions for Prometheus metrics registry feature (Forgejo #940).""" + +from __future__ import annotations + +import tempfile +from pathlib import Path + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.config.settings import Settings +from cleveragents.domain.models.observability.metrics import MetricCollector +from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter +from cleveragents.infrastructure.observability.prometheus_registry import ( + PrometheusRegistry, +) + +__all__: list[str] = [] + +VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM" + + +# --------------------------------------------------------------------------- +# When — registry creation +# --------------------------------------------------------------------------- + + +@when("I create a PrometheusRegistry") +def step_create_registry(context: Context) -> None: + context.registry = PrometheusRegistry() + context.registry_error = None + + +# --------------------------------------------------------------------------- +# Then — registry creation assertions +# --------------------------------------------------------------------------- + + +@then("the registry should be created successfully") +def step_registry_created(context: Context) -> None: + assert context.registry is not None + assert isinstance(context.registry, PrometheusRegistry) + + +@then("the registry should have 14 instruments registered") +def step_registry_14_instruments(context: Context) -> None: + assert len(context.registry.instruments) == 14, ( + f"Expected 14 instruments, got {len(context.registry.instruments)}: " + f"{list(context.registry.instruments.keys())}" + ) + + +@then("the instrument for {key} should be a Histogram") +def step_instrument_is_histogram(context: Context, key: str) -> None: + try: + from prometheus_client import Histogram # type: ignore[import-untyped] + except ImportError: + context.scenario.skip("prometheus_client not available") + return + instrument = context.registry.instruments.get(key) + assert instrument is not None, f"No instrument found for key '{key}'" + assert isinstance(instrument, Histogram), ( + f"Expected Histogram for '{key}', got {type(instrument).__name__}" + ) + + +@then("the instrument for {key} should be a Counter") +def step_instrument_is_counter(context: Context, key: str) -> None: + try: + from prometheus_client import Counter # type: ignore[import-untyped] + except ImportError: + context.scenario.skip("prometheus_client not available") + return + instrument = context.registry.instruments.get(key) + assert instrument is not None, f"No instrument found for key '{key}'" + assert isinstance(instrument, Counter), ( + f"Expected Counter for '{key}', got {type(instrument).__name__}" + ) + + +@then("the instrument for {key} should be a Gauge") +def step_instrument_is_gauge(context: Context, key: str) -> None: + try: + from prometheus_client import Gauge # type: ignore[import-untyped] + except ImportError: + context.scenario.skip("prometheus_client not available") + return + instrument = context.registry.instruments.get(key) + assert instrument is not None, f"No instrument found for key '{key}'" + assert isinstance(instrument, Gauge), ( + f"Expected Gauge for '{key}', got {type(instrument).__name__}" + ) + + +# --------------------------------------------------------------------------- +# Given — registry with plan_id +# --------------------------------------------------------------------------- + + +@given("a PrometheusRegistry") +def step_given_registry(context: Context) -> None: + context.registry = PrometheusRegistry() + + +@given('a plan_id "{plan_id}"') +def step_given_plan_id_quoted(context: Context, plan_id: str) -> None: + context.plan_id = plan_id + + +# --------------------------------------------------------------------------- +# When — recording metrics +# --------------------------------------------------------------------------- + + +@when("I record a PLAN_DURATION_MS histogram entry with value {value:f}") +def step_record_plan_duration(context: Context, value: float) -> None: + entry = MetricCollector.plan_duration(context.plan_id, value) + context.registry.record(entry) + + +@when("I record a LLM_CALL_COUNT counter entry with value {value:f}") +def step_record_llm_call_count(context: Context, value: float) -> None: + entry = MetricCollector.llm_call_count(context.plan_id, value) + context.registry.record(entry) + + +@when("I record a PLAN_DECISION_COUNT gauge entry with value {value:f}") +def step_record_plan_decision_count(context: Context, value: float) -> None: + entry = MetricCollector.plan_decision_count(context.plan_id, value) + context.registry.record(entry) + + +@when("I record a LLM_TOTAL_TOKENS counter entry with value {value:f}") +def step_record_llm_total_tokens(context: Context, value: float) -> None: + entry = MetricCollector.llm_total_tokens(context.plan_id, value) + context.registry.record(entry) + + +@when("I record a ACTOR_INVOCATION_COUNT counter entry with value {value:f}") +def step_record_actor_invocation_count(context: Context, value: float) -> None: + entry = MetricCollector.actor_invocation_count(context.plan_id, value) + context.registry.record(entry) + + +# --------------------------------------------------------------------------- +# When — generate_text +# --------------------------------------------------------------------------- + + +@when("I call generate_text") +def step_call_generate_text(context: Context) -> None: + context.generate_text_result = context.registry.generate_text() + + +# --------------------------------------------------------------------------- +# Then — generate_text assertions +# --------------------------------------------------------------------------- + + +@then("the generate_text result should be bytes") +def step_result_is_bytes(context: Context) -> None: + assert isinstance(context.generate_text_result, bytes), ( + f"Expected bytes, got {type(context.generate_text_result).__name__}" + ) + + +@then("the generate_text result should be non-empty bytes") +def step_result_is_non_empty_bytes(context: Context) -> None: + assert isinstance(context.generate_text_result, bytes) + assert len(context.generate_text_result) > 0 + + +@then('the Prometheus text output should contain "{text}"') +def step_prometheus_output_contains(context: Context, text: str) -> None: + output = context.registry.generate_text().decode("utf-8") + assert text in output, ( + f"Expected '{text}' in Prometheus output, but got:\n{output[:500]}" + ) + + +# --------------------------------------------------------------------------- +# Given / When / Then — file export +# --------------------------------------------------------------------------- + + +@given("a PrometheusRegistry with a temp metrics file") +def step_given_registry_with_temp_file(context: Context) -> None: + context.temp_dir = tempfile.mkdtemp() + context.metrics_file = Path(context.temp_dir) / "metrics.prom" + context.registry = PrometheusRegistry(metrics_file=context.metrics_file) + + +@when("I export metrics to file") +def step_export_to_file(context: Context) -> None: + context.registry.export_to_file() + + +@then("the metrics file should exist") +def step_metrics_file_exists(context: Context) -> None: + assert context.metrics_file.exists(), ( + f"Metrics file {context.metrics_file} does not exist" + ) + + +@then('the metrics file should contain "{text}"') +def step_metrics_file_contains(context: Context, text: str) -> None: + content = context.metrics_file.read_text(encoding="utf-8") + assert text in content, f"Expected '{text}' in metrics file, got:\n{content[:500]}" + + +@when("I call export_to_file without a path") +def step_export_to_file_no_path(context: Context) -> None: + try: + context.registry.export_to_file() + context.export_error = None + except RuntimeError as exc: + context.export_error = exc + + +@then("a RuntimeError should be raised") +def step_runtime_error_raised(context: Context) -> None: + assert context.export_error is not None, "Expected RuntimeError but none was raised" + assert isinstance(context.export_error, RuntimeError) + + +# --------------------------------------------------------------------------- +# Settings integration +# --------------------------------------------------------------------------- + + +@given("settings with metrics_export_prometheus enabled") +def step_settings_prometheus_enabled(context: Context) -> None: + import os + + Settings._instance = None + os.environ["CLEVERAGENTS_METRICS_EXPORT_PROMETHEUS"] = "true" + try: + context.settings = Settings(database_url="sqlite:///:memory:") + finally: + del os.environ["CLEVERAGENTS_METRICS_EXPORT_PROMETHEUS"] + + +@given("default settings for prometheus") +def step_default_settings(context: Context) -> None: + Settings._instance = None + context.settings = Settings(database_url="sqlite:///:memory:") + + +@when("I create a MetricsEmitter from settings") +def step_create_emitter_from_settings(context: Context) -> None: + context.emitter = MetricsEmitter.from_settings(context.settings) + + +@then("the emitter prometheus_enabled should be True") +def step_emitter_prometheus_enabled_true(context: Context) -> None: + assert context.emitter.prometheus_enabled is True + + +@then("the emitter prometheus_enabled should be False") +def step_emitter_prometheus_enabled_false(context: Context) -> None: + assert context.emitter.prometheus_enabled is False diff --git a/features/steps/structlog_config_steps.py b/features/steps/structlog_config_steps.py new file mode 100644 index 000000000..1fcf7f7cb --- /dev/null +++ b/features/steps/structlog_config_steps.py @@ -0,0 +1,170 @@ +"""Step definitions for JSON structlog configuration feature (Forgejo #940).""" + +from __future__ import annotations + +from typing import Any + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.config.settings import Settings +from cleveragents.infrastructure.observability.structlog_config import ( + configure_structlog, + get_correlation_id, + inject_correlation_id, + set_correlation_id, +) + +__all__: list[str] = [] + + +# --------------------------------------------------------------------------- +# When — configure_structlog +# --------------------------------------------------------------------------- + + +@when('I call configure_structlog with log_format "{log_format}"') +def step_configure_structlog_format(context: Context, log_format: str) -> None: + context.configure_error = None + try: + configure_structlog(log_format=log_format) + except Exception as exc: + context.configure_error = exc + + +@when('I call configure_structlog with log_level "{log_level}"') +def step_configure_structlog_level(context: Context, log_level: str) -> None: + context.configure_error = None + try: + configure_structlog(log_level=log_level) + except Exception as exc: + context.configure_error = exc + + +@then("no exception should be raised from configure_structlog") +def step_no_exception_raised(context: Context) -> None: + error = getattr(context, "configure_error", None) + assert error is None, f"Unexpected exception: {error}" + + +# --------------------------------------------------------------------------- +# When / Then — correlation ID context variable +# --------------------------------------------------------------------------- + + +@when('I set the correlation ID to "{cid}"') +def step_set_correlation_id(context: Context, cid: str) -> None: + set_correlation_id(cid) + + +@when("I set the correlation ID to None") +def step_set_correlation_id_none(context: Context) -> None: + set_correlation_id(None) + + +@given("the correlation ID is cleared") +def step_clear_correlation_id(context: Context) -> None: + set_correlation_id(None) + + +@given('the correlation ID is set to "{cid}"') +def step_given_correlation_id_set(context: Context, cid: str) -> None: + set_correlation_id(cid) + + +@then('get_correlation_id should return "{expected}"') +def step_get_correlation_id_returns(context: Context, expected: str) -> None: + actual = get_correlation_id() + assert actual == expected, f"Expected '{expected}', got '{actual}'" + + +@then("get_correlation_id should return None") +def step_get_correlation_id_returns_none(context: Context) -> None: + assert get_correlation_id() is None, f"Expected None, got '{get_correlation_id()}'" + + +# --------------------------------------------------------------------------- +# When / Then — inject_correlation_id processor +# --------------------------------------------------------------------------- + + +@when("I run the inject_correlation_id processor on an empty event dict") +def step_run_processor_empty_dict(context: Context) -> None: + context.event_dict: dict[str, Any] = {} + context.event_dict = inject_correlation_id(None, "info", context.event_dict) + + +@when( + 'I run the inject_correlation_id processor on an event dict with correlation_id "{existing}"' +) +def step_run_processor_existing_cid(context: Context, existing: str) -> None: + context.event_dict = {"correlation_id": existing} + context.event_dict = inject_correlation_id(None, "info", context.event_dict) + + +@then('the event dict should contain key "{key}" with value "{value}"') +def step_event_dict_has_key_value(context: Context, key: str, value: str) -> None: + assert key in context.event_dict, ( + f"Key '{key}' not found in event dict: {context.event_dict}" + ) + assert context.event_dict[key] == value, ( + f"Expected event_dict['{key}'] == '{value}', got '{context.event_dict[key]}'" + ) + + +@then('the event dict should not contain key "{key}"') +def step_event_dict_missing_key(context: Context, key: str) -> None: + assert key not in context.event_dict, ( + f"Key '{key}' unexpectedly found in event dict: {context.event_dict}" + ) + + +# --------------------------------------------------------------------------- +# Settings integration +# --------------------------------------------------------------------------- + + +@when("I create default settings for structlog") +def step_create_default_settings(context: Context) -> None: + Settings._instance = None + context.settings = Settings(database_url="sqlite:///:memory:") + + +@when('I create settings with log_format "{log_format}"') +def step_create_settings_with_log_format(context: Context, log_format: str) -> None: + import os + + Settings._instance = None + os.environ["CLEVERAGENTS_LOG_FORMAT"] = log_format + try: + context.settings = Settings(database_url="sqlite:///:memory:") + finally: + del os.environ["CLEVERAGENTS_LOG_FORMAT"] + + +@then('settings.log_format should equal "{expected}"') +def step_settings_log_format_equals(context: Context, expected: str) -> None: + assert context.settings.log_format == expected, ( + f"Expected log_format '{expected}', got '{context.settings.log_format}'" + ) + + +@then("settings.metrics_prometheus_port should be an integer") +def step_settings_prometheus_port_is_int(context: Context) -> None: + assert isinstance(context.settings.metrics_prometheus_port, int), ( + f"Expected int, got {type(context.settings.metrics_prometheus_port).__name__}" + ) + + +@then("settings.audit_jsonl_path should be a string") +def step_settings_audit_jsonl_path_is_str(context: Context) -> None: + assert isinstance(context.settings.audit_jsonl_path, str), ( + f"Expected str, got {type(context.settings.audit_jsonl_path).__name__}" + ) + + +@then("settings.metrics_file_path should be a string") +def step_settings_metrics_file_path_is_str(context: Context) -> None: + assert isinstance(context.settings.metrics_file_path, str), ( + f"Expected str, got {type(context.settings.metrics_file_path).__name__}" + ) diff --git a/pyproject.toml b/pyproject.toml index ad642ab01..06962be53 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,6 +47,7 @@ dependencies = [ "jsonschema>=4.20.0", # JSON Schema validation for tool inputs/outputs "tomlkit>=0.13.0", # TOML writing with comment preservation for config CLI "tenacity>=8.2.0", # Retry framework for service layer resilience + "prometheus-client>=0.20.0", # Prometheus metrics export (#940) ] [project.optional-dependencies] diff --git a/src/cleveragents/config/settings.py b/src/cleveragents/config/settings.py index e361b6216..2ad236c5f 100644 --- a/src/cleveragents/config/settings.py +++ b/src/cleveragents/config/settings.py @@ -505,6 +505,33 @@ class Settings(BaseSettings): description="Enable Prometheus metrics export endpoint.", ) + # Observability configuration (#940) + metrics_prometheus_port: int = Field( + default=9090, + ge=1, + le=65535, + validation_alias=AliasChoices("CLEVERAGENTS_METRICS_PROMETHEUS_PORT"), + description="HTTP port for Prometheus /metrics endpoint (server mode).", + ) + metrics_file_path: str = Field( + default="metrics.prom", + validation_alias=AliasChoices("CLEVERAGENTS_METRICS_FILE_PATH"), + description="File path for Prometheus text-format export (local mode).", + ) + audit_jsonl_path: str = Field( + default="audit.jsonl", + validation_alias=AliasChoices("CLEVERAGENTS_AUDIT_JSONL_PATH"), + description="File path for the automatic audit bridge JSONL output.", + ) + log_format: str = Field( + default="text", + validation_alias=AliasChoices("CLEVERAGENTS_LOG_FORMAT"), + description=( + "Log output format: 'json' for structured JSON lines, " + "'text' for human-readable console output." + ), + ) + # Retry policy configuration (M6 - #313) retry_max_attempts: int = Field( default=3, diff --git a/src/cleveragents/infrastructure/observability/__init__.py b/src/cleveragents/infrastructure/observability/__init__.py index 9243ed1cf..af739c459 100644 --- a/src/cleveragents/infrastructure/observability/__init__.py +++ b/src/cleveragents/infrastructure/observability/__init__.py @@ -1,13 +1,39 @@ -"""Observability infrastructure for metric emission. +"""Observability infrastructure package. -Provides ``MetricsEmitter`` for structured log emission of -``MetricEntry`` objects in local mode. +Provides: -Based on Forgejo issue #579. +* ``MetricsEmitter`` — structured log emission of ``MetricEntry`` objects. +* ``PrometheusRegistry`` — Prometheus metrics registry with counters/histograms/gauges. +* ``AuditBridge`` — automatic audit bridge writing security events to ``audit.jsonl``. +* ``configure_structlog`` — JSON structlog formatter configuration. + +Based on Forgejo issues #579 and #940. """ from __future__ import annotations +from cleveragents.infrastructure.observability.audit_bridge import ( + AUDIT_EVENT_TYPES, + AuditBridge, + AuditRecord, +) from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter +from cleveragents.infrastructure.observability.prometheus_registry import ( + PrometheusRegistry, +) +from cleveragents.infrastructure.observability.structlog_config import ( + configure_structlog, + get_correlation_id, + set_correlation_id, +) -__all__ = ["MetricsEmitter"] +__all__ = [ + "AUDIT_EVENT_TYPES", + "AuditBridge", + "AuditRecord", + "MetricsEmitter", + "PrometheusRegistry", + "configure_structlog", + "get_correlation_id", + "set_correlation_id", +] diff --git a/src/cleveragents/infrastructure/observability/audit_bridge.py b/src/cleveragents/infrastructure/observability/audit_bridge.py new file mode 100644 index 000000000..e2f26a644 --- /dev/null +++ b/src/cleveragents/infrastructure/observability/audit_bridge.py @@ -0,0 +1,304 @@ +"""Automatic audit bridge: event subscriber writing to ``audit.jsonl``. + +The ``AuditBridge`` subscribes to all domain events that carry +``audit=True`` semantics (the :data:`AUDIT_EVENT_TYPES` set) and writes +a structured JSON line to ``audit.jsonl`` for each matching event. + +Each audit record conforms to the :class:`AuditRecord` schema: + +* ``timestamp`` — ISO-8601 UTC string +* ``event_type`` — dot-separated event type string +* ``actor`` — actor name or ``None`` +* ``action`` — action extracted from event details or event type +* ``resource`` — resource identifier from event details or ``None`` +* ``outcome`` — ``"success"`` or ``"failure"`` based on event type +* ``metadata`` — sanitised copy of event details (secrets redacted) +* ``correlation_id`` — ULID linking related events +* ``plan_id`` — associated plan ULID or ``None`` +* ``session_id`` — associated session ULID or ``None`` + +Based on Forgejo issue #940. +""" + +from __future__ import annotations + +import json +import logging +from datetime import UTC +from pathlib import Path +from typing import Any + +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.protocol import EventBus +from cleveragents.infrastructure.events.types import EventType +from cleveragents.shared.redaction import redact_dict + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Audit event set — events that are automatically written to audit.jsonl +# --------------------------------------------------------------------------- + +#: Domain event types that are automatically bridged to the audit log. +#: These correspond to security-relevant and compliance-critical operations. +AUDIT_EVENT_TYPES: frozenset[EventType] = frozenset( + { + EventType.PLAN_APPLIED, + EventType.PLAN_CANCELLED, + EventType.PLAN_ERRORED, + EventType.RESOURCE_MODIFIED, + EventType.CORRECTION_APPLIED, + EventType.CONFIG_CHANGED, + EventType.ENTITY_DELETED, + EventType.SESSION_CREATED, + EventType.AUTH_SUCCESS, + EventType.AUTH_FAILURE, + EventType.INVARIANT_VIOLATED, + EventType.INVARIANT_ENFORCED, + EventType.DECISION_APPROVED, + EventType.DECISION_CORRECTED, + } +) + +# Event types whose outcome is classified as "failure" +_FAILURE_EVENT_TYPES: frozenset[EventType] = frozenset( + { + EventType.PLAN_ERRORED, + EventType.AUTH_FAILURE, + EventType.INVARIANT_VIOLATED, + } +) + + +class AuditRecord: + """Structured audit log record. + + Attributes: + timestamp: ISO-8601 UTC timestamp string. + event_type: Dot-separated event type string. + actor: Actor name or ``None``. + action: Action description derived from event details or type. + resource: Resource identifier or ``None``. + outcome: ``"success"`` or ``"failure"``. + metadata: Sanitised event details dict. + correlation_id: ULID linking related events. + plan_id: Associated plan ULID or ``None``. + session_id: Associated session ULID or ``None``. + """ + + __slots__ = ( + "action", + "actor", + "correlation_id", + "event_type", + "metadata", + "outcome", + "plan_id", + "resource", + "session_id", + "timestamp", + ) + + def __init__( + self, + *, + timestamp: str, + event_type: str, + actor: str | None, + action: str, + resource: str | None, + outcome: str, + metadata: dict[str, Any], + correlation_id: str, + plan_id: str | None, + session_id: str | None, + ) -> None: + self.timestamp = timestamp + self.event_type = event_type + self.actor = actor + self.action = action + self.resource = resource + self.outcome = outcome + self.metadata = metadata + self.correlation_id = correlation_id + self.plan_id = plan_id + self.session_id = session_id + + def to_dict(self) -> dict[str, Any]: + """Serialise the record to a plain dict suitable for JSON output.""" + return { + "timestamp": self.timestamp, + "event_type": self.event_type, + "actor": self.actor, + "action": self.action, + "resource": self.resource, + "outcome": self.outcome, + "metadata": self.metadata, + "correlation_id": self.correlation_id, + "plan_id": self.plan_id, + "session_id": self.session_id, + } + + def to_json(self) -> str: + """Serialise the record to a JSON string (single line).""" + return json.dumps(self.to_dict(), default=str) + + +def _build_audit_record(event: DomainEvent) -> AuditRecord: + """Build an :class:`AuditRecord` from a :class:`DomainEvent`. + + Args: + event: The domain event to convert. + + Returns: + A structured audit record. + """ + details = dict(event.details) if event.details else {} + sanitised = redact_dict(details) + + # Derive action from details or fall back to event type + action: str = ( + str(details.get("action_name", "")) + or str(details.get("action", "")) + or str(event.event_type) + ) + + # Derive resource from details + resource: str | None = ( + str(details.get("resource_id", "")) or str(details.get("resource", "")) or None + ) + if resource == "": + resource = None + + outcome = "failure" if event.event_type in _FAILURE_EVENT_TYPES else "success" + + return AuditRecord( + timestamp=event.timestamp.astimezone(UTC).isoformat(), + event_type=str(event.event_type), + actor=event.actor_name, + action=action, + resource=resource, + outcome=outcome, + metadata=sanitised, + correlation_id=event.correlation_id, + plan_id=event.plan_id, + session_id=event.session_id, + ) + + +class AuditBridge: + """Event subscriber that writes audit-flagged events to ``audit.jsonl``. + + On construction the bridge subscribes to every event type in + :data:`AUDIT_EVENT_TYPES` on the provided *event_bus*. When a + matching event is received it is converted to an :class:`AuditRecord` + and appended as a JSON line to the configured audit file. + + Errors during file I/O are logged but never propagated so that a + failing audit write does not interrupt normal event processing. + + Attributes: + _audit_path: Path to the ``audit.jsonl`` file. + _event_bus: The event bus this bridge is subscribed to. + _records: In-memory list of records written (for testing). + """ + + def __init__( + self, + *, + event_bus: EventBus, + audit_path: Path | str = Path("audit.jsonl"), + capture_in_memory: bool = False, + ) -> None: + """Initialise the audit bridge and subscribe to audit event types. + + Args: + event_bus: The event bus to subscribe to. + audit_path: Path to the audit JSONL file. + capture_in_memory: If ``True``, also keep records in + ``self.records`` (useful for testing without file I/O). + """ + self._audit_path = Path(audit_path) + self._event_bus = event_bus + self._capture_in_memory = capture_in_memory + self._records: list[AuditRecord] = [] + + for event_type in AUDIT_EVENT_TYPES: + event_bus.subscribe(event_type, self._handle_event) + + # ------------------------------------------------------------------ + # Internal handler + # ------------------------------------------------------------------ + + def _handle_event(self, event: DomainEvent) -> None: + """Handle an incoming audit-flagged domain event. + + Args: + event: The domain event to audit. + """ + try: + record = _build_audit_record(event) + if self._capture_in_memory: + self._records.append(record) + self._write_record(record) + except Exception as exc: + logger.warning( + "audit_bridge.handle_error", + extra={ + "event_type": str(event.event_type), + "error": str(exc), + }, + ) + + def _write_record(self, record: AuditRecord) -> None: + """Append *record* as a JSON line to the audit file. + + Args: + record: The audit record to persist. + """ + try: + self._audit_path.parent.mkdir(parents=True, exist_ok=True) + with self._audit_path.open("a", encoding="utf-8") as fh: + fh.write(record.to_json() + "\n") + except OSError as exc: + logger.error( + "audit_bridge.write_error", + extra={ + "path": str(self._audit_path), + "error": str(exc), + }, + ) + + # ------------------------------------------------------------------ + # Public helpers + # ------------------------------------------------------------------ + + @property + def records(self) -> list[AuditRecord]: + """In-memory audit records (only populated when ``capture_in_memory=True``).""" + return list(self._records) + + def read_records(self) -> list[dict[str, Any]]: + """Read all records from the audit JSONL file. + + Returns: + List of parsed audit record dicts. + """ + if not self._audit_path.exists(): + return [] + records: list[dict[str, Any]] = [] + with self._audit_path.open("r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if line: + try: + records.append(json.loads(line)) + except json.JSONDecodeError: + logger.warning( + "audit_bridge.malformed_line", + extra={"line": line[:100]}, + ) + return records + + +__all__ = ["AUDIT_EVENT_TYPES", "AuditBridge", "AuditRecord"] diff --git a/src/cleveragents/infrastructure/observability/prometheus_registry.py b/src/cleveragents/infrastructure/observability/prometheus_registry.py new file mode 100644 index 000000000..949a2f2de --- /dev/null +++ b/src/cleveragents/infrastructure/observability/prometheus_registry.py @@ -0,0 +1,245 @@ +"""Prometheus metrics registry for CleverAgents observability. + +Provides a ``PrometheusRegistry`` that wraps ``prometheus_client`` to +expose counters, histograms, and gauges for the 14 operational metric +keys defined in :mod:`cleveragents.domain.models.observability.metrics`. + +The registry supports two export modes: + +* **Server mode** — exposes a ``/metrics`` HTTP endpoint via + ``prometheus_client.start_http_server``. +* **Local mode** — writes metrics to a file via + ``prometheus_client.write_to_textfile``. + +Based on Forgejo issue #940. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import prometheus_client # type: ignore[import-untyped] +from prometheus_client import ( # type: ignore[import-untyped] + CollectorRegistry, + Counter, + Gauge, + Histogram, + generate_latest, + start_http_server, + write_to_textfile, +) + +if TYPE_CHECKING: + from cleveragents.domain.models.observability.metrics import MetricEntry + +logger = logging.getLogger(__name__) + + +# Histogram bucket boundaries for latency metrics (milliseconds) +_LATENCY_BUCKETS = ( + 10.0, + 50.0, + 100.0, + 250.0, + 500.0, + 1000.0, + 2500.0, + 5000.0, + 10000.0, + float("inf"), +) + +# Histogram bucket boundaries for cost metrics (USD) +_COST_BUCKETS = ( + 0.001, + 0.005, + 0.01, + 0.05, + 0.1, + 0.5, + 1.0, + 5.0, + float("inf"), +) + + +class PrometheusRegistry: + """Prometheus metrics registry for CleverAgents operational metrics. + + Registers one Prometheus instrument per ``OperationalMetricKey``: + + * ``HISTOGRAM`` keys → :class:`prometheus_client.Histogram` + * ``COUNTER`` keys → :class:`prometheus_client.Counter` + * ``GAUGE`` keys → :class:`prometheus_client.Gauge` + + All instruments are registered on an isolated + :class:`prometheus_client.CollectorRegistry` so that tests can create + multiple instances without collisions. + + Attributes: + _registry: The underlying Prometheus collector registry. + _instruments: Mapping from metric key string to Prometheus instrument. + _http_port: Port for the HTTP metrics server (server mode). + _metrics_file: Path for file-based export (local mode). + _server_started: Whether the HTTP server has been started. + """ + + def __init__( + self, + *, + http_port: int | None = None, + metrics_file: Path | str | None = None, + ) -> None: + """Initialise the registry and register all instruments. + + Args: + http_port: If set, ``start_server()`` will expose metrics on + this port (server mode). + metrics_file: If set, ``export_to_file()`` will write metrics + to this path (local mode). + """ + self._http_port = http_port + self._metrics_file = Path(metrics_file) if metrics_file else None + self._server_started = False + self._instruments: dict[str, Any] = {} + self._registry: CollectorRegistry = CollectorRegistry() + self._register_instruments() + + # ------------------------------------------------------------------ + # Instrument registration + # ------------------------------------------------------------------ + + def _register_instruments(self) -> None: + """Register all 14 operational metric instruments.""" + from cleveragents.domain.models.observability.metrics import ( + METRIC_DEFINITIONS, + MetricType, + ) + + for key, defn in METRIC_DEFINITIONS.items(): + metric_name = f"cleveragents_{key.value.replace('.', '_')}" + labels = ["plan_id"] + + if defn.metric_type == MetricType.HISTOGRAM: + buckets = ( + _COST_BUCKETS + if "cost" in key.value or "usd" in key.value + else _LATENCY_BUCKETS + ) + self._instruments[key.value] = Histogram( + metric_name, + defn.description or key.value, + labels, + buckets=buckets, + registry=self._registry, + ) + elif defn.metric_type == MetricType.COUNTER: + self._instruments[key.value] = Counter( + metric_name, + defn.description or key.value, + labels, + registry=self._registry, + ) + elif defn.metric_type == MetricType.GAUGE: + self._instruments[key.value] = Gauge( + metric_name, + defn.description or key.value, + labels, + registry=self._registry, + ) + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def record(self, entry: MetricEntry) -> None: + """Record a single ``MetricEntry`` in the Prometheus registry. + + Dispatches to the appropriate Prometheus instrument based on the + entry's ``metric_type``. Unknown keys are silently ignored. + + Args: + entry: The metric entry to record. + """ + instrument = self._instruments.get(str(entry.key)) + if instrument is None: + logger.debug( + "prometheus_registry.unknown_key", + extra={"key": str(entry.key)}, + ) + return + + plan_id = entry.plan_id or "unknown" + + from cleveragents.domain.models.observability.metrics import MetricType + + if entry.metric_type == MetricType.HISTOGRAM: + instrument.labels(plan_id=plan_id).observe(entry.value) + elif entry.metric_type == MetricType.COUNTER: + instrument.labels(plan_id=plan_id).inc(entry.value) + elif entry.metric_type == MetricType.GAUGE: + instrument.labels(plan_id=plan_id).set(entry.value) + + def generate_text(self) -> bytes: + """Generate Prometheus text-format metrics output. + + Returns: + UTF-8 encoded Prometheus text exposition format. + """ + return generate_latest(self._registry) + + def start_server(self, port: int | None = None) -> None: + """Start the Prometheus HTTP metrics server. + + Args: + port: Override the port configured at construction time. + + Raises: + RuntimeError: If no port is configured. + RuntimeError: If the server has already been started. + """ + effective_port = port or self._http_port + if effective_port is None: + raise RuntimeError("No HTTP port configured for Prometheus metrics server.") + if self._server_started: + raise RuntimeError("Prometheus metrics server is already running.") + start_http_server(effective_port, registry=self._registry) + self._server_started = True + logger.info( + "prometheus_registry.server_started", + extra={"port": effective_port}, + ) + + def export_to_file(self, path: Path | str | None = None) -> None: + """Write metrics to a text file in Prometheus format. + + Args: + path: Override the file path configured at construction time. + + Raises: + RuntimeError: If no file path is configured. + """ + effective_path = Path(path) if path else self._metrics_file + if effective_path is None: + raise RuntimeError("No metrics file path configured for file-based export.") + effective_path.parent.mkdir(parents=True, exist_ok=True) + write_to_textfile(str(effective_path), self._registry) + logger.debug( + "prometheus_registry.exported_to_file", + extra={"path": str(effective_path)}, + ) + + @property + def instruments(self) -> dict[str, Any]: + """Read-only view of registered Prometheus instruments.""" + return dict(self._instruments) + + @property + def server_started(self) -> bool: + """Whether the HTTP metrics server has been started.""" + return self._server_started + + +__all__ = ["PrometheusRegistry"] diff --git a/src/cleveragents/infrastructure/observability/structlog_config.py b/src/cleveragents/infrastructure/observability/structlog_config.py new file mode 100644 index 000000000..27924af64 --- /dev/null +++ b/src/cleveragents/infrastructure/observability/structlog_config.py @@ -0,0 +1,152 @@ +"""JSON structlog formatter and configuration. + +Provides ``configure_structlog`` to set up :mod:`structlog` with either +JSON output (for production / machine-readable logs) or coloured +console output (for development). + +The format is controlled by the ``log_format`` setting: + +* ``"json"`` — structured JSON lines, one per log record. +* ``"text"`` (default) — human-readable coloured console output. + +Log correlation IDs are injected into every log record via the +``inject_correlation_id`` processor when a correlation ID is present in +the current context. + +Based on Forgejo issue #940. +""" + +from __future__ import annotations + +import logging +import sys +from contextvars import ContextVar +from typing import Any + +import structlog + +# --------------------------------------------------------------------------- +# Correlation ID context variable +# --------------------------------------------------------------------------- + +#: Thread/async-safe storage for the current log correlation ID. +_correlation_id_var: ContextVar[str | None] = ContextVar( + "log_correlation_id", default=None +) + + +def set_correlation_id(correlation_id: str | None) -> None: + """Set the current log correlation ID in the context. + + Args: + correlation_id: The correlation ID to inject into log records, + or ``None`` to clear it. + """ + _correlation_id_var.set(correlation_id) + + +def get_correlation_id() -> str | None: + """Return the current log correlation ID from the context. + + Returns: + The current correlation ID, or ``None`` if not set. + """ + return _correlation_id_var.get() + + +# --------------------------------------------------------------------------- +# Custom structlog processors +# --------------------------------------------------------------------------- + + +def inject_correlation_id( + logger: Any, + method: str, + event_dict: dict[str, Any], +) -> dict[str, Any]: + """Structlog processor that injects the current correlation ID. + + If a correlation ID is set in the current context it is added to the + log record under the key ``"correlation_id"``. Existing values are + not overwritten. + + Args: + logger: The structlog logger (unused). + method: The log method name (unused). + event_dict: The current log record dict. + + Returns: + The (possibly mutated) event dict. + """ + cid = _correlation_id_var.get() + if cid is not None and "correlation_id" not in event_dict: + event_dict["correlation_id"] = cid + return event_dict + + +# --------------------------------------------------------------------------- +# Configuration entry point +# --------------------------------------------------------------------------- + + +def configure_structlog(*, log_format: str = "text", log_level: str = "INFO") -> None: + """Configure :mod:`structlog` for the application. + + Sets up the shared processors, renderer, and stdlib integration. + Safe to call multiple times — subsequent calls reconfigure structlog. + + Args: + log_format: ``"json"`` for JSON output, ``"text"`` for console. + log_level: Standard log level string (e.g. ``"INFO"``). + """ + level = getattr(logging, log_level.upper(), logging.INFO) + + # Shared processors applied to every log record + shared_processors: list[Any] = [ + structlog.contextvars.merge_contextvars, + inject_correlation_id, + structlog.stdlib.add_logger_name, + structlog.stdlib.add_log_level, + structlog.processors.TimeStamper(fmt="iso", utc=True), + structlog.processors.StackInfoRenderer(), + ] + + if log_format == "json": + renderer: Any = structlog.processors.JSONRenderer() + else: + renderer = structlog.dev.ConsoleRenderer(colors=sys.stderr.isatty()) + + structlog.configure( + processors=[ + *shared_processors, + structlog.stdlib.ProcessorFormatter.wrap_for_formatter, + ], + wrapper_class=structlog.make_filtering_bound_logger(level), + context_class=dict, + logger_factory=structlog.stdlib.LoggerFactory(), + cache_logger_on_first_use=True, + ) + + formatter = structlog.stdlib.ProcessorFormatter( + foreign_pre_chain=shared_processors, + processors=[ + structlog.stdlib.ProcessorFormatter.remove_processors_meta, + renderer, + ], + ) + + handler = logging.StreamHandler(sys.stderr) + handler.setFormatter(formatter) + + root_logger = logging.getLogger() + root_logger.handlers.clear() + root_logger.addHandler(handler) + root_logger.setLevel(level) + + +__all__ = [ + "configure_structlog", + "get_correlation_id", + "inject_correlation_id", + "set_correlation_id", +]