fix(audit): clamp timestamps forward on clock step-back (AUDIT-001)

src/cmcp_gateway/audit/chain.py

@@ -117,10 +117,15 @@ def append(
         workflow_id: str | None = None,
     ) -> AuditEntry:
         prev_hash = self._entries[-1].entry_hash if self._entries else "genesis"
+        now = datetime.now(tz=UTC)
+        if self._entries:
+            prev_ts = datetime.fromisoformat(self._entries[-1].timestamp_utc)
+            if now < prev_ts:
+                now = prev_ts
         entry = AuditEntry(
             entry_id=str(uuid4()),
             sequence_number=len(self._entries),
-            timestamp_utc=datetime.now(tz=UTC).isoformat(),
+            timestamp_utc=now.isoformat(),
             session_id=self._session_id,
             call_id=call_id,
             entry_type=entry_type,

tests/unit/test_audit.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+from datetime import UTC, datetime, timedelta
+from unittest.mock import patch
+
 from cmcp_gateway.audit.chain import AuditChain
 from cmcp_gateway.audit.keys import SigningKey
 
@@ -119,3 +122,37 @@ def test_audit_chain_entries_returns_copy():
     entries = chain.entries
     entries.clear()
     assert chain.length == 1  # original not affected
+
+
+# ── AUDIT-001: monotonic timestamp guard ──────────────────────────────────────
+
+def test_timestamp_clamped_when_clock_steps_backward():
+    """AUDIT-001: if wall-clock steps back, timestamp is clamped to previous entry's time."""
+    chain = AuditChain("sess-001")
+    first_ts = datetime.fromisoformat(chain.entries[0].timestamp_utc)
+
+    # Simulate a clock that returns a time 5 seconds in the past on the next call
+    backward = first_ts - timedelta(seconds=5)
+    with patch("cmcp_gateway.audit.chain.datetime") as mock_dt:
+        mock_dt.now.return_value = backward
+        mock_dt.fromisoformat = datetime.fromisoformat
+        entry = chain.append("tool_call", call_id="c1", tool_name="t", policy_decision="allow")
+
+    appended_ts = datetime.fromisoformat(entry.timestamp_utc)
+    assert appended_ts >= first_ts, (
+        f"timestamp {appended_ts} must not be before previous entry {first_ts}"
+    )
+
+
+def test_timestamp_not_clamped_when_clock_moves_forward():
+    """AUDIT-001: normal forward-moving timestamps are preserved as-is."""
+    chain = AuditChain("sess-001")
+    first_ts = datetime.fromisoformat(chain.entries[0].timestamp_utc)
+
+    forward = first_ts + timedelta(seconds=10)
+    with patch("cmcp_gateway.audit.chain.datetime") as mock_dt:
+        mock_dt.now.return_value = forward
+        mock_dt.fromisoformat = datetime.fromisoformat
+        entry = chain.append("tool_call", call_id="c1", tool_name="t", policy_decision="allow")
+
+    assert datetime.fromisoformat(entry.timestamp_utc) == forward