nanobot/tests/agent/test_dream.py

"""Tests for Dream memory consolidation — build_dream_prompt and cursor management."""

import pytest

from nanobot.agent.memory import MemoryStore
from nanobot.config.schema import ModelPresetConfig
from nanobot.providers.base import LLMResponse
from nanobot.security.workspace_access import (
    bind_workspace_scope,
    default_workspace_scope,
    reset_workspace_scope,
)
from nanobot.utils.prompt_templates import render_template


@pytest.fixture
def store(tmp_path):
    s = MemoryStore(tmp_path)
    s.write_soul("# Soul\n- Helpful")
    s.write_memory("# Memory\n- Project X active")
    return s


class TestBuildDreamPrompt:
    def test_returns_none_when_no_history(self, store):
        assert store.build_dream_prompt() is None

    def test_returns_prompt_with_history(self, store):
        store.append_history("hello")
        result = store.build_dream_prompt()
        assert result is not None
        prompt, cursor = result
        assert cursor > 0
        assert "## Conversation History" in prompt
        assert "hello" in prompt

    def test_cursor_advances_only_new_entries(self, store):
        store.append_history("first")
        r1 = store.build_dream_prompt()
        assert r1 is not None
        _, c1 = r1

        # Cursor not yet advanced — same entries are still available
        assert store.build_dream_prompt() is not None

        # Advance cursor
        store.set_last_dream_cursor(c1)
        # Now no new entries
        assert store.build_dream_prompt() is None

        # Add new entry
        store.append_history("second")
        r2 = store.build_dream_prompt()
        assert r2 is not None
        _, c2 = r2
        assert c2 > c1

    def test_prompt_includes_skill_creator_path(self, store):
        store.append_history("test")
        result = store.build_dream_prompt()
        assert result is not None
        prompt, _ = result
        assert "skill-creator" in prompt

    def test_prompt_embeds_current_memory_file_contents(self, store):
        """Dream must see the real current file contents (Tier 4) so it edits the
        files, not a stale mental model."""
        store.append_history("hello")
        result = store.build_dream_prompt()
        assert result is not None
        prompt, _ = result
        assert "## Current Memory Files" in prompt
        assert "### SOUL.md" in prompt
        assert "### USER.md" in prompt
        assert "### memory/MEMORY.md" in prompt
        # Real current contents are embedded verbatim.
        assert "Project X active" in prompt
        assert "Helpful" in prompt

    def test_prompt_renders_missing_files_as_empty(self, tmp_path):
        store = MemoryStore(tmp_path)  # no durable files written
        store.append_history("hello")
        result = store.build_dream_prompt()
        assert result is not None
        prompt, _ = result
        assert "(empty)" in prompt

    def test_workspace_dream_prompt_overrides_default(self, store):
        store.dream_prompt_file.parent.mkdir(parents=True)
        store.dream_prompt_file.write_text(
            "Custom Dream prompt.",
            encoding="utf-8",
        )
        store.append_history("keep this fact")

        result = store.build_dream_prompt()

        assert result is not None
        prompt, _ = result
        assert prompt.startswith("Custom Dream prompt.")
        assert "memory consolidation engine" not in prompt
        assert "## Conversation History" in prompt
        assert "keep this fact" in prompt

    def test_workspace_dream_prompt_override_is_capped(self, store):
        store.dream_prompt_file.parent.mkdir(parents=True)
        store.dream_prompt_file.write_text("x" * 40_000, encoding="utf-8")
        store.append_history("keep this fact")

        result = store.build_dream_prompt()

        assert result is not None
        prompt, _ = result
        assert "x" * 40_000 not in prompt
        assert "... (truncated)" in prompt
        assert "## Conversation History" in prompt
        assert "keep this fact" in prompt

    def test_empty_workspace_dream_prompt_uses_default(self, store):
        store.dream_prompt_file.parent.mkdir(parents=True)
        store.dream_prompt_file.write_text("  \n", encoding="utf-8")
        store.append_history("test")

        result = store.build_dream_prompt()

        assert result is not None
        prompt, _ = result
        assert "memory consolidation engine" in prompt

    def test_truncates_long_entries(self, store):
        long_content = "x" * 2000
        store.append_history(long_content)
        result = store.build_dream_prompt()
        assert result is not None
        prompt, _ = result
        # The full 2000 chars should not appear — truncated to 500
        assert long_content not in prompt
        assert "x" * 500 in prompt

    def test_batches_oldest_unprocessed_entries_first(self, store):
        for i in range(25):
            store.append_history(f"entry-{i + 1:02d}")

        result = store.build_dream_prompt(max_entries=20)
        assert result is not None
        prompt, cursor = result

        assert cursor == 20
        assert "entry-01" in prompt
        assert "entry-20" in prompt
        assert "entry-21" not in prompt

        store.set_last_dream_cursor(cursor)
        next_result = store.build_dream_prompt(max_entries=20)
        assert next_result is not None
        next_prompt, next_cursor = next_result
        assert next_cursor == 25
        assert "entry-21" in next_prompt
        assert "entry-25" in next_prompt

    def test_skips_malformed_history_entries(self, store):
        """Dream prompt building should tolerate externally corrupted JSONL rows."""
        store.history_file.write_text(
            '{"cursor": 1, "timestamp": "2026-04-01 10:00"}\n'
            '{"cursor": 2, "timestamp": "2026-04-01 10:01", "content": "usable memory"}\n',
            encoding="utf-8",
        )

        result = store.build_dream_prompt()

        assert result is not None
        prompt, cursor = result
        assert cursor == 2
        assert "usable memory" in prompt

    def test_dream_prompt_consumes_consolidator_attribute_tags(self):
        prompt = render_template(
            "agent/dream.md",
            strip=True,
            skill_creator_path="skills/skill-creator/SKILL.md",
        )

        assert "History attribute tags" in prompt
        assert "[skip]: audit-only" in prompt
        assert "[correction]: replace the older conflicting fact" in prompt
        assert "Always strip these bracketed tags from saved memory content" in prompt


class TestDreamTools:
    def test_dream_tools_are_restricted_to_file_edits(self, store):
        tools = store.build_dream_tools()

        assert set(tools.tool_names) == {
            "apply_patch",
            "edit_file",
            "read_file",
            "write_file",
        }

    @pytest.mark.asyncio
    async def test_dream_can_edit_canonical_memory_files(self, store):
        tools = store.build_dream_tools()

        memory_result = await tools.execute(
            "apply_patch",
            {
                "edits": [
                    {
                        "path": "memory/MEMORY.md",
                        "action": "replace",
                        "old_text": "Project X active",
                        "new_text": "Project Y active",
                    }
                ]
            },
        )
        soul_result = await tools.execute(
            "edit_file",
            {
                "path": "SOUL.md",
                "old_text": "Helpful",
                "new_text": "Precise",
            },
        )

        assert "Patch applied" in memory_result
        assert "Successfully edited" in soul_result
        assert "Project Y active" in store.memory_file.read_text(encoding="utf-8")
        assert "Precise" in store.soul_file.read_text(encoding="utf-8")

    @pytest.mark.asyncio
    async def test_dream_can_write_workspace_skills(self, store):
        tools = store.build_dream_tools()
        target = store.workspace / "skills" / "demo" / "SKILL.md"

        result = await tools.execute(
            "write_file",
            {
                "path": "skills/demo/SKILL.md",
                "content": "---\nname: demo\ndescription: Demo skill.\n---\n\nUse when needed.\n",
            },
        )

        assert "Successfully wrote" in result
        assert target.read_text(encoding="utf-8").startswith("---\nname: demo")

    @pytest.mark.asyncio
    async def test_dream_tools_keep_internal_write_scope_under_full_access(self, store):
        tools = store.build_dream_tools()
        scope = default_workspace_scope(store.workspace, restrict_to_workspace=False)
        outside = store.workspace.parent / f"{store.workspace.name}-outside"
        outside.mkdir()
        outside_target = outside / "escape.txt"
        skill_target = store.workspace / "skills" / "scoped" / "SKILL.md"

        token = bind_workspace_scope(scope)
        try:
            outside_result = await tools.execute(
                "write_file",
                {"path": str(outside_target), "content": "owned"},
            )
            skill_result = await tools.execute(
                "apply_patch",
                {
                    "edits": [
                        {
                            "path": "skills/scoped/SKILL.md",
                            "action": "add",
                            "new_text": "---\nname: scoped\n---\n",
                        }
                    ]
                },
            )
        finally:
            reset_workspace_scope(token)

        assert "outside allowed directory" in outside_result
        assert not outside_target.exists()
        assert "Patch applied" in skill_result
        assert skill_target.read_text(encoding="utf-8").startswith("---\nname: scoped")

    @pytest.mark.asyncio
    async def test_dream_cannot_modify_memory_internal_files(self, store):
        tools = store.build_dream_tools()
        store.history_file.write_text("before\n", encoding="utf-8")
        store._dream_cursor_file.write_text("1", encoding="utf-8")

        history_result = await tools.execute(
            "apply_patch",
            {
                "edits": [
                    {
                        "path": "memory/history.jsonl",
                        "action": "replace",
                        "old_text": "before",
                        "new_text": "after",
                    }
                ]
            },
        )
        cursor_result = await tools.execute(
            "edit_file",
            {
                "path": "memory/.dream_cursor",
                "old_text": "1",
                "new_text": "2",
            },
        )

        assert "outside allowed directory" in history_result
        assert "outside allowed directory" in cursor_result
        assert store.history_file.read_text(encoding="utf-8") == "before\n"
        assert store._dream_cursor_file.read_text(encoding="utf-8") == "1"

    @pytest.mark.asyncio
    async def test_dream_cannot_create_children_under_canonical_files(self, store):
        tools = store.build_dream_tools()

        memory_child = store.memory_file / "evil.txt"
        user_child = store.user_file / "evil.txt"
        memory_result = await tools.execute(
            "apply_patch",
            {
                "edits": [
                    {
                        "path": "memory/MEMORY.md/evil.txt",
                        "action": "add",
                        "new_text": "owned",
                    }
                ]
            },
        )
        user_result = await tools.execute(
            "edit_file",
            {
                "path": "USER.md/evil.txt",
                "old_text": "",
                "new_text": "owned",
            },
        )

        assert "outside allowed directory" in memory_result
        assert "outside allowed directory" in user_result
        assert not memory_child.exists()
        assert not user_child.exists()


class TestEphemeralDirect:
    """Tests for the ephemeral flag that skips history.jsonl writes for Dream."""

    @pytest.fixture
    def _make_loop(self, tmp_path):
        """Factory fixture that builds a minimal AgentLoop with mocked deps."""
        from unittest.mock import AsyncMock, MagicMock, patch

        from nanobot.agent.loop import AgentLoop
        from nanobot.agent.memory import MemoryStore
        from nanobot.bus.queue import MessageBus

        store = MemoryStore(tmp_path)
        store.write_soul("# Soul")
        store.write_memory("# Memory")

        bus = MessageBus()
        provider = MagicMock()
        provider.get_default_model.return_value = "test-model"
        provider.supports_tools = True
        provider.generation = MagicMock(max_tokens=4096)
        provider.chat_with_retry = AsyncMock(
            return_value=LLMResponse(content="done", tool_calls=[], finish_reason="stop", usage={})
        )

        with (
            patch("nanobot.agent.loop.SessionManager"),
            patch("nanobot.agent.loop.SubagentManager") as mock_sub,
            patch("nanobot.agent.loop.Consolidator") as mock_consolidator_cls,
        ):
            mock_sub.return_value.cancel_by_session = AsyncMock(return_value=0)
            mock_consolidator_cls.return_value.maybe_consolidate_by_tokens = AsyncMock()
            loop = AgentLoop(
                bus=bus,
                provider=provider,
                workspace=tmp_path,
                context_window_tokens=8000,
            )

        return loop, store

    def test_dream_runtime_uses_preset_without_changing_default(self, _make_loop):
        loop, _ = _make_loop
        loop.runtime_resolver._model_presets = {
            "dream": ModelPresetConfig(model="dream-model"),
        }
        loop.dream_model_preset = "dream"

        runtime = loop.dream_runtime()

        assert runtime is not None
        assert runtime.model == "dream-model"
        assert runtime.model_preset == "dream"
        assert loop.model == "test-model"
        assert loop.model_preset is None

    async def test_ephemeral_skips_raw_archive(self, tmp_path, _make_loop):
        """When ephemeral=True, raw_archive must not be called."""
        from unittest.mock import patch

        loop, store = _make_loop

        with patch.object(loop.context.memory, "raw_archive") as mock_archive:
            await loop.process_direct(
                "test", session_key="dream:test", ephemeral=True,
            )
            mock_archive.assert_not_called()

    async def test_non_ephemeral_runs_normally(self, tmp_path, _make_loop):
        """Without ephemeral, the normal path returns the model response."""
        loop, store = _make_loop
        response = await loop.process_direct("test", session_key="cli:normal")

        assert response is not None
        assert response.content == "done"
        loop.provider.chat_with_retry.assert_awaited()

    async def test_ephemeral_sets_ctx_flag(self, tmp_path, _make_loop):
        """Verify that ephemeral=True is forwarded to TurnContext."""
        from unittest.mock import patch

        loop, store = _make_loop

        captured = {}

        original_save = loop._persist_turn

        async def patched_save(ctx):
            captured["ephemeral"] = ctx.ephemeral
            return await original_save(ctx)

        with patch.object(loop, "_persist_turn", side_effect=patched_save):
            await loop.process_direct(
                "test", session_key="dream:check", ephemeral=True,
            )

        assert captured.get("ephemeral") is True

    async def test_default_ephemeral_is_false(self, tmp_path, _make_loop):
        """By default ephemeral is False in TurnContext."""
        from unittest.mock import patch

        loop, store = _make_loop

        captured = {}

        original_save = loop._persist_turn

        async def patched_save(ctx):
            captured["ephemeral"] = ctx.ephemeral
            return await original_save(ctx)

        with patch.object(loop, "_persist_turn", side_effect=patched_save):
            await loop.process_direct("test", session_key="cli:normal")

        assert captured.get("ephemeral") is False

    async def test_ephemeral_skips_consolidator(self, tmp_path, _make_loop):
        """When ephemeral=True, consolidator.maybe_consolidate_by_tokens is not called."""
        from unittest.mock import patch

        loop, store = _make_loop

        with patch.object(
            loop.consolidator, "maybe_consolidate_by_tokens",
        ) as mock_consolidate:
            await loop.process_direct(
                "test", session_key="dream:consolidate-test", ephemeral=True,
            )
            mock_consolidate.assert_not_called()

    async def test_ephemeral_response_reports_stop_reason(self, tmp_path, _make_loop):
        loop, store = _make_loop
        loop.provider.chat_with_retry.return_value = LLMResponse(
            content="provider error",
            finish_reason="error",
        )

        resp = await loop.process_direct(
            "test", session_key="dream:error", ephemeral=True,
        )

        assert resp is not None
        assert resp.metadata["_stop_reason"] == "error"
        assert MemoryStore.dream_run_completed(resp) is False

    async def test_dream_turn_can_skip_unbatched_recent_history(self, tmp_path):
        """Dream must only see the batch selected by build_dream_prompt."""
        from unittest.mock import MagicMock

        from nanobot.agent.loop import AgentLoop
        from nanobot.bus.queue import MessageBus

        store = MemoryStore(tmp_path)
        for i in range(60):
            store.append_history(f"entry-{i + 1:02d}")

        result = store.build_dream_prompt(max_entries=20)
        assert result is not None
        prompt, cursor = result
        assert cursor == 20

        captured: dict[str, list[dict]] = {}
        provider = MagicMock()
        provider.get_default_model.return_value = "test-model"
        provider.supports_tools = True
        provider.generation = MagicMock(max_tokens=4096)

        async def chat_with_retry(**kwargs):
            captured["messages"] = kwargs["messages"]
            return LLMResponse(content="done", finish_reason="stop")

        provider.chat_with_retry = chat_with_retry
        loop = AgentLoop(
            bus=MessageBus(),
            provider=provider,
            workspace=tmp_path,
            context_window_tokens=8000,
        )

        await loop.process_direct(
            prompt,
            session_key="dream:test",
            ephemeral=True,
            tools=store.build_dream_tools(),
        )

        messages = captured["messages"]
        system_prompt = messages[0]["content"]
        request_text = "\n".join(str(message.get("content", "")) for message in messages)
        assert "# Recent History" not in system_prompt
        assert "entry-01" in request_text
        assert "entry-20" in request_text
        assert "entry-21" not in request_text
        assert "entry-60" not in request_text


class TestEphemeralHooks:
    """When ephemeral=True, extra hooks must not fire."""

    @pytest.fixture
    def _make_loop_with_spy(self, tmp_path):
        """Build an AgentLoop with a spy hook to verify hook firing behavior."""
        from unittest.mock import AsyncMock, MagicMock, patch

        from nanobot.agent.hook import AgentHook
        from nanobot.agent.loop import AgentLoop
        from nanobot.bus.queue import MessageBus

        bus = MessageBus()
        provider = MagicMock()
        provider.get_default_model.return_value = "test-model"
        provider.supports_tools = True
        provider.generation = MagicMock(max_tokens=4096)
        provider.chat_with_retry = AsyncMock(
            return_value=MagicMock(
                content="done", finish_reason="stop", tool_calls=[], usage={},
            )
        )

        spy = MagicMock(spec=AgentHook)
        spy.wants_streaming.return_value = False
        spy.before_iteration = AsyncMock()
        spy.after_iteration = AsyncMock()

        with (
            patch("nanobot.agent.loop.SessionManager"),
            patch("nanobot.agent.loop.SubagentManager") as mock_sub,
            patch("nanobot.agent.loop.Consolidator") as mock_consolidator_cls,
        ):
            mock_sub.return_value.cancel_by_session = AsyncMock(return_value=0)
            mock_consolidator_cls.return_value.maybe_consolidate_by_tokens = AsyncMock()
            loop = AgentLoop(
                bus=bus,
                provider=provider,
                workspace=tmp_path,
                context_window_tokens=8000,
                hooks=[spy],
            )

        return loop, spy

    async def test_extra_hooks_skipped_when_ephemeral(self, tmp_path, _make_loop_with_spy):
        """When ephemeral=True, extra hooks must not fire."""
        loop, spy = _make_loop_with_spy

        await loop.process_direct(
            "test", session_key="dream:hook-test", ephemeral=True,
        )
        spy.before_iteration.assert_not_called()
        spy.after_iteration.assert_not_called()

    async def test_extra_hooks_fire_for_normal_sessions(self, tmp_path, _make_loop_with_spy):
        """Without ephemeral, extra hooks should fire normally."""
        loop, spy = _make_loop_with_spy

        await loop.process_direct("test", session_key="cli:normal")
        spy.before_iteration.assert_called()

class TestDreamCommitMessage:
    def test_commit_message_reflects_real_diff_not_narrative(self, tmp_path):
        """The Dream commit message must mirror the real git diff and ignore the
        LLM's narrative, so ``/dream-log`` can never lie.

        Regression for the hallucinated-commit bug: commit ``a72ca2a`` claimed a
        "Medical Research" section that never reached the diff.
        """
        import subprocess

        store = MemoryStore(tmp_path)
        store.write_soul("# Soul")
        store.write_memory("# Memory")
        store.git.init()
        store.git.auto_commit("initial state")

        # A real edit to a tracked content file.
        store.write_memory("# Memory\n- DMSO research notes")

        # A lying narrative the old code would have appended verbatim.
        lying = "Added a Medical Research section (Mastic Gum, DMSO) to MEMORY.md"
        diff_body = store.dream_content_diff()
        assert diff_body, "real edit must be detected"
        assert "DMSO research notes" in diff_body
        assert lying not in diff_body

        msg = MemoryStore.build_dream_commit_message(
            "dream: periodic memory consolidation", diff_body,
        )
        assert lying not in msg
        assert "DMSO research notes" in msg

        sha = store.git.auto_commit(msg)
        assert sha is not None
        log = subprocess.check_output(
            ["git", "log", "-1", "--format=%B"],
            cwd=str(tmp_path), text=True,
        ).strip()
        assert "dream: periodic memory consolidation" in log
        assert "DMSO research notes" in log
        assert lying not in log

    def test_commit_message_is_bare_prefix_when_no_changes(self, tmp_path):
        """A no-op Dream run yields only the prefix — never a narrated summary."""
        store = MemoryStore(tmp_path)
        store.write_soul("# Soul")
        store.write_memory("# Memory")
        store.git.init()
        store.git.auto_commit("initial state")

        # No edits at all.
        assert store.dream_content_diff() == ""
        msg = MemoryStore.build_dream_commit_message(
            "dream: manual run", store.dream_content_diff(),
        )
        assert msg == "dream: manual run"

    def test_build_commit_message_ignores_none_and_empty_body(self):
        assert MemoryStore.build_dream_commit_message("dream: x", "") == "dream: x"
        assert MemoryStore.build_dream_commit_message("dream: x", None) == "dream: x"
        assert MemoryStore.build_dream_commit_message("dream: x", "  ") == "dream: x"
        assert (
            MemoryStore.build_dream_commit_message("dream: x", "SOUL.md: +1 -0")
            == "dream: x\n\nSOUL.md: +1 -0"
        )


class TestDreamContentDiff:
    """The ground-truth signal that gates cursor advance and commit messages."""

    def test_empty_when_git_not_initialized(self, store):
        assert store.dream_content_diff() == ""

    def test_empty_when_no_tracked_changes(self, store):
        store.git.init()
        store.git.auto_commit("initial")
        assert store.dream_content_diff() == ""

    def test_ignores_platform_line_ending_normalization(self, store, monkeypatch):
        import subprocess

        system_config = store.workspace / "system.gitconfig"
        global_config = store.workspace / "global.gitconfig"
        system_config.touch()
        global_config.touch()
        subprocess.run(
            [
                "git", "config", "--file", str(system_config),
                "core.autocrlf", "false",
            ],
            check=True,
            capture_output=True,
        )
        monkeypatch.delenv("GIT_CONFIG_NOSYSTEM", raising=False)
        monkeypatch.setenv("GIT_CONFIG_SYSTEM", str(system_config))
        monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(global_config))

        store.soul_file.write_bytes(b"# Soul\r\n- Helpful")
        store.memory_file.write_bytes(b"# Memory\r\n- Project X active")
        store.git.init()

        status = subprocess.check_output(
            ["git", "status", "--porcelain", "--", "SOUL.md", "memory/MEMORY.md"],
            cwd=store.workspace,
            text=True,
        )
        assert status == ""
        assert store.dream_content_diff() == ""

    @pytest.mark.parametrize(
        ("before", "after", "changed"),
        [
            (b"# Memory\r\n", b"# Memory\n", False),
            (b"# Memory\n", b"# Memory", True),
            (b"# Memory\r", b"# Memory\n", True),
        ],
    )
    def test_only_ignores_crlf_lf_changes(self, store, before, after, changed):
        store.memory_file.write_bytes(before)
        store.git.init()

        store.memory_file.write_bytes(after)

        assert bool(store.dream_content_diff()) is changed

    def test_reflects_real_content_edits(self, store):
        store.git.init()
        store.git.auto_commit("initial")
        store.write_memory("# Memory\n- DMSO research notes")
        diff = store.dream_content_diff()
        assert diff
        assert "memory/MEMORY.md" in diff
        assert "DMSO research notes" in diff

    def test_ignores_cursor_only_changes(self, store):
        """Advancing the cursor must not count as a productive content edit."""
        store.git.init()
        store.git.auto_commit("initial")
        store.set_last_dream_cursor(99)  # only memory/.dream_cursor changes
        assert store.dream_content_diff() == ""