ATOCore/tests/test_memory.py

"""Tests for Memory Core."""

import os
import tempfile

import pytest

import atocore.config as _config
from atocore.models.database import init_db


@pytest.fixture(autouse=True)
def isolated_db():
    """Give each test a completely isolated database."""
    tmpdir = tempfile.mkdtemp()
    os.environ["ATOCORE_DATA_DIR"] = tmpdir

    # Replace the global settings so all modules see the new data_dir
    _config.settings = _config.Settings()

    # Also reset any module-level references to the old settings
    import atocore.models.database
    # database.py now uses _config.settings dynamically, so no patch needed

    init_db()
    yield tmpdir


def test_create_memory(isolated_db):
    from atocore.memory.service import create_memory
    mem = create_memory("identity", "User is a mechanical engineer specializing in optics")
    assert mem.memory_type == "identity"
    assert mem.status == "active"
    assert mem.confidence == 1.0


def test_create_memory_invalid_type(isolated_db):
    from atocore.memory.service import create_memory
    with pytest.raises(ValueError, match="Invalid memory type"):
        create_memory("invalid_type", "some content")


def test_create_memory_dedup(isolated_db):
    from atocore.memory.service import create_memory
    m1 = create_memory("identity", "User is an engineer")
    m2 = create_memory("identity", "User is an engineer")
    assert m1.id == m2.id


def test_create_memory_dedup_is_project_scoped(isolated_db):
    from atocore.memory.service import create_memory
    m1 = create_memory("project", "Uses SQLite for local state", project="atocore")
    m2 = create_memory("project", "Uses SQLite for local state", project="openclaw")
    assert m1.id != m2.id


def test_project_is_persisted_and_filterable(isolated_db):
    from atocore.memory.service import create_memory, get_memories
    create_memory("project", "Uses SQLite for local state", project="atocore")
    create_memory("project", "Uses Postgres in production", project="openclaw")

    atocore_memories = get_memories(memory_type="project", project="atocore")
    assert len(atocore_memories) == 1
    assert atocore_memories[0].project == "atocore"


def test_get_memories_all(isolated_db):
    from atocore.memory.service import create_memory, get_memories
    create_memory("identity", "User is an engineer")
    create_memory("preference", "Prefers Python with type hints")
    create_memory("knowledge", "Zerodur has near-zero thermal expansion")

    mems = get_memories()
    assert len(mems) == 3


def test_get_memories_by_type(isolated_db):
    from atocore.memory.service import create_memory, get_memories
    create_memory("identity", "User is an engineer")
    create_memory("preference", "Prefers concise code")
    create_memory("preference", "Uses FastAPI for APIs")

    mems = get_memories(memory_type="preference")
    assert len(mems) == 2


def test_get_memories_active_only(isolated_db):
    from atocore.memory.service import create_memory, get_memories, invalidate_memory
    m = create_memory("knowledge", "Fact about optics")
    invalidate_memory(m.id)

    assert len(get_memories(active_only=True)) == 0
    assert len(get_memories(active_only=False)) == 1


def test_get_memories_min_confidence(isolated_db):
    from atocore.memory.service import create_memory, get_memories
    create_memory("knowledge", "High confidence fact", confidence=0.9)
    create_memory("knowledge", "Low confidence fact", confidence=0.3)

    high = get_memories(min_confidence=0.5)
    assert len(high) == 1
    assert high[0].confidence == 0.9


def test_update_memory(isolated_db):
    from atocore.memory.service import create_memory, get_memories, update_memory
    mem = create_memory("knowledge", "Initial fact")
    update_memory(mem.id, content="Updated fact", confidence=0.8)

    mems = get_memories()
    assert len(mems) == 1
    assert mems[0].content == "Updated fact"
    assert mems[0].confidence == 0.8


def test_update_memory_rejects_duplicate_active_memory(isolated_db):
    from atocore.memory.service import create_memory, update_memory
    import pytest

    first = create_memory("knowledge", "Canonical fact", project="atocore")
    second = create_memory("knowledge", "Different fact", project="atocore")

    with pytest.raises(ValueError, match="duplicate active memory"):
        update_memory(second.id, content="Canonical fact")


def test_create_memory_validates_confidence(isolated_db):
    from atocore.memory.service import create_memory
    import pytest

    with pytest.raises(ValueError, match="Confidence must be between 0.0 and 1.0"):
        create_memory("knowledge", "Out of range", confidence=1.5)


def test_invalidate_memory(isolated_db):
    from atocore.memory.service import create_memory, get_memories, invalidate_memory
    mem = create_memory("knowledge", "Wrong fact")
    invalidate_memory(mem.id)
    assert len(get_memories(active_only=True)) == 0


def test_supersede_memory(isolated_db):
    from atocore.memory.service import create_memory, get_memories, supersede_memory
    mem = create_memory("knowledge", "Old fact")
    supersede_memory(mem.id)

    mems = get_memories(active_only=False)
    assert len(mems) == 1
    assert mems[0].status == "superseded"


def test_memories_for_context(isolated_db):
    from atocore.memory.service import create_memory, get_memories_for_context
    create_memory("identity", "User is a senior mechanical engineer")
    create_memory("preference", "Prefers Python with type hints")

    text, chars = get_memories_for_context(memory_types=["identity", "preference"], budget=500)
    assert "--- AtoCore Memory ---" in text
    assert "[identity]" in text
    assert "[preference]" in text
    assert chars > 0


def test_memories_for_context_reserves_room_for_each_type(isolated_db):
    from atocore.memory.service import create_memory, get_memories_for_context
    create_memory("identity", "Identity entry that is intentionally long so it could consume the whole budget on its own")
    create_memory("preference", "Preference entry that should still appear")

    text, _ = get_memories_for_context(memory_types=["identity", "preference"], budget=120)
    assert "[preference]" in text


def test_memories_for_context_respects_actual_serialized_budget(isolated_db):
    from atocore.memory.service import create_memory, get_memories_for_context
    create_memory("identity", "Identity text that should fit the wrapper-aware memory budget calculation")
    create_memory("preference", "Preference text that should also fit")

    text, chars = get_memories_for_context(memory_types=["identity", "preference"], budget=140)
    assert chars == len(text)
    assert chars <= 140


def test_memories_for_context_empty(isolated_db):
    from atocore.memory.service import get_memories_for_context
    text, chars = get_memories_for_context()
    assert text == ""
    assert chars == 0


# --- Phase 10: auto-promotion + candidate expiry ---


def _get_memory_by_id(memory_id):
    """Helper: fetch a single memory by ID."""
    from atocore.models.database import get_connection
    with get_connection() as conn:
        row = conn.execute("SELECT * FROM memories WHERE id = ?", (memory_id,)).fetchone()
    return dict(row) if row else None


def test_auto_promote_reinforced_basic(isolated_db):
    from atocore.memory.service import (
        auto_promote_reinforced,
        create_memory,
        reinforce_memory,
    )

    mem_obj = create_memory("knowledge", "Zerodur has near-zero CTE", status="candidate", confidence=0.7)
    mid = mem_obj.id
    # reinforce_memory only touches active memories, so we need to
    # promote first to reinforce, then demote back to candidate —
    # OR just bump reference_count + last_referenced_at directly
    from atocore.models.database import get_connection
    from datetime import datetime, timezone
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    with get_connection() as conn:
        conn.execute(
            "UPDATE memories SET reference_count = 3, last_referenced_at = ? WHERE id = ?",
            (now, mid),
        )

    promoted = auto_promote_reinforced(min_reference_count=3, min_confidence=0.7)
    assert mid in promoted
    mem = _get_memory_by_id(mid)
    assert mem["status"] == "active"


def test_auto_promote_reinforced_ignores_low_refs(isolated_db):
    from atocore.memory.service import auto_promote_reinforced, create_memory
    from atocore.models.database import get_connection
    from datetime import datetime, timezone

    mem_obj = create_memory("knowledge", "Some knowledge", status="candidate", confidence=0.7)
    mid = mem_obj.id
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    with get_connection() as conn:
        conn.execute(
            "UPDATE memories SET reference_count = 1, last_referenced_at = ? WHERE id = ?",
            (now, mid),
        )

    promoted = auto_promote_reinforced(min_reference_count=3, min_confidence=0.7)
    assert mid not in promoted
    mem = _get_memory_by_id(mid)
    assert mem["status"] == "candidate"


def test_expire_stale_candidates(isolated_db):
    from atocore.memory.service import create_memory, expire_stale_candidates
    from atocore.models.database import get_connection

    mem_obj = create_memory("knowledge", "Old unreferenced fact", status="candidate")
    mid = mem_obj.id
    with get_connection() as conn:
        conn.execute(
            "UPDATE memories SET created_at = datetime('now', '-30 days') WHERE id = ?",
            (mid,),
        )

    expired = expire_stale_candidates(max_age_days=14)
    assert mid in expired
    mem = _get_memory_by_id(mid)
    assert mem["status"] == "invalid"


# --- Phase 4: memory_audit log ---


def test_audit_create_logs_entry(isolated_db):
    from atocore.memory.service import create_memory, get_memory_audit

    mem = create_memory("knowledge", "test content for audit", actor="test-harness")
    audit = get_memory_audit(mem.id)
    assert len(audit) >= 1
    latest = audit[0]
    assert latest["action"] == "created"
    assert latest["actor"] == "test-harness"
    assert latest["after"]["content"] == "test content for audit"


def test_audit_promote_logs_entry(isolated_db):
    from atocore.memory.service import create_memory, get_memory_audit, promote_memory

    mem = create_memory("knowledge", "candidate for promote", status="candidate")
    promote_memory(mem.id, actor="test-triage")
    audit = get_memory_audit(mem.id)
    actions = [a["action"] for a in audit]
    assert "promoted" in actions
    promote_entry = next(a for a in audit if a["action"] == "promoted")
    assert promote_entry["actor"] == "test-triage"
    assert promote_entry["before"]["status"] == "candidate"
    assert promote_entry["after"]["status"] == "active"


def test_audit_reject_logs_entry(isolated_db):
    from atocore.memory.service import create_memory, get_memory_audit, reject_candidate_memory

    mem = create_memory("knowledge", "candidate for reject", status="candidate")
    reject_candidate_memory(mem.id, actor="test-triage", note="stale")
    audit = get_memory_audit(mem.id)
    actions = [a["action"] for a in audit]
    assert "rejected" in actions
    reject_entry = next(a for a in audit if a["action"] == "rejected")
    assert reject_entry["note"] == "stale"


def test_audit_update_captures_before_after(isolated_db):
    from atocore.memory.service import create_memory, get_memory_audit, update_memory

    mem = create_memory("knowledge", "original content", confidence=0.5)
    update_memory(mem.id, content="updated content", confidence=0.9, actor="human-edit")
    audit = get_memory_audit(mem.id)
    update_entries = [a for a in audit if a["action"] == "updated"]
    assert len(update_entries) >= 1
    u = update_entries[0]
    assert u["before"]["content"] == "original content"
    assert u["after"]["content"] == "updated content"
    assert u["before"]["confidence"] == 0.5
    assert u["after"]["confidence"] == 0.9


def test_audit_reinforce_logs_entry(isolated_db):
    from atocore.memory.service import create_memory, get_memory_audit, reinforce_memory

    mem = create_memory("knowledge", "reinforced mem", confidence=0.5)
    reinforce_memory(mem.id, confidence_delta=0.02)
    audit = get_memory_audit(mem.id)
    actions = [a["action"] for a in audit]
    assert "reinforced" in actions


def test_recent_audit_returns_cross_memory_entries(isolated_db):
    from atocore.memory.service import create_memory, get_recent_audit

    m1 = create_memory("knowledge", "mem one content", actor="harness")
    m2 = create_memory("knowledge", "mem two content", actor="harness")
    recent = get_recent_audit(limit=10)
    ids = {e["memory_id"] for e in recent}
    assert m1.id in ids and m2.id in ids


# --- Phase 3: domain_tags + valid_until ---


def test_create_memory_with_tags_and_valid_until(isolated_db):
    from atocore.memory.service import create_memory

    mem = create_memory(
        "knowledge",
        "CTE gradient dominates WFE at F/1.2",
        domain_tags=["optics", "thermal", "materials"],
        valid_until="2027-01-01",
    )
    assert mem.domain_tags == ["optics", "thermal", "materials"]
    assert mem.valid_until == "2027-01-01"


def test_create_memory_normalizes_tags(isolated_db):
    from atocore.memory.service import create_memory

    mem = create_memory(
        "knowledge",
        "some content here",
        domain_tags=["  Optics  ", "OPTICS", "Thermal", ""],
    )
    # Duplicates and empty removed; lowercased; stripped
    assert mem.domain_tags == ["optics", "thermal"]


def test_update_memory_sets_tags_and_valid_until(isolated_db):
    from atocore.memory.service import create_memory, update_memory
    from atocore.models.database import get_connection

    mem = create_memory("knowledge", "some content for update test")
    assert update_memory(
        mem.id,
        domain_tags=["controls", "firmware"],
        valid_until="2026-12-31",
    )
    with get_connection() as conn:
        row = conn.execute("SELECT domain_tags, valid_until FROM memories WHERE id = ?", (mem.id,)).fetchone()
    import json as _json
    assert _json.loads(row["domain_tags"]) == ["controls", "firmware"]
    assert row["valid_until"] == "2026-12-31"


def test_get_memories_for_context_excludes_expired(isolated_db):
    """Expired active memories must not land in context packs."""
    from atocore.memory.service import create_memory, get_memories_for_context

    # Active but expired
    create_memory(
        "knowledge",
        "stale snapshot from long ago period",
        valid_until="2020-01-01",
        confidence=1.0,
    )
    # Active and valid
    create_memory(
        "knowledge",
        "durable engineering insight stays valid forever",
        confidence=1.0,
    )

    text, _ = get_memories_for_context(memory_types=["knowledge"], budget=600)
    assert "durable engineering" in text
    assert "stale snapshot" not in text


def test_context_builder_tag_boost_orders_results(isolated_db):
    """Memories with tags matching query should rank higher."""
    from atocore.memory.service import create_memory, get_memories_for_context

    create_memory("knowledge", "generic content has no obvious overlap with topic", confidence=0.8, domain_tags=[])
    create_memory("knowledge", "generic content has no obvious overlap topic here", confidence=0.8, domain_tags=["optics"])

    text, _ = get_memories_for_context(
        memory_types=["knowledge"],
        budget=2000,
        query="tell me about optics",
    )
    # Tagged memory should appear before the untagged one
    idx_tagged = text.find("overlap topic here")
    idx_untagged = text.find("overlap with topic")
    assert idx_tagged != -1
    assert idx_untagged != -1
    assert idx_tagged < idx_untagged


def test_project_memory_ranking_ignores_scope_noise(isolated_db):
    """Project words should not crowd out the actual query intent."""
    from atocore.memory.service import create_memory, get_memories_for_context

    create_memory(
        "project",
        "Norman is the end operator for p06-polisher and requires an explicit manual mode to operate the machine.",
        project="p06-polisher",
        confidence=0.7,
    )
    create_memory(
        "project",
        "Polisher Control firmware spec document titled 'Fulum Polisher Machine Control Firmware Spec v1' lives in PKM.",
        project="p06-polisher",
        confidence=0.7,
    )
    create_memory(
        "project",
        "Machine design principle: works fully offline and independently; network connection is for remote access only",
        project="p06-polisher",
        confidence=0.5,
    )
    create_memory(
        "project",
        "Use Tailscale mesh for RPi remote access to provide SSH, file transfer, and NAT traversal without port forwarding.",
        project="p06-polisher",
        confidence=0.5,
    )

    text, _ = get_memories_for_context(
        memory_types=["project"],
        project="p06-polisher",
        budget=360,
        query="how do we access the polisher machine remotely",
    )

    assert "Tailscale" in text
    assert text.find("remote access only") < text.find("Tailscale")
    assert "manual mode" not in text


def test_project_memory_ranking_prefers_multiple_intent_hits(isolated_db):
    """A rich memory with several query hits should beat a terse one-hit memory."""
    from atocore.memory.service import create_memory, get_memories_for_context

    create_memory(
        "project",
        "CGH vendor selected for p05. Active integration coordination with Katie/AOM.",
        project="p05-interferometer",
        confidence=0.7,
    )
    create_memory(
        "knowledge",
        "Vendor-summary current signal: 4D is the strongest technical Twyman-Green candidate; "
        "a certified used Zygo Verifire SV around $55k emerged as a strong value path.",
        project="p05-interferometer",
        confidence=0.9,
    )

    text, _ = get_memories_for_context(
        memory_types=["project", "knowledge"],
        project="p05-interferometer",
        budget=220,
        query="what is the current vendor signal for the interferometer procurement",
    )

    assert "4D" in text
    assert "Zygo" in text


def test_project_memory_query_ranks_beyond_confidence_prefilter(isolated_db):
    """Query-time ranking should see older low-confidence but exact-intent memories."""
    from atocore.memory.service import create_memory, get_memories_for_context

    for idx in range(35):
        create_memory(
            "project",
            f"High confidence p06 filler memory {idx}: Polisher Control planning note.",
            project="p06-polisher",
            confidence=0.9,
        )
    create_memory(
        "project",
        "Use Tailscale mesh for RPi remote access to provide SSH, file transfer, and NAT traversal without port forwarding.",
        project="p06-polisher",
        confidence=0.5,
    )

    text, _ = get_memories_for_context(
        memory_types=["project"],
        project="p06-polisher",
        budget=360,
        query="how do we access the polisher machine remotely",
    )

    assert "Tailscale" in text


def test_project_memory_query_prefers_exact_cam_fact(isolated_db):
    from atocore.memory.service import create_memory, get_memories_for_context

    create_memory(
        "project",
        "Polisher Control firmware spec document titled 'Fulum Polisher Machine Control Firmware Spec v1' lives in PKM.",
        project="p06-polisher",
        confidence=0.9,
    )
    create_memory(
        "project",
        "Polisher Control doc must cover manual mode for Norman as a required deliverable per the plan.",
        project="p06-polisher",
        confidence=0.9,
    )
    create_memory(
        "project",
        "Cam amplitude and offset are mechanically set by operator and read via encoders; no actuators control them.",
        project="p06-polisher",
        confidence=0.5,
    )

    text, _ = get_memories_for_context(
        memory_types=["project"],
        project="p06-polisher",
        budget=300,
        query="how is cam amplitude controlled on the polisher",
    )

    assert "encoders" in text


def test_expire_stale_candidates_keeps_reinforced(isolated_db):
    from atocore.memory.service import create_memory, expire_stale_candidates
    from atocore.models.database import get_connection

    mem_obj = create_memory("knowledge", "Referenced fact", status="candidate")
    mid = mem_obj.id
    with get_connection() as conn:
        conn.execute(
            "UPDATE memories SET reference_count = 1, "
            "created_at = datetime('now', '-30 days') WHERE id = ?",
            (mid,),
        )

    expired = expire_stale_candidates(max_age_days=14)
    assert mid not in expired
    mem = _get_memory_by_id(mid)
    assert mem["status"] == "candidate"


# ---------------------------------------------------------------------------
# Wave 1 (2026-04-29) — counts come from SQL, not from the top-N sample.
# Exposed by Codex audit when prod /admin/dashboard reported 315 active
# while /admin/integrity-check reported 1091. The dashboard was building
# its counts from a confidence-sorted limit=500 fetch.
# ---------------------------------------------------------------------------


def test_get_memory_count_summary_returns_full_table_aggregates(isolated_db):
    """Counts come from SQL aggregates, not a sampled fetch."""
    from atocore.memory.service import (
        create_memory,
        get_memory_count_summary,
        invalidate_memory,
    )

    # Create more rows than any reasonable sampling LIMIT so any
    # LIMIT-based counter would visibly disagree with reality.
    for i in range(120):
        create_memory(
            "knowledge",
            f"fact-{i}",
            project="p04-gigabit",
            confidence=0.9,
            status="active",
        )
    for i in range(7):
        create_memory("knowledge", f"cand-{i}", status="candidate")
    invalid_obj = create_memory("knowledge", "to-invalidate", status="active")
    invalidate_memory(invalid_obj.id)

    summary = get_memory_count_summary()
    assert summary["total"] == 120 + 7 + 1
    assert summary["by_status"]["active"] == 120
    assert summary["by_status"]["candidate"] == 7
    assert summary["by_status"]["invalid"] == 1
    assert summary["active"]["total"] == 120
    assert summary["active"]["by_type"] == {"knowledge": 120}
    assert summary["active"]["by_project"] == {"p04-gigabit": 120}


def test_get_memory_returns_single_row_or_none(isolated_db):
    from atocore.memory.service import create_memory, get_memory

    mem = create_memory("knowledge", "single-row test")
    fetched = get_memory(mem.id)
    assert fetched is not None
    assert fetched.id == mem.id
    assert get_memory("non-existent-id") is None


def test_update_memory_can_change_project_with_canonicalization(
    isolated_db, project_registry
):
    """update_memory(project=...) canonicalizes aliases and writes audit."""
    project_registry(("p04-gigabit", ("p04", "gigabit")))
    from atocore.memory.service import (
        create_memory,
        get_memory,
        get_memory_audit,
        update_memory,
    )

    mem = create_memory("knowledge", "retargetable fact", project="atocore")
    ok = update_memory(mem.id, project="p04")  # alias
    assert ok is True

    refreshed = get_memory(mem.id)
    assert refreshed.project == "p04-gigabit"  # canonical, not "p04"

    audit_rows = get_memory_audit(mem.id, limit=10)
    update_rows = [r for r in audit_rows if r.get("action") == "updated"]
    assert update_rows, f"expected an updated audit row, got {audit_rows}"
    head = update_rows[0]
    assert head["before"]["project"] == "atocore"
    assert head["after"]["project"] == "p04-gigabit"


def test_update_memory_project_unchanged_when_not_passed(isolated_db):
    from atocore.memory.service import create_memory, get_memory, update_memory

    mem = create_memory("knowledge", "untouched project", project="p06-polisher")
    update_memory(mem.id, content="edited content")
    assert get_memory(mem.id).project == "p06-polisher"


def test_update_memory_to_empty_project_detects_global_duplicate(isolated_db):
    """Codex P3: when retargeting to project='' (global), the duplicate
    check must scope to the new project. If a global active memory with
    the same content already exists, the update must raise."""
    import pytest as _pytest
    from atocore.memory.service import create_memory, update_memory

    create_memory("knowledge", "shared global fact", project="")
    scoped = create_memory("knowledge", "shared global fact", project="p04-gigabit")

    with _pytest.raises(ValueError, match="duplicate active memory"):
        update_memory(scoped.id, project="")


def test_auto_triage_suggested_project_put_body_uses_project_key():
    """Regression: the auto_triage caller used to PUT {"content": ...}
    which silently dropped the suggested project change. The fix sends
    {"project": suggested}. Inspect the script source so we don't have
    to spin up a live triage run."""
    from pathlib import Path

    src = Path(__file__).resolve().parents[1] / "scripts" / "auto_triage.py"
    text = src.read_text(encoding="utf-8")
    # The block that PUTs to /memory/{mid} for a suggested_project fix
    assert 'json.dumps({"project": suggested})' in text, (
        "auto_triage.py must PUT {\"project\": suggested} so the "
        "suggested-project correction actually applies. See Wave 1."
    )
    # And must not be back to the old shape
    assert 'json.dumps({"content": cand["content"]})' not in text