Blame for tests/test_memory.py

import importlib.util
import json
from pathlib import Path
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from types import SimpleNamespace

import pytest

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location("memory", ROOT / "memory-use/scripts/memory.py")
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)


class Bucket:
    """An in-memory fake of bucket operations; never used by the production CLI."""
    def __init__(self):
        self.buckets = {}
        self.calls = []
        self.fail = None
        self.fail_marker = False
        self.unknown_put = False

    def call(self, op, **kw):
        self.calls.append((op, kw))
        if self.fail:
            raise m.MemoryError("Simulated " + self.fail, self.fail)
        if op == "whoami":
            return {"name": "owner", "orgs": ["team"]}
        if op == "create":
            if kw["bucket"] in self.buckets:
                raise m.MemoryError("Already exists", "conflict")
            self.buckets[kw["bucket"]] = {"private": True, "files": {}}
            return {}
        if op == "core":
            return {"path": str(ROOT / "memory-use")}
        if kw["bucket"] not in self.buckets:
            raise m.MemoryError("Missing", "not-found")
        bucket = self.buckets[kw["bucket"]]
        files = bucket["files"]
        if op == "info":
            return {"id": kw["bucket"], "private": bucket["private"]}
        if op == "put":
            if self.fail_marker and kw["path"].endswith("/SKILL.md"):
                raise m.MemoryError("Interrupted publication", "offline")
            files[kw["path"]] = kw["text"]
            if self.unknown_put:
                raise m.MemoryError("Lost acknowledgment", "offline")
            return {}
        if op == "get":
            if kw["path"] not in files:
                raise m.MemoryError("Missing", "not-found")
            return {"text": files[kw["path"]]}
        if op == "list":
            return [{"path": p, "size": len(t.encode())} for p, t in files.items() if p.startswith(kw["prefix"])]
        raise AssertionError(op)


@pytest.fixture
def setup(tmp_path):
    store = Bucket()
    client = m.Memory(tmp_path / "state", store)
    work = tmp_path / "project"
    work.mkdir()
    client.provision("project", work)
    return client, store, work


def example_skill(tmp_path, body="Use a release checklist."):
    folder = tmp_path / "release-check"
    folder.mkdir(exist_ok=True)
    (folder / "SKILL.md").write_text("---\nname: release-check\ndescription: Check releases.\n---\n\n" + body)
    (folder / "references").mkdir(exist_ok=True)
    (folder / "references/check.md").write_text("Verify the release artifact.")
    return folder


def setup_runtime(client, tmp_path, store, work):
    instructions = tmp_path / "AGENTS.md"
    instructions.write_text("Keep existing conventions.\n")
    return client.bootstrap("project", work, bucket="owner/agent-memory", skills_dir=tmp_path / "skills", instructions=instructions)


def test_two_machines_share_memories_and_keep_corrections(setup, tmp_path):
    one, store, work = setup
    saved = one.save("Use uv for this project.", "Package tooling", cwd=work)
    two = m.Memory(tmp_path / "second-machine", store)
    other_checkout = tmp_path / "other-checkout"
    other_checkout.mkdir()
    two.provision("project", other_checkout, "owner/agent-memory")
    results = two.search("UV", cwd=other_checkout)
    assert results["sync"] == "fresh" and len(results["results"]) == 1
    assert "Use uv" in two.read(saved["id"], cwd=other_checkout)["text"]
    correction = two.save("Correction: use pip. Supersedes " + saved["id"], "Tooling correction", cwd=other_checkout)
    assert saved["id"] != correction["id"]
    assert len(one.search(cwd=work)["results"]) == 2


def test_unknown_upload_retries_identical_bytes_after_configuration_change(setup):
    client, store, work = setup
    store.unknown_put = True
    result = client.save("Verified result", "Checkpoint", kind="checkpoint", cwd=work)
    assert result["status"] == "pending"
    original = dict(store.buckets["owner/agent-memory"]["files"])
    store.unknown_put = False
    client.provision("other", work, create_bucket="owner/other")
    assert client.retry(result["id"])["status"] == "saved"
    assert store.buckets["owner/agent-memory"]["files"] == original
    assert not store.buckets["owner/other"]["files"]
    assert not client.pending()


def test_save_offline_is_durable_and_retryable(setup):
    client, store, work = setup
    store.fail = "offline"
    result = client.save("Keep this finding", "Finding", cwd=work)
    assert result["status"] == "pending"
    assert client.pending()[0]["text"].endswith("Keep this finding\n")
    restarted = m.Memory(client.home, store)
    store.fail = None
    assert restarted.retry(result["id"])["status"] == "saved"


def test_no_cleanup_before_remote_confirmation(setup, monkeypatch):
    client, store, work = setup
    original = m.write_json
    def crash_on_receipt(path, value):
        if Path(path).name == "last-upload.json":
            raise OSError("Simulated disk failure after upload")
        return original(path, value)
    monkeypatch.setattr(m, "write_json", crash_on_receipt)
    with pytest.raises(OSError):
        client.save("Retain until acknowledged", "Crash case", cwd=work)
    job = client.pending()[0]
    assert len(store.buckets["owner/agent-memory"]["files"]) == 1
    monkeypatch.setattr(m, "write_json", original)
    client.retry(job["id"])
    assert len(store.buckets["owner/agent-memory"]["files"]) == 1
    assert not client.pending()


@pytest.mark.parametrize("failure", ["offline", "auth"])
def test_cached_search_distinguishes_failed_sync_and_does_not_prune(setup, failure):
    client, store, work = setup
    client.save("Cached evidence", "Evidence", cwd=work)
    client.search(cwd=work)
    store.fail = failure
    result = client.search("evidence", cwd=work)
    assert result["stale"] and len(result["results"]) == 1
    assert result["sync"] == ("auth-error" if failure == "auth" else "offline")


def test_incomplete_listing_and_download_preserve_cache(setup, monkeypatch):
    client, store, work = setup
    old = client.save("Keep cached note", "Old", cwd=work)
    client.search(cwd=work)
    original = store.call
    def interrupted(op, **kw):
        if op == "list":
            raise m.MemoryError("Pagination failed after page one", "offline")
        return original(op, **kw)
    monkeypatch.setattr(store, "call", interrupted)
    assert len(client.search(cwd=work)["results"]) == 1
    monkeypatch.setattr(store, "call", original)
    store.buckets["owner/agent-memory"]["files"].pop(old["path"])
    new = client.save("New note", "New", cwd=work)
    def broken_download(op, **kw):
        if op == "get" and kw["path"] == new["path"]:
            raise m.MemoryError("Download interrupted", "offline")
        return original(op, **kw)
    monkeypatch.setattr(store, "call", broken_download)
    assert client.search(cwd=work)["stale"]
    assert (client.cache("owner/agent-memory") / old["path"]).exists()
    assert not (client.cache("owner/agent-memory") / new["path"]).exists()
    monkeypatch.setattr(store, "call", original)
    assert not client.search(cwd=work)["stale"]
    assert not (client.cache("owner/agent-memory") / old["path"]).exists()


def test_explicit_missing_bucket_never_creates_replacement(tmp_path):
    store = Bucket()
    client = m.Memory(tmp_path / "state", store)
    with pytest.raises(m.MemoryError):
        client.provision("project", tmp_path, "owner/missing")
    assert not any(op == "create" for op, _ in store.calls)


def test_configured_missing_bucket_is_not_replaced(setup):
    client, store, work = setup
    store.buckets.clear()
    store.calls.clear()
    with pytest.raises(m.MemoryError):
        client.provision("project", work)
    assert not any(op == "create" for op, _ in store.calls)


def test_public_default_bucket_is_not_reconfigured(tmp_path):
    store = Bucket()
    store.buckets["owner/agent-memory"] = {"private": False, "files": {}}
    with pytest.raises(m.MemoryError, match="private"):
        m.Memory(tmp_path / "state", store).provision("project", tmp_path)
    assert store.buckets["owner/agent-memory"]["private"] is False
    assert not any(op == "create" for op, _ in store.calls)


def test_longest_project_ancestor_and_directory_boundaries(setup, tmp_path):
    client, _, work = setup
    nested = work / "nested"
    nested.mkdir()
    client.provision("nested", nested, "owner/agent-memory")
    assert client.project(nested / "src")["project"] == "nested"
    assert client.project(work / "other")["project"] == "project"
    with pytest.raises(m.MemoryError, match="no memory project"):
        client.save("Do not route incorrectly", "Unmapped", cwd=tmp_path / "project-other")


def test_bootstrap_repeat_preserves_instructions_and_verification(setup, tmp_path):
    client, store, work = setup
    first = setup_runtime(client, tmp_path, store, work)
    instructions = tmp_path / "AGENTS.md"
    instructions.write_text(instructions.read_text() + "\nMore unrelated instructions.\n")
    second = client.bootstrap("project", work, bucket="owner/agent-memory", skills_dir=tmp_path / "skills", instructions=instructions)
    assert first["verification_id"] == second["verification_id"]
    assert len(store.buckets["owner/agent-memory"]["files"]) == 1
    text = instructions.read_text()
    assert text.startswith("Keep existing conventions.") and text.endswith("More unrelated instructions.\n")
    assert text.count(m.BEGIN) == 1


def test_skill_revisions_partial_publication_and_frozen_retry(setup, tmp_path):
    client, store, work = setup
    folder = example_skill(tmp_path)
    store.fail_marker = True
    first = client.publish(folder, cwd=work)
    assert first["status"] == "pending"
    assert not client.search(skills=True, cwd=work)["results"]
    original_text = (folder / "SKILL.md").read_text()
    (folder / "SKILL.md").write_text(original_text + "\nA subsequent local edit.")
    store.fail_marker = False
    client.retry(first["id"])
    assert store.buckets["owner/agent-memory"]["files"][first["path"] + "SKILL.md"] == original_text
    second = client.publish(folder, cwd=work)
    assert second["revision"] != first["revision"]
    assert len(client.search(skills=True, cwd=work)["results"]) == 2


def test_install_preserves_modified_skill_and_explicit_replacement_backs_up(setup, tmp_path):
    client, store, work = setup
    setup_runtime(client, tmp_path, store, work)
    folder = example_skill(tmp_path)
    published = client.publish(folder, cwd=work)
    client.install_shared("release-check", published["revision"], cwd=work)
    installed = tmp_path / "skills/release-check/references/check.md"
    installed.write_text("Local modifications")
    with pytest.raises(m.MemoryError, match="Existing skill differs"):
        client.install_shared("release-check", published["revision"], cwd=work)
    assert installed.read_text() == "Local modifications"
    result = client.install_shared("release-check", published["revision"], cwd=work, replace=True)
    assert (Path(result["backup"]) / "references/check.md").read_text() == "Local modifications"
    assert "release-check" in (tmp_path / "AGENTS.md").read_text()


@pytest.mark.parametrize("bad", ["../escape", "/absolute", "a/../../escape", "a\\b", "a//b", "./file", "C:/file"])
def test_unsafe_paths_rejected(bad):
    with pytest.raises(m.MemoryError):
        m.validate_skill({"SKILL.md": "---\nname: example\ndescription: Example\n---\nBody", bad: "bad"})


def test_symlink_publish_and_install_destination_rejected(setup, tmp_path):
    client, store, work = setup
    folder = example_skill(tmp_path)
    (folder / "link").symlink_to(tmp_path / "outside")
    with pytest.raises(m.MemoryError, match="symlink"):
        client.publish(folder, cwd=work)
    (folder / "link").unlink()
    root = tmp_path / "linked-skills"
    root.symlink_to(folder, target_is_directory=True)
    with pytest.raises(m.MemoryError, match="Symlinks"):
        client.install_files(m.skill_files(folder), "release-check", root, {})


def test_core_name_cannot_be_published_or_installed_from_bucket(setup):
    client, _, work = setup
    with pytest.raises(m.MemoryError, match="pinned Space"):
        client.publish(ROOT / "memory-use", cwd=work)
    with pytest.raises(m.MemoryError, match="cannot replace"):
        client.install_shared("memory-use", "abc", cwd=work)


def test_reminder_has_no_network_or_transcript_reads(setup, tmp_path):
    client, store, work = setup
    store.calls.clear()
    payload = {"hook_event_name": "UserPromptSubmit", "cwd": str(work), "transcript_path": "/must/not/read", "prompt": "Do not echo my message"}
    result = m.hook_output(client, payload)
    assert result["hookSpecificOutput"]["hookEventName"] == "UserPromptSubmit"
    assert "Do not echo" not in json.dumps(result)
    assert not store.calls
    assert m.hook_output(client, {**payload, "cwd": str(tmp_path / "unmapped")}) == {}


def test_hook_merge_remove_preserves_unrelated_handlers(setup, tmp_path, monkeypatch):
    client, store, work = setup
    setup_runtime(client, tmp_path, store, work)
    config = tmp_path / "hooks.json"
    unrelated = {"type": "command", "command": "echo unrelated"}
    initial = {"description": "Existing hooks", "hooks": {"SessionStart": [{"matcher": "resume", "hooks": [unrelated]}], "Stop": [{"hooks": [unrelated]}]}}
    config.write_text(json.dumps(initial))
    monkeypatch.setattr(m.subprocess, "run", lambda *a, **k: SimpleNamespace(stdout="codex-cli 0.162.1"))
    m.hooks(client, "install", config)
    m.hooks(client, "install", config)
    value = json.loads(config.read_text())
    assert len(value["hooks"]["SessionStart"]) == 2
    m.hooks(client, "remove", config)
    assert json.loads(config.read_text()) == initial


def test_credentials_rejected_and_errors_do_not_echo_sdk_secrets(setup, monkeypatch):
    client, _, work = setup
    token = "hf_" + "x" * 30
    monkeypatch.setenv("HF_TOKEN", token)
    with pytest.raises(m.MemoryError, match="credential"):
        client.save(token, "Bad note", cwd=work)
    exc = Exception("Authorization: " + token)
    assert token not in str(m.error_from_exception(exc))


def test_cli_json_errors_and_hook_fail_open(tmp_path):
    script = ROOT / "memory-use/scripts/memory.py"
    command = [sys.executable, str(script), "--home", str(tmp_path / "state")]
    result = subprocess.run(command + ["search", "query", "--json"], capture_output=True, text=True, cwd=tmp_path)
    assert result.returncode == 1
    assert json.loads(result.stdout)["code"] == "unmapped"
    hook = subprocess.run(command + ["hook"], input="invalid JSON", capture_output=True, text=True)
    assert hook.returncode == 0
    assert "systemMessage" in json.loads(hook.stdout)


def test_native_network_child_has_hard_deadline(tmp_path, monkeypatch):
    hanging = tmp_path / "hanging.py"
    hanging.write_text("import time\ntime.sleep(10)\n")
    monkeypatch.setattr(m, "__file__", str(hanging))
    with pytest.raises(m.MemoryError, match="deadline"):
        m.HFStore(timeout=0.05).call("whoami")


def test_concurrent_saves_survive_with_distinct_objects(setup):
    client, store, work = setup
    with ThreadPoolExecutor(max_workers=4) as pool:
        futures = [pool.submit(client.save, f"Finding {i}", f"Concurrent {i}", cwd=work) for i in range(12)]
        results = [f.result() for f in futures]
    assert len({r["id"] for r in results}) == 12
    assert all(r["status"] == "saved" for r in results)
    assert len(store.buckets["owner/agent-memory"]["files"]) == 12
    assert not client.pending()


def test_remote_traversal_skill_never_changes_installation(setup, tmp_path):
    client, store, work = setup
    setup_runtime(client, tmp_path, store, work)
    prefix = "skills/hostile/revision/"
    store.buckets["owner/agent-memory"]["files"].update({
        prefix + "SKILL.md": "---\nname: hostile\ndescription: Test\n---\nInstructions",
        prefix + "../../../escaped": "Bad path",
    })
    with pytest.raises(m.MemoryError, match="Unsafe"):
        client.install_shared("hostile", "revision", cwd=work)
    assert not (tmp_path / "skills/hostile").exists()
    assert not (tmp_path / "escaped").exists()


def test_missing_authentication_has_clear_local_message(monkeypatch):
    import huggingface_hub
    monkeypatch.setattr(huggingface_hub, "get_token", lambda: None)
    with pytest.raises(m.MemoryError, match="No HF credentials") as result:
        m.hf_operation({"op": "whoami"})
    assert result.value.code == "auth"


def test_fresh_read_never_silently_falls_back_to_cache(setup):
    client, store, work = setup
    saved = client.save("Cached", "Fresh check", cwd=work)
    client.read(saved["id"], cwd=work)
    store.fail = "offline"
    assert client.read(saved["id"], cwd=work)["stale"]
    with pytest.raises(m.MemoryError):
        client.read(saved["id"], cwd=work, fresh=True)