fix(vikingdb): normalize created_at filters in API key client

docs/design/openviking-usage-count-record-implementation-plan.md

@@ -31,7 +31,7 @@
 
 ```json
 {
-  "event_time": "2026-08-05 11:30:00",
+  "event_time": "2026-08-05T11:30:00Z",
   "tenant_id": "resource_id:ov-resource-id;account_id:2101858484;user_id:user-1;resource_uri:viking://user/user-1/memories/experiences/exchange.md",
   "event_name": "experience.recall.count",
   "object_id": "ue_recall",

docs/design/openviking-usage-reporter-sink-design.md

@@ -154,7 +154,7 @@ HTTP 请求。日志文件使用 UTC 小时滚动，默认保留 168 个小时
 一行：
 
 ```json
-{"event_time":"2026-08-05 11:30:00","tenant_id":"resource_id:ov-xxx;account_id:new;user_id:test;resource_uri:viking://user/test/memories/experiences/example.md","event_name":"experience.recall.count","object_id":"ue_<sha256>","count":1,"tags":{"resource_type":"experience"}}
+{"event_time":"2026-08-05T11:30:00Z","tenant_id":"resource_id:ov-xxx;account_id:new;user_id:test;resource_uri:viking://user/test/memories/experiences/example.md","event_name":"experience.recall.count","object_id":"ue_<sha256>","count":1,"tags":{"resource_type":"experience"}}
 ```
 
 字段映射：

docs/en/guides/01-configuration.md

@@ -1603,10 +1603,10 @@ Set the environment variable named by `resource_id_env` before starting the serv
 Each line has the following form:
 
 ```json
-{"event_time":"2026-08-05 11:30:00","tenant_id":"resource_id:ov-example;account_id:default;user_id:default;resource_uri:viking://user/default/memories/experiences/example.md","event_name":"experience.recall.count","object_id":"ue_<sha256>","count":1,"tags":{"resource_type":"experience"}}
+{"event_time":"2026-08-05T11:30:00Z","tenant_id":"resource_id:ov-example;account_id:default;user_id:default;resource_uri:viking://user/default/memories/experiences/example.md","event_name":"experience.recall.count","object_id":"ue_<sha256>","count":1,"tags":{"resource_type":"experience"}}
 ```
 
-`event_time` is UTC. `tenant_id` combines the deployment resource ID, event account, user, and Experience URI. `memory.recalled` maps to `experience.recall.count`, while `memory.injected` maps to `experience.inject.count`. `object_id` is the stable Usage Event ID. Downstream consumers must deduplicate by the composite `(tenant_id, object_id)` key rather than by `object_id` globally. Aggregate usage with `sum(count)` after filtering by `tenant_id`, `event_name`, and the desired `event_time` range. File collection and downstream delivery remain best-effort.
+`event_time` is an RFC 3339 UTC timestamp with an explicit `Z` suffix. `tenant_id` combines the deployment resource ID, event account, user, and Experience URI. `memory.recalled` maps to `experience.recall.count`, while `memory.injected` maps to `experience.inject.count`. `object_id` is the stable Usage Event ID. Downstream consumers must deduplicate by the composite `(tenant_id, object_id)` key rather than by `object_id` globally. Aggregate usage with `sum(count)` after filtering by `tenant_id`, `event_name`, and the desired `event_time` range. File collection and downstream delivery remain best-effort.
 
 Supported add target URIs:
 

docs/zh/guides/01-configuration.md

@@ -1684,10 +1684,10 @@ openviking add-resource ./docs --exclude "*.tmp"
 每行格式如下：
 
 ```json
-{"event_time":"2026-08-05 11:30:00","tenant_id":"resource_id:ov-example;account_id:default;user_id:default;resource_uri:viking://user/default/memories/experiences/example.md","event_name":"experience.recall.count","object_id":"ue_<sha256>","count":1,"tags":{"resource_type":"experience"}}
+{"event_time":"2026-08-05T11:30:00Z","tenant_id":"resource_id:ov-example;account_id:default;user_id:default;resource_uri:viking://user/default/memories/experiences/example.md","event_name":"experience.recall.count","object_id":"ue_<sha256>","count":1,"tags":{"resource_type":"experience"}}
 ```
 
-`event_time` 使用 UTC 时间。`tenant_id` 由部署 resource ID、事件所属的 account、user 和 Experience URI 拼接。`memory.recalled` 映射为 `experience.recall.count`，`memory.injected` 映射为 `experience.inject.count`。`object_id` 是稳定的 Usage Event ID。下游必须使用 `(tenant_id, object_id)` 复合键去重，不能跨 tenant 仅按 `object_id` 全局去重。查询时按 `tenant_id`、`event_name` 和 `event_time` 范围过滤，再通过 `sum(count)` 汇总。文件采集和下游投递仍为 best-effort。
+`event_time` 使用 RFC 3339 UTC 时间格式，并显式携带 `Z` 后缀。`tenant_id` 由部署 resource ID、事件所属的 account、user 和 Experience URI 拼接。`memory.recalled` 映射为 `experience.recall.count`，`memory.injected` 映射为 `experience.inject.count`。`object_id` 是稳定的 Usage Event ID。下游必须使用 `(tenant_id, object_id)` 复合键去重，不能跨 tenant 仅按 `object_id` 全局去重。查询时按 `tenant_id`、`event_name` 和 `event_time` 范围过滤，再通过 `sum(count)` 汇总。文件采集和下游投递仍为 best-effort。
 
 支持的 add target URI：
 

openviking/storage/vectordb/collection/volcengine_api_key_collection.py

@@ -150,8 +150,22 @@ def _sanitize_list_payload(cls, obj: List[Any]) -> List[Any]:
                 sanitized_list.append(y)
         return sanitized_list
 
+    @classmethod
+    def _normalize_date_time_filter(cls, obj: Any) -> Any:
+        if isinstance(obj, list):
+            return [cls._normalize_date_time_filter(item) for item in obj]
+        if not isinstance(obj, dict):
+            return obj
+
+        normalized = {key: cls._normalize_date_time_filter(value) for key, value in obj.items()}
+        if normalized.get("op") == "range" and normalized.get("field") == "created_at":
+            normalized["op"] = "time_range"
+        return normalized
+
     def _data_post(self, path: str, data: Dict[str, Any]):
         safe_data = self._sanitize_payload(data)
+        if isinstance(safe_data, dict) and "filter" in safe_data:
+            safe_data["filter"] = self._normalize_date_time_filter(safe_data["filter"])
         response = self.data_client.do_req("POST", path, req_body=safe_data)
         if response.status_code != 200:
             raise self._build_response_error(response, path)

openviking/usage_reporter/file_log_sink.py

@@ -123,7 +123,7 @@ def _format_event_time(value: str) -> str:
     parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
     if parsed.tzinfo is None:
         parsed = parsed.replace(tzinfo=timezone.utc)
-    return parsed.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+    return parsed.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
 
 
 def _build_tenant_id(

tests/storage/test_volcengine_clients.py

@@ -373,6 +373,64 @@ def _fake_do_req(method, req_path=None, req_params=None, req_body=None):
     }
 
 
+def test_volcengine_api_key_collection_uses_date_time_filter_operator(monkeypatch):
+    captured = {}
+
+    class _Response:
+        status_code = 200
+
+        @staticmethod
+        def json():
+            return {"result": {"agg": {"_total": 1}}}
+
+    from openviking.storage.vectordb.collection.volcengine_api_key_collection import (
+        VolcengineApiKeyCollection,
+    )
+
+    collection = VolcengineApiKeyCollection(
+        api_key="vk-test-token",
+        region="cn-beijing",
+        meta_data={"ProjectName": "default", "CollectionName": "context", "IndexName": "default"},
+    )
+
+    def _fake_do_req(method, req_path=None, req_params=None, req_body=None):
+        captured["path"] = req_path
+        captured["req_body"] = req_body
+        return _Response()
+
+    monkeypatch.setattr(collection.data_client, "do_req", _fake_do_req)
+
+    collection.aggregate_data(
+        index_name="default",
+        filters={
+            "op": "and",
+            "conds": [
+                {
+                    "op": "range",
+                    "field": "created_at",
+                    "gte": "2026-08-10T00:00:00+00:00",
+                    "lt": "2026-08-11T00:00:00+00:00",
+                },
+                {"op": "range", "field": "level", "gte": 1},
+            ],
+        },
+    )
+
+    assert captured["path"] == "/api/vikingdb/data/agg"
+    assert captured["req_body"]["filter"] == {
+        "op": "and",
+        "conds": [
+            {
+                "op": "time_range",
+                "field": "created_at",
+                "gte": "2026-08-10T00:00:00+00:00",
+                "lt": "2026-08-11T00:00:00+00:00",
+            },
+            {"op": "range", "field": "level", "gte": 1},
+        ],
+    }
+
+
 def test_volcengine_adapter_update_data_returns_ids():
     adapter = VolcengineCollectionAdapter(
         ak="test-ak",

tests/unit/usage_reporter/test_file_log_sink.py

@@ -18,6 +18,7 @@ def __init__(
         event_id: str = "ue_recall",
         event_type: str = "memory.recalled",
         session_id: str = "session-1",
+        occurred_at: str = "2026-08-05T19:30:00+08:00",
         resource_uri: str = ("viking://user/default/memories/experiences/生成请假邮件通用模版.md"),
     ) -> None:
         self._record = {
@@ -28,7 +29,7 @@ def __init__(
             "user_id": "default",
             "session_id": session_id,
             "task_id": "task-1",
-            "occurred_at": "2026-08-05T19:30:00+08:00",
+            "occurred_at": occurred_at,
             "resource_uri": resource_uri,
             "resource_type": "experience",
             "evidence": {
@@ -66,7 +67,7 @@ async def test_file_log_sink_writes_usage_event_record(tmp_path):
     lines = log_path.read_text(encoding="utf-8").splitlines()
     assert len(lines) == 1
     assert _parse_line(lines[0]) == {
-        "event_time": "2026-08-05 11:30:00",
+        "event_time": "2026-08-05T11:30:00Z",
         "tenant_id": (
             "resource_id:ov-test;account_id:default;user_id:default;resource_uri:"
             "viking://user/default/memories/experiences/生成请假邮件通用模版.md"
@@ -78,6 +79,28 @@ async def test_file_log_sink_writes_usage_event_record(tmp_path):
     }
 
 
+@pytest.mark.parametrize(
+    ("occurred_at", "expected"),
+    [
+        ("2026-08-05T11:30:00Z", "2026-08-05T11:30:00Z"),
+        ("2026-08-05T11:30:00", "2026-08-05T11:30:00Z"),
+        ("2026-08-05T07:30:00-04:00", "2026-08-05T11:30:00Z"),
+    ],
+)
+@pytest.mark.asyncio
+async def test_file_log_sink_emits_rfc3339_utc_event_time(tmp_path, occurred_at, expected):
+    log_path = tmp_path / "usage.log"
+    sink = FileLogUsageSink(path=log_path)
+
+    try:
+        await sink.write(events=[FakeUsageEvent(occurred_at=occurred_at)])
+    finally:
+        sink.close()
+
+    record = _parse_line(log_path.read_text(encoding="utf-8").strip())
+    assert record["event_time"] == expected
+
+
 @pytest.mark.asyncio
 async def test_file_log_sink_maps_injection_event_name(tmp_path):
     log_path = tmp_path / "usage.log"