CHANGELOG.md

@@ -73,6 +73,26 @@
   `@tdd_bug_<N>` 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

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)

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",
 ]

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)

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"

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"
+    )

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}"
+    )

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()

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

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]} <command>", file=sys.stderr)
+        print(f"Commands: {list(_COMMANDS)}", file=sys.stderr)
+        sys.exit(1)
+    _COMMANDS[sys.argv[1]]()

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
 
 

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",

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."""

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",
+]

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 ───────────────────────────────────────────────────

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."<project_name>"].<key>`` 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:

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
 
     # ------------------------------------------------------------------

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)

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")

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,

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:

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(

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,

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 ""
 

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,

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"