feat(events): add user_identity field to DomainEvent and propagate through event pipeline (#1257)

Co-authored-by: Brent Edwards <brent.edwards@cleverthis.com>
Co-committed-by: Brent Edwards <brent.edwards@cleverthis.com>

features/observability/audit_service_wiring.feature

@@ -118,3 +118,18 @@ Feature: AuditService EventBus Wiring
     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"
+
+  Scenario: User identity from DomainEvent field is forwarded to audit entry
+    Given an event bus with an audit subscriber
+    When a session_created event with user_identity field "field-user@example.com" is emitted
+    Then the audit entry should have user_identity "field-user@example.com"
+
+  Scenario: DomainEvent field user_identity takes precedence over details
+    Given an event bus with an audit subscriber
+    When a session_created event with both user_identity field "field@example.com" and details identity "details@example.com" is emitted
+    Then the audit entry should have user_identity "field@example.com"
+
+  Scenario: Audit entry user_identity is None when not provided anywhere
+    Given an event bus with an audit subscriber
+    When a plan_applied event is emitted
+    Then the audit entry user_identity should be None

features/observability/event_system_taxonomy.feature

@@ -102,6 +102,7 @@ Feature: Event System Domain Event Taxonomy
     And the taxonomy event should have field "session_id"
     And the taxonomy event should have field "actor_name"
     And the taxonomy event should have field "project_name"
+    And the taxonomy event should have field "user_identity"
     And the taxonomy event should have field "details"
 
   Scenario: DomainEvent correlation_id is a 26-char ULID string
@@ -130,6 +131,18 @@ Feature: Event System Domain Event Taxonomy
     When I create a DomainEvent for taxonomy with typed details
     Then the taxonomy event details should contain the expected keys
 
+  Scenario: DomainEvent user_identity defaults to None
+    When I create a DomainEvent for taxonomy with event_type "plan.created"
+    Then the taxonomy event user_identity should be None
+
+  Scenario: DomainEvent accepts explicit user_identity
+    When I create a DomainEvent for taxonomy with user_identity "alice@example.com"
+    Then the taxonomy event user_identity should be "alice@example.com"
+
+  Scenario: DomainEvent with user_identity serializes to JSON round-trip
+    When I create a DomainEvent for taxonomy with user_identity "bob@example.com"
+    Then the taxonomy event should serialize to JSON and deserialize back identically
+
   # ---------------------------------------------------------------------------
   # EventBus Protocol
   # ---------------------------------------------------------------------------

features/steps/audit_wiring_extended_steps.py

@@ -177,3 +177,47 @@ def step_audit_entry_user_identity(context: Context, expected_uid: str) -> None:
     assert entries[0].user_identity == expected_uid, (
         f"Expected user_identity={expected_uid!r}, got {entries[0].user_identity!r}"
     )
+
+
+# ── #715: DomainEvent.user_identity field propagation ────────────
+
+
+@when('a session_created event with user_identity field "{uid}" is emitted')
+def step_emit_session_with_user_identity_field(context: Context, uid: str) -> None:
+    context.event_bus.emit(
+        DomainEvent(
+            event_type=EventType.SESSION_CREATED,
+            session_id="SES-FIELD",
+            user_identity=uid,
+            details={"mode": "interactive"},
+        )
+    )
+
+
+@when(
+    'a session_created event with both user_identity field "{field_uid}" '
+    'and details identity "{details_uid}" is emitted'
+)
+def step_emit_session_with_both_identities(
+    context: Context, field_uid: str, details_uid: str
+) -> None:
+    context.event_bus.emit(
+        DomainEvent(
+            event_type=EventType.SESSION_CREATED,
+            session_id="SES-BOTH",
+            user_identity=field_uid,
+            details={
+                "mode": "interactive",
+                "user_identity": details_uid,
+            },
+        )
+    )
+
+
+@then("the audit entry user_identity should be None")
+def step_audit_entry_user_identity_none(context: Context) -> None:
+    entries = context.audit_service.list_entries(event_type="plan_applied")
+    assert len(entries) >= 1, "No plan_applied entries found"
+    assert entries[0].user_identity is None, (
+        f"Expected user_identity=None, got {entries[0].user_identity!r}"
+    )

features/steps/event_system_taxonomy_steps.py

@@ -160,6 +160,28 @@ def step_taxonomy_session_id_set(ctx: Context) -> None:
     assert ctx.taxonomy_event.session_id is not None
 
 
+@then("the taxonomy event user_identity should be None")
+def step_taxonomy_event_user_identity_none(ctx: Context) -> None:
+    assert ctx.taxonomy_event.user_identity is None, (
+        f"Expected user_identity=None, got {ctx.taxonomy_event.user_identity!r}"
+    )
+
+
+@when('I create a DomainEvent for taxonomy with user_identity "{uid}"')
+def step_create_taxonomy_event_with_user_identity(ctx: Context, uid: str) -> None:
+    ctx.taxonomy_event = DomainEvent(
+        event_type=EventType.PLAN_CREATED,
+        user_identity=uid,
+    )
+
+
+@then('the taxonomy event user_identity should be "{expected}"')
+def step_taxonomy_event_user_identity_value(ctx: Context, expected: str) -> None:
+    assert ctx.taxonomy_event.user_identity == expected, (
+        f"Expected user_identity={expected!r}, got {ctx.taxonomy_event.user_identity!r}"
+    )
+
+
 @when("I create a DomainEvent for taxonomy with typed details")
 def step_create_taxonomy_event_with_details(ctx: Context) -> None:
     ctx.taxonomy_event = DomainEvent(

robot/audit_service_wiring.robot

@@ -57,3 +57,33 @@ 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
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    user_identity_field
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    IDENTITY_FIELD_OK
+
+User Identity Fallback From Details
+    [Documentation]    Backward compat: user_identity extracted from details when field is None
+    [Tags]    observability    audit    identity
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    user_identity_details_fallback
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    IDENTITY_FALLBACK_OK
+
+User Identity Field Precedence Over Details
+    [Documentation]    DomainEvent.user_identity field takes precedence over details dict
+    [Tags]    observability    audit    identity
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    user_identity_field_precedence
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    IDENTITY_PRECEDENCE_OK

robot/helper_audit_wiring.py

@@ -211,6 +211,106 @@ def e2e_plan_lifecycle() -> None:
     print("E2E_OK")
 
 
+def user_identity_field() -> None:
+    """Verify DomainEvent.user_identity propagates to audit entry."""
+    svc = _make_audit_service()
+    bus = ReactiveEventBus()
+    AuditEventSubscriber(audit_service=svc, event_bus=bus)
+
+    bus.emit(
+        DomainEvent(
+            event_type=EventType.SESSION_CREATED,
+            session_id="SES-FIELD-INT",
+            user_identity="integration-user@example.com",
+            details={"mode": "interactive"},
+        )
+    )
+
+    entries = svc.list_entries(event_type="session_created")
+    if not entries:
+        print("FAIL: no session_created entry found", file=sys.stderr)
+        sys.exit(1)
+    if entries[0].user_identity != "integration-user@example.com":
+        print(
+            f"FAIL: user_identity={entries[0].user_identity!r}, "
+            f"expected 'integration-user@example.com'",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    print("IDENTITY_FIELD_OK")
+
+
+def user_identity_details_fallback() -> None:
+    """Verify backward compat: user_identity from details when field is None."""
+    svc = _make_audit_service()
+    bus = ReactiveEventBus()
+    AuditEventSubscriber(audit_service=svc, event_bus=bus)
+
+    bus.emit(
+        DomainEvent(
+            event_type=EventType.SESSION_CREATED,
+            session_id="SES-FALLBACK-INT",
+            details={
+                "mode": "interactive",
+                "user_identity": "fallback-user@example.com",
+            },
+        )
+    )
+
+    entries = svc.list_entries(event_type="session_created")
+    if not entries:
+        print("FAIL: no session_created entry found", file=sys.stderr)
+        sys.exit(1)
+    if entries[0].user_identity != "fallback-user@example.com":
+        print(
+            f"FAIL: user_identity={entries[0].user_identity!r}, "
+            f"expected 'fallback-user@example.com'",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    print("IDENTITY_FALLBACK_OK")
+
+
+def user_identity_field_precedence() -> None:
+    """Verify field takes precedence over details for user_identity."""
+    svc = _make_audit_service()
+    bus = ReactiveEventBus()
+    AuditEventSubscriber(audit_service=svc, event_bus=bus)
+
+    bus.emit(
+        DomainEvent(
+            event_type=EventType.SESSION_CREATED,
+            session_id="SES-PREC-INT",
+            user_identity="field-wins@example.com",
+            details={
+                "mode": "interactive",
+                "user_identity": "details-loses@example.com",
+            },
+        )
+    )
+
+    entries = svc.list_entries(event_type="session_created")
+    if not entries:
+        print("FAIL: no session_created entry found", file=sys.stderr)
+        sys.exit(1)
+    if entries[0].user_identity != "field-wins@example.com":
+        print(
+            f"FAIL: user_identity={entries[0].user_identity!r}, "
+            f"expected 'field-wins@example.com'",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    # Also verify the details dict does NOT contain user_identity
+    # (it should have been stripped to avoid duplication).
+    if "user_identity" in entries[0].details:
+        print(
+            "FAIL: user_identity still present in details after field precedence",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    print("IDENTITY_PRECEDENCE_OK")
+
+
 # ---------------------------------------------------------------------------
 # Dispatch
 # ---------------------------------------------------------------------------
@@ -221,6 +321,9 @@ _COMMANDS = {
     "ignore_non_security": ignore_non_security,
     "container_wiring": container_wiring,
     "e2e_plan_lifecycle": e2e_plan_lifecycle,
+    "user_identity_field": user_identity_field,
+    "user_identity_details_fallback": user_identity_details_fallback,
+    "user_identity_field_precedence": user_identity_field_precedence,
 }
 
 if __name__ == "__main__":

src/cleveragents/application/services/audit_event_subscriber.py

@@ -99,12 +99,18 @@ class AuditEventSubscriber:
         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)
+            # SEC-1: Prefer the first-class ``user_identity`` field added
+            # in #715.  Fall back to extracting it from the ``details``
+            # dict for backward compatibility with callers that still
+            # embed the identity there (pre-#715 convention).
+            user_identity: str | None = event.user_identity
+            if user_identity is None:
+                user_identity = raw_details.pop("user_identity", None)
+            else:
+                # Remove from details to avoid duplication in the
+                # persisted payload — the value lives in the dedicated
+                # ``user_identity`` audit column.
+                raw_details.pop("user_identity", None)
 
             # N1: Audit persistence must never respect the display-layer
             # ``show_secrets`` flag — force redaction regardless of the

src/cleveragents/infrastructure/events/models.py

@@ -34,6 +34,8 @@ class DomainEvent(BaseModel):
         session_id: ULID of the session, if any.
         actor_name: Name of the actor that emitted the event, if any.
         project_name: Name of the project context, if any.
+        user_identity: Authenticated user identity, if available.
+            ``None`` in local (unauthenticated) mode.
         details: Event-type-specific payload (arbitrary key/value pairs).
     """
 
@@ -62,6 +64,13 @@ class DomainEvent(BaseModel):
     project_name: str | None = Field(
         default=None, description="Project context of this event."
     )
+    user_identity: str | None = Field(
+        default=None,
+        description=(
+            "Authenticated user identity associated with this event, "
+            "if available.  ``None`` in local (unauthenticated) mode."
+        ),
+    )
     details: dict[str, Any] = Field(
         default_factory=dict,
         description="Event-type-specific payload.",