From 3e3e9b4b5d43d63054216a6144fdaec8a21e48f0 Mon Sep 17 00:00:00 2001 From: CoreRasurae Date: Tue, 10 Mar 2026 00:33:58 +0000 Subject: [PATCH] feat(observability): wire AuditService.record() into domain services via EventBus auto-dispatch Implements AuditEventSubscriber that subscribes to all 9 security-relevant EventType members and persists redacted audit entries via AuditService.record(). Key components: - AuditEventSubscriber: bridges EventBus and AuditService (SEC7) - SECURITY_EVENT_MAP: maps EventType enum to audit type strings - Redaction via redact_dict() on event details before persistence - Graceful error handling: failures logged, never propagated Post-review fixes applied: - BUG-1: Remove dead correlation_id null-check guard (DomainEvent.correlation_id is always non-None via ULID default_factory) - SEC-2: Redact exception messages in warning logs via redact_value() to prevent potential leakage of sensitive internal state (e.g. DB connection strings) - PERF-3: Pre-generate unique DomainEvent instances in ASV benchmark setup to avoid skew from reusing a single frozen object - Wire event_bus from the DI container into CorrectionService (plan.py), ConfigService (config.py, skill.py x2, server.py), and PersistentSessionService (session.py) at their CLI construction sites. Closes #581 --- CHANGELOG.md | 20 + benchmarks/bench_audit_service.py | 122 ++++++ features/mocks/__init__.py | 2 + .../mocks/transient_fail_audit_service.py | 30 ++ .../audit_service_wiring.feature | 120 ++++++ features/steps/audit_service_wiring_steps.py | 397 ++++++++++++++++++ features/steps/audit_wiring_extended_steps.py | 179 ++++++++ .../steps/skill_cli_coverage_boost_steps.py | 4 +- robot/audit_service_wiring.robot | 59 +++ robot/helper_audit_wiring.py | 221 ++++++++++ src/cleveragents/application/container.py | 57 ++- .../application/services/__init__.py | 4 + .../application/services/actor_service.py | 32 +- .../services/audit_event_subscriber.py | 151 +++++++ .../application/services/audit_service.py | 62 ++- .../application/services/config_service.py | 35 +- .../services/correction_service.py | 45 ++ .../services/plan_lifecycle_service.py | 113 ++++- .../application/services/project_service.py | 35 +- .../application/services/session_service.py | 42 +- src/cleveragents/cli/commands/config.py | 8 +- src/cleveragents/cli/commands/plan.py | 2 +- src/cleveragents/cli/commands/server.py | 21 +- src/cleveragents/cli/commands/skill.py | 8 +- .../infrastructure/events/reactive.py | 15 +- .../infrastructure/events/types.py | 13 + 26 files changed, 1738 insertions(+), 59 deletions(-) create mode 100644 benchmarks/bench_audit_service.py create mode 100644 features/mocks/transient_fail_audit_service.py create mode 100644 features/observability/audit_service_wiring.feature create mode 100644 features/steps/audit_service_wiring_steps.py create mode 100644 features/steps/audit_wiring_extended_steps.py create mode 100644 robot/audit_service_wiring.robot create mode 100644 robot/helper_audit_wiring.py create mode 100644 src/cleveragents/application/services/audit_event_subscriber.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2cf9fe53e..9b944affe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -73,6 +73,26 @@ `@tdd_bug_` prerequisites. Implemented via `Scenario.run()` monkey-patch in `before_all`. Includes 34 Behave BDD scenarios (19 tag-validation, 14 infrastructure, and 1 demo) and 12 Robot Framework integration test cases. (#627) +- Wired `AuditService.record()` into domain services via EventBus auto-dispatch. + Created `AuditEventSubscriber` that subscribes to 9 security-relevant event types + (`plan_applied`, `plan_cancelled`, `resource_modified`, `correction_applied`, + `config_changed`, `entity_deleted`, `session_created`, `auth_success`, + `auth_failure`) and persists them via `AuditService.record()` with secret masking + applied to all audit log details (always `show_secrets=False`). Subscriber enriches + audit entries with `session_id` and `correlation_id` from the domain event for + traceability. Wired `PlanLifecycleService` to emit `PLAN_APPLIED` and + `PLAN_CANCELLED` events with all project names in event details. Added 5 new + `EventType` enum members. Registered subscriber as eagerly-initialized singleton + in DI container. `AuditService` now accepts an explicit `database_url` parameter + so it shares the same database as the rest of the application. All `EventBus.emit()` + call sites are wrapped in try/except guards with structured logging, and + `ReactiveEventBus` isolates per-handler failures so one failing subscriber cannot + block others. `server connect` now emits per-setting `CONFIG_CHANGED` audit events + via `set_value()`. `SessionService.delete()` emits `ENTITY_DELETED`. Exception + messages in the DI container bootstrap are redacted before logging. + `CorrectionService` is registered as a singleton in the DI container. Includes + 23 Behave BDD scenarios, 5 Robot Framework integration tests, and ASV benchmarks. + (#581) ### Added diff --git a/benchmarks/bench_audit_service.py b/benchmarks/bench_audit_service.py new file mode 100644 index 000000000..b6af169fc --- /dev/null +++ b/benchmarks/bench_audit_service.py @@ -0,0 +1,122 @@ +"""ASV benchmarks for audit service write throughput. + +Measures the overhead of the AuditEventSubscriber pipeline: +EventBus.emit() -> redact_dict() -> AuditService.record() -> SQLite. +""" + +from __future__ import annotations + +import importlib +import sys +from pathlib import Path +from typing import ClassVar + +_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 sqlalchemy import create_engine # noqa: E402 +from sqlalchemy.orm import sessionmaker # noqa: E402 + +from cleveragents.application.services.audit_event_subscriber import ( # noqa: E402 + AuditEventSubscriber, +) +from cleveragents.application.services.audit_service import AuditService # noqa: E402 +from cleveragents.config.settings import Settings # noqa: E402 +from cleveragents.infrastructure.database.models import Base # noqa: E402 +from cleveragents.infrastructure.events.models import DomainEvent # noqa: E402 +from cleveragents.infrastructure.events.reactive import ( # noqa: E402 + ReactiveEventBus, +) +from cleveragents.infrastructure.events.types import EventType # noqa: E402 + + +def _make_audit_service() -> AuditService: + """Create an in-memory AuditService for benchmarking.""" + Settings._instance = None + settings = Settings(database_url="sqlite:///:memory:") + engine = create_engine("sqlite:///:memory:", echo=False) + Base.metadata.create_all(engine) + session = sessionmaker(bind=engine)() + return AuditService(settings=settings, session=session) + + +class AuditServiceBenchmarks: + """Benchmarks for audit event recording throughput.""" + + timeout = 120 + params: ClassVar[list[int]] = [1, 10, 100, 1000] + param_names: ClassVar[list[str]] = ["num_events"] + + def setup(self, num_events: int) -> None: + # Recreate fresh DB/bus per ASV timing round to prevent row + # accumulation across repetitions (PERF-4). + self.svc = _make_audit_service() + self.bus = ReactiveEventBus() + self.subscriber = AuditEventSubscriber( + audit_service=self.svc, + event_bus=self.bus, + ) + # Pre-generate unique events so each iteration has a distinct + # timestamp and correlation_id, avoiding benchmark skew from + # reusing a single frozen object (PERF-3). + self.events = [ + DomainEvent( + event_type=EventType.PLAN_APPLIED, + plan_id=f"BENCH-{i:04d}", + details={"action_name": "local/benchmark"}, + ) + for i in range(num_events) + ] + + def time_record_events(self, num_events: int) -> None: + for event in self.events: + self.bus.emit(event) + + +class AuditDirectRecordBenchmarks: + """Benchmarks for direct AuditService.record() calls.""" + + timeout = 120 + params: ClassVar[list[int]] = [1, 10, 100, 1000] + param_names: ClassVar[list[str]] = ["num_events"] + + def setup(self, num_events: int) -> None: + self.svc = _make_audit_service() + + def time_direct_record(self, num_events: int) -> None: + for _ in range(num_events): + self.svc.record( + event_type="plan_applied", + plan_id="BENCH-001", + details={"action_name": "local/benchmark"}, + ) + + +class AuditRedactionBenchmarks: + """Benchmarks for the redaction overhead in audit recording.""" + + timeout = 120 + + def setup(self) -> None: + self.svc = _make_audit_service() + self.bus = ReactiveEventBus() + self.subscriber = AuditEventSubscriber( + audit_service=self.svc, + event_bus=self.bus, + ) + self.event_with_secrets = DomainEvent( + event_type=EventType.CONFIG_CHANGED, + details={ + "api_key": "sk-proj-ABCDEFGHIJ1234567890", + "token": "tok_ABCDEF1234567890", + "normal_field": "safe-value", + }, + ) + + def time_record_with_redaction(self) -> None: + self.bus.emit(self.event_with_secrets) diff --git a/features/mocks/__init__.py b/features/mocks/__init__.py index fece31add..ad8729563 100644 --- a/features/mocks/__init__.py +++ b/features/mocks/__init__.py @@ -9,6 +9,7 @@ from .fake_provider import FakeProviderInfo, FakeProviderRegistry from .lsp_transport_mock import MockLspTransport, parse_lsp_responses from .mock_ai_provider import MockAIProvider from .mock_mcp_transport import MockMCPTransport +from .transient_fail_audit_service import TransientFailAuditService __all__ = [ "FakeProviderInfo", @@ -16,5 +17,6 @@ __all__ = [ "MockAIProvider", "MockLspTransport", "MockMCPTransport", + "TransientFailAuditService", "parse_lsp_responses", ] diff --git a/features/mocks/transient_fail_audit_service.py b/features/mocks/transient_fail_audit_service.py new file mode 100644 index 000000000..600dc3497 --- /dev/null +++ b/features/mocks/transient_fail_audit_service.py @@ -0,0 +1,30 @@ +"""Transient-failure wrapper for AuditService used in BDD tests. + +Wraps a real ``AuditService`` instance and forces the **first** +``record()`` call to raise, simulating a transient database error. +Subsequent calls delegate to the real service. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from cleveragents.application.services.audit_service import AuditService + + +class TransientFailAuditService: + """Wraps AuditService to fail on the first record() call, then succeed.""" + + def __init__(self, real_service: AuditService) -> None: + self._real = real_service + self._call_count = 0 + + def record(self, **kwargs: Any) -> Any: + self._call_count += 1 + if self._call_count == 1: + raise RuntimeError("Transient DB error") + return self._real.record(**kwargs) + + def __getattr__(self, name: str) -> Any: + return getattr(self._real, name) diff --git a/features/observability/audit_service_wiring.feature b/features/observability/audit_service_wiring.feature new file mode 100644 index 000000000..ce31d810a --- /dev/null +++ b/features/observability/audit_service_wiring.feature @@ -0,0 +1,120 @@ +Feature: AuditService EventBus Wiring + As a security-conscious system + I want domain operations to automatically generate audit log entries + So that all security-relevant events are tracked + + Scenario: Plan applied event generates audit log entry + Given an event bus with an audit subscriber + When a plan_applied event is emitted + Then the audit log should contain a plan_applied entry + + Scenario: Plan cancelled event generates audit log entry + Given an event bus with an audit subscriber + When a plan_cancelled event is emitted + Then the audit log should contain a plan_cancelled entry + + Scenario: Config changed event generates audit log entry + Given an event bus with an audit subscriber + When a config_changed event is emitted + Then the audit log should contain a config_changed entry + + Scenario: Entity deleted event generates audit log entry + Given an event bus with an audit subscriber + When an entity_deleted event is emitted + Then the audit log should contain an entity_deleted entry + + Scenario: Session created event generates audit log entry + Given an event bus with an audit subscriber + When a session_created event is emitted + Then the audit log should contain a session_created entry + + Scenario: Correction applied event generates audit log entry + Given an event bus with an audit subscriber + When a correction_applied event is emitted + Then the audit log should contain a correction_applied entry + + Scenario: Resource modified event generates audit log entry + Given an event bus with an audit subscriber + When a resource_modified event is emitted + Then the audit log should contain a resource_modified entry + + Scenario: Auth success event generates audit log entry + Given an event bus with an audit subscriber + When an auth_success event is emitted + Then the audit log should contain an auth_success entry + + Scenario: Auth failure event generates audit log entry + Given an event bus with an audit subscriber + When an auth_failure event is emitted + Then the audit log should contain an auth_failure entry + + Scenario: Sensitive data in event details is redacted in audit log + Given an event bus with an audit subscriber + When an event with sensitive details is emitted + Then the audit log entry should have redacted values + + Scenario: Multiple events are all logged + Given an event bus with an audit subscriber + When multiple security events are emitted + Then all events should appear in the audit log + + Scenario: Non-security events are not logged + Given an event bus with an audit subscriber + When a non-security event is emitted + Then the audit log should not contain the event + + Scenario: Subscriber maps event type enum values to audit type strings + Given an event bus with an audit subscriber + When a plan_applied event is emitted + Then the audit entry event_type should be the string "plan_applied" + + Scenario: Event plan_id is propagated to audit entry + Given an event bus with an audit subscriber + When a plan_applied event with plan_id "PLAN-123" is emitted + Then the audit entry should have plan_id "PLAN-123" + + Scenario: Event actor_name is propagated to audit entry + Given an event bus with an audit subscriber + When a config_changed event with actor_name "admin-user" is emitted + Then the audit entry should have actor_name "admin-user" + + Scenario: Event session_id is propagated to audit entry details + Given an event bus with an audit subscriber + When a session_created event with session_id "SES-456" is emitted + Then the audit entry details should contain session_id "SES-456" + + Scenario: Event correlation_id is propagated to audit entry details + Given an event bus with an audit subscriber + When a plan_applied event with correlation_id "CORR-789" is emitted + Then the audit entry details should contain correlation_id "CORR-789" + + Scenario: Subscriber handles recording errors gracefully + Given an event bus with a failing audit service + When a plan_applied event is emitted on the failing bus + Then no audit subscriber exception should propagate + + Scenario: Subscriber recovers after a recording failure + Given an event bus with a transiently failing audit service + When a plan_applied event is emitted causing a transient failure + And a plan_cancelled event is emitted after the transient failure + Then the audit log should contain the plan_cancelled entry after recovery + + Scenario: Multiple non-security event types are filtered out + Given an event bus with an audit subscriber + When non-security events PLAN_CREATED and ACTOR_INVOKED are emitted + Then the audit log should remain empty + + Scenario: Nested sensitive data in event details is redacted + Given an event bus with an audit subscriber + When an event with nested sensitive details is emitted + Then the nested audit log values should be redacted + + Scenario: Auto-generated correlation_id is propagated to audit entry + Given an event bus with an audit subscriber + When a plan_applied event without explicit correlation_id is emitted + Then the audit entry details should contain a non-empty correlation_id + + Scenario: User identity is extracted from event details to audit column + Given an event bus with an audit subscriber + When a session_created event with user_identity in details is emitted + Then the audit entry should have user_identity "test-user@example.com" diff --git a/features/steps/audit_service_wiring_steps.py b/features/steps/audit_service_wiring_steps.py new file mode 100644 index 000000000..9b9619354 --- /dev/null +++ b/features/steps/audit_service_wiring_steps.py @@ -0,0 +1,397 @@ +"""Step definitions for AuditService EventBus wiring (Forgejo #581). + +Tests that the :class:`AuditEventSubscriber` correctly bridges +:class:`EventBus` events to :class:`AuditService.record()` with +secret masking, event type mapping, and error resilience. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +from behave import given, then, when +from behave.runner import Context +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from cleveragents.application.services.audit_event_subscriber import ( + AuditEventSubscriber, +) +from cleveragents.application.services.audit_service import AuditService +from cleveragents.config.settings import Settings +from cleveragents.infrastructure.database.models import Base +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.reactive import ReactiveEventBus +from cleveragents.infrastructure.events.types import EventType + +__all__: list[str] = [] + +# ── Helpers ─────────────────────────────────────────────────────── + + +def _make_settings() -> Settings: + """Create a fresh Settings instance for testing.""" + Settings._instance = None + return Settings(database_url="sqlite:///:memory:") + + +def _fresh_audit_service() -> AuditService: + """Create an AuditService backed by an in-memory SQLite database.""" + settings = _make_settings() + engine = create_engine("sqlite:///:memory:", echo=False) + Base.metadata.create_all(engine) + session = sessionmaker(bind=engine)() + return AuditService(settings=settings, session=session) + + +# ── Given ───────────────────────────────────────────────────────── + + +@given("an event bus with an audit subscriber") +def step_event_bus_with_audit_subscriber(context: Context) -> None: + context.audit_service = _fresh_audit_service() + context.event_bus = ReactiveEventBus() + context.subscriber = AuditEventSubscriber( + audit_service=context.audit_service, + event_bus=context.event_bus, + ) + + +@given("an event bus with a failing audit service") +def step_event_bus_with_failing_audit_service(context: Context) -> None: + context.event_bus = ReactiveEventBus() + mock_service = MagicMock(spec=AuditService) + mock_service.record.side_effect = RuntimeError("DB unavailable") + context.subscriber = AuditEventSubscriber( + audit_service=mock_service, + event_bus=context.event_bus, + ) + + +# ── When ────────────────────────────────────────────────────────── + + +@when("a plan_applied event is emitted") +def step_emit_plan_applied(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_APPLIED, + plan_id="PLAN-001", + details={"action_name": "local/deploy"}, + ) + ) + + +@when("a plan_cancelled event is emitted") +def step_emit_plan_cancelled(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_CANCELLED, + plan_id="PLAN-002", + details={"reason": "user requested"}, + ) + ) + + +@when("a config_changed event is emitted") +def step_emit_config_changed(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.CONFIG_CHANGED, + details={"key": "log_level", "value": "DEBUG"}, + ) + ) + + +@when("an entity_deleted event is emitted") +def step_emit_entity_deleted(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.ENTITY_DELETED, + details={"entity": "resource/old-file"}, + ) + ) + + +@when("a session_created event is emitted") +def step_emit_session_created(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.SESSION_CREATED, + session_id="SES-001", + details={"mode": "interactive"}, + ) + ) + + +@when("a correction_applied event is emitted") +def step_emit_correction_applied(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.CORRECTION_APPLIED, + plan_id="PLAN-003", + details={"correction": "fix import"}, + ) + ) + + +@when("a resource_modified event is emitted") +def step_emit_resource_modified(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.RESOURCE_MODIFIED, + details={"path": "src/main.py", "action": "write"}, + ) + ) + + +@when("an auth_success event is emitted") +def step_emit_auth_success(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.AUTH_SUCCESS, + details={"user": "admin", "method": "token"}, + ) + ) + + +@when("an auth_failure event is emitted") +def step_emit_auth_failure(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.AUTH_FAILURE, + details={"user": "unknown", "reason": "invalid credentials"}, + ) + ) + + +@when("an event with sensitive details is emitted") +def step_emit_event_with_sensitive_details(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.CONFIG_CHANGED, + details={ + "key": "openai_api_key", + "api_key": "sk-proj-ABCDEFGHIJ1234567890", + "normal_field": "visible", + }, + ) + ) + + +@when("multiple security events are emitted") +def step_emit_multiple_events(context: Context) -> None: + context.event_bus.emit(DomainEvent(event_type=EventType.PLAN_APPLIED, plan_id="P1")) + context.event_bus.emit( + DomainEvent(event_type=EventType.PLAN_CANCELLED, plan_id="P2") + ) + context.event_bus.emit( + DomainEvent(event_type=EventType.SESSION_CREATED, session_id="S1") + ) + + +@when("a non-security event is emitted") +def step_emit_non_security_event(context: Context) -> None: + context.event_bus.emit( + DomainEvent(event_type=EventType.PLAN_CREATED, plan_id="P-nonsec") + ) + + +@when('a plan_applied event with plan_id "{plan_id}" is emitted') +def step_emit_plan_applied_with_plan_id(context: Context, plan_id: str) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_APPLIED, + plan_id=plan_id, + details={"action_name": "local/test"}, + ) + ) + + +@when('a config_changed event with actor_name "{actor_name}" is emitted') +def step_emit_config_changed_with_actor(context: Context, actor_name: str) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.CONFIG_CHANGED, + actor_name=actor_name, + details={"key": "debug_enabled", "value": "true"}, + ) + ) + + +@when('a session_created event with session_id "{session_id}" is emitted') +def step_emit_session_created_with_session_id( + context: Context, session_id: str +) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.SESSION_CREATED, + session_id=session_id, + details={"mode": "interactive"}, + ) + ) + + +@when('a plan_applied event with correlation_id "{correlation_id}" is emitted') +def step_emit_plan_applied_with_correlation_id( + context: Context, correlation_id: str +) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_APPLIED, + plan_id="PLAN-CORR", + correlation_id=correlation_id, + details={"action_name": "local/test"}, + ) + ) + + +@when("a plan_applied event is emitted on the failing bus") +def step_emit_on_failing_bus(context: Context) -> None: + context.exception_raised = False + try: + context.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_APPLIED, + plan_id="PLAN-FAIL", + ) + ) + except Exception: + context.exception_raised = True + + +# ── Then ────────────────────────────────────────────────────────── + + +@then("the audit log should contain a plan_applied entry") +def step_audit_has_plan_applied(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="plan_applied") + assert len(entries) == 1, f"Expected plan_applied entry, got {len(entries)}" + + +@then("the audit log should contain a plan_cancelled entry") +def step_audit_has_plan_cancelled(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="plan_cancelled") + assert len(entries) == 1, f"Expected plan_cancelled entry, got {len(entries)}" + + +@then("the audit log should contain a config_changed entry") +def step_audit_has_config_changed(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="config_changed") + assert len(entries) == 1, f"Expected config_changed entry, got {len(entries)}" + + +@then("the audit log should contain an entity_deleted entry") +def step_audit_has_entity_deleted(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="entity_deleted") + assert len(entries) == 1, f"Expected entity_deleted entry, got {len(entries)}" + + +@then("the audit log should contain a session_created entry") +def step_audit_has_session_created(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="session_created") + assert len(entries) == 1, f"Expected session_created entry, got {len(entries)}" + + +@then("the audit log should contain a correction_applied entry") +def step_audit_has_correction_applied(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="correction_applied") + assert len(entries) == 1, f"Expected correction_applied entry, got {len(entries)}" + + +@then("the audit log should contain a resource_modified entry") +def step_audit_has_resource_modified(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="resource_modified") + assert len(entries) == 1, f"Expected resource_modified entry, got {len(entries)}" + + +@then("the audit log should contain an auth_success entry") +def step_audit_has_auth_success(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="auth_success") + assert len(entries) == 1, f"Expected auth_success entry, got {len(entries)}" + + +@then("the audit log should contain an auth_failure entry") +def step_audit_has_auth_failure(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="auth_failure") + assert len(entries) == 1, f"Expected auth_failure entry, got {len(entries)}" + + +@then("the audit log entry should have redacted values") +def step_audit_entry_has_redacted(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="config_changed") + assert len(entries) >= 1, "Expected at least one config_changed entry" + details = entries[0].details + assert details.get("api_key") == "***REDACTED***", ( + f"Expected api_key to be redacted, got {details.get('api_key')!r}" + ) + assert details.get("normal_field") == "visible", ( + f"Expected normal_field to be 'visible', got {details.get('normal_field')!r}" + ) + + +@then("all events should appear in the audit log") +def step_all_events_in_audit_log(context: Context) -> None: + total = context.audit_service.count() + assert total == 3, f"Expected 3 audit entries, got {total}" + + +@then("the audit log should not contain the event") +def step_audit_log_no_event(context: Context) -> None: + total = context.audit_service.count() + assert total == 0, f"Expected 0 audit entries, got {total}" + + +@then('the audit entry event_type should be the string "{expected_type}"') +def step_audit_entry_type_string(context: Context, expected_type: str) -> None: + entries = context.audit_service.list_entries(event_type=expected_type) + assert len(entries) >= 1, f"No entries with event_type={expected_type!r}" + assert entries[0].event_type == expected_type + + +@then('the audit entry should have plan_id "{expected_plan_id}"') +def step_audit_entry_plan_id(context: Context, expected_plan_id: str) -> None: + entries = context.audit_service.list_entries(event_type="plan_applied") + assert len(entries) >= 1 + assert entries[0].plan_id == expected_plan_id, ( + f"Expected plan_id={expected_plan_id!r}, got {entries[0].plan_id!r}" + ) + + +@then('the audit entry should have actor_name "{expected_actor}"') +def step_audit_entry_actor_name(context: Context, expected_actor: str) -> None: + entries = context.audit_service.list_entries(event_type="config_changed") + assert len(entries) >= 1 + assert entries[0].actor_name == expected_actor, ( + f"Expected actor_name={expected_actor!r}, got {entries[0].actor_name!r}" + ) + + +@then('the audit entry details should contain session_id "{expected_sid}"') +def step_audit_entry_details_session_id(context: Context, expected_sid: str) -> None: + entries = context.audit_service.list_entries(event_type="session_created") + assert len(entries) >= 1, "No session_created entries found" + details = entries[0].details + assert details.get("session_id") == expected_sid, ( + f"Expected session_id={expected_sid!r}, got {details.get('session_id')!r}" + ) + + +@then('the audit entry details should contain correlation_id "{expected_cid}"') +def step_audit_entry_details_correlation_id( + context: Context, expected_cid: str +) -> None: + entries = context.audit_service.list_entries(event_type="plan_applied") + assert len(entries) >= 1, "No plan_applied entries found" + details = entries[0].details + assert details.get("correlation_id") == expected_cid, ( + f"Expected correlation_id={expected_cid!r}, " + f"got {details.get('correlation_id')!r}" + ) + + +@then("no audit subscriber exception should propagate") +def step_no_audit_subscriber_exception(context: Context) -> None: + assert not getattr(context, "exception_raised", True), ( + "An exception propagated from the audit subscriber" + ) diff --git a/features/steps/audit_wiring_extended_steps.py b/features/steps/audit_wiring_extended_steps.py new file mode 100644 index 000000000..71411ea0e --- /dev/null +++ b/features/steps/audit_wiring_extended_steps.py @@ -0,0 +1,179 @@ +"""Extended step definitions for AuditService EventBus wiring (Forgejo #581). + +Covers additional coverage and security scenarios: +- COV-6: Subscriber recovery after transient failure +- COV-5: Multiple non-security event type filtering +- TFLAW-3: Nested sensitive data redaction +- TFLAW-4: Auto-generated correlation_id propagation +- SEC-1: User identity extraction from details + +Split from ``audit_service_wiring_steps.py`` to stay within the +500-line file limit (CONTRIBUTING.md). +""" + +from __future__ import annotations + +from typing import cast + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.application.services.audit_event_subscriber import ( + AuditEventSubscriber, +) +from cleveragents.application.services.audit_service import AuditService +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.reactive import ReactiveEventBus +from cleveragents.infrastructure.events.types import EventType +from features.mocks.transient_fail_audit_service import TransientFailAuditService +from features.steps.audit_service_wiring_steps import _fresh_audit_service + +__all__: list[str] = [] + +# ── COV-6: Subscriber recovery after failure ───────────────────── + + +@given("an event bus with a transiently failing audit service") +def step_event_bus_with_transient_failure(context: Context) -> None: + context.audit_service = _fresh_audit_service() + context.event_bus = ReactiveEventBus() + context.fail_once_service = TransientFailAuditService(context.audit_service) + context.subscriber = AuditEventSubscriber( + audit_service=cast(AuditService, context.fail_once_service), + event_bus=context.event_bus, + ) + + +@when("a plan_applied event is emitted causing a transient failure") +def step_emit_causing_transient_failure(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_APPLIED, + plan_id="PLAN-TRANSIENT", + details={"action_name": "local/fail-once"}, + ) + ) + + +@when("a plan_cancelled event is emitted after the transient failure") +def step_emit_after_transient(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_CANCELLED, + plan_id="PLAN-RECOVER", + details={"reason": "recovery test"}, + ) + ) + + +@then("the audit log should contain the plan_cancelled entry after recovery") +def step_audit_has_cancelled_after_recovery(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="plan_cancelled") + assert len(entries) >= 1, ( + f"Expected plan_cancelled entry after recovery, got {len(entries)}" + ) + + +# ── COV-5: Multiple non-security event type filtering ──────────── + + +@when("non-security events PLAN_CREATED and ACTOR_INVOKED are emitted") +def step_emit_multiple_non_security(context: Context) -> None: + context.event_bus.emit( + DomainEvent(event_type=EventType.PLAN_CREATED, plan_id="P-ns1") + ) + context.event_bus.emit( + DomainEvent(event_type=EventType.ACTOR_INVOKED, details={"actor": "test"}) + ) + + +@then("the audit log should remain empty") +def step_audit_log_empty(context: Context) -> None: + total = context.audit_service.count() + assert total == 0, f"Expected 0 audit entries, got {total}" + + +# ── TFLAW-3: Nested sensitive data redaction ───────────────────── + + +@when("an event with nested sensitive details is emitted") +def step_emit_nested_sensitive(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.CONFIG_CHANGED, + details={ + "config": { + "api_key": "sk-proj-NESTED1234567890", + "database": {"password": "s3cret!"}, + }, + "normal_field": "visible", + }, + ) + ) + + +@then("the nested audit log values should be redacted") +def step_nested_redacted(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="config_changed") + assert len(entries) >= 1, "Expected config_changed entry" + details = entries[0].details + config = details.get("config", {}) + assert config.get("api_key") == "***REDACTED***", ( + f"Expected nested api_key redacted, got {config.get('api_key')!r}" + ) + db = config.get("database", {}) + assert db.get("password") == "***REDACTED***", ( + f"Expected nested password redacted, got {db.get('password')!r}" + ) + assert details.get("normal_field") == "visible" + + +# ── TFLAW-4: Auto-generated correlation_id propagation ─────────── + + +@when("a plan_applied event without explicit correlation_id is emitted") +def step_emit_auto_correlation(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_APPLIED, + plan_id="PLAN-AUTO-CORR", + details={"action_name": "local/auto-corr"}, + ) + ) + + +@then("the audit entry details should contain a non-empty correlation_id") +def step_audit_has_auto_correlation_id(context: Context) -> None: + entries = context.audit_service.list_entries(event_type="plan_applied") + assert len(entries) >= 1, "No plan_applied entries found" + details = entries[0].details + cid = details.get("correlation_id") + assert cid is not None and len(str(cid)) > 0, ( + f"Expected non-empty auto-generated correlation_id, got {cid!r}" + ) + + +# ── SEC-1: User identity extraction from details ───────────────── + + +@when("a session_created event with user_identity in details is emitted") +def step_emit_session_with_user_identity(context: Context) -> None: + context.event_bus.emit( + DomainEvent( + event_type=EventType.SESSION_CREATED, + session_id="SES-UI", + details={ + "mode": "interactive", + "user_identity": "test-user@example.com", + }, + ) + ) + + +@then('the audit entry should have user_identity "{expected_uid}"') +def step_audit_entry_user_identity(context: Context, expected_uid: str) -> None: + entries = context.audit_service.list_entries(event_type="session_created") + assert len(entries) >= 1, "No session_created entries found" + assert entries[0].user_identity == expected_uid, ( + f"Expected user_identity={expected_uid!r}, got {entries[0].user_identity!r}" + ) diff --git a/features/steps/skill_cli_coverage_boost_steps.py b/features/steps/skill_cli_coverage_boost_steps.py index 0d557b55c..54abb4207 100644 --- a/features/steps/skill_cli_coverage_boost_steps.py +++ b/features/steps/skill_cli_coverage_boost_steps.py @@ -105,7 +105,7 @@ def step_boost_mock_db_components(context: Context) -> None: # Patch at the source modules since the function uses lazy imports p1 = patch( - "cleveragents.application.container.get_container", + "cleveragents.cli.commands.skill.get_container", return_value=mock_container, ) p2 = patch( @@ -134,7 +134,7 @@ def step_boost_mock_db_components(context: Context) -> None: def step_boost_container_raises(context: Context) -> None: """Patch get_container to raise so the except branch is hit.""" p1 = patch( - "cleveragents.application.container.get_container", + "cleveragents.cli.commands.skill.get_container", side_effect=RuntimeError("DB unavailable"), ) p1.start() diff --git a/robot/audit_service_wiring.robot b/robot/audit_service_wiring.robot new file mode 100644 index 000000000..ff82e1858 --- /dev/null +++ b/robot/audit_service_wiring.robot @@ -0,0 +1,59 @@ +*** Settings *** +Documentation Integration test for AuditService EventBus wiring +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment With Database Isolation +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_audit_wiring.py + +*** Test Cases *** +Audit Subscriber Records Events + [Documentation] Verify AuditEventSubscriber records events from EventBus + [Tags] observability audit + ${result}= Run Process ${PYTHON} ${HELPER} record_events + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} AUDIT_OK + +Audit Subscriber Redacts Sensitive Data + [Documentation] Verify sensitive data is redacted in audit entries + [Tags] observability audit security + ${result}= Run Process ${PYTHON} ${HELPER} redact_secrets + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} REDACT_OK + +Audit Subscriber Ignores Non-Security Events + [Documentation] Verify non-security events are not recorded + [Tags] observability audit + ${result}= Run Process ${PYTHON} ${HELPER} ignore_non_security + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} FILTER_OK + +Container Wires Audit Subscriber + [Documentation] Verify DI container registers AuditEventSubscriber + [Tags] observability audit di + ${result}= Run Process ${PYTHON} ${HELPER} container_wiring + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} CONTAINER_OK + +E2E Plan Lifecycle Through Audit Pipeline + [Documentation] End-to-end: PlanLifecycleService cancel_plan -> EventBus -> AuditEventSubscriber -> AuditService DB + [Tags] observability audit e2e + ${result}= Run Process ${PYTHON} ${HELPER} e2e_plan_lifecycle + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} E2E_OK diff --git a/robot/helper_audit_wiring.py b/robot/helper_audit_wiring.py new file mode 100644 index 000000000..ed7a75ec9 --- /dev/null +++ b/robot/helper_audit_wiring.py @@ -0,0 +1,221 @@ +"""Helper script for audit_service_wiring.robot integration tests. + +Each subcommand is a self-contained check that prints a sentinel on success. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +# Ensure local source tree is importable +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +from sqlalchemy import create_engine # noqa: E402 +from sqlalchemy.orm import sessionmaker # noqa: E402 + +from cleveragents.application.services.audit_event_subscriber import ( # noqa: E402 + SECURITY_EVENT_MAP, + AuditEventSubscriber, +) +from cleveragents.application.services.audit_service import ( # noqa: E402 + AuditService, +) +from cleveragents.application.services.plan_lifecycle_service import ( # noqa: E402 + PlanLifecycleService, +) +from cleveragents.config.settings import Settings # noqa: E402 +from cleveragents.infrastructure.database.models import Base # noqa: E402 +from cleveragents.infrastructure.events.models import DomainEvent # noqa: E402 +from cleveragents.infrastructure.events.reactive import ( # noqa: E402 + ReactiveEventBus, +) +from cleveragents.infrastructure.events.types import EventType # noqa: E402 + + +def _make_audit_service() -> AuditService: + """Create an in-memory AuditService for testing.""" + Settings._instance = None + settings = Settings(database_url="sqlite:///:memory:") + engine = create_engine("sqlite:///:memory:", echo=False) + Base.metadata.create_all(engine) + session = sessionmaker(bind=engine)() + return AuditService(settings=settings, session=session) + + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + + +def record_events() -> None: + """Verify subscriber records security events through AuditService.""" + svc = _make_audit_service() + bus = ReactiveEventBus() + AuditEventSubscriber(audit_service=svc, event_bus=bus) + + # Emit all 9 security-relevant event types + for event_type in SECURITY_EVENT_MAP: + bus.emit(DomainEvent(event_type=event_type, plan_id="TEST-001")) + + count = svc.count() + if count != len(SECURITY_EVENT_MAP): + print( + f"FAIL: expected {len(SECURITY_EVENT_MAP)} entries, got {count}", + file=sys.stderr, + ) + sys.exit(1) + print("AUDIT_OK") + + +def redact_secrets() -> None: + """Verify sensitive data in event details is redacted.""" + svc = _make_audit_service() + bus = ReactiveEventBus() + AuditEventSubscriber(audit_service=svc, event_bus=bus) + + bus.emit( + DomainEvent( + event_type=EventType.CONFIG_CHANGED, + details={ + "api_key": "sk-proj-ABCDEFGHIJ1234567890", + "normal": "visible", + }, + ) + ) + + entries = svc.list_entries(event_type="config_changed") + if not entries: + print("FAIL: no config_changed entry found", file=sys.stderr) + sys.exit(1) + + details = entries[0].details + if details.get("api_key") != "***REDACTED***": + print( + f"FAIL: api_key not redacted: {details.get('api_key')!r}", + file=sys.stderr, + ) + sys.exit(1) + if details.get("normal") != "visible": + print( + f"FAIL: normal field changed: {details.get('normal')!r}", + file=sys.stderr, + ) + sys.exit(1) + print("REDACT_OK") + + +def ignore_non_security() -> None: + """Verify non-security events are not recorded.""" + svc = _make_audit_service() + bus = ReactiveEventBus() + AuditEventSubscriber(audit_service=svc, event_bus=bus) + + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED, plan_id="P-nonsec")) + + count = svc.count() + if count != 0: + print(f"FAIL: expected 0 entries, got {count}", file=sys.stderr) + sys.exit(1) + print("FILTER_OK") + + +def container_wiring() -> None: + """Verify DI container registers a functional AuditEventSubscriber. + + Goes beyond type checking: emits an event through the container's + EventBus and verifies it reaches the AuditService via the subscriber. + """ + from cleveragents.application.container import Container + + container = Container() + subscriber = container.audit_event_subscriber() + if not isinstance(subscriber, AuditEventSubscriber): + print( + f"FAIL: expected AuditEventSubscriber, got {type(subscriber).__name__}", + file=sys.stderr, + ) + sys.exit(1) + + # Functional verification: emit through container's EventBus + bus = container.event_bus() + bus.emit(DomainEvent(event_type=EventType.PLAN_APPLIED, plan_id="WIRE-001")) + audit_svc = container.audit_service() + count = audit_svc.count() + if count < 1: + print( + f"FAIL: expected >=1 audit entries after emit, got {count}", + file=sys.stderr, + ) + sys.exit(1) + print("CONTAINER_OK") + + +def e2e_plan_lifecycle() -> None: + """E2E: PlanLifecycleService -> EventBus -> AuditEventSubscriber -> DB. + + Exercises the full domain-to-audit pipeline: + 1. Create PlanLifecycleService with an EventBus (in-memory mode). + 2. Wire AuditEventSubscriber on the same bus. + 3. Create an action, use it to create a plan. + 4. Cancel the plan (triggers PLAN_CANCELLED event). + 5. Verify a plan_cancelled audit entry exists in the DB. + """ + svc = _make_audit_service() + bus = ReactiveEventBus() + AuditEventSubscriber(audit_service=svc, event_bus=bus) + + settings = Settings._instance or Settings(database_url="sqlite:///:memory:") + plc = PlanLifecycleService(settings=settings, event_bus=bus) + + # Create and register an action + plc.create_action( + name="local/e2e-audit-test", + description="E2E test action", + definition_of_done="verify audit wiring", + strategy_actor="test-strategy", + execution_actor="test-execution", + ) + + # Use the action to create a plan + plan = plc.use_action("local/e2e-audit-test") + + # Cancel the plan -- this emits PLAN_CANCELLED via the EventBus + plan_id = plan.identity.plan_id + plc.cancel_plan(plan_id, reason="e2e audit verification") + + entries = svc.list_entries(event_type="plan_cancelled") + if not entries: + print( + "FAIL: no plan_cancelled audit entry after cancel_plan()", file=sys.stderr + ) + sys.exit(1) + if entries[0].plan_id != plan_id: + print( + f"FAIL: plan_id mismatch: {entries[0].plan_id!r} != {plan_id!r}", + file=sys.stderr, + ) + sys.exit(1) + print("E2E_OK") + + +# --------------------------------------------------------------------------- +# Dispatch +# --------------------------------------------------------------------------- + +_COMMANDS = { + "record_events": record_events, + "redact_secrets": redact_secrets, + "ignore_non_security": ignore_non_security, + "container_wiring": container_wiring, + "e2e_plan_lifecycle": e2e_plan_lifecycle, +} + +if __name__ == "__main__": + if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: + print(f"Usage: {sys.argv[0]} ", file=sys.stderr) + print(f"Commands: {list(_COMMANDS)}", file=sys.stderr) + sys.exit(1) + _COMMANDS[sys.argv[1]]() diff --git a/src/cleveragents/application/container.py b/src/cleveragents/application/container.py index 53d96a102..edf1ea736 100644 --- a/src/cleveragents/application/container.py +++ b/src/cleveragents/application/container.py @@ -6,11 +6,16 @@ Uses dependency-injector for managing service instances. from pathlib import Path +import structlog from dependency_injector import containers, providers from cleveragents.actor.registry import ActorRegistry from cleveragents.application.reactive_registry_adapter import register_registry_agents from cleveragents.application.services.actor_service import ActorService +from cleveragents.application.services.audit_event_subscriber import ( + AuditEventSubscriber, +) +from cleveragents.application.services.audit_service import AuditService from cleveragents.application.services.autonomy_controller import ( AutonomyController, ) @@ -22,6 +27,7 @@ from cleveragents.application.services.context_service import ContextService from cleveragents.application.services.context_tiers import ( ContextTierService, ) +from cleveragents.application.services.correction_service import CorrectionService from cleveragents.application.services.cost_budget_service import CostBudgetService from cleveragents.application.services.decision_service import DecisionService from cleveragents.application.services.decomposition_service import ( @@ -88,6 +94,9 @@ from cleveragents.langgraph.bridge import RxPyLangGraphBridge from cleveragents.providers.registry import ProviderRegistry, get_provider_registry from cleveragents.reactive.route_bridge import RouteBridge from cleveragents.reactive.stream_router import ReactiveStreamRouter +from cleveragents.shared.redaction import redact_value + +_logger = structlog.get_logger(__name__) def get_ai_provider( @@ -288,6 +297,7 @@ def _build_trace_service( def _build_session_service( database_url: str, + event_bus: ReactiveEventBus | None = None, ) -> PersistentSessionService: """Build a PersistentSessionService with auto-committing repositories. @@ -326,7 +336,7 @@ def _build_session_service( factory = sessionmaker(bind=engine, expire_on_commit=False) session_repo = SessionRepository(session_factory=factory, auto_commit=True) message_repo = SessionMessageRepository(session_factory=factory, auto_commit=True) - return PersistentSessionService(session_repo, message_repo) + return PersistentSessionService(session_repo, message_repo, event_bus=event_bus) class Container(containers.DeclarativeContainer): @@ -372,6 +382,7 @@ class Container(containers.DeclarativeContainer): ProjectService, settings=settings, unit_of_work=unit_of_work, + event_bus=event_bus, ) vector_store_service = providers.Factory( @@ -391,6 +402,7 @@ class Container(containers.DeclarativeContainer): ActorService, settings=settings, unit_of_work=unit_of_work, + event_bus=event_bus, ) actor_registry = providers.Factory( @@ -498,6 +510,7 @@ class Container(containers.DeclarativeContainer): session_service = providers.Factory( _build_session_service, database_url=database_url, + event_bus=event_bus, ) # Metrics Emitter - structured metric emission (Forgejo #579) @@ -558,6 +571,31 @@ class Container(containers.DeclarativeContainer): event_bus=event_bus, ) + # Audit Service - Singleton (one shared audit store per process). + # N6: Pass database_url from the container so audit entries are + # written to the same database as all other services, not to the + # Settings default ("sqlite:///cleveragents.db"). + audit_service = providers.Singleton( + AuditService, + settings=settings, + database_url=database_url, + ) + + # Audit Event Subscriber - Singleton (wires EventBus -> AuditService) + audit_event_subscriber = providers.Singleton( + AuditEventSubscriber, + audit_service=audit_service, + event_bus=event_bus, + ) + + # Correction Service - registered in the DI container so that + # event_bus is always wired and CORRECTION_APPLIED audit events + # are emitted in production (N9). + correction_service = providers.Singleton( + CorrectionService, + event_bus=event_bus, + ) + # Reactive routing stream_router = providers.Singleton(ReactiveStreamRouter) langgraph_bridge = providers.Singleton( @@ -584,11 +622,28 @@ def get_container() -> Container: """Get the global container instance. Creates container on first call, returns same instance thereafter. + Eagerly instantiates the :class:`AuditEventSubscriber` singleton so + that EventBus subscriptions are registered at startup and + security-relevant events are automatically routed to the audit log. """ global _container if _container is None: _container = Container() # Don't initialize database here - let services handle it + # Eagerly wire the audit subscriber so EventBus subscriptions + # are active from startup (lazy singletons are only created on + # first access; without this call the subscriber would never + # register its handlers). + try: + _container.audit_event_subscriber() + except Exception as exc: + _logger.warning( + "audit_subscriber_deferred", + reason="Database not yet initialised; audit subscriptions " + "will be registered on first AuditEventSubscriber access.", + error_type=type(exc).__name__, + error_message=redact_value(str(exc)), + ) return _container diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index d716f248c..23d442ef8 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -19,6 +19,9 @@ from cleveragents.application.services.acms_phase3 import ( ProvenancePreambleGenerator, RelevanceCoherenceOrderer, ) +from cleveragents.application.services.audit_event_subscriber import ( + AuditEventSubscriber, +) from cleveragents.application.services.autonomy_controller import ( AutonomyController, ) @@ -211,6 +214,7 @@ __all__ = [ "ApplyValidationSummary", "ArceStrategy", "AttachmentScope", + "AuditEventSubscriber", "AutonomyController", "AutonomyGuardrailService", "BreadthDepthNavigatorStrategy", diff --git a/src/cleveragents/application/services/actor_service.py b/src/cleveragents/application/services/actor_service.py index dd68d8c05..1b6c4335b 100644 --- a/src/cleveragents/application/services/actor_service.py +++ b/src/cleveragents/application/services/actor_service.py @@ -4,7 +4,9 @@ from __future__ import annotations import os from datetime import datetime -from typing import Any +from typing import TYPE_CHECKING, Any + +import structlog from cleveragents.config.settings import Settings from cleveragents.core.exceptions import ( @@ -14,14 +16,27 @@ from cleveragents.core.exceptions import ( ) from cleveragents.domain.models.core import Actor from cleveragents.infrastructure.database.unit_of_work import UnitOfWork +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.types import EventType + +if TYPE_CHECKING: + from cleveragents.infrastructure.events.protocol import EventBus + +_logger = structlog.get_logger(__name__) class ActorService: """Coordinate actor persistence and default selection rules.""" - def __init__(self, settings: Settings, unit_of_work: UnitOfWork): + def __init__( + self, + settings: Settings, + unit_of_work: UnitOfWork, + event_bus: EventBus | None = None, + ): self.settings = settings self.unit_of_work = unit_of_work + self._event_bus = event_bus def _normalize_name(self, name: str, *, allow_built_in: bool = False) -> str: normalized = name.strip() @@ -132,6 +147,19 @@ class ActorService: ctx.actors.delete(normalized) except ValueError as exc: # Built-in or default guard raise BusinessRuleViolation(str(exc)) from exc + if self._event_bus is not None: + try: + self._event_bus.emit( + DomainEvent( + event_type=EventType.ENTITY_DELETED, + details={ + "entity_type": "actor", + "entity_name": normalized, + }, + ) + ) + except Exception: + _logger.warning("audit_emit_failed", event_type="ENTITY_DELETED") def set_default_actor(self, name: str) -> Actor: """Mark a built-in or custom actor as the default entry.""" diff --git a/src/cleveragents/application/services/audit_event_subscriber.py b/src/cleveragents/application/services/audit_event_subscriber.py new file mode 100644 index 000000000..39338688a --- /dev/null +++ b/src/cleveragents/application/services/audit_event_subscriber.py @@ -0,0 +1,151 @@ +"""Audit event subscriber (SEC7 EventBus wiring). + +Bridges the :class:`EventBus` and :class:`AuditService` by subscribing to +all security-relevant domain event types and persisting them via +``AuditService.record()``. Follows the Single Responsibility Principle: +the subscriber only translates events into audit entries; it does not own +the audit storage. + +Sensitive data in event ``details`` is redacted via +:func:`cleveragents.shared.redaction.redact_dict` before persistence. + +Based on: + - docs/specification.md §Audit Logging (SEC7) + - Forgejo issue #581 +""" + +from __future__ import annotations + +import structlog + +from cleveragents.application.services.audit_service import AuditService +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, redact_value + +_logger = structlog.get_logger(__name__) + +# Mapping from :class:`EventType` enum members to the underscore-delimited +# ``event_type`` strings accepted by ``AuditService.record()``. +# +# Design note (SPEC-1 / #581): Only security-relevant event types are +# persisted to the audit log. Non-security events (e.g. PLAN_CREATED, +# ACTOR_INVOKED) flow through the EventBus but are intentionally *not* +# recorded here, keeping the audit table focused on compliance-critical +# operations per docs/specification.md §Audit Logging. +SECURITY_EVENT_MAP: dict[EventType, str] = { + EventType.PLAN_APPLIED: "plan_applied", + EventType.PLAN_CANCELLED: "plan_cancelled", + # NOTE: RESOURCE_MODIFIED has no producing service yet — the tool + # execution framework does not emit this event. The subscriber handler + # is intentionally registered so that audit entries are automatically + # created once tool-write event emission is implemented. + EventType.RESOURCE_MODIFIED: "resource_modified", + EventType.CORRECTION_APPLIED: "correction_applied", + EventType.CONFIG_CHANGED: "config_changed", + EventType.ENTITY_DELETED: "entity_deleted", + EventType.SESSION_CREATED: "session_created", + # NOTE: AUTH_SUCCESS and AUTH_FAILURE have no producing service yet — + # server-mode authentication is not implemented. The subscriber handlers + # are registered so that audit entries are automatically created once + # server auth emits these events. + EventType.AUTH_SUCCESS: "auth_success", + EventType.AUTH_FAILURE: "auth_failure", +} + +#: The set of :class:`EventType` values that are considered security-relevant. +SECURITY_EVENT_TYPES: frozenset[EventType] = frozenset(SECURITY_EVENT_MAP) + + +class AuditEventSubscriber: + """Subscribes to security-relevant domain events and persists them via AuditService. + + On construction the subscriber registers a handler for each event type + in :data:`SECURITY_EVENT_MAP` with the provided *event_bus*. When an + event is received the handler: + + 1. Extracts ``plan_id``, ``actor_name``, ``project_name``, and + ``details`` from the :class:`DomainEvent`. + 2. Applies :func:`~cleveragents.shared.redaction.redact_dict` to mask + secrets in the details dict. + 3. Calls ``audit_service.record()`` with the mapped event type string. + + Args: + audit_service: The :class:`AuditService` used for persistence. + event_bus: The :class:`EventBus` to subscribe to. + """ + + def __init__(self, audit_service: AuditService, event_bus: EventBus) -> None: + self._audit_service = audit_service + self._event_bus = event_bus + for event_type in SECURITY_EVENT_TYPES: + event_bus.subscribe(event_type, self._handle_event) + + def _handle_event(self, event: DomainEvent) -> None: + """Redact sensitive data and persist an audit log entry. + + Enriches the recorded details with ``session_id`` and + ``correlation_id`` from the :class:`DomainEvent` when present, + ensuring full traceability in the audit log. + + Failures are logged but never propagated — audit recording must + not break the event pipeline. + """ + audit_event_type = SECURITY_EVENT_MAP.get(event.event_type) + if audit_event_type is None: + return + + try: + raw_details = dict(event.details) + + # SEC-1: Extract user_identity from event details when provided + # by domain services. The DomainEvent model does not carry a + # dedicated ``user_identity`` field, so services embed it in the + # ``details`` dict and the subscriber hoists it to the top-level + # audit column for indexed queries. + user_identity: str | None = raw_details.pop("user_identity", None) + + # N1: Audit persistence must never respect the display-layer + # ``show_secrets`` flag — force redaction regardless of the + # global toggle so that secrets are always masked in the + # persistent audit log. + redacted_details = redact_dict(raw_details, show_secrets=False) + + # N11: Inject session_id and correlation_id *after* redaction + # to avoid false-positive pattern matching on ULIDs. + if event.session_id is not None: + redacted_details["session_id"] = event.session_id + # correlation_id always has a value (ULID default_factory), + # so enrich unconditionally. + redacted_details["correlation_id"] = event.correlation_id + + # Apply redaction to user_identity to prevent accidental secret + # leakage (e.g. tokens embedded in identity strings). + redacted_identity: str | None = ( + redact_value(user_identity) if user_identity is not None else None + ) + + self._audit_service.record( + event_type=audit_event_type, + plan_id=event.plan_id, + project_name=event.project_name, + actor_name=event.actor_name, + user_identity=redacted_identity, + details=redacted_details, + ) + except Exception as exc: + _logger.warning( + "audit_event_recording_failed", + event_type=audit_event_type, + plan_id=event.plan_id, + error_type=type(exc).__name__, + error_message=redact_value(str(exc)), + ) + + +__all__ = [ + "SECURITY_EVENT_MAP", + "SECURITY_EVENT_TYPES", + "AuditEventSubscriber", +] diff --git a/src/cleveragents/application/services/audit_service.py b/src/cleveragents/application/services/audit_service.py index 91e2d31f1..f637fec79 100644 --- a/src/cleveragents/application/services/audit_service.py +++ b/src/cleveragents/application/services/audit_service.py @@ -35,6 +35,8 @@ VALID_EVENT_TYPES: frozenset[str] = frozenset( "config_changed", "entity_deleted", "session_created", + "auth_success", + "auth_failure", } ) @@ -95,18 +97,44 @@ class AuditService: self, settings: Settings, session: Session | None = None, + database_url: str | None = None, ) -> None: if not isinstance(settings, Settings): raise TypeError("settings must be a Settings instance") self._settings = settings + self._database_url = database_url self._owns_session = session is None - if session is not None: - self._session = session - else: - engine = create_engine(settings.database_url, echo=False) + # When a session is injected (e.g. in tests), use it directly. + # Otherwise defer engine + table creation to the first call that + # actually needs the database (_ensure_session). This avoids + # running ``Base.metadata.create_all()`` at container-init time + # which (a) is expensive (~50-150 ms for 38 tables on each fresh + # SQLite file) and (b) pre-creates the per-scenario DB file, + # bypassing the template-DB fast-path in the test harness. + self._session: Session | None = session + + # ── Lazy session bootstrap ─────────────────────────────────── + + def _ensure_session(self) -> Session: + """Return the active SQLAlchemy session, creating it on first call. + + When no session was injected via the constructor the engine and + tables are created here rather than in ``__init__``. This keeps + the constructor cheap (important when the DI container eagerly + instantiates the service) and avoids pre-creating the database + file before the test-harness template-copy mechanism has run. + + Uses ``database_url`` (if provided at construction) to ensure + audit entries are written to the same database as all other + services. Falls back to ``settings.database_url``. + """ + if self._session is None: + url = self._database_url or self._settings.database_url + engine = create_engine(url, echo=False) Base.metadata.create_all(engine) factory = sessionmaker(bind=engine) self._session = factory() + return self._session # ── Context manager ────────────────────────────────────────── @@ -122,7 +150,7 @@ class AuditService: Only closes sessions that were created by this service (i.e. not externally-provided test sessions). Safe to call multiple times. """ - if self._owns_session: + if self._owns_session and self._session is not None: self._session.close() # ── Record ─────────────────────────────────────────────────── @@ -163,9 +191,10 @@ class AuditService: details=details_json, created_at=now, ) - self._session.add(row) - self._session.commit() - self._session.refresh(row) + session = self._ensure_session() + session.add(row) + session.commit() + session.refresh(row) return self._row_to_entry(row) @@ -195,7 +224,8 @@ class AuditService: Returns: List of matching entries, most recent first. """ - query = self._session.query(AuditLogModel) + session = self._ensure_session() + query = session.query(AuditLogModel) if plan_id is not None: query = query.filter(AuditLogModel.plan_id == plan_id) @@ -219,18 +249,15 @@ class AuditService: Returns: The entry, or ``None`` if not found. """ - row = ( - self._session.query(AuditLogModel) - .filter(AuditLogModel.id == audit_id) - .first() - ) + session = self._ensure_session() + row = session.query(AuditLogModel).filter(AuditLogModel.id == audit_id).first() if row is None: return None return self._row_to_entry(row) def count(self) -> int: """Return the total number of audit log entries.""" - return self._session.query(AuditLogModel).count() + return self._ensure_session().query(AuditLogModel).count() # ── Retention / prune ──────────────────────────────────────── @@ -253,12 +280,13 @@ class AuditService: if days <= 0: return 0 cutoff = (datetime.now(tz=UTC) - timedelta(days=days)).strftime(_TIMESTAMP_FMT) + session = self._ensure_session() result = ( - self._session.query(AuditLogModel) + session.query(AuditLogModel) .filter(AuditLogModel.created_at < cutoff) .delete(synchronize_session="fetch") ) - self._session.commit() + session.commit() return result # ── Helpers ─────────────────────────────────────────────────── diff --git a/src/cleveragents/application/services/config_service.py b/src/cleveragents/application/services/config_service.py index c13fae1d2..050642f5e 100644 --- a/src/cleveragents/application/services/config_service.py +++ b/src/cleveragents/application/services/config_service.py @@ -16,10 +16,19 @@ import tomllib from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any +import structlog import tomlkit +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.types import EventType + +if TYPE_CHECKING: + from cleveragents.infrastructure.events.protocol import EventBus + +_logger = structlog.get_logger(__name__) + # --------------------------------------------------------------------------- # Public types # --------------------------------------------------------------------------- @@ -1102,9 +1111,11 @@ class ConfigService: *, config_dir: Path | None = None, config_path: Path | None = None, + event_bus: EventBus | None = None, ) -> None: self._config_dir: Path = config_dir or _DEFAULT_CONFIG_DIR self._config_path: Path = config_path or (self._config_dir / "config.toml") + self._event_bus = event_bus # -- public registry API -------------------------------------------------- @@ -1151,8 +1162,23 @@ class ConfigService: def set_value(self, key: str, value: Any) -> None: """Persist a single key-value pair into the global TOML config.""" data = self.read_config() + old_value = data.get(key) data[key] = value self.write_config(data) + if self._event_bus is not None: + try: + self._event_bus.emit( + DomainEvent( + event_type=EventType.CONFIG_CHANGED, + details={ + "key": key, + "old_value": old_value, + "new_value": value, + }, + ) + ) + except Exception: + _logger.warning("audit_emit_failed", event_type="CONFIG_CHANGED") # -- validation ----------------------------------------------------------- @@ -1382,6 +1408,13 @@ class ConfigService: The value is stored at ``[project.""].`` in the global config file. + Note: This method does **not** emit a ``CONFIG_CHANGED`` event. + Per the spec (§Audit Logging), ``config_changed`` is triggered by + ``agents config set`` which operates on global configuration only. + Project-scoped config changes are governed by ``agents project + context set`` and are not classified as security-relevant audit + events in the current spec revision. + Parameters ---------- project_name: diff --git a/src/cleveragents/application/services/correction_service.py b/src/cleveragents/application/services/correction_service.py index 8d0cbb353..937b1f035 100644 --- a/src/cleveragents/application/services/correction_service.py +++ b/src/cleveragents/application/services/correction_service.py @@ -11,6 +11,7 @@ from __future__ import annotations from collections import deque from datetime import UTC, datetime +from typing import TYPE_CHECKING import structlog from ulid import ULID @@ -26,6 +27,11 @@ from cleveragents.domain.models.core.correction import ( CorrectionResult, CorrectionStatus, ) +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.types import EventType + +if TYPE_CHECKING: + from cleveragents.infrastructure.events.protocol import EventBus logger = structlog.get_logger(__name__) @@ -51,12 +57,49 @@ class CorrectionService: def __init__( self, checkpoint_service: CheckpointService | None = None, + event_bus: EventBus | None = None, ) -> None: self._corrections: dict[str, CorrectionRequest] = {} self._impacts: dict[str, CorrectionImpact] = {} self._attempts: dict[str, list[CorrectionAttempt]] = {} self._results: dict[str, CorrectionResult] = {} self._checkpoint_service = checkpoint_service + self._event_bus = event_bus + + # ------------------------------------------------------------------ + # Event emission helper + # ------------------------------------------------------------------ + + def _emit_correction_applied( + self, + correction_id: str, + request: CorrectionRequest, + result: CorrectionResult, + ) -> None: + """Emit a ``CORRECTION_APPLIED`` event when the result is successful.""" + if self._event_bus is not None and result.status == CorrectionStatus.APPLIED: + try: + self._event_bus.emit( + DomainEvent( + event_type=EventType.CORRECTION_APPLIED, + plan_id=request.plan_id, + details={ + "correction_id": correction_id, + "target_decision_id": request.target_decision_id, + "mode": request.mode.value + if hasattr(request.mode, "value") + else str(request.mode), + "guidance": request.guidance, + }, + ) + ) + except Exception: + logger.warning( + "event_bus_emit_failed", + event_type="CORRECTION_APPLIED", + correction_id=correction_id, + exc_info=True, + ) # ------------------------------------------------------------------ # Creation @@ -296,6 +339,7 @@ class CorrectionService: correction_id=correction_id, status=result.status, ) + self._emit_correction_applied(correction_id, request, result) return result # ------------------------------------------------------------------ @@ -362,6 +406,7 @@ class CorrectionService: correction_id=correction_id, status=result.status, ) + self._emit_correction_applied(correction_id, request, result) return result # ------------------------------------------------------------------ diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index 619b81847..0e9f78a8f 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/src/cleveragents/application/services/plan_lifecycle_service.py @@ -728,16 +728,24 @@ class PlanLifecycleService: phase=plan.phase.value, ) if self.event_bus is not None: - self.event_bus.emit( - DomainEvent( - event_type=EventType.PLAN_CREATED, - plan_id=plan_id, - details={ - "action_name": action_full_name, - "phase": plan.phase.value, - }, + try: + self.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_CREATED, + plan_id=plan_id, + details={ + "action_name": action_full_name, + "phase": plan.phase.value, + }, + ) + ) + except Exception: + self._logger.warning( + "event_bus_emit_failed", + event_type="PLAN_CREATED", + plan_id=plan_id, + exc_info=True, ) - ) return plan @@ -976,18 +984,26 @@ class PlanLifecycleService: phase=plan.phase.value, ) if self.event_bus is not None: - self.event_bus.emit( - DomainEvent( - event_type=EventType.PLAN_PHASE_CHANGED, - plan_id=plan_id, - details={ - "phase": plan.phase.value, - "processing_state": plan.processing_state.value - if plan.processing_state - else None, - }, + try: + self.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_PHASE_CHANGED, + plan_id=plan_id, + details={ + "phase": plan.phase.value, + "processing_state": plan.processing_state.value + if plan.processing_state + else None, + }, + ) + ) + except Exception: + self._logger.warning( + "event_bus_emit_failed", + event_type="PLAN_PHASE_CHANGED", + plan_id=plan_id, + exc_info=True, ) - ) # Enqueue async job when async execution is enabled self._maybe_enqueue_async_job(plan_id, "execute") @@ -1162,6 +1178,35 @@ class PlanLifecycleService: plan_id=plan_id, phase=plan.phase.value, ) + if self.event_bus is not None: + try: + project_names = [ + link.project_name for link in (plan.project_links or []) + ] + # BUG-2: The top-level ``project_name`` column in the audit_log + # table stores only the first project for multi-project plans. + # All project names are captured in ``details["project_names"]`` + # for completeness. This is a known schema limitation. + self.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_APPLIED, + plan_id=plan_id, + actor_name=plan.created_by, + project_name=project_names[0] if project_names else None, + details={ + "action_name": plan.action_name, + "phase": plan.phase.value, + "project_names": project_names, + }, + ) + ) + except Exception: + self._logger.warning( + "event_bus_emit_failed", + event_type="PLAN_APPLIED", + plan_id=plan_id, + exc_info=True, + ) # F1-r6 fix: wire container cleanup to plan completion hook. self._cleanup_devcontainers(plan_id) @@ -1243,6 +1288,34 @@ class PlanLifecycleService: self._commit_plan(plan) self._logger.info("Plan cancelled", plan_id=plan_id, reason=reason) + if self.event_bus is not None: + try: + project_names = [ + link.project_name for link in (plan.project_links or []) + ] + self.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_CANCELLED, + plan_id=plan_id, + actor_name=plan.created_by, + project_name=project_names[0] if project_names else None, + details={ + "reason": reason or "", + "project_names": project_names, + }, + ) + ) + except Exception: + self._logger.warning( + "event_bus_emit_failed", + event_type="PLAN_CANCELLED", + plan_id=plan_id, + exc_info=True, + ) + # COV-3: Spec says plan_cancelled should also include + # "resources released", but that data is not available at + # this point -- downstream cleanup happens in separate + # services. Tracked for future enhancement. # F1-r6 fix: wire container cleanup to plan cancellation hook. self._cleanup_devcontainers(plan_id) diff --git a/src/cleveragents/application/services/project_service.py b/src/cleveragents/application/services/project_service.py index bedbb820b..3480d69fe 100644 --- a/src/cleveragents/application/services/project_service.py +++ b/src/cleveragents/application/services/project_service.py @@ -4,10 +4,14 @@ This service handles project initialization, configuration, and management. Uses repository pattern and Unit of Work for persistence (ADR-007). """ +from __future__ import annotations + import os from datetime import datetime from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any + +import structlog from cleveragents.application.services.context_service import DEFAULT_IGNORE_PATTERNS from cleveragents.config.settings import Settings @@ -18,6 +22,13 @@ from cleveragents.core.exceptions import ( ) from cleveragents.domain.models.core import Plan, PlanStatus, Project, ProjectSettings from cleveragents.infrastructure.database.unit_of_work import UnitOfWork +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.types import EventType + +if TYPE_CHECKING: + from cleveragents.infrastructure.events.protocol import EventBus + +_logger = structlog.get_logger(__name__) class ProjectService: @@ -27,15 +38,22 @@ class ProjectService: managing projects throughout their lifecycle. """ - def __init__(self, settings: Settings, unit_of_work: UnitOfWork): + def __init__( + self, + settings: Settings, + unit_of_work: UnitOfWork, + event_bus: EventBus | None = None, + ): """Initialize the project service. Args: settings: Application settings unit_of_work: Unit of Work for database transactions + event_bus: Optional EventBus for domain event emission. """ self.settings = settings self.unit_of_work = unit_of_work + self._event_bus = event_bus # Optional search root to limit filesystem discovery (used in tests) self.search_root: Path | None = None @@ -461,3 +479,16 @@ class ProjectService: with self.unit_of_work.transaction() as ctx: if project.id: ctx.projects.delete(project.id) + if project.id and self._event_bus is not None: + try: + self._event_bus.emit( + DomainEvent( + event_type=EventType.ENTITY_DELETED, + details={ + "entity_type": "project", + "entity_name": project.name, + }, + ) + ) + except Exception: + _logger.warning("audit_emit_failed", event_type="ENTITY_DELETED") diff --git a/src/cleveragents/application/services/session_service.py b/src/cleveragents/application/services/session_service.py index 7bdfe167a..1bf4e7c54 100644 --- a/src/cleveragents/application/services/session_service.py +++ b/src/cleveragents/application/services/session_service.py @@ -11,8 +11,9 @@ from __future__ import annotations import hashlib import json from datetime import datetime -from typing import Any +from typing import TYPE_CHECKING, Any +import structlog from ulid import ULID from cleveragents.application.services.prompt_sanitizer import PromptSanitizer @@ -30,6 +31,13 @@ from cleveragents.infrastructure.database.repositories import ( SessionMessageRepository, SessionRepository, ) +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.types import EventType + +if TYPE_CHECKING: + from cleveragents.infrastructure.events.protocol import EventBus + +_logger = structlog.get_logger(__name__) class PersistentSessionService(SessionService): @@ -45,16 +53,19 @@ class PersistentSessionService(SessionService): self, session_repo: SessionRepository, message_repo: SessionMessageRepository, + event_bus: EventBus | None = None, ) -> None: """Initialise with repository instances. Args: session_repo: Repository for session CRUD. message_repo: Repository for message operations. + event_bus: Optional EventBus for domain event emission. """ self._session_repo = session_repo self._message_repo = message_repo self._sanitizer = PromptSanitizer() + self._event_bus = event_bus def create(self, actor_name: str | None = None) -> Session: """Create a new session. @@ -70,6 +81,21 @@ class PersistentSessionService(SessionService): actor_name=actor_name, ) self._session_repo.create(session) + if self._event_bus is not None: + try: + self._event_bus.emit( + DomainEvent( + event_type=EventType.SESSION_CREATED, + session_id=session.session_id, + actor_name=actor_name, + details={ + "session_id": session.session_id, + "actor_name": actor_name, + }, + ) + ) + except Exception: + _logger.warning("audit_emit_failed", event_type="SESSION_CREATED") return session def get(self, session_id: str) -> Session: @@ -109,6 +135,20 @@ class PersistentSessionService(SessionService): deleted = self._session_repo.delete(session_id) if not deleted: raise SessionNotFoundError(f"Session '{session_id}' not found") + if self._event_bus is not None: + try: + self._event_bus.emit( + DomainEvent( + event_type=EventType.ENTITY_DELETED, + session_id=session_id, + details={ + "entity_type": "session", + "entity_name": session_id, + }, + ) + ) + except Exception: + _logger.warning("audit_emit_failed", event_type="ENTITY_DELETED") def append_message( self, diff --git a/src/cleveragents/cli/commands/config.py b/src/cleveragents/cli/commands/config.py index 81a89e99b..d89fe9046 100644 --- a/src/cleveragents/cli/commands/config.py +++ b/src/cleveragents/cli/commands/config.py @@ -33,6 +33,7 @@ from rich.console import Console from rich.panel import Panel from rich.table import Table +from cleveragents.application.container import get_container from cleveragents.application.services.config_service import ( _REGISTRY, ConfigService, @@ -63,7 +64,12 @@ _FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich) def _get_service() -> ConfigService: """Return a ``ConfigService`` wired to the standard config paths.""" - return ConfigService(config_dir=_CONFIG_DIR, config_path=_CONFIG_PATH) + container = get_container() + return ConfigService( + config_dir=_CONFIG_DIR, + config_path=_CONFIG_PATH, + event_bus=container.event_bus(), + ) def _is_secret_key(key: str) -> bool: diff --git a/src/cleveragents/cli/commands/plan.py b/src/cleveragents/cli/commands/plan.py index e5482513f..b839deb31 100644 --- a/src/cleveragents/cli/commands/plan.py +++ b/src/cleveragents/cli/commands/plan.py @@ -2418,7 +2418,7 @@ def correct_decision( # Fetch influence DAG edges influence_edges = decision_svc.get_influence_edges(resolved_plan_id) - svc = CorrectionService() + svc = CorrectionService(event_bus=container.event_bus()) # Create the correction request request = svc.request_correction( diff --git a/src/cleveragents/cli/commands/server.py b/src/cleveragents/cli/commands/server.py index 317fb7fa3..7d66dcfce 100644 --- a/src/cleveragents/cli/commands/server.py +++ b/src/cleveragents/cli/commands/server.py @@ -16,6 +16,7 @@ from rich.console import Console from rich.panel import Panel from cleveragents.a2a.server_config import ServerConnectionConfig +from cleveragents.application.container import get_container from cleveragents.application.services.config_service import ConfigService from cleveragents.cli.formatting import OutputFormat, format_output @@ -40,7 +41,12 @@ def _get_config_service() -> ConfigService: """ config_dir = Path.home() / ".cleveragents" config_path = config_dir / "config.toml" - return ConfigService(config_dir=config_dir, config_path=config_path) + container = get_container() + return ConfigService( + config_dir=config_dir, + config_path=config_path, + event_bus=container.event_bus(), + ) def resolve_server_mode() -> str: @@ -109,13 +115,14 @@ def server_connect( console.print(f"[red]Invalid server configuration:[/red] {exc}") raise typer.Exit(code=1) from exc - # Persist to config file + # Persist to config file via set_value() so that each change emits + # a CONFIG_CHANGED audit event. Using write_config() would bypass + # the EventBus and leave no audit trail for these security-relevant + # settings (server URL, namespace, TLS verification). svc = _get_config_service() - config_data = svc.read_config() - config_data["server.url"] = config.server_url - config_data["server.namespace"] = config.namespace - config_data["server.tls-verify"] = config.tls_verify - svc.write_config(config_data) + svc.set_value("server.url", config.server_url) + svc.set_value("server.namespace", config.namespace) + svc.set_value("server.tls-verify", config.tls_verify) result: dict[str, Any] = { "server_url": config.server_url, diff --git a/src/cleveragents/cli/commands/skill.py b/src/cleveragents/cli/commands/skill.py index 284c8e9e4..c110599d8 100644 --- a/src/cleveragents/cli/commands/skill.py +++ b/src/cleveragents/cli/commands/skill.py @@ -46,6 +46,7 @@ from rich.console import Console from rich.panel import Panel from rich.table import Table +from cleveragents.application.container import get_container from cleveragents.application.services.skill_service import SkillService from cleveragents.cli.formatting import OutputFormat, format_output from cleveragents.domain.models.core.skill import ( @@ -83,7 +84,6 @@ def _get_skill_service() -> SkillService: from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker - from cleveragents.application.container import get_container from cleveragents.infrastructure.database.repositories import ( SkillRepository, ) @@ -853,7 +853,8 @@ def tools( ConfigService, ) - config_svc = ConfigService() + _container = get_container() + config_svc = ConfigService(event_bus=_container.event_bus()) resolved = config_svc.resolve("skills.agent_skills_paths") raw_paths: str = str(resolved.value) if resolved.value else "" if raw_paths: @@ -992,7 +993,8 @@ def refresh( ConfigService, ) - config_svc = ConfigService() + _container = get_container() + config_svc = ConfigService(event_bus=_container.event_bus()) resolved = config_svc.resolve("skills.agent_skills_paths") raw_paths: str = str(resolved.value) if resolved.value else "" diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py index b9a11e9cd..7f1262fe0 100644 --- a/src/cleveragents/infrastructure/events/reactive.py +++ b/src/cleveragents/infrastructure/events/reactive.py @@ -20,12 +20,15 @@ from __future__ import annotations from collections.abc import Callable +import structlog from rx.core.observable.observable import Observable from rx.subject.subject import Subject from cleveragents.infrastructure.events.models import DomainEvent from cleveragents.infrastructure.events.types import EventType +_logger = structlog.get_logger(__name__) + class ReactiveEventBus: """In-process event bus using RxPY for real-time distribution. @@ -71,7 +74,17 @@ class ReactiveEventBus: ) self._subject.on_next(event) for handler in self._subscriptions.get(event.event_type, []): - handler(event) + try: + handler(event) + except Exception as exc: + _logger.warning( + "event_handler_failed", + event_type=event.event_type.value + if hasattr(event.event_type, "value") + else str(event.event_type), + handler=getattr(handler, "__qualname__", repr(handler)), + error_type=type(exc).__name__, + ) def subscribe( self, diff --git a/src/cleveragents/infrastructure/events/types.py b/src/cleveragents/infrastructure/events/types.py index f3d391bb1..f513d026d 100644 --- a/src/cleveragents/infrastructure/events/types.py +++ b/src/cleveragents/infrastructure/events/types.py @@ -58,6 +58,19 @@ class EventType(StrEnum): RESOURCE_MODIFIED = "resource.modified" RESOURCE_INDEXED = "resource.indexed" + # --- Correction --- + CORRECTION_APPLIED = "correction.applied" + + # --- Configuration --- + CONFIG_CHANGED = "config.changed" + + # --- Entity lifecycle --- + ENTITY_DELETED = "entity.deleted" + + # --- Authentication (server-mode only) --- + AUTH_SUCCESS = "auth.success" + AUTH_FAILURE = "auth.failure" + # --- Sandbox --- SANDBOX_CREATED = "sandbox.created" SANDBOX_COMMITTED = "sandbox.committed" -- 2.52.0