features/observability/audit_service_wiring.feature

@@ -133,3 +133,18 @@ Feature: AuditService EventBus Wiring
     Given an event bus with an audit subscriber
     When a plan_applied event is emitted
     Then the audit entry user_identity should be None
+
+  Scenario: Event timestamp is preserved in audit entry via subscriber
+    Given an event bus with an audit subscriber
+    When a plan_applied event with a known timestamp is emitted
+    Then the audit entry created_at should match the event timestamp
+
+  Scenario: AuditService.record with explicit timestamp
+    Given a direct audit service instance
+    When record is called with an explicit timestamp
+    Then the audit entry created_at should equal the explicit timestamp
+
+  Scenario: AuditService.record without timestamp auto-generates created_at
+    Given a direct audit service instance
+    When record is called without a timestamp
+    Then the audit entry created_at should be close to now

features/steps/audit_wiring_extended_steps.py

@@ -13,6 +13,7 @@ Split from ``audit_service_wiring_steps.py`` to stay within the
 
 from __future__ import annotations
 
+from datetime import UTC, datetime, timedelta
 from typing import cast
 
 from behave import given, then, when
@@ -221,3 +222,76 @@ def step_audit_entry_user_identity_none(context: Context) -> None:
     assert entries[0].user_identity is None, (
         f"Expected user_identity=None, got {entries[0].user_identity!r}"
     )
+
+
+# ── #719: DomainEvent.timestamp preservation ─────────────────────
+
+
+@when("a plan_applied event with a known timestamp is emitted")
+def step_emit_plan_applied_with_known_timestamp(context: Context) -> None:
+    context.known_timestamp = datetime(2024, 6, 15, 10, 30, 0, tzinfo=UTC)
+    context.event_bus.emit(
+        DomainEvent(
+            event_type=EventType.PLAN_APPLIED,
+            plan_id="PLAN-TS",
+            timestamp=context.known_timestamp,
+        )
+    )
+
+
+@then("the audit entry created_at should match the event timestamp")
+def step_audit_entry_matches_event_timestamp(context: Context) -> None:
+    entries = context.audit_service.list_entries(event_type="plan_applied")
+    assert len(entries) >= 1, "No plan_applied entries found"
+    expected = context.known_timestamp.strftime("%Y-%m-%dT%H:%M:%S.%f")
+    assert entries[0].created_at == expected, (
+        f"Expected created_at={expected!r}, got {entries[0].created_at!r}"
+    )
+
+
+@given("a direct audit service instance")
+def step_direct_audit_service(context: Context) -> None:
+    context.audit_service = _fresh_audit_service()
+
+
+@when("record is called with an explicit timestamp")
+def step_record_with_explicit_timestamp(context: Context) -> None:
+    context.explicit_timestamp = datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)
+    context.audit_service.record(
+        event_type="plan_applied",
+        plan_id="PLAN-EXPLICIT-TS",
+        timestamp=context.explicit_timestamp,
+    )
+
+
+@then("the audit entry created_at should equal the explicit timestamp")
+def step_audit_entry_equals_explicit_timestamp(context: Context) -> None:
+    entries = context.audit_service.list_entries(event_type="plan_applied")
+    assert len(entries) >= 1, "No plan_applied entries found"
+    expected = context.explicit_timestamp.strftime("%Y-%m-%dT%H:%M:%S.%f")
+    assert entries[0].created_at == expected, (
+        f"Expected created_at={expected!r}, got {entries[0].created_at!r}"
+    )
+
+
+@when("record is called without a timestamp")
+def step_record_without_timestamp(context: Context) -> None:
+    context.before_record = datetime.now(tz=UTC)
+    context.audit_service.record(
+        event_type="plan_applied",
+        plan_id="PLAN-AUTO-TS",
+    )
+    context.after_record = datetime.now(tz=UTC)
+
+
+@then("the audit entry created_at should be close to now")
+def step_audit_entry_close_to_now(context: Context) -> None:
+    entries = context.audit_service.list_entries(event_type="plan_applied")
+    assert len(entries) >= 1, "No plan_applied entries found"
+    recorded = datetime.fromisoformat(entries[0].created_at).replace(tzinfo=UTC)
+    tolerance = timedelta(seconds=60)
+    lower = context.before_record - tolerance
+    upper = context.after_record + tolerance
+    assert lower <= recorded <= upper, (
+        f"created_at={recorded!r} not within 60s of call time"
+    )

robot/audit_service_wiring.robot

@@ -96,3 +96,13 @@ Auth Middleware Emits Audit Events
     Log    ${result.stderr}
     Should Be Equal As Integers    ${result.rc}    0
     Should Contain    ${result.stdout}    AUTH_PIPELINE_OK
+
+Audit Subscriber Preserves Event Timestamp
+    [Documentation]    DomainEvent.timestamp propagates to AuditLogEntry.created_at via AuditEventSubscriber
+    [Tags]    observability    audit
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    preserve_event_timestamp
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    TIMESTAMP_OK

robot/helper_audit_wiring.py

@@ -326,6 +326,38 @@ def user_identity_field_precedence() -> None:
     print("IDENTITY_PRECEDENCE_OK")
 
 
+def preserve_event_timestamp() -> None:
+    """Verify DomainEvent.timestamp is preserved in the audit entry created_at."""
+    from datetime import UTC, datetime
+
+    svc = _make_audit_service()
+    bus = ReactiveEventBus()
+    AuditEventSubscriber(audit_service=svc, event_bus=bus)
+
+    known_ts = datetime(2024, 6, 15, 10, 30, 0, tzinfo=UTC)
+    bus.emit(
+        DomainEvent(
+            event_type=EventType.PLAN_APPLIED,
+            plan_id="PLAN-TS-INT",
+            timestamp=known_ts,
+        )
+    )
+
+    entries = svc.list_entries(event_type="plan_applied")
+    if not entries:
+        print("FAIL: no plan_applied entry found", file=sys.stderr)
+        sys.exit(1)
+
+    expected = known_ts.strftime("%Y-%m-%dT%H:%M:%S.%f")
+    if entries[0].created_at != expected:
+        print(
+            f"FAIL: created_at={entries[0].created_at!r}, expected {expected!r}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    print("TIMESTAMP_OK")
+
+
 def auth_middleware_pipeline() -> None:
     """E2E: TokenAuthMiddleware -> EventBus -> AuditEventSubscriber -> DB."""
     svc = _make_audit_service()
@@ -435,6 +467,7 @@ _COMMANDS = {
     "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,
     "auth_middleware_pipeline": auth_middleware_pipeline,
 }
 

robot/security_audit.robot

@@ -92,4 +92,9 @@ Audit Documentation Exists
     Should Contain    ${content}    agents audit list
     Should Contain    ${content}    agents audit show
     Should Contain    ${content}    agents audit prune
+
+Audit Service Record Accepts Timestamp Parameter
+    [Documentation]    Verify AuditService.record() has a timestamp parameter (issue #719)
+    ${content}=    Get File    ${AUDIT_SERVICE}
+    Should Contain    ${content}    timestamp: datetime | None = None
     Should Contain    ${content}    audit_retention_days

src/cleveragents/application/services/audit_event_subscriber.py

@@ -133,6 +133,7 @@ class AuditEventSubscriber:
                 actor_name=event.actor_name,
                 user_identity=redacted_identity,
                 details=redacted_details,
+                timestamp=event.timestamp,
             )
         except Exception as exc:
             _logger.warning(

src/cleveragents/application/services/audit_service.py

@@ -340,6 +340,7 @@ class AuditService:
         actor_name: str | None = None,
         user_identity: str | None = None,
         details: dict[str, Any] | None = None,
+        timestamp: datetime | None = None,
     ) -> AuditLogEntry:
         """Write an audit event to the database.
 
@@ -352,6 +353,12 @@ class AuditService:
         is written and committed before this method returns, and the
         returned entry has the real database-assigned ``id``.
 
+        Args:
+            timestamp: Optional UTC datetime to use as ``created_at``.
+                When provided, the audit entry records the original event
+                creation time rather than the time ``record()`` was called.
+                Defaults to ``datetime.now(tz=UTC)`` when ``None``.
+
         Raises:
             ValueError: If *event_type* is not one of the spec-defined
                 event types in :data:`VALID_EVENT_TYPES`.
@@ -365,7 +372,11 @@ class AuditService:
                 f"Unknown event_type {event_type!r}; "
                 f"expected one of {sorted(VALID_EVENT_TYPES)}"
             )
-        now = datetime.now(tz=UTC).strftime(_TIMESTAMP_FMT)
+        created_at = (
+            timestamp.strftime(_TIMESTAMP_FMT)
+            if timestamp is not None
+            else datetime.now(tz=UTC).strftime(_TIMESTAMP_FMT)
+        )
         details_json = json.dumps(details or {}, default=str)
 
         if self._async_mode and self._queue is not None:
@@ -376,7 +387,7 @@ class AuditService:
                 actor_name=actor_name,
                 user_identity=user_identity,
                 details_json=details_json,
-                created_at=now,
+                created_at=created_at,
             )
             self._queue.put(payload)
             # Return a placeholder entry — id is not yet known.
@@ -388,7 +399,7 @@ class AuditService:
                 actor_name=actor_name,
                 user_identity=user_identity,
                 details=details or {},
-                created_at=now,
+                created_at=created_at,
             )
 
         # Synchronous path (original behaviour).
@@ -399,7 +410,7 @@ class AuditService:
             actor_name=actor_name,
             user_identity=user_identity,
             details=details_json,
-            created_at=now,
+            created_at=created_at,
         )
         session = self._ensure_session()
         session.add(row)