fix(audit): forward DomainEvent.timestamp to AuditService.record()

- Replace 8 # type: ignore[arg-type] suppressions in _row_to_entry() with cast() calls per project type-safety standards
- Add argument validation for timestamp parameter (TypeError for non-datetime, ValueError for naive datetime)
- Restore container_wiring integration test to use _ensure_session() + Base.metadata.create_all() + _async_mode=False pattern (broken by prior rebase)
- Restore user_identity_* integration test functions and robot tests that were incorrectly removed during prior rebase
- Fix sys.path insertion in helper_audit_wiring.py to ensure repo src takes precedence over system-level /app/src installation
- Add preserve_event_timestamp integration test to robot suite

ISSUES CLOSED: #719

robot/audit_service_wiring.robot

@@ -58,6 +58,36 @@ E2E Plan Lifecycle Through Audit Pipeline
     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
+
 Audit Subscriber Preserves Event Timestamp
     [Documentation]    Verify DomainEvent.timestamp is forwarded to audit entry created_at (#719)
     [Tags]    observability    audit    timestamp

robot/helper_audit_wiring.py

@@ -9,9 +9,14 @@ import sys
 from datetime import UTC, datetime
 from pathlib import Path
 
-# Ensure local source tree is importable
+# Ensure local source tree is importable and takes precedence over any
+# system-level installation (e.g. /app/src from the dev container).
 _SRC = str(Path(__file__).resolve().parents[1] / "src")
-if _SRC not in sys.path:
+if sys.path and sys.path[0] != _SRC:
+    # Remove any existing entry for _SRC to avoid duplicates, then
+    # re-insert at position 0 so the repo's source takes precedence.
+    if _SRC in sys.path:
+        sys.path.remove(_SRC)
     sys.path.insert(0, _SRC)
 
 from sqlalchemy import create_engine  # noqa: E402
@@ -151,9 +156,21 @@ def container_wiring() -> None:
         sys.exit(1)
 
     # Functional verification: emit through container's EventBus
+    # and verify the event reaches the AuditService.
+    audit_svc = container.audit_service()
+
+    # The container's in-memory DB has no tables yet — create them
+    # so the audit service can persist entries.
+    db_session = audit_svc._ensure_session()
+    audit_engine = db_session.get_bind()
+    Base.metadata.create_all(audit_engine)
+
+    # Disable async mode so writes happen synchronously and we can
+    # verify the count immediately without race conditions.
+    audit_svc._async_mode = False
+
     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(
@@ -212,11 +229,110 @@ 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
 # ---------------------------------------------------------------------------
 
-
 def preserve_event_timestamp() -> None:
     """Verify DomainEvent.timestamp is preserved in audit entry created_at (#719).
 
@@ -260,6 +376,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,
     "preserve_event_timestamp": preserve_event_timestamp,
 }
 

src/cleveragents/application/services/audit_service.py

@@ -44,7 +44,7 @@ import queue
 import threading
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
-from typing import Any, Self
+from typing import Any, Self, cast
 
 import structlog
 from sqlalchemy import create_engine
@@ -360,6 +360,9 @@ class AuditService:
         Raises:
             ValueError: If *event_type* is not one of the spec-defined
                 event types in :data:`VALID_EVENT_TYPES`.
+            TypeError: If *timestamp* is not a ``datetime`` instance or
+                ``None``.
+            ValueError: If *timestamp* is a naive datetime (no ``tzinfo``).
 
         Returns:
             The created :class:`AuditLogEntry`.  In async mode the ``id``
@@ -370,6 +373,17 @@ class AuditService:
                 f"Unknown event_type {event_type!r}; "
                 f"expected one of {sorted(VALID_EVENT_TYPES)}"
             )
+        if timestamp is not None:
+            if not isinstance(timestamp, datetime):
+                raise TypeError(
+                    f"timestamp must be a datetime instance or None, "
+                    f"got {type(timestamp).__name__!r}"
+                )
+            if timestamp.tzinfo is None:
+                raise ValueError(
+                    "timestamp must be timezone-aware (UTC); "
+                    "got a naive datetime with no tzinfo"
+                )
         created_at = (
             timestamp.strftime(_TIMESTAMP_FMT)
             if timestamp is not None
@@ -512,18 +526,26 @@ class AuditService:
 
     @staticmethod
     def _row_to_entry(row: AuditLogModel) -> AuditLogEntry:
-        """Convert a SQLAlchemy model row to a domain data class."""
+        """Convert a SQLAlchemy model row to a domain data class.
+
+        ``AuditLogModel`` uses legacy ``Column()`` declarations with
+        ``__allow_unmapped__ = True``, so Pyright cannot infer the
+        concrete types of the ORM accessor attributes.  ``cast()`` is
+        used here to assert the expected types explicitly, which is the
+        project-standard alternative to ``# type: ignore`` suppressions.
+        """
         try:
-            details = json.loads(row.details) if row.details else {}  # type: ignore[arg-type]
+            details_raw = cast("str | None", row.details)
+            details = json.loads(details_raw) if details_raw else {}
         except (json.JSONDecodeError, TypeError):
             details = {}
         return AuditLogEntry(
-            id=row.id,  # type: ignore[arg-type]
-            event_type=row.event_type,  # type: ignore[arg-type]
-            plan_id=row.plan_id,  # type: ignore[arg-type]
-            project_name=row.project_name,  # type: ignore[arg-type]
-            actor_name=row.actor_name,  # type: ignore[arg-type]
-            user_identity=row.user_identity,  # type: ignore[arg-type]
+            id=cast(int, row.id),
+            event_type=cast(str, row.event_type),
+            plan_id=cast("str | None", row.plan_id),
+            project_name=cast("str | None", row.project_name),
+            actor_name=cast("str | None", row.actor_name),
+            user_identity=cast("str | None", row.user_identity),
             details=details,
-            created_at=row.created_at,  # type: ignore[arg-type]
+            created_at=cast(str, row.created_at),
         )