smart-org/apps/api/tests/test_engine_prompt.py

"""The prompt builder: a stable prefix, a bounded tail.

What these tests defend:

* the prefix order is the plan's order (rules → protocol → toolkit → app
  profile → memory), because that order is what makes prompt caching useful
* the task brief is capped, and the history truncation drops the **oldest tool
  results first** — a model that forgets the rules is worse than one that
  forgot what it read three calls ago
* nothing secret reaches the prompt: the builder only ever sees the strings it
  is handed, and the runner never hands it settings
"""

from __future__ import annotations

from app.application.engine.prompt import (
    PLATFORM_RULES,
    TOOLKIT_DESCRIPTION,
    TurnContext,
    TurnMessage,
    build_turn_prompt,
)
from app.core.config import settings
from app.domain.workforce.models import RoleKey
from app.domain.workforce.pipeline import Mode, Stage


def _ctx(**kwargs) -> TurnContext:
    base = {
        "organization_id": "org-1",
        "task_id": "task-1",
        "project_id": "proj-1",
        "task_title": "ساخت صفحهٔ ورود",
        "task_description": "توضیح کار",
        "stage": Stage.PENDING_IMPLEMENTATION,
        "role": RoleKey.DEVELOPER,
        "mode": Mode.WORK,
        "app_profile_text": "زبان: پایتون\nدستور تست: pytest -q",
        "memory_context": "این سازمان تست‌ها را با pytest اجرا می‌کند.",
    }
    base.update(kwargs)
    return TurnContext(**base)


def test_the_prefix_keeps_the_plan_order():
    prompt = build_turn_prompt(_ctx())
    system = prompt.system
    positions = [
        system.index(PLATFORM_RULES.splitlines()[0]),  # 1. platform rules
        system.index("تو برنامه‌نویس هستی"),  # 2. the role protocol
        system.index(TOOLKIT_DESCRIPTION.splitlines()[0]),  # 3. the toolkit
        system.index("دستور تست: pytest -q"),  # 4. the app profile
        system.index("این سازمان"),  # 5. org memory
    ]
    assert positions == sorted(positions)
    assert prompt.messages[-1].role == "user"  # 6/7. brief and history last


def test_the_toolkit_is_only_shown_to_roles_that_have_tools():
    developer = build_turn_prompt(_ctx())
    manager = build_turn_prompt(
        _ctx(role=RoleKey.MANAGER, mode=Mode.TRIAGE, stage=Stage.PENDING_MANAGER_REVIEW)
    )
    assert "READ_FILE" in developer.system
    assert "READ_FILE" not in manager.system


def test_the_task_brief_is_capped():
    prompt = build_turn_prompt(_ctx(task_description="x" * 5000))
    brief = prompt.messages[-1].content
    # The description is cut, the title survives.
    assert "ساخت صفحهٔ ورود" in brief
    assert len(brief) < 2000


def test_stage_context_only_goes_to_the_role_that_needs_it():
    """A reviewer sees the diff; a developer sees the spec; nobody sees both."""
    developer = build_turn_prompt(
        _ctx(spec="برنامهٔ برنامه‌ریز", diff_text="diff --git a/x b/x")
    )
    assert "برنامهٔ برنامه‌ریز" in developer.messages[-1].content
    assert "diff --git" not in developer.messages[-1].content

    reviewer = build_turn_prompt(
        _ctx(
            role=RoleKey.REVIEWER,
            mode=Mode.WORK,
            stage=Stage.PENDING_REVIEW,
            spec="برنامهٔ برنامه‌ریز",
            diff_text="diff --git a/x b/x",
        )
    )
    assert "diff --git" in reviewer.messages[-1].content
    assert "برنامهٔ برنامه‌ریز" not in reviewer.messages[-1].content


def test_history_drops_the_oldest_results_first():
    big = "y" * 4000
    history = [
        TurnMessage(kind="result", text=big),  # oldest: dropped first
        TurnMessage(kind="result", text=big),
        TurnMessage(kind="result", text="نتیجهٔ تازه و مهم"),
    ]
    settings.engine_prompt_char_cap = 9000  # keep the prefix small
    try:
        prompt = build_turn_prompt(_ctx(history=list(history)))
    finally:
        settings.engine_prompt_char_cap = 24000
    joined = "\n".join(m.content for m in prompt.messages)
    assert "نتیجهٔ تازه و مهم" in joined
    assert joined.count("y" * 4000) <= 1
    assert prompt.total_chars <= 9000 + 4000


def test_a_tiny_cap_still_keeps_the_rules_and_the_brief():
    settings.engine_prompt_char_cap = 3000
    try:
        prompt = build_turn_prompt(
            _ctx(history=[TurnMessage(kind="result", text="z" * 20000)])
        )
    finally:
        settings.engine_prompt_char_cap = 24000
    # The prefix and the brief survive; only history space is sacrificed.
    assert "قوانین پلتفرم" in prompt.system
    assert "ساخت صفحهٔ ورود" in prompt.messages[-1].content


def test_the_prompt_never_contains_the_settings(monkeypatch):
    monkeypatch.setattr(settings, "jwt_secret", "top-secret")
    monkeypatch.setattr(settings, "database_url", "sqlite:///private.db")
    prompt = build_turn_prompt(
        _ctx(history=[TurnMessage(kind="result", text="نتیجه")])
    )
    text = prompt.system + "\n" + "\n".join(m.content for m in prompt.messages)
    assert "top-secret" not in text
    assert "private.db" not in text
    assert PLATFORM_RULES.splitlines()[0] in text


def test_manager_modes_get_their_own_protocol_hint():
    triage = build_turn_prompt(
        _ctx(role=RoleKey.MANAGER, mode=Mode.TRIAGE, stage=Stage.PENDING_MANAGER_REVIEW)
    )
    final = build_turn_prompt(
        _ctx(role=RoleKey.MANAGER, mode=Mode.FINAL, stage=Stage.PENDING_FINAL_REVIEW)
    )
    assert "ASSIGN" in triage.system
    assert "ASSIGN" not in final.system
    assert "COMPLETE" in final.system


def test_the_toolkit_description_lists_every_tool_tag():
    for tool in ("READ_FILE", "WRITE_FILE", "EDIT_FILE", "LIST_DIR", "SEARCH", "RUN"):
        assert tool in TOOLKIT_DESCRIPTION