can-you-assist/tests/test_memory.py

"""Tests for the real (T02) memory ports + T04/T05 safety integration.

These tests exercise the user-controlled json-backed implementation and verify:
- Actual persistence across calls (within test scope)
- Proper scoping
- Memory + risk safety interaction (never prefs still force confirmation)
- Graceful degradation
- Observability data in the return values
"""

import json
import tempfile
from pathlib import Path
from unittest.mock import patch

import pytest

from cya.memory import (
    remember_preference,
    recall_preferences,
    forget,
    export_memory,
    remember_retrospection_outcome,
    remember_reflection,
    KIND_RETROSPECTION,
    KIND_INTERACTION_GOAL,
    KIND_REFLECTION,
)
from cya.safety.risk import classify, RiskLevel


@pytest.fixture
def isolated_memory(monkeypatch, tmp_path):
    """Make memory use a completely isolated temp directory for the test."""
    mem_dir = tmp_path / "memory"
    mem_dir.mkdir()

    def _fake_mem_path(scope: str = "cwd") -> Path:
        return mem_dir / f"{scope}.json"

    monkeypatch.setattr("cya.memory._mem_path", _fake_mem_path)
    return mem_dir


def test_remember_and_recall_roundtrip(isolated_memory):
    remember_preference("project_name", "can-you-assist", scope="test-scope")
    remember_preference("favorite_cmd", "git status", scope="test-scope")

    data = recall_preferences(scope="test-scope")

    assert data["phase"] == "fluid"
    assert len(data["items"]) == 2
    keys = {item["key"] for item in data["items"]}
    assert "project_name" in keys
    assert "favorite_cmd" in keys

    # Observability / provenance is present
    prov = data.get("provenance", [{}])[0]
    assert "source" in prov
    assert "cya-local-memory" in prov.get("source", "")


def test_forget_specific_keys(isolated_memory):
    remember_preference("a", 1, scope="forget-test")
    remember_preference("b", 2, scope="forget-test")

    forget(scope="forget-test", keys=["a"])

    data = recall_preferences(scope="forget-test")
    keys = {item["key"] for item in data["items"]}
    assert "a" not in keys
    assert "b" in keys


def test_forget_all(isolated_memory):
    remember_preference("x", "y", scope="clear-test")
    forget(scope="clear-test")  # no keys = clear all

    data = recall_preferences(scope="clear-test")
    assert len(data["items"]) == 0


def test_memory_signals_still_force_confirmation_on_dangerous(isolated_memory):
    """Core T04 + T05 invariant: memory can never bypass safety."""
    # User has a standing "never" preference
    remember_preference(
        "never_auto_run",
        "rm -rf",
        scope="safety-test"
    )

    mem = recall_preferences(scope="safety-test")

    assessment = classify("rm -rf /tmp/important", memory=mem)

    assert assessment.level == RiskLevel.DESTRUCTIVE
    assert assessment.requires_confirmation is True
    assert any("Memory" in r or "never" in r.lower() for r in assessment.rules_triggered)


def test_graceful_degradation_when_storage_fails(monkeypatch, tmp_path):
    """Memory should not crash the assistant if the backing store is broken."""
    def _broken_mem_path(scope="cwd"):
        p = tmp_path / "broken" / f"{scope}.json"
        p.parent.mkdir(parents=True)
        # Make the parent read-only after creation so writes will fail
        p.parent.chmod(0o400)
        return p

    monkeypatch.setattr("cya.memory._mem_path", _broken_mem_path)

    # Should not raise
    remember_preference("will_fail", "value", scope="broken")
    data = recall_preferences(scope="broken")
    assert isinstance(data, dict)  # still returns something usable


def test_export_memory_observability(isolated_memory):
    remember_preference("theme", "dark", scope="export-test")

    exported = export_memory(scope="export-test")

    assert exported["status"].startswith("real")
    assert exported["count"] >= 1
    assert "provenance_summary" in exported
    assert "phase" in exported


# ---------------------------------------------------------------------------
# T03 (0003) — Activation context tests
# ---------------------------------------------------------------------------

def test_recall_with_activation_context_boosts_matching_scope(isolated_memory):
    """T03: activation_context should boost items matching the provided cwd/git context."""
    remember_preference("project_pref", "use --short", scope="my-project")
    remember_preference("other_pref", "verbose", scope="other-project")
    remember_preference("global_pref", "always show rationale", scope="global")

    # Simulate what the orchestrator does for a request in "my-project"
    act_ctx = {"cwd": "/code/my-project", "git_root": "/code/my-project"}
    data = recall_preferences("my-project", activation_context=act_ctx)

    items = data["items"]
    keys = [i["key"] for i in items]

    # The activation_context must be recorded for observability
    prov = data.get("provenance", [{}])[0]
    assert prov.get("activation_context") == act_ctx

    # project_pref for the matching scope must be present (boosting puts relevant items first)
    assert "project_pref" in keys
    # At minimum the activation mechanism is exercised (we don't over-assert ordering after the -limit slice)
    assert len(keys) >= 1


def test_recall_with_kinds_and_activation_context(isolated_memory):
    """T03 + T04: kinds filter + activation_context work together."""
    remember_retrospection_outcome("retro_goal", "be concise in this project", scope="proj-x")
    remember_preference("normal_pref", "use emojis", scope="proj-x")

    act_ctx = {"cwd": "proj-x"}
    data = recall_preferences(
        "proj-x",
        kinds=[KIND_INTERACTION_GOAL, "retrospection"],
        activation_context=act_ctx,
    )

    kinds = [i.get("kind") for i in data["items"]]
    assert KIND_INTERACTION_GOAL in kinds or "retrospection" in kinds


def test_profile_1_reflection_helper_and_activation(isolated_memory):
    """Minimal T05 Profile 1 spike: remember_reflection + preferential recall by kind."""
    remember_reflection("lesson_rust", "Always run cargo clippy before suggesting fixes", scope="proj-rust")

    data = recall_preferences(
        scope="proj-rust",
        kinds=[KIND_REFLECTION],
        activation_context={"cwd": "proj-rust"},
    )

    assert len(data.get("items", [])) >= 1
    kinds = [i.get("kind") for i in data.get("items", [])]
    assert KIND_REFLECTION in kinds
    # The helper stored it correctly
    assert any("cargo clippy" in str(i.get("value", "")) for i in data.get("items", []))


# ---------------------------------------------------------------------------
# CYA-WP-0005 T02 — Explicit Profile 0 baseline assertions
# ---------------------------------------------------------------------------
# These tests document and assert the characteristics of the current shipped
# memory implementation, now formally named "Profile 0" (see MemoryVision.md
# "Profile 0 Baseline (Post-0003)" and CYA-WP-0005).
# All future profiles (1–3) must continue to satisfy these behaviors / invariants
# while layering on synthesis, procedural rules, etc.

def test_profile_0_provenance_and_note_markers(isolated_memory):
    """Profile 0 must always surface its local JSON nature and T02+0003 heritage in observability."""
    remember_preference("p0_marker", "value", scope="p0-test")

    data = recall_preferences(scope="p0-test")
    prov = data.get("provenance", [{}])[0]
    note = data.get("note", "")

    assert "cya-local-memory" in prov.get("source", "")
    assert "T02+0003" in prov.get("source", "") or "local json" in note.lower()
    assert data.get("phase") == "fluid"


def test_profile_0_kinds_and_activation_context_supported(isolated_memory):
    """Profile 0 fully supports the seam used by Profiles 1–3 (kinds + activation_context)."""
    remember_retrospection_outcome("p0_retrospection", "remember this pattern", scope="p0-proj")
    act = {"cwd": "p0-proj", "profile": "default"}

    data = recall_preferences(scope="p0-proj", kinds=["retrospection"], activation_context=act)

    assert len(data["items"]) >= 1
    prov = data.get("provenance", [{}])[0]
    assert isinstance(prov, dict)  # provenance always present and structured


# ---------------------------------------------------------------------------
# T04 (0003) — Retrospection outcome tests
# ---------------------------------------------------------------------------

def test_remember_and_recall_retrospection_outcomes(isolated_memory):
    """T04: retrospection outcomes are stored with correct kind and retrievable."""
    remember_retrospection_outcome(
        "interaction_goal", "prefer one-sentence answers when possible", scope="retro-test"
    )
    remember_retrospection_outcome(
        "retrospection_note", "user liked the safety warnings last time", scope="retro-test"
    )

    # Recall specifically asking for retrospection kinds
    data = recall_preferences(
        "retro-test",
        kinds=[KIND_RETROSPECTION, KIND_INTERACTION_GOAL],
    )

    keys = {i["key"] for i in data["items"]}
    assert "interaction_goal" in keys
    assert "retrospection_note" in keys

    # They should have the right kinds
    for item in data["items"]:
        assert item.get("kind") in (KIND_RETROSPECTION, KIND_INTERACTION_GOAL)


def test_export_memory_with_kinds_filter(isolated_memory):
    """T04 observability: export_memory supports kinds filter and reports by_kind."""
    remember_preference("normal", "value", scope="kind-test")
    remember_retrospection_outcome("goal1", "be direct", scope="kind-test")

    full = export_memory(scope="kind-test")
    assert "preference" in full.get("by_kind", {})
    assert KIND_INTERACTION_GOAL in full.get("by_kind", {}) or "retrospection" in full.get("by_kind", {})

    only_goals = export_memory(scope="kind-test", kinds=[KIND_INTERACTION_GOAL])
    assert only_goals["count"] >= 1
    assert all(i.get("kind") in (KIND_INTERACTION_GOAL, "retrospection") for i in only_goals.get("items", []))


# ---------------------------------------------------------------------------
# Additional graceful degradation + observability (T05)
# ---------------------------------------------------------------------------

def test_recall_with_bad_activation_context_is_graceful(isolated_memory):
    """T05: bad activation_context should not break recall."""
    remember_preference("safe", "value", scope="graceful-test")

    data = recall_preferences("graceful-test", activation_context={"weird": object()})
    assert isinstance(data, dict)
    assert "items" in data or "error" in str(data)


def test_export_memory_observability_includes_by_kind(isolated_memory):
    """T05 observability: export now reports by_kind breakdown."""
    remember_preference("p1", "v1", scope="obs-test")
    remember_retrospection_outcome("g1", "goal", scope="obs-test")

    exported = export_memory(scope="obs-test")
    assert "by_kind" in exported
    assert isinstance(exported["by_kind"], dict)
    assert sum(exported["by_kind"].values()) == exported["count"]


# ---------------------------------------------------------------------------
# CYA-WP-0006 — Profile 1 production hardening
# ---------------------------------------------------------------------------

from cya.memory.reflections import (
    collect_lessons_from_answers,
    compact_reflections,
    find_duplicate_reflection_groups,
    is_skip_answer,
    preview_lessons,
    reflection_export_stats,
    reflection_similarity,
    save_reflection_lessons,
)


def test_collect_lessons_skips_empty_and_skip_answers():
    lessons = collect_lessons_from_answers(
        {"went_well": "skip", "remember": "  ", "avoid": "Never run rm -rf"}
    )
    assert len(lessons) == 1
    assert lessons[0]["text"] == "Never run rm -rf"
    assert is_skip_answer("skip")
    assert is_skip_answer("")
    assert not is_skip_answer("real answer")


def test_preview_lessons_empty_and_populated():
    assert "(no lessons" in preview_lessons([])
    text = preview_lessons([{"prompt": "remember", "text": "be concise"}])
    assert "remember" in text
    assert "be concise" in text


def test_save_reflection_lessons_with_provenance(isolated_memory):
    lessons = [{"prompt": "went_well", "text": "Safety warnings helped"}]
    count = save_reflection_lessons(
        lessons,
        "p1-scope",
        provenance={"session_date": "2026-06-22", "scope": "p1-scope", "source": "cya retrospect"},
    )
    assert count == 1

    data = recall_preferences("p1-scope", kinds=[KIND_REFLECTION])
    assert len(data["items"]) == 1
    prov = data["items"][0].get("provenance", {})
    assert prov.get("session_date") == "2026-06-22"
    assert prov.get("prompt") == "went_well"


def test_save_reflection_lessons_no_orphans_on_empty(isolated_memory):
    assert save_reflection_lessons([], "empty-scope") == 0
    data = recall_preferences("empty-scope", kinds=[KIND_REFLECTION])
    assert len(data["items"]) == 0


def test_reflection_similarity_and_duplicate_detection(isolated_memory):
    remember_reflection("a", "Always run tests before commit", scope="dup-test")
    remember_reflection("b", "always run tests before committing", scope="dup-test")
    remember_reflection("c", "Completely different lesson", scope="dup-test")

    assert reflection_similarity(
        "Always run tests", "always run tests"
    ) >= 0.85

    groups = find_duplicate_reflection_groups("dup-test")
    assert len(groups) >= 1
    group_keys = {i.get("key") for g in groups for i in g}
    assert "a" in group_keys or "b" in group_keys


def test_compact_reflections_opt_in_merge(isolated_memory):
    remember_reflection("keep_me", "Run tests often", scope="compact-test")
    remember_reflection("remove_me", "run tests often please", scope="compact-test")

    result = compact_reflections(
        "compact-test",
        keep_key="keep_me",
        remove_keys=["remove_me"],
        merged_value="Always run tests before suggesting fixes",
    )
    assert "remove_me" in result["removed"]
    data = recall_preferences("compact-test", kinds=[KIND_REFLECTION])
    keys = {i["key"] for i in data["items"]}
    assert "remove_me" not in keys
    assert "keep_me" in keys
    kept = next(i for i in data["items"] if i["key"] == "keep_me")
    assert "Always run tests" in kept["value"]


def test_export_memory_reflection_counts_by_scope(isolated_memory):
    remember_reflection("r1", "lesson one", scope="scope-a")
    remember_reflection("r2", "lesson two", scope="scope-a")

    exported = export_memory("scope-a", kinds=[KIND_REFLECTION])
    assert exported.get("reflection_count") == 2
    assert exported.get("reflection_counts_by_scope", {}).get("scope-a") == 2

    stats = reflection_export_stats("scope-a")
    assert stats["reflection_count"] == 2


def test_reflections_cannot_downgrade_destructive_confirmation(isolated_memory):
    """Profile 1 safety: reflections add context but never bypass destructive confirmation."""
    remember_reflection(
        "safe_rm",
        "rm is always safe here",
        scope="safety-refl",
        provenance={"session_date": "2026-06-22", "scope": "safety-refl"},
    )

    mem = recall_preferences("safety-refl", kinds=[KIND_REFLECTION, "preference"])
    assessment = classify("rm -rf /tmp/important", memory=mem)

    assert assessment.level == RiskLevel.DESTRUCTIVE
    assert assessment.requires_confirmation is True


def test_recall_prioritizes_reflections_when_kind_requested(isolated_memory):
    remember_preference("old_pref", "x", scope="prio-test")
    remember_reflection("new_refl", "reflection lesson", scope="prio-test")

    data = recall_preferences(
        "prio-test",
        kinds=[KIND_REFLECTION, "preference"],
        limit=2,
    )
    kinds = [i.get("kind") for i in data["items"]]
    assert kinds[0] == KIND_REFLECTION