harness_zbt/backend/tests/test_agent_memory.py

from datetime import UTC, datetime

import pytest

from zbt.core.errors import AppError
from zbt.domains.agent.memory import (
    CustomerMemoryService,
    InMemoryCustomerMemoryIndex,
    InMemoryCustomerMemoryRepository,
)

NOW = datetime(2026, 7, 28, 8, 0, tzinfo=UTC)


def test_customer_memory_is_semantically_recalled_only_for_its_owner() -> None:
    service = CustomerMemoryService(
        InMemoryCustomerMemoryRepository(),
        InMemoryCustomerMemoryIndex(),
        lambda: NOW,
    )
    saved = service.save(
        owner_id="customer-a",
        category="PREFERENCE",
        content="我优先考虑给父母配置医疗险",
    )

    recalled = service.search(
        owner_id="customer-a",
        query="父母适合买什么医疗保险",
        limit=3,
    )
    other_customer = service.search(
        owner_id="customer-b",
        query="父母适合买什么医疗保险",
        limit=3,
    )

    assert recalled["items"][0]["memory_id"] == saved["memory_id"]
    assert recalled["items"][0]["content"] == "我优先考虑给父母配置医疗险"
    assert other_customer["items"] == []


def test_customer_memory_rejects_sensitive_identifiers() -> None:
    service = CustomerMemoryService(
        InMemoryCustomerMemoryRepository(),
        InMemoryCustomerMemoryIndex(),
        lambda: NOW,
    )

    with pytest.raises(AppError) as error:
        service.save(
            owner_id="customer-a",
            category="PREFERENCE",
            content="请记住我的手机号是13800138000",
        )

    assert error.value.code == "CUSTOMER_MEMORY_SENSITIVE"