FabledScribe/tests/test_cited_record_status.py

"""A record you merely CITE carries its status (#4154, milestone 409 step 8).

WHY THIS EXISTS

Step 1 (#4010) made placement cheap for a task whose status CHANGES: the write
returns where it sits, and the report is written from that. It does nothing for
a task a reply only mentions. In this milestone's own step-6 review the session
reported "#4014 is the open step of milestone 409". #4014 had been done for
four days; the open step was #4015. The id did not come from a read — it came
from a retrieval hint, which carries an id, a kind and a title and says nothing
about status, while `list_milestones` said "8 of 9" and would not say which one.

Two surfaces, one principle: the status arrives with the id.

  1. A milestone summary row names its next open step, so the listing that
     prompts the question also answers it.
  2. An injected menu line renders a task's status, so a finished step cannot
     read as live work.

THE ONE THAT MATTERS MOST

`test_the_listing_and_placement_agree_on_what_open_means` — two surfaces now
answer "what is next" and they must not drift. It is written as a behavioural
cross-check rather than `assert milestones.OPEN_STEP_STATUSES is placement._OPEN`,
which shares one object today and would therefore pass no matter what either
side did with it (rule 167: a guard has to be able to fail). This one fails if
either side changes its ordering or its notion of "open" alone.
"""
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from sqlalchemy import true

from scribe.services import milestones as ms
from scribe.services import placement as pl
from tests.helpers import fake_note, make_mock_session


def _milestone_row(mid: int, project_id: int = 5):
    """A Milestone as the first query returns it — .to_dict() feeds the entry."""
    row = MagicMock()
    row.id, row.project_id, row.user_id = mid, project_id, 7
    row.updated_at = None
    row.to_dict.return_value = {"id": mid, "title": f"M{mid}", "project_id": project_id}
    return row


def _sessions(results: list, counter: list[int], seen_sql: list | None = None):
    """A patched `async_session` handing out queued results and counting queries.

    Every query in `get_project_milestone_summaries` pulls the next entry, so
    the ORDER of `results` pins the order of the queries — which is what makes
    the query-count assertion meaningful rather than incidental.

    `seen_sql` collects the rendered statements. The ordering these surfaces
    have to agree on lives in an ORDER BY, which no amount of feeding rows to a
    mock can exercise — a stand-in hands back whatever order the test chose.
    """
    session = make_mock_session()

    async def _execute(stmt=None, *_a, **_kw):
        counter[0] += 1
        if seen_sql is not None:
            seen_sql.append(str(stmt))
        rows = results.pop(0) if results else []
        r = MagicMock()
        r.fetchall = MagicMock(return_value=rows)
        r.scalars = MagicMock(return_value=MagicMock(all=lambda: rows))
        return r

    session.execute = _execute
    return MagicMock(return_value=session)


def _order_by(sql: str) -> str:
    """The ORDER BY tail of a rendered statement, normalised."""
    _, _, tail = sql.upper().partition("ORDER BY")
    return " ".join(tail.split())


async def _summaries(milestones, counts, open_steps, counter=None, seen_sql=None):
    counter = counter if counter is not None else [0]
    results = [milestones, counts, open_steps]
    with patch.object(ms, "async_session", _sessions(results, counter, seen_sql)), \
         patch.object(ms.access_svc, "readable_notes_clause",
                      MagicMock(return_value=true())):
        return await ms.get_project_milestone_summaries(7, [5])


async def _placement_sql(seen_sql: list):
    """Run `task_placement` far enough to render its sibling-steps query."""
    session = make_mock_session()
    milestone = MagicMock(id=409, project_id=5, user_id=7, title="M", status="active")

    async def _execute(stmt=None, *_a, **_kw):
        seen_sql.append(str(stmt))
        r = MagicMock()
        r.scalars = MagicMock(return_value=MagicMock(
            first=lambda: milestone, all=lambda: [],
        ))
        return r

    session.execute = _execute
    with patch.object(pl, "async_session", MagicMock(return_value=session)), \
         patch.object(pl.access_svc, "can_read_project", AsyncMock(return_value=False)), \
         patch.object(pl.access_svc, "readable_notes_clause",
                      MagicMock(return_value=true())):
        await pl.task_placement(7, SimpleNamespace(
            id=1, project_id=5, milestone_id=409, status="todo",
        ))


@pytest.mark.asyncio
async def test_a_milestone_row_names_its_next_open_step():
    """The listing that says "8 of 9" now says WHICH one, in the same read."""
    rows = await _summaries(
        [_milestone_row(409)],
        [(409, "done", 8), (409, "todo", 1)],
        [(409, 4154, "Step 8 — a cited record carries its status", "todo")],
    )
    assert rows[5][0]["next_step"] == {
        "id": 4154,
        "title": "Step 8 — a cited record carries its status",
        "status": "todo",
    }


@pytest.mark.asyncio
async def test_a_finished_plan_names_no_next_step_rather_than_omitting_the_key():
    """Always present, None included.

    A key that disappears when the answer is "nothing left" makes a reader
    test for its absence to learn the answer, and a reader who forgets is back
    to guessing — which is the failure this step exists for.
    """
    rows = await _summaries([_milestone_row(416)], [(416, "done", 9)], [])
    assert rows[5][0]["next_step"] is None
    assert "next_step" in rows[5][0]


@pytest.mark.asyncio
async def test_the_earliest_open_step_wins_not_the_earliest_step():
    """The done-first case, which is the shape every part-finished plan has.

    Steps arrive in creation order, so a plan whose first two are closed must
    name the third. Naming the first would reproduce the exact misreport: a
    step that IS in the milestone, that IS plausible, and that is finished.
    """
    rows = await _summaries(
        [_milestone_row(409)],
        [(409, "done", 2), (409, "todo", 2)],
        # The query filters to open steps, so the closed ones never appear —
        # this asserts the ORDER of what does: earliest open, not last written.
        [(409, 4015, "Step 6", "in_progress"), (409, 4154, "Step 8", "todo")],
    )
    assert rows[5][0]["next_step"]["id"] == 4015
    assert rows[5][0]["next_step"]["status"] == "in_progress"


@pytest.mark.asyncio
async def test_the_batch_does_not_fan_out_per_milestone():
    """#2384's shape must not come back through the new query.

    Three queries for one milestone and three for forty — the cost is in the
    number of QUERIES, not the number of plans. A per-milestone "what's next"
    lookup would produce identical output and reproduce the pool exhaustion.
    """
    counter = [0]
    await _summaries(
        [_milestone_row(i) for i in range(40)],
        [(i, "todo", 1) for i in range(40)],
        [(i, 1000 + i, f"S{i}", "todo") for i in range(40)],
        counter=counter,
    )
    assert counter[0] == 3, f"{counter[0]} queries for 40 milestones"


@pytest.mark.asyncio
async def test_both_step_queries_take_the_same_visibility_clause():
    """Rule 78, and a correctness point on top of it.

    `next_step` names a step; the counts beside it say how many there are. Read
    through different clauses, a row could name a step its own progress numbers
    exclude — one row describing two different milestones.
    """
    seen = []

    def _clause(uid):
        seen.append(uid)
        return true()

    results = [[_milestone_row(409)], [(409, "todo", 1)], [(409, 1, "S", "todo")]]
    with patch.object(ms, "async_session", _sessions(results, [0])), \
         patch.object(ms.access_svc, "readable_notes_clause", _clause):
        await ms.get_project_milestone_summaries(7, [5])

    # Built ONCE and reused, so the two queries cannot be given different ones.
    assert seen == [7]


@pytest.mark.asyncio
async def test_the_listing_and_placement_agree_on_what_open_means():
    """THE GUARD. Two surfaces answer "what is next"; they must not drift.

    Behavioural on purpose — see the module docstring. `placement` answers for
    a session that knows which step it is on, the listing for one that does
    not, so the two are only equal when the current step is the first. That is
    the case checked here, and it fails if either side's ordering or its open
    set moves without the other.
    """
    steps = [
        SimpleNamespace(id=1, title="Step 1", status="done"),
        SimpleNamespace(id=2, title="Step 2", status="done"),
        SimpleNamespace(id=3, title="Step 3", status="todo"),
        SimpleNamespace(id=4, title="Step 4", status="todo"),
    ]
    # From the first step, placement's "the next open one after this" and the
    # listing's "the earliest open one" are the same question.
    from_placement = pl._next_open(steps, current_id=1)

    listing_sql: list = []
    rows = await _summaries(
        [_milestone_row(409)],
        [(409, "done", 2), (409, "todo", 2)],
        [(409, s.id, s.title, s.status) for s in steps if s.status in ms.OPEN_STEP_STATUSES],
        seen_sql=listing_sql,
    )
    assert rows[5][0]["next_step"] == from_placement

    # And they agree on ORDERING, which the rows above cannot show: a stand-in
    # session hands back whatever order this test chose, so "the first row
    # wins" would pass against any ORDER BY at all. The database does the
    # sorting in production, so the assertion belongs on the statement.
    placement_sql: list = []
    await _placement_sql(placement_sql)
    steps_query = next(q for q in placement_sql if "ORDER BY" in q.upper())
    assert _order_by(listing_sql[-1]) == _order_by(steps_query)
    assert _order_by(listing_sql[-1]), "the open-steps query has no ORDER BY at all"


def test_a_task_line_in_the_menu_says_where_the_work_stands():
    """The other half: an id and a title with no status is what got cited."""
    from scribe.services.plugin_context import _record_kind

    assert _record_kind(fake_note(is_task=True, status="done")) == "task (done)"
    assert _record_kind(fake_note(is_task=True, status="todo")) == "task (todo)"
    assert _record_kind(
        fake_note(is_task=True, task_kind="issue", status="in_progress")
    ) == "issue (in_progress)"


def test_a_record_with_no_status_gains_no_parenthesis():
    """Only tasks. A note or a snippet has no status to be wrong about, and a
    marker that appeared on every line would stop being read."""
    from scribe.services.plugin_context import _record_kind

    assert _record_kind(fake_note(note_type="snippet")) == "snippet"
    assert _record_kind(fake_note(note_type="process")) == "process"
    assert _record_kind(fake_note()) == "note"


def test_the_listing_tools_say_what_next_step_is_and_how_to_use_it():
    """The contract every MCP client reads (rule 119, decision #4027).

    The field is only worth adding if a caller knows it is there. Without this
    the docstring can be tidied to a parameter list and the one surface that
    reaches a non-Claude-Code client goes quiet about it.
    """
    from tests.helpers import tool_doc

    for module, name in (
        ("scribe.mcp.tools.milestones", "list_milestones"),
        ("scribe.mcp.tools.projects", "enter_project"),
        ("scribe.mcp.tools.projects", "get_project"),
    ):
        doc = tool_doc(module, name).lower()
        assert "next_step" in doc, f"{name} does not mention next_step"
        # Naming the field is not enough — a reader has to be told that a null
        # is an answer, or an absent next step reads as data not yet loaded.
        assert "null" in doc or "none" in doc, f"{name} does not say when it is empty"


def test_the_reply_skill_says_a_cited_record_still_gets_read():
    """The other half of the fix is a practice, and it has one home.

    Presence, not absence — the skill legitimately discusses recall in order to
    warn against it, so an absence check here would be satisfied by the warning
    itself (snippet #3352).
    """
    import pathlib

    skill = pathlib.Path(__file__).resolve().parents[1] / "plugin/skills/reporting-back/SKILL.md"
    text = " ".join(skill.read_text().split()).lower()
    assert "a record you only mention is a record to read" in text
    # And it points at the field rather than restating how to compute it.
    assert "next_step" in text