FabledScribe/tests/test_task_work_log_surface.py

"""The work log is READABLE from the tools an agent opens a task with (#4241).

`add_task_log` wrote to a surface no agent could read back: the entries
reached the web UI through routes/task_logs.py and nothing else. So a session
opening a task saw only the body — a claim written once, before the work —
with the record written during it invisible beside it. A stale body had
nothing to contradict it, and shipped work got rebuilt.

These pin the read, not the write. The payload shape is tested against the
real `work_log_payload`; the three tools are tested by re-patching the arm
that tests/conftest.py stubs autouse, so each one is exercised live here and
nowhere else.
"""
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from scribe.mcp.tools.milestones import get_milestone
from scribe.mcp.tools.tasks import (
    elide, get_task, list_tasks, work_log_payload,
)
from scribe.services.notes import brief_row
# Bound at IMPORT time, which is before any fixture runs — so these names
# keep pointing at the real implementations even though conftest's autouse
# _no_task_log_arm replaces the module attributes for every test. Reaching
# them as `task_logs.logs_for_task` would get the stub and test nothing.
from scribe.services.task_logs import (
    count_logs_for_task,
    log_counts_for_tasks,
    logs_for_task,
)
from tests.helpers import fake_task, make_mock_session

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


def fake_log(log_id: int = 1, content: str = "did the thing", **attrs):
    """A stand-in TaskLog. `to_dict` is what the payload builder consumes."""
    row = {
        "id": log_id,
        "task_id": 1,
        "user_id": 7,
        "content": content,
        "duration_minutes": None,
        "created_at": "2026-09-21T12:00:00+00:00",
        "updated_at": "2026-09-21T12:00:00+00:00",
    }
    row.update(attrs)
    log = MagicMock()
    log.to_dict = MagicMock(return_value=row)
    log.content = row["content"]
    log.id = row["id"]
    return log


def _live_arm(logs=(), total=0, counts=None):
    """Re-patch conftest's autouse stub so the arm under test is live."""
    return (
        patch("scribe.services.task_logs.logs_for_task",
              AsyncMock(return_value=list(logs))),
        patch("scribe.services.task_logs.count_logs_for_task",
              AsyncMock(return_value=total)),
        patch("scribe.services.task_logs.log_counts_for_tasks",
              AsyncMock(return_value=counts or {})),
    )


# ---------------------------------------------------------------------------
# The payload shape
# ---------------------------------------------------------------------------

def test_no_logs_reads_as_a_count_not_a_missing_key():
    """"This task has no record" and "you were not shown the record" are
    different answers, and a caller has to be able to tell them apart."""
    out = work_log_payload([], 0, 800)
    assert out == {"total": 0, "entries": []}
    # No advice on an empty log: there is no body/record disagreement to warn
    # about, and a standing paragraph on every task is noise.
    assert "advice" not in out


def test_entries_carry_the_advice_that_the_log_outranks_the_body():
    out = work_log_payload([fake_log()], 1, 800)
    assert "advice" in out
    advice = out["advice"].lower()
    assert "claim" in advice and "record" in advice
    # The operative instruction: which one to believe when they disagree.
    assert "later" in advice


def test_elision_keeps_the_end_where_the_conclusion_lives():
    """The whole reason this is not `text[:n]`.

    Prose does not put its conclusion first. An entry that opens with what was
    attempted and closes with "so this shipped in 04775c3" loses the one
    sentence that answers the question if the cut is taken from the head — and
    a `truncated: true` flag tells a reader that something went, never whether
    it mattered. Both ends survive, and the gap says how much is missing.
    """
    text = "Tried the form column. " + ("m" * 2000) + " CONCLUSION: shipped in 04775c3."
    out, cut = elide(text, 300)
    assert cut is True
    assert out.startswith("Tried the form column.")
    assert out.rstrip().endswith("shipped in 04775c3.")
    assert "characters omitted" in out


def test_elision_states_how_much_went():
    out, _ = elide("a" * 1000, 100)
    assert "900 characters omitted" in out


def test_short_text_is_returned_untouched():
    out, cut = elide("brief", 300)
    assert out == "brief"
    assert cut is False


def test_the_newest_entry_arrives_whole_and_older_ones_are_headlines():
    """The newest entry answers "where does this stand", which is the question
    the block exists for — so it is not competing for budget with history."""
    newest = fake_log(9, content="N" * 3000)
    older = fake_log(8, content="O" * 3000)
    out = work_log_payload([newest, older], 2, chars=800, latest_chars=4000)
    assert out["entries"][0]["content"] == "N" * 3000
    assert "truncated" not in out["entries"][0]
    assert out["entries"][1]["truncated"] is True
    assert out["entries"][1]["full_length"] == 3000


def test_even_the_newest_entry_is_capped_somewhere():
    out = work_log_payload([fake_log(content="x" * 9000)], 1, chars=800,
                           latest_chars=4000)
    entry = out["entries"][0]
    assert entry["truncated"] is True
    assert entry["full_length"] == 9000
    assert "read_all" in out


def test_read_all_says_the_window_was_chosen_by_recency_not_relevance():
    """A pointer to the fuller read is only useful if the reader knows the
    shown part was not selected for being the pertinent part."""
    out = work_log_payload([fake_log(content="x" * 9000)], 1, chars=800,
                           latest_chars=4000)
    assert "relevance" in out["read_all"]


def test_total_is_the_record_and_entries_are_the_window():
    """The bug this guards: reporting len(entries) as the total, so "the last
    three of nine" is indistinguishable from "three"."""
    out = work_log_payload([fake_log(1), fake_log(2), fake_log(3)], 9, 800)
    assert out["total"] == 9
    assert len(out["entries"]) == 3
    assert out["not_shown"] == 6
    assert "get_task" in out["read_all"]


def test_an_untruncated_full_window_advertises_nothing_further():
    out = work_log_payload([fake_log(1), fake_log(2)], 2, 800)
    assert "not_shown" not in out
    assert "read_all" not in out


# ---------------------------------------------------------------------------
# get_task — the tool that failed
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_get_task_returns_the_work_log():
    fake = fake_task(id=4208, title="stale body", parent_id=None)
    p1, p2, p3 = _live_arm(logs=[fake_log(content="partially shipped in 04775c3")],
                           total=1)
    with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
               AsyncMock(return_value=(fake, "owner"))), p1, p2, p3:
        out = await get_task(task_id=4208)
    assert "work_log" in out
    assert out["work_log"]["total"] == 1
    assert "04775c3" in out["work_log"]["entries"][0]["content"]


@pytest.mark.asyncio
async def test_get_task_says_a_task_has_no_log_rather_than_omitting_the_key():
    fake = fake_task(id=1, parent_id=None)
    p1, p2, p3 = _live_arm(logs=[], total=0)
    with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
               AsyncMock(return_value=(fake, "owner"))), p1, p2, p3:
        out = await get_task(task_id=1)
    assert out["work_log"] == {"total": 0, "entries": []}


@pytest.mark.asyncio
async def test_get_task_counts_every_entry_not_just_the_window():
    fake = fake_task(id=1, parent_id=None)
    window = [fake_log(i) for i in (9, 8, 7)]
    p1, p2, p3 = _live_arm(logs=window, total=9)
    with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
               AsyncMock(return_value=(fake, "owner"))), p1, p2, p3:
        out = await get_task(task_id=1)
    assert out["work_log"]["total"] == 9
    assert out["work_log"]["not_shown"] == 6


@pytest.mark.asyncio
async def test_get_task_default_window_is_three_entries():
    fake = fake_task(id=1, parent_id=None)
    limit_seen = {}

    async def _capture_limit(uid, task_id, limit=0):
        limit_seen["limit"] = limit
        return []

    with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
               AsyncMock(return_value=(fake, "owner"))), \
         patch("scribe.services.task_logs.logs_for_task", _capture_limit), \
         patch("scribe.services.task_logs.count_logs_for_task",
               AsyncMock(return_value=0)):
        await get_task(task_id=1)
    assert limit_seen["limit"] == 3


@pytest.mark.asyncio
async def test_log_limit_zero_asks_for_every_entry_and_skips_the_count_query():
    """0 means "all" here because it means "no cap" in the service it calls.
    With no cap the rows ARE the total, so a second query would be waste."""
    fake = fake_task(id=1, parent_id=None)
    counter = AsyncMock(return_value=99)
    p1, _, p3 = _live_arm(logs=[fake_log(1), fake_log(2)])
    with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
               AsyncMock(return_value=(fake, "owner"))), \
         p1, p3, patch("scribe.services.task_logs.count_logs_for_task", counter):
        out = await get_task(task_id=1, log_limit=0)
    assert out["work_log"]["total"] == 2
    assert counter.await_count == 0


@pytest.mark.asyncio
async def test_get_task_docstring_tells_a_reader_the_log_outranks_the_body():
    """#2846: the docstring IS the agent-facing contract. A `work_log` key
    that nothing tells the reader to prefer over a stale body is the same
    failure one layer up."""
    doc = (get_task.__doc__ or "").lower()
    assert "work_log" in doc
    assert "body" in doc


# ---------------------------------------------------------------------------
# The list surfaces — which rows carry a record
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_list_tasks_rows_carry_a_zero_filled_log_count():
    rows = [fake_task(id=1, milestone_id=None), fake_task(id=2, milestone_id=None)]
    _, _, p3 = _live_arm(counts={1: 4})
    with patch("scribe.mcp.tools.tasks.notes_svc.list_notes",
               AsyncMock(return_value=(rows, 2))), p3:
        out = await list_tasks()
    assert out["tasks"][0]["log_count"] == 4
    # Zero-filled, not omitted: a missing key would read as "no logs" on every
    # row rather than on the rows that have none.
    assert out["tasks"][1]["log_count"] == 0


@pytest.mark.asyncio
async def test_list_tasks_counts_the_page_in_one_query():
    rows = [fake_task(id=i, milestone_id=None) for i in range(1, 6)]
    counter = AsyncMock(return_value={})
    with patch("scribe.mcp.tools.tasks.notes_svc.list_notes",
               AsyncMock(return_value=(rows, 5))), \
         patch("scribe.services.task_logs.log_counts_for_tasks", counter):
        await list_tasks()
    assert counter.await_count == 1
    assert sorted(counter.await_args.args[1]) == [1, 2, 3, 4, 5]


@pytest.mark.asyncio
async def test_get_milestone_steps_carry_a_log_count():
    """A step's status is set by hand, and a plan is exactly where that goes
    stale — so a plan reader needs to see which steps have a record."""
    milestone = MagicMock(id=385, title="Lessons", project_id=2)
    milestone.to_dict = MagicMock(return_value={"id": 385, "title": "Lessons"})
    steps = [fake_task(id=3735, milestone_id=385),
             fake_task(id=3736, milestone_id=385)]
    counter = AsyncMock(return_value={3735: 2})
    with patch("scribe.mcp.tools.milestones.milestones_svc.get_milestone",
               AsyncMock(return_value=milestone)), \
         patch("scribe.mcp.tools.milestones.milestones_svc.get_milestone_progress",
               AsyncMock(return_value={})), \
         patch("scribe.mcp.tools.milestones.notes_svc.list_notes",
               AsyncMock(return_value=(steps, 2))), \
         patch("scribe.mcp.tools.milestones.rulebooks_svc.get_applicable_rules",
               AsyncMock(return_value=[])), \
         patch("scribe.mcp.tools.milestones.rulebooks_svc.rules_payload",
               MagicMock(return_value={})), \
         patch("scribe.services.task_logs.log_counts_for_tasks", counter):
        out = await get_milestone(milestone_id=385)
    assert out["steps"][0]["log_count"] == 2
    assert out["steps"][1]["log_count"] == 0


def test_brief_row_without_a_mapping_is_unchanged():
    """The other three brief_row callers (list_notes, list_system_records)
    pass no counts and must not grow a misleading zero."""
    row = brief_row(fake_task(id=1, milestone_id=None, due_date=None))
    assert "log_count" not in row


# ---------------------------------------------------------------------------
# The service read — scoped by who may read the TASK (rule #78)
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_logs_for_task_is_not_filtered_by_who_wrote_the_entry():
    """`list_logs` filters TaskLog.user_id == user_id, which hands a shared
    collaborator an empty list that reads as "no work has been done". The work
    log belongs to the task."""
    session = make_mock_session()
    session.execute = AsyncMock(return_value=MagicMock(
        scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[])))
    ))
    with patch("scribe.services.task_logs.can_read_note",
               AsyncMock(return_value=True)), \
         patch("scribe.services.task_logs.async_session",
               MagicMock(return_value=session)):
        await logs_for_task(7, 42)
    # The WHERE clause specifically — `user_id` is a selected COLUMN of the
    # row and appears in every SELECT, so asserting against the whole compiled
    # statement tests the projection rather than the scoping.
    where = str(session.execute.await_args.args[0].whereclause)
    assert "task_logs.task_id" in where
    assert "task_logs.user_id" not in where


@pytest.mark.asyncio
async def test_logs_for_task_refuses_a_task_the_caller_cannot_read():
    """Unscoped would have been the mirror-image hole: rule #78 is about
    routing the question through the access layer, in both directions."""
    opened = MagicMock()
    with patch("scribe.services.task_logs.can_read_note",
               AsyncMock(return_value=False)), \
         patch("scribe.services.task_logs.async_session", opened):
        out = await logs_for_task(7, 42)
    assert out == []
    assert opened.call_count == 0


@pytest.mark.asyncio
async def test_count_for_an_unreadable_task_is_zero_not_a_leak():
    opened = MagicMock()
    with patch("scribe.services.task_logs.can_read_note",
               AsyncMock(return_value=False)), \
         patch("scribe.services.task_logs.async_session", opened):
        assert await count_logs_for_task(7, 42) == 0
    assert opened.call_count == 0


@pytest.mark.asyncio
async def test_log_counts_for_tasks_scopes_by_readability_in_the_same_query():
    """A per-row can_read_note would be the N+1 this function exists to avoid,
    so the permission goes in as set membership instead."""
    session = make_mock_session()
    session.execute = AsyncMock(return_value=MagicMock(
        all=MagicMock(return_value=[(1, 3)])
    ))
    with patch("scribe.services.task_logs.async_session",
               MagicMock(return_value=session)):
        out = await log_counts_for_tasks(7, [1, 2])
    assert out == {1: 3}
    stmt = str(session.execute.await_args.args[0])
    assert "notes" in stmt.lower()


@pytest.mark.asyncio
async def test_no_ids_asks_the_database_nothing():
    from scribe.services import task_logs as svc

    opened = MagicMock()
    with patch("scribe.services.task_logs.async_session", opened):
        assert await log_counts_for_tasks(7, []) == {}
    assert opened.call_count == 0