fix: replace deprecated datetime.utcnow() with timezone-aware alternative

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app/curator.py

@@ -6,7 +6,7 @@ The prompt determines behavior based on current date.
 """
 import logging
 import os
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import List, Dict, Any, Optional
 from pathlib import Path
 import httpx
@@ -49,7 +49,7 @@ class Curator:
         Otherwise runs daily mode (processes recent 24h only).
         The prompt determines behavior based on current date.
         """
-        current_date = datetime.utcnow()
+        current_date = datetime.now(timezone.utc)
         is_monthly = current_date.day == 1
         mode = "MONTHLY" if is_monthly else "DAILY"
         
@@ -169,7 +169,7 @@ Remember: Respond with ONLY valid JSON. No markdown, no explanations, just the J
             return True
         try:
             mem_time = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
-            cutoff = datetime.utcnow() - timedelta(hours=hours)
+            cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=hours)
             return mem_time.replace(tzinfo=None) > cutoff
         except (ValueError, TypeError):
             logger.debug(f"Could not parse timestamp: {timestamp}")

app/proxy_handler.py

@@ -48,17 +48,17 @@ def debug_log(category: str, message: str, data: dict = None):
     if not config.debug:
         return
     
-    from datetime import datetime
+    from datetime import datetime, timezone
     
     # Create logs directory
     log_dir = DEBUG_LOG_DIR
     log_dir.mkdir(parents=True, exist_ok=True)
     
-    today = datetime.utcnow().strftime("%Y-%m-%d")
+    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
     log_path = log_dir / f"debug_{today}.log"
     
     entry = {
-        "timestamp": datetime.utcnow().isoformat() + "Z",
+        "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
         "category": category,
         "message": message
     }

app/qdrant_service.py

@@ -2,7 +2,7 @@
 from qdrant_client import AsyncQdrantClient
 from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue
 from typing import List, Dict, Any, Optional
-from datetime import datetime
+from datetime import datetime, timezone
 import uuid
 import logging
 import httpx
@@ -54,7 +54,7 @@ class QdrantService:
         point_id = str(uuid.uuid4())
         embedding = await self.get_embedding(content)
         
-        timestamp = datetime.utcnow().isoformat() + "Z"
+        timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
         text = content
         if role == "user":
             text = f"User: {content}"
@@ -85,7 +85,7 @@ class QdrantService:
         """Store a complete Q&A turn as one document."""
         await self._ensure_collection()
         
-        timestamp = datetime.utcnow().isoformat() + "Z"
+        timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
         text = f"User: {user_question}\nAssistant: {assistant_answer}\nTimestamp: {timestamp}"
         
         point_id = str(uuid.uuid4())

app/utils.py

@@ -3,7 +3,7 @@ from .config import config
 import tiktoken
 import os
 from typing import List, Dict, Optional
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
 # Use cl100k_base encoding (GPT-4 compatible)
@@ -56,7 +56,7 @@ def truncate_by_tokens(text: str, max_tokens: int) -> str:
 
 def filter_memories_by_time(memories: List[Dict], hours: int = 24) -> List[Dict]:
     """Filter memories from the last N hours."""
-    cutoff = datetime.utcnow() - timedelta(hours=hours)
+    cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=hours)
     filtered = []
     for mem in memories:
         ts = mem.get("timestamp")

tests/test_curator.py

@@ -2,7 +2,7 @@
 import pytest
 import json
 import os
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -77,14 +77,14 @@ class TestIsRecent:
     def test_memory_within_window(self):
         """Memory timestamped 1 hour ago is recent (within 24h)."""
         curator, _ = make_curator()
-        ts = (datetime.utcnow() - timedelta(hours=1)).isoformat() + "Z"
+        ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=1)).isoformat() + "Z"
         memory = {"timestamp": ts}
         assert curator._is_recent(memory, hours=24) is True
 
     def test_memory_outside_window(self):
         """Memory timestamped 48 hours ago is not recent."""
         curator, _ = make_curator()
-        ts = (datetime.utcnow() - timedelta(hours=48)).isoformat() + "Z"
+        ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=48)).isoformat() + "Z"
         memory = {"timestamp": ts}
         assert curator._is_recent(memory, hours=24) is False
 
@@ -109,7 +109,7 @@ class TestIsRecent:
     def test_boundary_edge_just_inside(self):
         """Memory at exactly hours-1 minutes ago should be recent."""
         curator, _ = make_curator()
-        ts = (datetime.utcnow() - timedelta(hours=23, minutes=59)).isoformat() + "Z"
+        ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=23, minutes=59)).isoformat() + "Z"
         memory = {"timestamp": ts}
         assert curator._is_recent(memory, hours=24) is True
 

tests/test_utils.py

@@ -90,20 +90,20 @@ class TestFilterMemoriesByTime:
 
     def test_includes_recent_memory(self):
         """Memory with timestamp in the last 24h should be included."""
-        from datetime import datetime, timedelta
+        from datetime import datetime, timedelta, timezone
         from app.utils import filter_memories_by_time
 
-        ts = (datetime.utcnow() - timedelta(hours=1)).isoformat()
+        ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=1)).isoformat()
         memories = [{"timestamp": ts, "text": "recent"}]
         result = filter_memories_by_time(memories, hours=24)
         assert len(result) == 1
 
     def test_excludes_old_memory(self):
         """Memory older than cutoff should be excluded."""
-        from datetime import datetime, timedelta
+        from datetime import datetime, timedelta, timezone
         from app.utils import filter_memories_by_time
 
-        ts = (datetime.utcnow() - timedelta(hours=48)).isoformat()
+        ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=48)).isoformat()
         memories = [{"timestamp": ts, "text": "old"}]
         result = filter_memories_by_time(memories, hours=24)
         assert len(result) == 0
@@ -132,10 +132,10 @@ class TestFilterMemoriesByTime:
 
     def test_z_suffix_timestamp(self):
         """ISO timestamp with Z suffix should be handled correctly."""
-        from datetime import datetime, timedelta
+        from datetime import datetime, timedelta, timezone
         from app.utils import filter_memories_by_time
 
-        ts = (datetime.utcnow() - timedelta(hours=1)).isoformat() + "Z"
+        ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=1)).isoformat() + "Z"
         memories = [{"timestamp": ts, "text": "recent with Z"}]
         result = filter_memories_by_time(memories, hours=24)
         assert len(result) == 1