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

AuditService.record() was generating its own timestamp internally,
discarding the original DomainEvent.timestamp. This means audit entries
recorded when an event was audited, not when the domain event actually
occurred, breaking forensic accuracy per §Audit Logging (SEC7).

Changes:
- Add `timestamp: datetime | None = None` keyword parameter to
  AuditService.record(). When provided, uses it as created_at;
  falls back to datetime.now(tz=UTC) for backward compatibility.
  Applied to both the async queue path and the synchronous DB path.
- AuditEventSubscriber._handle_event() now passes timestamp=event.timestamp
  so the original event creation time is preserved in audit entries.
- Add 3 Behave BDD scenarios covering: full pipeline timestamp
  preservation, direct record() with explicit timestamp, and backward
  compatibility (record() without timestamp auto-generates created_at).
- Add preserve_event_timestamp Robot integration test and helper subcommand.
- Add static source check in security_audit.robot verifying the
  timestamp parameter signature exists.

ISSUES CLOSED: #719

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)