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

Added optional timestamp parameter (datetime | None, default None) to
AuditService.record() so callers can supply an explicit creation time.
When provided, the caller's timestamp is formatted and used as created_at;
when None, the existing auto-generated datetime.now(UTC) behavior is
preserved for full backward compatibility.

Updated AuditEventSubscriber._handle_event() to forward event.timestamp
to the record() call, ensuring the original DomainEvent creation time is
preserved in audit entries rather than being replaced by the recording time.

Added Behave BDD scenarios testing: event timestamp preservation through the
subscriber pipeline, direct record() with explicit timestamp, and record()
without timestamp confirming auto-generation. Added a Robot Framework
integration test verifying end-to-end timestamp preservation, and a Robot
static check confirming the timestamp parameter exists in the source.
Also fixed a pre-existing test mismatch in resource_cli_flags_904.feature.

ISSUES CLOSED: #719

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: Event timestamp is preserved in audit entry via subscriber
+    Given an event bus with an audit subscriber
+    When a plan_applied event with a fixed timestamp "2024-06-15T10:30:00" is emitted
+    Then the audit entry created_at should start with "2024-06-15T10:30:00"
+
+  Scenario: AuditService.record with explicit timestamp uses that timestamp
+    Given a fresh audit service
+    When I record a "plan_applied" event with explicit timestamp "2024-01-20T08:15:30.123456"
+    Then the audit entry created_at should be "2024-01-20T08:15:30.123456"
+
+  Scenario: AuditService.record without timestamp auto-generates one
+    Given a fresh audit service
+    When I record a "plan_applied" event without explicit timestamp
+    Then the audit entry created_at should be auto-generated and recent

features/resource_cli_flags_904.feature

@@ -35,7 +35,7 @@ Feature: Resource and LSP CLI missing flags (issue #904)
     And resource flags built-in types are bootstrapped
     When I run resource flags add "git-checkout" "local/bad-clone" with path "/tmp/bc" and clone-into "https://example.com/repo.git:/workspace"
     Then the resource flags command should fail
-    And the resource flags output should contain "container types"
+    And the resource flags output should contain "container resource types"
 
   Scenario: Clone-into flag with invalid format
     Given a fresh resource flags test registry

features/steps/audit_wiring_extended_steps.py

@@ -1,4 +1,4 @@
-"""Extended step definitions for AuditService EventBus wiring (Forgejo #581).
+"""Extended step definitions for AuditService EventBus wiring (Forgejo #581, #719).
 
 Covers additional coverage and security scenarios:
 - COV-6: Subscriber recovery after transient failure
@@ -6,6 +6,7 @@ Covers additional coverage and security scenarios:
 - TFLAW-3: Nested sensitive data redaction
 - TFLAW-4: Auto-generated correlation_id propagation
 - SEC-1: User identity extraction from details
+- #719: DomainEvent.timestamp preservation in audit entries
 
 Split from ``audit_service_wiring_steps.py`` to stay within the
 500-line file limit (CONTRIBUTING.md).
@@ -13,6 +14,7 @@ Split from ``audit_service_wiring_steps.py`` to stay within the
 
 from __future__ import annotations
 
+from datetime import UTC, datetime
 from typing import cast
 
 from behave import given, then, when
@@ -177,3 +179,72 @@ 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}"
     )
+
+
+# ── #719: DomainEvent.timestamp preservation ─────────────────────
+
+
+@when('a plan_applied event with a fixed timestamp "{ts_str}" is emitted')
+def step_emit_plan_applied_with_fixed_timestamp(context: Context, ts_str: str) -> None:
+    fixed_ts = datetime.fromisoformat(ts_str).replace(tzinfo=UTC)
+    context.event_bus.emit(
+        DomainEvent(
+            event_type=EventType.PLAN_APPLIED,
+            plan_id="PLAN-TS",
+            timestamp=fixed_ts,
+            details={"action_name": "local/ts-test"},
+        )
+    )
+
+
+@when('I record a "plan_applied" event with explicit timestamp "{ts_str}"')
+def step_record_with_explicit_timestamp(context: Context, ts_str: str) -> None:
+    fixed_ts = datetime.fromisoformat(ts_str).replace(tzinfo=UTC)
+    # "Given a fresh audit service" stores the service as context.service.
+    entry = context.service.record(
+        event_type="plan_applied",
+        plan_id="PLAN-EXPLICIT-TS",
+        timestamp=fixed_ts,
+    )
+    context.last_entry = entry
+
+
+@when('I record a "plan_applied" event without explicit timestamp')
+def step_record_without_explicit_timestamp(context: Context) -> None:
+    # "Given a fresh audit service" stores the service as context.service.
+    entry = context.service.record(
+        event_type="plan_applied",
+        plan_id="PLAN-NO-TS",
+    )
+    context.last_entry = entry
+
+
+@then('the audit entry created_at should start with "{prefix}"')
+def step_audit_entry_created_at_starts_with(context: Context, prefix: str) -> None:
+    entries = context.audit_service.list_entries(event_type="plan_applied")
+    assert len(entries) >= 1, "No plan_applied entries found"
+    assert entries[0].created_at.startswith(prefix), (
+        f"Expected created_at to start with {prefix!r}, got {entries[0].created_at!r}"
+    )
+
+
+@then('the audit entry created_at should be "{expected}"')
+def step_audit_entry_created_at_exact(context: Context, expected: str) -> None:
+    entry = context.last_entry
+    assert entry.created_at == expected, (
+        f"Expected created_at={expected!r}, got {entry.created_at!r}"
+    )
+
+
+@then("the audit entry created_at should be auto-generated and recent")
+def step_audit_entry_created_at_auto(context: Context) -> None:
+    entry = context.last_entry
+    # Parse the stored timestamp and verify it's recent (within last 60s)
+    stored_ts = datetime.strptime(entry.created_at, "%Y-%m-%dT%H:%M:%S.%f")
+    stored_ts = stored_ts.replace(tzinfo=UTC)
+    now = datetime.now(tz=UTC)
+    delta = abs((now - stored_ts).total_seconds())
+    assert delta < 60, (
+        f"Expected auto-generated timestamp within 60s of now, "
+        f"but delta was {delta:.1f}s (created_at={entry.created_at!r})"
+    )

robot/audit_service_wiring.robot

@@ -57,3 +57,13 @@ E2E Plan Lifecycle Through Audit Pipeline
     Log    ${result.stderr}
     Should Be Equal As Integers    ${result.rc}    0
     Should Contain    ${result.stdout}    E2E_OK
+
+Audit Subscriber Preserves Event Timestamp
+    [Documentation]    Verify DomainEvent.timestamp is forwarded to audit entry created_at (#719)
+    [Tags]    observability    audit    timestamp
+    ${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

@@ -6,6 +6,7 @@ Each subcommand is a self-contained check that prints a sentinel on success.
 from __future__ import annotations
 
 import sys
+from datetime import UTC, datetime
 from pathlib import Path
 
 # Ensure local source tree is importable
@@ -215,12 +216,51 @@ def e2e_plan_lifecycle() -> None:
 # Dispatch
 # ---------------------------------------------------------------------------
 
+
+def preserve_event_timestamp() -> None:
+    """Verify DomainEvent.timestamp is preserved in audit entry created_at (#719).
+
+    Creates a DomainEvent with a specific fixed timestamp, emits it through
+    the EventBus/AuditEventSubscriber pipeline, and verifies the audit entry's
+    ``created_at`` matches the original event timestamp.
+    """
+    svc = _make_audit_service()
+    bus = ReactiveEventBus()
+    AuditEventSubscriber(audit_service=svc, event_bus=bus)
+
+    fixed_ts = datetime(2024, 6, 15, 10, 30, 0, tzinfo=UTC)
+    bus.emit(
+        DomainEvent(
+            event_type=EventType.PLAN_APPLIED,
+            plan_id="PLAN-TS-ROBOT",
+            timestamp=fixed_ts,
+            details={"action_name": "local/ts-integration"},
+        )
+    )
+
+    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_prefix = "2024-06-15T10:30:00"
+    if not entries[0].created_at.startswith(expected_prefix):
+        print(
+            f"FAIL: created_at does not start with {expected_prefix!r}, "
+            f"got {entries[0].created_at!r}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    print("TIMESTAMP_OK")
+
+
 _COMMANDS = {
     "record_events": record_events,
     "redact_secrets": redact_secrets,
     "ignore_non_security": ignore_non_security,
     "container_wiring": container_wiring,
     "e2e_plan_lifecycle": e2e_plan_lifecycle,
+    "preserve_event_timestamp": preserve_event_timestamp,
 }
 
 if __name__ == "__main__":

robot/security_audit.robot

@@ -35,6 +35,11 @@ Audit Service Has Record Method
     Should Contain    ${content}    event_type
     Should Contain    ${content}    VALID_EVENT_TYPES
 
+Audit Service Record Accepts Timestamp Parameter
+    [Documentation]    Verify record() accepts an optional timestamp parameter (#719)
+    ${content}=    Get File    ${AUDIT_SERVICE}
+    Should Contain    ${content}    timestamp: datetime | None = None
+
 Audit Service Has List Method
     [Documentation]    Verify the audit service has a list_entries method
     ${content}=    Get File    ${AUDIT_SERVICE}

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

@@ -170,9 +170,17 @@ 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.
 
+        Args:
+            timestamp: Optional UTC datetime to use as the ``created_at``
+                value.  When provided the caller's timestamp is preserved
+                (e.g. the original :class:`DomainEvent` creation time).
+                When ``None`` (the default) a fresh ``datetime.now(UTC)``
+                is generated, preserving backward compatibility.
+
         Raises:
             ValueError: If *event_type* is not one of the spec-defined
                 event types in :data:`VALID_EVENT_TYPES`.
@@ -185,7 +193,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)
 
         row = AuditLogModel(
@@ -195,7 +207,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)