# Step 1-4: 写入不同发生时间和有效期的记忆
content_a = "早期事件"
content_b = "中期事件"
content_c = "晚期事件"
content_d = "限时事件"
time_early = datetime(2024, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
time_mid = datetime(2024, 1, 2, 10, 0, 0, tzinfo=timezone.utc)
time_late = datetime(2024, 1, 3, 10, 0, 0, tzinfo=timezone.utc)
# 记忆A、B、C：永久有效（不传invalid_at，默认None→T_INVALID_OPEN）
units_a = aw.api.add(content_a, scope, identity=identity, occurred_at=time_early)
units_b = aw.api.add(content_b, scope, identity=identity, occurred_at=time_mid)
units_c = aw.api.add(content_c, scope, identity=identity, occurred_at=time_late)
# 记忆D：先写入（永久有效），再用update设置固定失效时间
units_d = aw.api.add(content_d, scope, identity=identity, occurred_at=time_late)
time_invalid_d = datetime(2024, 6, 1, 0, 0, 0, tzinfo=timezone.utc)
patch_d = MemoryPatch(t_invalid=time_invalid_d)
units_d_updated = aw.api.update(units_d[0].id, scope, patch_d, identity=identity)
assert len(units_a) == 1, "write应返回1个unit"
assert len(units_b) == 1, "write应返回1个unit"
assert len(units_c) == 1, "write应返回1个unit"
assert len(units_d) == 1, "write应返回1个unit"
# 精确断言：验证write返回的unit.content与写入内容一致
assert units_a[0].content == content_a, "write返回的content不匹配"
assert units_b[0].content == content_b, "write返回的content不匹配"
assert units_c[0].content == content_c, "write返回的content不匹配"
assert units_d[0].content == content_d, "write返回的content不匹配"
# 验证update后记忆D的t_invalid被正确设置
assert units_d_updated.temporal.t_invalid is not None, "update后t_invalid应不为None"
case_logger.info(f"记忆D update后 t_invalid={units_d_updated.temporal.t_invalid}")
case_logger.info(f"记忆A: id={units_a[0].id}, occurred_at=2024-01-01, 永久有效")
case_logger.info(f"记忆B: id={units_b[0].id}, occurred_at=2024-01-02, 永久有效")
case_logger.info(f"记忆C: id={units_c[0].id}, occurred_at=2024-01-03, 永久有效")
case_logger.info(f"记忆D: id={units_d_updated.id}, occurred_at=2024-01-03, t_invalid=2024-06-01")
# ============ Step 5: t_event使用GT（大于）过滤验证 ============
# 使用记忆A、B、C验证t_event GT
# 使用记忆B的t_event时间戳作为阈值，验证GT严格大于语义
t_event_b_ms = int(time_mid.timestamp() * 1000) # 2024-01-02 10:00:00 UTC
result_gt = aw.api.search(
"事件", Context(scope), identity=identity,
filters={"t_event": {"gt": t_event_b_ms}}
)
ids_gt = {item.unit_id for item in result_gt.items}
assert len(ids_gt) == 1
case_logger.info(f"t_event GT过滤结果（阈值=记忆B的t_event）: {ids_gt}")
# 满足分支：t_event > 阈值的记忆（记忆C）
assert units_c[0].id in ids_gt, "t_event GT应返回记忆C（t_event > 阈值）"
# 不满足分支：t_event ≤ 阈值的记忆（记忆A、B）
assert units_b[0].id not in ids_gt, "t_event GT不应返回记忆B（t_event = 阈值，不满足>）"
assert units_a[0].id not in ids_gt, "t_event GT不应返回记忆A（t_event < 阈值）"
case_logger.info("t_event GT过滤验证通过：满足分支和不满足分支都已验证")
# ============ Step 6: t_valid使用LT（小于）过滤验证 ============
# 使用记忆A、B、C验证t_valid LT
# t_valid表示记忆有效期起始时间，默认为写入时间
# 获取记忆C的实际t_valid，使用它作为阈值
# 这样记忆A、B的t_valid < 阈值（满足分支），记忆C的t_valid ≥ 阈值（不满足分支）
t_valid_c = units_c[0].temporal.t_valid
t_valid_c_ms = int(t_valid_c.timestamp() * 1000) if hasattr(t_valid_c, 'timestamp') else t_valid_c
result_lt = aw.api.search(
"事件", Context(scope), identity=identity,
filters={"t_valid": {"lt": t_valid_c_ms}}
)
ids_lt = {item.unit_id for item in result_lt.items}
assert len(ids_lt) == 2
case_logger.info(f"t_valid LT过滤结果（阈值=记忆C的t_valid）: {ids_lt}")
# 满足分支：t_valid < 阈值的记忆
assert units_a[0].id in ids_lt, "t_valid LT应返回记忆A（t_valid < 阈值）"
assert units_b[0].id in ids_lt, "t_valid LT应返回记忆B（t_valid < 阈值）"
# 不满足分支：t_valid ≥ 阈值的记忆
assert units_c[0].id not in ids_lt, "t_valid LT不应返回记忆C（t_valid ≥ 阈值）"
case_logger.info("t_valid LT过滤验证通过：满足分支和不满足分支都已验证")
# ============ Step 7: t_invalid使用GTE（大于等于）过滤验证 ============
# 使用记忆A、B、C（永久有效）和记忆D（固定失效）验证t_invalid GTE
# t_invalid表示记忆有效期结束时间
# 记忆A、B、C：永久有效 → t_invalid = T_INVALID_OPEN（极大值）
# 记忆D：固定失效时间 → t_invalid = 2024-06-01（较小值）
# 使用阈值：2025-01-01
# 满足分支：t_invalid >= 2025-01-01（记忆A、B、C的T_INVALID_OPEN满足）
# 不满足分支：t_invalid < 2025-01-01（记忆D的2024-06-01不满足）
t_invalid_threshold = int(datetime(2025, 1, 1, 0, 0, 0,
tzinfo=timezone.utc).timestamp() * 1000)
result_gte = aw.api.search(
"事件", Context(scope), identity=identity,
filters={"t_invalid": {"gte": t_invalid_threshold}}
)
ids_gte = {item.unit_id for item in result_gte.items}
case_logger.info(f"t_invalid GTE过滤结果（阈值=2025-01-01）: {ids_gte}")
# 满足分支：t_invalid >= 2025-01-01的永久有效记忆
assert units_a[0].id in ids_gte, "永久有效记忆A应被t_invalid GTE召回"
assert units_b[0].id in ids_gte, "永久有效记忆B应被t_invalid GTE召回"
assert units_c[0].id in ids_gte, "永久有效记忆C应被t_invalid GTE召回"
# 不满足分支：t_invalid < 2025-01-01的固定失效记忆
assert units_d_updated.id not in ids_gte, "固定失效记忆D不应被t_invalid GTE召回（t_invalid=2024-06-01 < 2025-01-01）"
case_logger.info("t_invalid GTE过滤验证通过：满足分支和不满足分支都已验证")
case_logger.info("测试通过：t_event GT、t_valid LT、t_invalid GTE验证成功")
Checklist
🐞 问题详细描述
用例描述：
# 1. 写入记忆A（occurred_at=2024-01-01，永久有效）
# 2. 写入记忆B（occurred_at=2024-01-02，永久有效）
# 3. 写入记忆C（occurred_at=2024-01-03，永久有效）
# 4. 写入记忆D（永久有效），update设置固定失效时间2024-06-01
# 5. t_event使用GT过滤（验证满足/不满足分支）
预期结果：t_event过滤成功，返回满足条件的数据
实际结果：未正常返回满足条件的数据
失败截图：
ai分析如下：
用例如下：
-- coding: utf-8 --
"""E2E-S03: 时间字段验证
整合验证t_event、t_valid、t_invalid三个时间字段的过滤功能。
三个字段分别使用GT（大于）、LT（小于）、GTE（大于等于）算子进行过滤验证。
每个算子都验证满足分支和不满足分支。
"""
import pytest
from datetime import datetime, timezone
from jiuwen_memory.common.type_def import Context, Scope, T_INVALID_OPEN
from jiuwen_memory.control.types import MemoryPatch
@pytest.mark.positive
def test_metadata_filter_024(aw, case_logger):
"""
# !+================================================================
# 版权 (C) 2019-2020，华为技术有限公司 2012实验室中央软件院
# ==================================================================
# @CaseID: test_metadata_filter_024
# @description: 时间字段验证，t_event使用GT、t_valid使用LT、t_invalid使用GTE
# @precondition: 部署 agent-memory 项目环境，配置 TEST_* 真实大模型环境变量
# @step:
# 1. 写入记忆A（occurred_at=2024-01-01，永久有效）
# 2. 写入记忆B（occurred_at=2024-01-02，永久有效）
# 3. 写入记忆C（occurred_at=2024-01-03，永久有效）
# 4. 写入记忆D（永久有效），update设置固定失效时间2024-06-01
# 5. t_event使用GT过滤（验证满足/不满足分支）
# 6. t_valid使用LT过滤（验证满足/不满足分支）
# 7. t_invalid使用GTE过滤（验证满足/不满足分支）
# @Result:
# t_event GT、t_valid LT、t_invalid GTE算子语义正确，满足和不满足分支都被验证
# !!================================================================
"""
identity = Scope(org="test_org", user="test_user", agent="test_agent", session="filter024")
scope = Scope(org="test_org", user="test_user", agent="test_agent", session="filter024")
case_logger.info("MemoryApiAW 就绪（aw fixture）")
详细的环境信息描述
windows
其他辅助信息
版本信息
感谢您的贡献 🎉!