orchestrator/agent-team/tests/test_ws5_memory_handbook.py

"""WS5 — memory + handbook injection seams tests.

Covers:
* retriever.save_memory() writes to _box-drafts/ subdir
* retriever.retrieve() memory_dir param routes to correct dir
* handbook.load_handbook_conventions() safe no-op when dir absent
* handbook.load_handbook_conventions() reads files when dir exists
* clarifier_llm: context_provider absent -> identical prompt; present -> prepended
* planner: context_provider absent -> identical prompt; present -> prepended
* coordinator.default_plan_node_factory: threads context_provider into planner
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

import pytest

# retriever.py lives at the orchestrator root (parents[2] of this file
# which is at agent-team/tests/test_ws5_memory_handbook.py)
_ORCHESTRATOR_ROOT = Path(__file__).resolve().parents[2]
if str(_ORCHESTRATOR_ROOT) not in sys.path:
    sys.path.insert(0, str(_ORCHESTRATOR_ROOT))


# ---------------------------------------------------------------------------
# retriever: save_memory
# ---------------------------------------------------------------------------


def test_save_memory_writes_to_box_drafts(tmp_path: Path) -> None:
    from retriever import save_memory

    dest = save_memory("my-note", "hello world", memory_dir=tmp_path)
    assert dest == tmp_path / "_box-drafts" / "my-note.md"
    assert dest.exists()
    assert dest.read_text() == "hello world"


def test_save_memory_creates_subdir(tmp_path: Path) -> None:
    from retriever import save_memory

    save_memory("test", "content", memory_dir=tmp_path)
    assert (tmp_path / "_box-drafts").is_dir()


def test_save_memory_rejects_path_traversal(tmp_path: Path) -> None:
    from retriever import save_memory

    with pytest.raises(ValueError):
        save_memory("../evil", "bad", memory_dir=tmp_path)
    with pytest.raises(ValueError):
        save_memory("sub/dir", "bad", memory_dir=tmp_path)
    with pytest.raises(ValueError):
        save_memory("", "bad", memory_dir=tmp_path)


def test_save_memory_does_not_appear_in_live_dir(tmp_path: Path) -> None:
    from retriever import load_memories, save_memory

    save_memory("draft", "draft content", memory_dir=tmp_path)
    # load_memories reads the memory_dir top level, not _box-drafts/
    memories = load_memories(tmp_path)
    names = [m.name for m in memories]
    assert "draft" not in names


def test_retrieve_uses_memory_dir_param(tmp_path: Path) -> None:
    """retrieve(memory_dir=...) routes to the given dir (empty -> [])."""
    from retriever import retrieve

    # Empty dir -> no memories -> returns []
    result = retrieve("some task", memory_dir=tmp_path)
    assert result == []


# ---------------------------------------------------------------------------
# handbook: load_handbook_conventions
# ---------------------------------------------------------------------------


def test_handbook_no_op_when_dir_absent(tmp_path: Path) -> None:
    from agent_team.nodes.handbook import load_handbook_conventions

    missing = tmp_path / "nonexistent"
    result = load_handbook_conventions(missing)
    assert result == ""


def test_handbook_no_op_when_empty_dir(tmp_path: Path) -> None:
    from agent_team.nodes.handbook import load_handbook_conventions

    result = load_handbook_conventions(tmp_path)
    assert result == ""


def test_handbook_loads_md_files(tmp_path: Path) -> None:
    from agent_team.nodes.handbook import load_handbook_conventions

    (tmp_path / "conventions.md").write_text("# Conventions\nUse typed Python.")
    (tmp_path / "ci.md").write_text("# CI\nRun ruff + pytest.")
    result = load_handbook_conventions(tmp_path)
    assert "Use typed Python" in result
    assert "Run ruff + pytest" in result
    assert "## Sea Haven engineering-handbook conventions" in result


def test_handbook_never_raises_on_unreadable_file(tmp_path: Path) -> None:
    from agent_team.nodes.handbook import load_handbook_conventions

    (tmp_path / "ok.md").write_text("content")
    # Simulate an IOError by pointing to a non-directory path as the dir
    result = load_handbook_conventions(tmp_path / "ok.md")
    assert result == ""


# ---------------------------------------------------------------------------
# clarifier_llm: context_provider injection seam
# ---------------------------------------------------------------------------


def _make_state(**kwargs: Any) -> dict[str, Any]:
    return {"task": "add a login form", "thread_id": "t1", **kwargs}


def _fake_invoke(prompt: str, **kw: Any) -> Any:
    from agent_team.billing import ClaudeResult

    return ClaudeResult(
        text='{"confidence": 0.9, "questions": [], "rationale": "clear"}',
        mode="api",
        usage={},
        raw=None,
    )


def test_clarifier_no_context_provider_produces_base_prompt() -> None:
    from agent_team.nodes.clarifier_llm import ClaudeClarifier

    c = ClaudeClarifier(invoke=_fake_invoke)
    state = _make_state()
    prompt = c._build_prompt([], state)
    assert "You are the CLARIFIER" in prompt
    # No context header injected
    assert "Sea Haven" not in prompt


def test_clarifier_context_provider_prepends_context() -> None:
    from agent_team.nodes.clarifier_llm import ClaudeClarifier

    def _provider() -> str:
        return "## Context\nsome memory"

    c = ClaudeClarifier(invoke=_fake_invoke, context_provider=_provider)
    state = _make_state()
    prompt = c._build_prompt([], state)
    assert prompt.startswith("## Context\nsome memory")
    assert "You are the CLARIFIER" in prompt


def test_clarifier_no_provider_prompt_identical_to_baseline() -> None:
    """Absent context_provider -> exact same output as no-kwarg construction."""
    from agent_team.nodes.clarifier_llm import ClaudeClarifier

    state = _make_state()
    base = ClaudeClarifier(invoke=_fake_invoke)._build_prompt([], state)
    explicit_none = ClaudeClarifier(
        invoke=_fake_invoke, context_provider=None
    )._build_prompt([], state)
    assert base == explicit_none


def test_clarifier_provider_failure_is_silent() -> None:
    """A context_provider that raises must not propagate to the caller."""

    def _bad_provider() -> str:
        raise RuntimeError("network down")

    from agent_team.nodes.clarifier_llm import ClaudeClarifier

    c = ClaudeClarifier(invoke=_fake_invoke, context_provider=_bad_provider)
    state = _make_state()
    prompt = c._build_prompt([], state)
    assert "You are the CLARIFIER" in prompt


def test_build_claude_clarifier_callables_accepts_context_provider() -> None:
    from agent_team.nodes.clarifier_llm import build_claude_clarifier_callables

    called: list[int] = []

    def _provider() -> str:
        called.append(1)
        return "## ctx\nhi"

    assess, generate = build_claude_clarifier_callables(
        invoke=_fake_invoke, context_provider=_provider
    )
    state = _make_state()
    # Calling assess_confidence triggers a prompt build -> provider call
    assess([], state)
    assert called, "context_provider was not called"


# ---------------------------------------------------------------------------
# planner: build_plan_prompt context_provider
# ---------------------------------------------------------------------------


def _make_plan_state(**kw: Any) -> Any:
    from agent_team.task_model import PipelineState

    return PipelineState(task="add a login form", **kw)


def test_planner_no_context_provider_prompt_unchanged() -> None:
    from agent_team.nodes.planner import build_plan_prompt

    state = _make_plan_state()
    base = build_plan_prompt(state)
    explicit_none = build_plan_prompt(state, context_provider=None)
    assert base == explicit_none


def test_planner_context_provider_prepends_context() -> None:
    from agent_team.nodes.planner import build_plan_prompt

    def _provider() -> str:
        return "## Memory\nremember this"

    state = _make_plan_state()
    prompt = build_plan_prompt(state, context_provider=_provider)
    assert prompt.startswith("## Memory\nremember this")
    assert "You are the PLANNER" in prompt


def test_planner_provider_failure_silent() -> None:
    from agent_team.nodes.planner import build_plan_prompt

    def _bad() -> str:
        raise ValueError("oops")

    state = _make_plan_state()
    prompt = build_plan_prompt(state, context_provider=_bad)
    assert "You are the PLANNER" in prompt


# ---------------------------------------------------------------------------
# coordinator.default_plan_node_factory threads context_provider
# ---------------------------------------------------------------------------


def test_default_plan_node_factory_no_provider_returns_callable() -> None:
    from agent_team.coordinator import default_plan_node_factory

    node = default_plan_node_factory()
    assert callable(node)


def test_default_plan_node_factory_with_provider_returns_callable() -> None:
    from agent_team.coordinator import default_plan_node_factory

    node = default_plan_node_factory(context_provider=lambda: "ctx")
    assert callable(node)