feat(events): wire domain services to emit missing EventBus events

Add TokenAuthMiddleware to emit AUTH_SUCCESS/AUTH_FAILURE with spec-aligned audit details and wire it through the DI container using server.token resolution.

Add Behave and Robot coverage for auth event emission and end-to-end audit persistence, and update audit subscriber producer notes and changelog.

ISSUES CLOSED: #714

CHANGELOG.md

@@ -5,6 +5,8 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 Changed `wf10_batch.robot` to be less likely to create files, and
 `plan_generation_graph.robot` to give more test answers.
 
+## [Unreleased]
+
 - Hardened the TDD bug-fix quality gate for issue #629: PR parsing now
   requires whole-word closing keywords (avoids false positives like
   "prefixes #12"), TDD bug tag discovery now uses exact token matching
@@ -40,6 +42,20 @@ Changed `wf10_batch.robot` to be less likely to create files, and
   cleanup conflict with `cli_init_yes_flag_steps.py`); 2 new Behave
   scenarios covering multi-line PR description parsing and non-string
   `pr_diff` type guard.
+
+- Added `TokenAuthMiddleware` and DI wiring to emit missing auth domain
+  events (`AUTH_SUCCESS`, `AUTH_FAILURE`) during token checks, including
+  spec-aligned audit details (`user_identity`, `attempted_identity`,
+  `ip_address`, `token_prefix`, `failure_reason`) and best-effort publish
+  behavior. Auth token prefixes now persist to audit logs without
+  redaction-key collisions, and short tokens are masked (`***...`) to
+  prevent full token disclosure. Added Behave coverage and Robot
+  audit-pipeline integration tests for auth event persistence. (#714)
+
+- Added TDD bug-capture E2E tests for bug #1028 — ACMS indexing pipeline not
+  wired into CLI.  Four Robot Framework E2E tests prove ContextTierService starts
+  empty on every CLI invocation.  Tests use ``@tdd_expected_fail`` until the bug
+  fix is merged.  (#1029)
 - Added Fix-then-Revalidate orchestration loop for required validations:
   bounded retry with configurable limits (0--100 per Safety Profile),
   strategy revision escalation via `auto_strategy_revision` float
@@ -52,7 +68,6 @@ Changed `wf10_batch.robot` to be less likely to create files, and
   counter, spec-required `validation_summary` and
   `final_validation_results` fields on the result model, DI container
 
-## [Unreleased]
 - **fix(tui): rename ActorSelectionOverlay._render to _refresh_display (issue #11039)** — `ActorSelectionOverlay._render()` shadows Textual's `Widget._render()` which must return a `Strip`. In textual >=1.0, layout calls `get_content_height()` `self._render()` gets `None` `AttributeError: 'NoneType' object has no attribute 'get_height'`. Renamed the method to `_refresh_display()` and updated all four internal call sites (`show()`, `move_up()`, `move_down()`, `set_search()`) to use the new name.
 
 - **Structural Component Output Validation** (#8164): Replaces exact character matching with structural component checking for output validation. Implements three validators covering plan tree output, decision CLI dicts, and structured session snapshots. The `validate_plan_tree` function validates node dicts for required keys (`decision_id`, `type`, `sequence`, `question`, `children`), ULID format, correct types, and sibling ordering. The `validate_decision_dict` function validates decision CLI output against the `Decision.as_cli_dict()` schema with field presence, type, ULID pattern, confidence range [0..1], and boolean field checks. The `validate_structured_output` function validates the StructuredOutput envelope for `command`, `session_id` (ULID), status membership, `exit_code`, and elements integrity. A unified dispatcher (`validate_structured_component_output`) enables routing by target_type. BDD test coverage added in `features/structural_validation.feature`. [Epic #8137](https://git.cleverthis.com/cleveragents/cleveragents-core/issues/8137)

features/auth_middleware_events.feature

@@ -0,0 +1,63 @@
+Feature: Auth middleware emits security events
+  As a platform operator
+  I want authentication outcomes emitted as domain events
+  So the audit pipeline captures auth success and failure details
+
+  Scenario: Successful authentication emits AUTH_SUCCESS
+    Given an auth middleware with expected token "tok_secret_123"
+    And an auth recording event bus
+    When I authenticate with token "tok_secret_123" identity "alice@example.com" and ip "10.0.0.5"
+    Then the auth result should be true
+    And the auth event bus should have exactly 1 event
+    And the latest auth event type should be AUTH_SUCCESS
+    And the latest auth event should contain detail "user_identity" with value "alice@example.com"
+    And the latest auth event should contain detail "ip_address" with value "10.0.0.5"
+    And the latest auth event should contain detail "token_prefix" with value "tok_se..."
+
+  Scenario: Short token success masks token_prefix
+    Given an auth middleware with expected token "short"
+    And an auth recording event bus
+    When I authenticate with token "short" identity "alice@example.com" and ip "10.0.0.5"
+    Then the auth result should be true
+    And the latest auth event type should be AUTH_SUCCESS
+    And the latest auth event should contain detail "token_prefix" with value "***..."
+
+  Scenario: Failed authentication emits AUTH_FAILURE
+    Given an auth middleware with expected token "tok_secret_123"
+    And an auth recording event bus
+    When I authenticate with token "tok_wrong_999" identity "alice@example.com" and ip "10.0.0.5"
+    Then the auth result should be false
+    And the auth event bus should have exactly 1 event
+    And the latest auth event type should be AUTH_FAILURE
+    And the latest auth event should contain detail "attempted_identity" with value "alice@example.com"
+    And the latest auth event should contain detail "failure_reason" with value "invalid_token"
+    And the latest auth event should contain detail "token_prefix" with value "tok_wr..."
+
+  Scenario: Missing configured token emits AUTH_FAILURE
+    Given an auth middleware with no configured token
+    And an auth recording event bus
+    When I authenticate with token "tok_any_111" identity "bob@example.com" and ip "10.0.0.9"
+    Then the auth result should be false
+    And the latest auth event type should be AUTH_FAILURE
+    And the latest auth event should contain detail "failure_reason" with value "token_not_configured"
+
+  Scenario: Empty authentication token is rejected before emission
+    Given an auth middleware with expected token "tok_secret_123"
+    And an auth recording event bus
+    When I try to authenticate with an empty token
+    Then an auth ValueError should be raised
+    And the auth event bus should have exactly 0 event
+
+  Scenario: Auth middleware events persist through AuditEventSubscriber
+    Given auth middleware is wired to AuditEventSubscriber with expected token "tok_secret_123"
+    When I authenticate with token "tok_secret_123" identity "carol@example.com" and ip "10.1.2.3"
+    And I authenticate with token "tok_bad_123" identity "dave@example.com" and ip "10.1.2.4"
+    Then the auth result history should be "true,false"
+    And the audit log should contain 1 auth_success entry from middleware
+    And the latest auth_success audit entry should contain detail "ip_address" with value "10.1.2.3"
+    And the latest auth_success audit entry should contain detail "token_prefix" with value "tok_se..."
+    And the audit log should contain 1 auth_failure entry from middleware
+    And the latest auth_failure audit entry should contain detail "ip_address" with value "10.1.2.4"
+    And the latest auth_failure audit entry should contain detail "attempted_identity" with value "dave@example.com"
+    And the latest auth_failure audit entry should contain detail "failure_reason" with value "invalid_token"
+    And the latest auth_failure audit entry should contain detail "token_prefix" with value "tok_ba..."

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 .recording_event_bus import RecordingEventBus
 from .transient_fail_audit_service import TransientFailAuditService
 
 # NOTE: tdd_test_helpers is NOT re-exported here because it imports
@@ -22,6 +23,7 @@ __all__ = [
     "MockAIProvider",
     "MockLspTransport",
     "MockMCPTransport",
+    "RecordingEventBus",
     "TransientFailAuditService",
     "parse_lsp_responses",
 ]

features/mocks/recording_event_bus.py

@@ -0,0 +1,25 @@
+"""Minimal in-memory event bus used for Behave assertions."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+from cleveragents.infrastructure.events.models import DomainEvent
+from cleveragents.infrastructure.events.types import EventType
+
+
+class RecordingEventBus:
+    """Minimal in-memory event bus used for Behave assertions."""
+
+    def __init__(self) -> None:
+        self.events: list[DomainEvent] = []
+
+    def emit(self, event: DomainEvent) -> None:
+        self.events.append(event)
+
+    def subscribe(
+        self,
+        event_type: EventType,
+        handler: Callable[[DomainEvent], None],
+    ) -> None:
+        _ = (event_type, handler)

features/steps/auth_middleware_events_steps.py

@@ -0,0 +1,165 @@
+"""Step definitions for auth_middleware_events.feature."""
+
+from __future__ import annotations
+
+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.application.services.auth_middleware import TokenAuthMiddleware
+from cleveragents.config.settings import Settings
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.events.reactive import ReactiveEventBus
+from cleveragents.infrastructure.events.types import EventType
+from features.mocks.recording_event_bus import RecordingEventBus
+
+
+def _fresh_audit_service() -> AuditService:
+    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)
+
+
+@given('an auth middleware with expected token "{token}"')
+def step_auth_middleware_with_token(context: Context, token: str) -> None:
+    context.expected_token = token
+
+
+@given("an auth middleware with no configured token")
+def step_auth_middleware_no_token(context: Context) -> None:
+    context.expected_token = None
+
+
+@given("an auth recording event bus")
+def step_auth_recording_event_bus(context: Context) -> None:
+    context.event_bus = RecordingEventBus()
+    context.auth_middleware = TokenAuthMiddleware(
+        expected_token=context.expected_token,
+        event_bus=context.event_bus,
+    )
+    context.auth_result_history = []
+
+
+@given('auth middleware is wired to AuditEventSubscriber with expected token "{token}"')
+def step_auth_middleware_audit_pipeline(context: Context, token: str) -> None:
+    context.audit_service = _fresh_audit_service()
+    context.event_bus = ReactiveEventBus()
+    context.audit_subscriber = AuditEventSubscriber(
+        audit_service=context.audit_service,
+        event_bus=context.event_bus,
+    )
+    context.auth_middleware = TokenAuthMiddleware(
+        expected_token=token,
+        event_bus=context.event_bus,
+    )
+    context.auth_result_history = []
+
+
+@when('I authenticate with token "{token}" identity "{identity}" and ip "{ip_address}"')
+def step_authenticate(
+    context: Context, token: str, identity: str, ip_address: str
+) -> None:
+    context.auth_result = context.auth_middleware.authenticate(
+        token,
+        identity=identity,
+        ip_address=ip_address,
+    )
+    context.auth_result_history.append(context.auth_result)
+
+
+@when("I try to authenticate with an empty token")
+def step_try_authenticate_empty_token(context: Context) -> None:
+    context.auth_error = None
+    try:
+        context.auth_middleware.authenticate("", identity="empty@example.com")
+    except ValueError as exc:
+        context.auth_error = exc
+
+
+@then("the auth result should be true")
+def step_auth_result_true(context: Context) -> None:
+    assert context.auth_result is True
+
+
+@then("the auth result should be false")
+def step_auth_result_false(context: Context) -> None:
+    assert context.auth_result is False
+
+
+@then("the auth event bus should have exactly {count:d} event")
+def step_auth_event_count(context: Context, count: int) -> None:
+    assert len(context.event_bus.events) == count, (
+        f"Expected {count} events, got {len(context.event_bus.events)}"
+    )
+
+
+@then("the latest auth event type should be {event_type_name}")
+def step_latest_auth_event_type(context: Context, event_type_name: str) -> None:
+    expected = EventType[event_type_name]
+    actual = context.event_bus.events[-1].event_type
+    assert actual == expected, f"Expected event type {expected}, got {actual}"
+
+
+@then(
+    'the latest auth event should contain detail "{key}" with value "{expected_value}"'
+)
+def step_latest_auth_event_detail(
+    context: Context, key: str, expected_value: str
+) -> None:
+    details = context.event_bus.events[-1].details
+    assert details.get(key) == expected_value, (
+        f"Expected {key}={expected_value!r}, got {details.get(key)!r}"
+    )
+
+
+@then("an auth ValueError should be raised")
+def step_auth_value_error(context: Context) -> None:
+    assert context.auth_error is not None, (
+        "Expected ValueError but no exception was raised"
+    )
+    assert isinstance(context.auth_error, ValueError), (
+        f"Expected ValueError, got {type(context.auth_error).__name__}"
+    )
+
+
+@then('the auth result history should be "{history_csv}"')
+def step_auth_result_history(context: Context, history_csv: str) -> None:
+    expected = [part.strip().lower() == "true" for part in history_csv.split(",")]
+    assert context.auth_result_history == expected, (
+        f"Expected history {expected}, got {context.auth_result_history}"
+    )
+
+
+@then("the audit log should contain 1 auth_success entry from middleware")
+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 1 auth_success entry, got {len(entries)}"
+
+
+@then("the audit log should contain 1 auth_failure entry from middleware")
+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 1 auth_failure entry, got {len(entries)}"
+
+
+@then(
+    'the latest {audit_event_type} audit entry should contain detail "{key}" with value "{expected_value}"'
+)
+def step_latest_audit_entry_detail(
+    context: Context, audit_event_type: str, key: str, expected_value: str
+) -> None:
+    entries = context.audit_service.list_entries(event_type=audit_event_type)
+    assert entries, f"Expected at least one {audit_event_type} audit entry"
+    details = entries[0].details
+    assert details.get(key) == expected_value, (
+        f"Expected {audit_event_type}.{key}={expected_value!r}, "
+        f"got {details.get(key)!r}"
+    )

robot/audit_service_wiring.robot

@@ -57,7 +57,6 @@ E2E Plan Lifecycle Through Audit Pipeline
     Log    ${result.stderr}
     Should Be Equal As Integers    ${result.rc}    0
     Should Contain    ${result.stdout}    E2E_OK
-
 User Identity Field Propagates Through Audit Pipeline
     [Documentation]    DomainEvent.user_identity propagates via AuditEventSubscriber to AuditService DB entry
     [Tags]    observability    audit    identity
@@ -87,3 +86,13 @@ User Identity Field Precedence Over Details
     Log    ${result.stderr}
     Should Be Equal As Integers    ${result.rc}    0
     Should Contain    ${result.stdout}    IDENTITY_PRECEDENCE_OK
+
+Auth Middleware Emits Audit Events
+    [Documentation]    End-to-end: TokenAuthMiddleware emits AUTH_SUCCESS/AUTH_FAILURE through audit pipeline
+    [Tags]    observability    audit    auth
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    auth_middleware_pipeline
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    AUTH_PIPELINE_OK

robot/helper_audit_wiring.py

@@ -23,6 +23,9 @@ from cleveragents.application.services.audit_event_subscriber import (  # noqa:
 from cleveragents.application.services.audit_service import (  # noqa: E402
     AuditService,
 )
+from cleveragents.application.services.auth_middleware import (  # noqa: E402
+    TokenAuthMiddleware,
+)
 from cleveragents.application.services.plan_lifecycle_service import (  # noqa: E402
     PlanLifecycleService,
 )
@@ -323,6 +326,102 @@ def user_identity_field_precedence() -> None:
     print("IDENTITY_PRECEDENCE_OK")
 
 
+def auth_middleware_pipeline() -> None:
+    """E2E: TokenAuthMiddleware -> EventBus -> AuditEventSubscriber -> DB."""
+    svc = _make_audit_service()
+    bus = ReactiveEventBus()
+    AuditEventSubscriber(audit_service=svc, event_bus=bus)
+    middleware = TokenAuthMiddleware(expected_token="tok_secret_123", event_bus=bus)
+
+    success = middleware.authenticate(
+        "tok_secret_123",
+        identity="carol@example.com",
+        ip_address="10.1.2.3",
+    )
+    failure = middleware.authenticate(
+        "tok_bad_123",
+        identity="dave@example.com",
+        ip_address="10.1.2.4",
+    )
+    if not success or failure:
+        print(
+            "FAIL: unexpected auth boolean results "
+            f"success={success} failure={failure}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    success_entries = svc.list_entries(event_type="auth_success")
+    failure_entries = svc.list_entries(event_type="auth_failure")
+    if len(success_entries) != 1:
+        print(
+            f"FAIL: expected 1 auth_success entry, got {len(success_entries)}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    if len(failure_entries) != 1:
+        print(
+            f"FAIL: expected 1 auth_failure entry, got {len(failure_entries)}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    s_entry = success_entries[0]
+    f_entry = failure_entries[0]
+    s_details = s_entry.details
+    f_details = f_entry.details
+    if s_entry.user_identity != "carol@example.com":
+        print(
+            f"FAIL: unexpected auth_success identity {s_entry.user_identity!r}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    if f_entry.user_identity != "unauthenticated":
+        print(
+            f"FAIL: unexpected auth_failure identity {f_entry.user_identity!r}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    if s_details.get("ip_address") != "10.1.2.3":
+        print(
+            f"FAIL: unexpected auth_success ip {s_details.get('ip_address')!r}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    if s_details.get("token_prefix") != "tok_se...":
+        print(
+            "FAIL: auth_success token_prefix mismatch",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    if f_details.get("ip_address") != "10.1.2.4":
+        print(
+            f"FAIL: unexpected auth_failure ip {f_details.get('ip_address')!r}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    if f_details.get("attempted_identity") != "dave@example.com":
+        print(
+            "FAIL: auth_failure attempted_identity mismatch",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    if f_details.get("failure_reason") != "invalid_token":
+        print(
+            f"FAIL: unexpected auth_failure reason {f_details.get('failure_reason')!r}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    if f_details.get("token_prefix") != "tok_ba...":
+        print(
+            "FAIL: auth_failure token_prefix mismatch",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    print("AUTH_PIPELINE_OK")
+
+
 # ---------------------------------------------------------------------------
 # Dispatch
 # ---------------------------------------------------------------------------
@@ -336,6 +435,7 @@ _COMMANDS = {
     "user_identity_field": user_identity_field,
     "user_identity_details_fallback": user_identity_details_fallback,
     "user_identity_field_precedence": user_identity_field_precedence,
+    "auth_middleware_pipeline": auth_middleware_pipeline,
 }
 
 if __name__ == "__main__":

src/cleveragents/application/container.py

@@ -23,6 +23,7 @@ from cleveragents.application.services.audit_event_subscriber import (
     AuditEventSubscriber,
 )
 from cleveragents.application.services.audit_service import AuditService
+from cleveragents.application.services.auth_middleware import TokenAuthMiddleware
 from cleveragents.application.services.automation_profile_service import (
     AutomationProfileService,
 )
@@ -410,6 +411,25 @@ def _resolve_auto_reindex() -> bool:
         return True
 
 
+def _resolve_server_token() -> str | None:
+    """Resolve ``server.token`` from config, returning ``None`` when unset."""
+    try:
+        svc = ConfigService()
+        resolved = svc.resolve("server.token")
+        raw = resolved.value
+        if raw is None:
+            return None
+        token = str(raw).strip()
+        return token if token else None
+    except Exception as exc:
+        _logger.warning(
+            "server_token_resolution_failed",
+            error_type=type(exc).__name__,
+            error_message=redact_value(str(exc)),
+        )
+        return None
+
+
 def _build_analyzer_registry() -> AnalyzerRegistry:
     """Build an AnalyzerRegistry with built-in analyzers registered."""
     from cleveragents.domain.models.acms.python_analyzer import PythonAnalyzer
@@ -667,6 +687,13 @@ class Container(containers.DeclarativeContainer):
     # Event Bus - Singleton (one shared instance for the process)
     event_bus = providers.Singleton(ReactiveEventBus)
 
+    # Per-request auth middleware with fresh token resolution from config.
+    auth_middleware = providers.Factory(
+        TokenAuthMiddleware,
+        expected_token=providers.Callable(_resolve_server_token),
+        event_bus=event_bus,
+    )
+
     # Services - Factory (new instance per request with injected dependencies)
     project_service = providers.Factory(
         ProjectService,

src/cleveragents/application/services/audit_event_subscriber.py

@@ -37,19 +37,13 @@ _logger = structlog.get_logger(__name__)
 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.
+    # Emitted by ResourceFileWatcher when indexed resources change.
     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.
+    # Emitted by TokenAuthMiddleware during bearer-token checks.
     EventType.AUTH_SUCCESS: "auth_success",
     EventType.AUTH_FAILURE: "auth_failure",
 }

src/cleveragents/application/services/auth_middleware.py

@@ -0,0 +1,189 @@
+"""Authentication middleware that emits audit-friendly auth events.
+
+This service encapsulates bearer-token authentication checks and emits
+``AUTH_SUCCESS`` / ``AUTH_FAILURE`` domain events through an injected
+``EventBus`` so the audit pipeline can persist login outcomes.
+
+Spec alignment:
+    - docs/specification.md § Event System
+    - docs/specification.md § Audit Logging (auth_success / auth_failure)
+"""
+
+from __future__ import annotations
+
+import hmac
+from typing import TYPE_CHECKING
+
+import structlog
+
+from cleveragents.infrastructure.events.models import DomainEvent
+from cleveragents.infrastructure.events.types import EventType
+from cleveragents.shared.redaction import redact_value
+
+if TYPE_CHECKING:
+    from cleveragents.infrastructure.events.protocol import EventBus
+
+_logger = structlog.get_logger(__name__)
+_TOKEN_PREFIX_LEN = 6
+_SHORT_TOKEN_MASK = "***..."
+
+
+def _token_prefix(token: str) -> str:
+    """Return a safe token prefix for audit details.
+
+    Never returns the full token value. Tokens shorter than or equal to the
+    prefix length are collapsed to a constant mask to avoid full disclosure.
+    """
+    if len(token) <= _TOKEN_PREFIX_LEN:
+        return _SHORT_TOKEN_MASK
+    return f"{token[:_TOKEN_PREFIX_LEN]}..."
+
+
+def _normalize_optional_text(field_name: str, value: str | None) -> str | None:
+    """Normalize optional text fields with guard-first validation."""
+    if value is None:
+        return None
+    if not isinstance(value, str):
+        raise TypeError(f"{field_name} must be a string or None")
+    normalized = value.strip()
+    if not normalized:
+        raise ValueError(f"{field_name} must not be empty when provided")
+    return normalized
+
+
+class TokenAuthMiddleware:
+    """Authenticate bearer tokens and emit AUTH_* events.
+
+    This class is intentionally transport-agnostic. A future server-mode
+    middleware can delegate authentication decisions to this component and
+    inherit consistent event emission semantics.
+    """
+
+    def __init__(
+        self,
+        expected_token: str | None,
+        event_bus: EventBus | None = None,
+    ) -> None:
+        """Initialize middleware with an expected token and optional event bus.
+
+        Args:
+            expected_token: Configured bearer token for successful auth.
+            event_bus: Event bus used for AUTH_SUCCESS/AUTH_FAILURE emission.
+        """
+        self._expected_token = _normalize_optional_text(
+            "expected_token", expected_token
+        )
+        self._event_bus = event_bus
+
+    def authenticate(
+        self,
+        token: str,
+        *,
+        identity: str | None = None,
+        ip_address: str | None = None,
+    ) -> bool:
+        """Authenticate *token* and emit a corresponding auth event.
+
+        Args:
+            token: Bearer token supplied by the caller.
+            identity: Optional user identity / principal for audit context.
+            ip_address: Optional client IP address for audit context.
+
+        Returns:
+            ``True`` if authentication succeeds, otherwise ``False``.
+        """
+        if not isinstance(token, str):
+            raise TypeError("token must be a string")
+        normalized_token = token.strip()
+        if not normalized_token:
+            raise ValueError("token must not be empty")
+
+        normalized_identity = _normalize_optional_text("identity", identity)
+        normalized_ip = _normalize_optional_text("ip_address", ip_address)
+
+        if self._expected_token is None:
+            self._emit_auth_failure(
+                attempted_identity=normalized_identity,
+                ip_address=normalized_ip,
+                token_prefix=_token_prefix(normalized_token),
+                reason="token_not_configured",
+            )
+            return False
+
+        if hmac.compare_digest(normalized_token, self._expected_token):
+            self._emit_auth_success(
+                user_identity=normalized_identity,
+                ip_address=normalized_ip,
+                token_prefix=_token_prefix(normalized_token),
+            )
+            return True
+
+        self._emit_auth_failure(
+            attempted_identity=normalized_identity,
+            ip_address=normalized_ip,
+            token_prefix=_token_prefix(normalized_token),
+            reason="invalid_token",
+        )
+        return False
+
+    def _emit_auth_success(
+        self,
+        *,
+        user_identity: str | None,
+        ip_address: str | None,
+        token_prefix: str,
+    ) -> None:
+        """Emit AUTH_SUCCESS if an event bus is configured."""
+        if self._event_bus is None:
+            return
+        details: dict[str, str] = {
+            "user_identity": user_identity or "unknown",
+            "token_prefix": token_prefix,
+        }
+        if ip_address is not None:
+            details["ip_address"] = ip_address
+        self._emit_event(EventType.AUTH_SUCCESS, details)
+
+    def _emit_auth_failure(
+        self,
+        *,
+        attempted_identity: str | None,
+        ip_address: str | None,
+        token_prefix: str,
+        reason: str,
+    ) -> None:
+        """Emit AUTH_FAILURE if an event bus is configured."""
+        if self._event_bus is None:
+            return
+        details: dict[str, str] = {
+            "attempted_identity": attempted_identity or "unknown",
+            "user_identity": "unauthenticated",
+            "failure_reason": reason,
+            "token_prefix": token_prefix,
+        }
+        if ip_address is not None:
+            details["ip_address"] = ip_address
+        self._emit_event(EventType.AUTH_FAILURE, details)
+
+    def _emit_event(self, event_type: EventType, details: dict[str, str]) -> None:
+        """Emit one auth event and swallow publish-side failures."""
+        bus = self._event_bus
+        if bus is None:
+            return
+        try:
+            bus.emit(
+                DomainEvent(
+                    event_type=event_type,
+                    details=details,
+                )
+            )
+        except Exception as exc:  # pragma: no cover - defensive logging path
+            _logger.warning(
+                "auth_event_emit_failed",
+                event_type=event_type.value,
+                error_type=type(exc).__name__,
+                error_message=redact_value(str(exc)),
+            )
+
+
+__all__ = ["TokenAuthMiddleware"]

src/cleveragents/shared/redaction.py

@@ -42,6 +42,7 @@ _SENSITIVE_SUBSTRINGS: set[str] = {
 # though they contain a sensitive substring (e.g. "token_count").
 _FALSE_POSITIVE_KEYS: set[str] = {
     "token_count",
+    "token_prefix",
     "token_limit",
     "token_usage",
     "input_tokens",