FabledScribe/tests/test_milestone_summary_brief.py

"""The enter_project handshake stays a small primer (#4045).

enter_project once carried every milestone's full plan body, the whole project
record, full rule text and ~9k of design guidance. On a project with 39
milestones that came to ~222k characters, past what an MCP client accepts as
a tool result, so the session handshake arrived as a file to page through.
"""
import contextlib
import json
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from scribe.mcp.tools.milestones import list_milestones
from scribe.mcp.tools.projects import enter_project, get_project
from scribe.services.milestones import brief_milestone_summary
from scribe.services.milestones import unplanned_milestones as brief_unplanned
from tests.helpers import fake_project


pytestmark = pytest.mark.usefixtures("_bind_user")

PLAN = "A plan paragraph long enough to matter. " * 125  # ~5k chars
GOAL = "What the project is for. " * 50                  # ~1.2k chars


def _milestone(mid: int, status: str, touched_day: int) -> dict:
    """A summary row as get_project_milestone_summary returns it."""
    touched = f"2026-08-{touched_day:02d}T00:00:00+00:00"
    return {
        "id": mid, "user_id": 7, "project_id": 5, "title": f"M{mid}",
        "description": f"what M{mid} is for", "body": PLAN, "status": status,
        "order_index": mid, "created_at": "2026-01-01T00:00:00+00:00",
        "updated_at": "2026-01-01T00:00:00+00:00", "last_touched_at": touched,
        "total": 4, "completed": 2, "pct": 50.0,
        "status_counts": {"todo": 2, "in_progress": 0, "done": 2, "cancelled": 0},
    }


def _history(count: int) -> list[dict]:
    """`count` milestones, alternating done/active; higher id = touched later."""
    return [
        _milestone(i, "done" if i % 2 else "active", 1 + i % 28)
        for i in range(count)
    ]


def test_brief_rows_leave_out_the_plan_and_what_the_caller_already_knows():
    brief, omitted = brief_milestone_summary([_milestone(1, "active", 3)])
    assert omitted == 0
    assert brief == [{
        "id": 1, "title": "M1", "description": "what M1 is for", "status": "active",
        "order_index": 1, "total": 4, "completed": 2, "pct": 50.0,
        "status_counts": {"todo": 2, "in_progress": 0, "done": 2, "cancelled": 0},
    }]


def test_limit_keeps_the_most_recently_touched_whatever_their_status():
    """A plan can sit "active" for months; recency is what says it's current."""
    rows = [_milestone(1, "active", 1), _milestone(2, "done", 9),
            _milestone(3, "active", 5), _milestone(4, "done", 7)]
    brief, omitted = brief_milestone_summary(rows, limit=2)
    assert omitted == 2
    assert [r["id"] for r in brief] == [2, 4]   # most recent first


def test_no_limit_keeps_every_row_in_order():
    brief, omitted = brief_milestone_summary(_history(8))
    assert omitted == 0
    assert [r["id"] for r in brief] == list(range(8))


def _task(tid: int, milestone_id: int | None) -> MagicMock:
    t = MagicMock()
    t.id = tid
    t.title = f"T{tid}"
    t.status = "todo"
    t.milestone_id = milestone_id
    return t


def _enter_stubs(project, milestones: list[dict], tasks: list, *, rules=None, systems=None,
                 design=None):
    applicable = rules or {"rules": [], "project_rules": [], "truncated": False}
    return [
        patch("scribe.mcp.tools.projects.projects_svc.get_project",
              AsyncMock(return_value=project)),
        patch("scribe.mcp.tools.projects.rulebooks_svc.get_applicable_rules",
              AsyncMock(return_value=applicable)),
        patch("scribe.mcp.tools.projects.milestones_svc.get_project_milestone_summary",
              AsyncMock(return_value=milestones)),
        patch("scribe.mcp.tools.projects.notes_svc.list_notes",
              AsyncMock(return_value=(tasks, len(tasks)))),
        patch("scribe.mcp.tools.projects.systems_svc.list_systems",
              AsyncMock(return_value=systems or [])),
        patch("scribe.mcp.tools.projects.systems_tools.bootstrap_systems_ask",
              AsyncMock(return_value=None)),
        patch("scribe.mcp.tools.projects.design_systems_svc.design_context",
              AsyncMock(return_value=design)),
        patch("scribe.mcp.tools.projects.coverage_svc.cached_coverage",
              AsyncMock(return_value=None)),
        patch("scribe.mcp.tools.projects.spawn"),
    ]


async def _enter(*stubs):
    with contextlib.ExitStack() as stack:
        mocks = [stack.enter_context(cm) for cm in stubs]
        out = await enter_project(project_id=5)
    return out, mocks


@pytest.mark.asyncio
async def test_enter_project_stays_small_however_large_the_project():
    """The ceiling is the point: a long-lived project's handshake must not
    grow with its history. 200 milestones with 5k-character plans, 60 project
    rules and 40 Systems would be well over 1M characters in the old shape."""
    project = fake_project(id=5, design_system_id=9, goal=GOAL,
                           description="Background. " * 300)
    rules = {
        "rules": [{"id": i, "title": f"r{i}", "statement": PLAN} for i in range(50)],
        "project_rules": [{"id": 100 + i, "title": f"pr{i}", "statement": PLAN,
                           "when_to_apply": PLAN} for i in range(60)],
        "truncated": True,
    }
    systems = []
    for i in range(40):
        s = MagicMock()
        s.id, s.name, s.description = i, f"Area {i}", PLAN
        systems.append(s)
    design = {"id": 9, "title": "Kit", "description": "", "inherits_from": ["House"],
              "guidance": [{"design_system_id": 9, "title": "Kit", "guidance": PLAN * 2}],
              "token_count": 111, "token_groups": ["accent", "surface"]}
    tasks = [_task(1000 + i, i % 200) for i in range(10)]

    out, _ = await _enter(*_enter_stubs(project, _history(200), tasks, rules=rules,
                                        systems=systems, design=design))

    assert len(out["milestone_summary"]) == 5
    assert out["milestone_summary_omitted"].startswith("195 other milestone(s)")
    assert len(out["open_tasks"]) == 10
    assert "applicable_rules" not in out and "recent_notes" not in out
    assert "guidance" not in out["design_system"]
    # The fixed parts are bounded by their caps; what's left to grow is the
    # goal, the rule and System titles, and the ask keys when they apply.
    size = len(json.dumps(out, indent=2))
    assert size < 16_000, size


@pytest.mark.asyncio
async def test_open_tasks_name_their_milestone_even_when_it_is_not_listed():
    """The milestone list is capped at 5; a task's milestone can fall outside
    it, and its id must not arrive without a name."""
    milestones = _history(20)
    tasks = [_task(1, 0), _task(2, None)]   # milestone 0 is the least recent
    out, mocks = await _enter(*_enter_stubs(fake_project(id=5), milestones, tasks))

    assert 0 not in [m["id"] for m in out["milestone_summary"]]
    assert out["open_tasks"] == [
        {"id": 1, "title": "T1", "status": "todo", "milestone_id": 0, "milestone_title": "M0"},
        {"id": 2, "title": "T2", "status": "todo", "milestone_id": None, "milestone_title": None},
    ]
    list_notes = mocks[3]
    assert list_notes.await_args.kwargs["sort"] == "touched"
    assert list_notes.await_args.kwargs["limit"] == 10


@pytest.mark.asyncio
async def test_omitted_key_is_absent_when_nothing_was_left_out():
    """Attached only when it applies (#2483)."""
    out, _ = await _enter(*_enter_stubs(fake_project(id=5), _history(3), []))
    assert len(out["milestone_summary"]) == 3
    assert "milestone_summary_omitted" not in out


@pytest.mark.asyncio
async def test_get_project_lists_every_milestone_without_plans():
    with patch("scribe.mcp.tools.projects.projects_svc.get_project",
               AsyncMock(return_value=fake_project(id=5))), \
         patch("scribe.mcp.tools.projects.milestones_svc.get_project_milestone_summary",
               AsyncMock(return_value=_history(10))), \
         patch("scribe.mcp.tools.projects.rulebooks_svc.get_applicable_rules",
               AsyncMock(return_value={"rules": [], "truncated": False})):
        out = await get_project(project_id=5)
    assert len(out["milestone_summary"]) == 10
    assert all("body" not in m for m in out["milestone_summary"])


@pytest.mark.asyncio
async def test_list_milestones_lists_every_milestone_without_plans():
    """The call milestone_summary_omitted points to."""
    with patch("scribe.mcp.tools.milestones.milestones_svc.get_project_milestone_summary",
               AsyncMock(return_value=_history(30))):
        out = await list_milestones(project_id=5)
    assert len(out["milestones"]) == 30
    assert all("body" not in m for m in out["milestones"])


def test_a_planning_read_lists_rules_without_restating_them():
    """#4081: a project's listing is every global rule tagged to its areas, so
    the full rule_brief of each put start_planning at 92k characters. Planning
    reads name the rules; get_rule reads one."""
    from scribe.services.rulebooks import rules_payload

    applicable = {
        "rules": [{"id": i, "title": f"r{i}", "statement": PLAN, "when_to_apply": PLAN,
                   "topic_title": "git", "relations": [{"note": PLAN}]} for i in range(50)]
                 + [{"id": 900, "title": "partner", "statement": PLAN, "via": "co_surfaces"}],
        "project_rules": [{"id": 100 + i, "title": f"pr{i}", "statement": PLAN}
                          for i in range(30)],
        "truncated": True,
    }
    with patch("scribe.services.rulebooks.record_rule_surfaced") as surfaced:
        out = rules_payload(applicable, user_id=7, source="start_planning")

    assert out["applicable_rules"][0] == {"id": 0, "title": "r0", "topic_title": "git"}
    assert out["applicable_rules"][-1] == {"id": 900, "title": "partner", "via": "co_surfaces"}
    assert out["project_rules"][0] == {"id": 100, "title": "pr0"}
    assert out["applicable_rules_truncated"] is True
    assert len(surfaced.call_args.kwargs["rule_ids"]) == 81
    assert len(json.dumps(out)) < 6_000, len(json.dumps(out))


# ── Milestones with no steps are open work (milestone 415) ─────────────────


def _planless(mid: int, status: str = "active") -> dict:
    """A roadmap milestone: a description and no steps, never touched since."""
    row = _milestone(mid, status, touched_day=1)
    row.update(total=0, completed=0, pct=0.0,
               status_counts={"todo": 0, "in_progress": 0, "done": 0, "cancelled": 0})
    return row


def test_unplanned_lists_active_milestones_with_no_steps_in_roadmap_order():
    rows = [_planless(3), _milestone(4, "active", 9), _planless(5, "done"), _planless(6)]
    kept, omitted = brief_unplanned(rows)
    assert [r["id"] for r in kept] == [3, 6]          # not the one with steps, not the done one
    assert kept[0] == {"id": 3, "title": "M3", "description": "what M3 is for"}
    assert omitted == 0


def test_unplanned_respects_exclusions_and_the_cap():
    rows = [_planless(i) for i in range(15)]
    kept, omitted = brief_unplanned(rows, exclude_ids={0, 1}, limit=10)
    assert [r["id"] for r in kept] == list(range(2, 12))
    assert omitted == 3


@pytest.mark.asyncio
async def test_enter_project_names_a_roadmap_the_recent_list_cannot_reach():
    """The FabledLibrarian shape: plans written as step-less milestones sat
    beside newer milestones that did their work, and the handshake — five most
    recently touched — could never show them."""
    rows = _history(8) + [_planless(100), _planless(101), _planless(102, "done")]
    out, _ = await _enter(*_enter_stubs(fake_project(id=5), rows, []))

    assert [m["id"] for m in out["unplanned_milestones"]] == [100, 101]
    listed = {m["id"] for m in out["milestone_summary"]}
    assert not listed & {m["id"] for m in out["unplanned_milestones"]}
    assert "unplanned_milestones_omitted" not in out


@pytest.mark.asyncio
async def test_no_unplanned_key_when_every_milestone_has_steps():
    out, _ = await _enter(*_enter_stubs(fake_project(id=5), _history(4), []))
    assert "unplanned_milestones" not in out


@pytest.mark.asyncio
async def test_a_long_roadmap_does_not_rebuild_the_payload():
    rows = _history(5) + [_planless(1000 + i) for i in range(40)]
    out, _ = await _enter(*_enter_stubs(fake_project(id=5), rows, []))
    assert len(out["unplanned_milestones"]) == 10
    assert out["unplanned_milestones_omitted"].startswith("30 more")