fix(tasks): add_task_log wrote to a surface no agent could read back (#4241)

The work log reached the web UI through routes/task_logs.py and nothing
else. get_task returned only the body — a claim written once, before the
work — with the record written during it invisible beside it. So a stale
body arrived with nothing to contradict it, and this session rebuilt work
that had already shipped, with the evidence sitting in the task's own logs.

Read side, scoped through the access layer (rule 78):
  - logs_for_task / count_logs_for_task / log_counts_for_tasks in
    services/task_logs.py. Scoped by who may read the TASK rather than by
    who wrote the entry: list_logs filters TaskLog.user_id == user_id,
    which hands a shared collaborator an empty list reading as "no work
    has been done". The page query folds readable_notes_clause into the
    same statement so the permission does not become an N+1.
  - get_task returns work_log; list_tasks and get_milestone steps carry
    log_count, zero-filled so "none" is a count and not a missing key.

Elision keeps both ends. The newest entry arrives whole to 4000 chars
because it answers "where does this stand"; older ones are shortened from
the MIDDLE, never the head. A head cut selects what a reader sees by
character position, which is uncorrelated with what matters — an entry
closing with "so this shipped in 04775c3" loses the one sentence that
answers the question, and a truncated flag says something went, never
whether it mattered. The gap states how many characters it covers.

conftest gains an autouse stub for the new read arm, same reasoning as
_no_rule_arm: three widely-called tools grew a database read, and the
existing call sites should not each have to learn about it.

Raised while reviewing this: search() has the same shape and worse —
body[:240] with no marker at all, while the chunk that actually matched
sits unused in the row that won. Filed as #4243, not fixed here.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy

src/scribe/mcp/tools/milestones.py

@@ -15,6 +15,7 @@ from scribe.mcp._context import current_user_id
 from scribe.services import dedup as dedup_svc
 from scribe.services import milestones as milestones_svc
 from scribe.services import notes as notes_svc
+from scribe.services import task_logs as task_logs_svc
 from scribe.services import rulebooks as rulebooks_svc
 from scribe.services import trash as trash_svc
 from scribe.services.record_refs import refuse_guessed_ids
@@ -53,6 +54,11 @@ async def get_milestone(milestone_id: int) -> dict:
     what it is and where it stands, not its whole body — read a step in full
     with get_task(id). A plan with forty long steps is otherwise too big to
     arrive inline, and the design is in the milestone body.
+
+    Each step also carries `log_count`. A step's STATUS is set by hand and a
+    plan is exactly where that goes stale; the count says which steps have a
+    work log that would say otherwise, so `get_task(id)` goes to the step
+    with a record rather than to each one in turn (#4241).
     """
     uid = current_user_id()
     milestone = await milestones_svc.get_milestone(uid, milestone_id)
@@ -65,11 +71,19 @@ async def get_milestone(milestone_id: int) -> dict:
     applicable = await rulebooks_svc.get_applicable_rules(
         project_id=milestone.project_id, user_id=uid,
     )
+    log_counts = await task_logs_svc.log_counts_for_tasks(
+        uid, [int(t.id) for t in steps]
+    )
     out = milestone.to_dict()
     out.update(progress)
     return {
         "milestone": out,
-        "steps": [notes_svc.brief_row(t, {milestone.id: milestone.title}) for t in steps],
+        "steps": [
+            notes_svc.brief_row(
+                t, {milestone.id: milestone.title}, log_counts=log_counts
+            )
+            for t in steps
+        ],
         **rulebooks_svc.rules_payload(applicable, user_id=uid, source="get_milestone"),
     }
 

src/scribe/mcp/tools/tasks.py

@@ -3,7 +3,13 @@
 Tasks are notes with a non-null `status` — same model, different filter.
 Wrappers call services/notes.py for CRUD with is_task=True and add the
 task-specific fields (status, priority, due_date, parent_id), plus
-services/task_logs.py for add_task_log.
+services/task_logs.py for add_task_log AND for reading those logs back —
+`get_task` returns them and `list_tasks` counts them. For most of this
+module's life `add_task_log` wrote to a surface no agent could read: the
+entries reached the web UI and nothing else, so a session opening a task
+saw only the body — a claim written before the work — with the record
+written during it invisible beside it. A stale body then had nothing to
+contradict it, and shipped work got rebuilt (#4241).
 
 There is no delete_task — matches the existing fable-mcp surface.
 Cancel by updating status to "cancelled".
@@ -41,6 +47,92 @@ from scribe.services.note_usage import record_pulled
 from scribe.services.record_refs import refuse_guessed_ids
 
 
+# A work log entry is prose, often long — the discipline asks for what was
+# decided and why, not a line of status. Two caps, not one, because the entries
+# are not equally useful: the NEWEST answers "where does this actually stand",
+# which is the question the block exists for, so it arrives whole up to a
+# generous ceiling. Older entries are there to say what happened and when, and
+# a headline does that.
+_WORK_LOG_ENTRIES = 3
+_WORK_LOG_CHARS = 800
+_WORK_LOG_LATEST_CHARS = 4000
+
+_WORK_LOG_ADVICE = (
+    "The body is a CLAIM, written once before the work. These entries are the "
+    "RECORD, written during and after it. Where the two disagree the log is "
+    "later — read it before acting on what the body says the status is."
+)
+
+
+def elide(text: str, budget: int) -> tuple[str, bool]:
+    """Cut to `budget` characters from the MIDDLE, keeping both ends.
+
+    A head-only cut — `text[:800]` — decides what a reader sees by character
+    position, which is uncorrelated with what matters. Prose does not put its
+    conclusion first: a log entry that opens with what was tried and closes
+    with "so this shipped in 04775c3" loses exactly the sentence that answers
+    the question, and the reader cannot tell, because a truncation marker says
+    that something was removed and never whether it mattered.
+
+    So keep the opening (what this entry is about) AND the closing (where it
+    landed), and say in between how much went. Two thirds to the head because
+    that is where the subject is established; the tail needs less to carry a
+    conclusion. `budget` of 0 means no cut.
+    """
+    if budget <= 0 or len(text) <= budget:
+        return text, False
+    head_len = max(1, budget * 2 // 3)
+    tail_len = max(1, budget - head_len)
+    omitted = len(text) - head_len - tail_len
+    head = text[:head_len].rstrip()
+    tail = text[-tail_len:].lstrip()
+    return f"{head}\n\n[… {omitted} characters omitted …]\n\n{tail}", True
+
+
+def work_log_payload(
+    logs: list, total: int, chars: int, latest_chars: int = _WORK_LOG_LATEST_CHARS
+) -> dict:
+    """The `work_log` block: recent entries newest-first, plus what was elided.
+
+    `total` is the count of ALL entries, not of `logs` — a reader has to be
+    able to tell "this task has no record" from "you were shown the last three
+    of nine", and those are the same response if the count comes from the
+    entries handed over.
+    """
+    entries = []
+    for i, log in enumerate(logs):
+        row = log.to_dict() if hasattr(log, "to_dict") else dict(log)
+        row.pop("updated_at", None)
+        content = row.get("content") or ""
+        # The first row IS the newest — logs_for_task orders descending.
+        budget = latest_chars if i == 0 else chars
+        if chars <= 0:
+            budget = 0
+        text, cut = elide(content, budget)
+        row["content"] = text
+        if cut:
+            row["full_length"] = len(content)
+            row["truncated"] = True
+        entries.append(row)
+
+    out: dict = {"total": total, "entries": entries}
+    if total == 0:
+        return out
+
+    out["advice"] = _WORK_LOG_ADVICE
+    not_shown = total - len(entries)
+    if not_shown > 0:
+        out["not_shown"] = not_shown
+    if not_shown > 0 or any(e.get("truncated") for e in entries):
+        out["read_all"] = (
+            "Entries were shortened or omitted. "
+            "get_task(task_id, log_limit=0, log_chars=0) returns every entry "
+            "in full — reach for it rather than judging from what is here, "
+            "which was selected by recency and length, not by relevance."
+        )
+    return out
+
+
 async def list_tasks(
     limit: int = 20,
     offset: int = 0,
@@ -75,10 +167,27 @@ async def list_tasks(
         offset=max(0, offset),
     )
     titles = await milestones_svc.titles_for({n.milestone_id for n in rows})
-    return {"tasks": [notes_svc.brief_row(n, titles) for n in rows], "total": total}
+    # One aggregate for the page, zero-filled: a reader scanning a list needs
+    # to know WHICH rows carry a record before choosing what to open, and a
+    # missing key would read as "no logs" on every row rather than on the
+    # rows that have none.
+    log_counts = await task_logs_svc.log_counts_for_tasks(
+        uid, [int(n.id) for n in rows]
+    )
+    return {
+        "tasks": [
+            notes_svc.brief_row(n, titles, log_counts=log_counts) for n in rows
+        ],
+        "total": total,
+    }
 
 
-async def get_task(task_id: int, project_id: int = 0) -> dict:
+async def get_task(
+    task_id: int,
+    project_id: int = 0,
+    log_limit: int = _WORK_LOG_ENTRIES,
+    log_chars: int = _WORK_LOG_CHARS,
+) -> dict:
     """Fetch a single Scribe task by ID.
 
     Returns id, title, body, status, priority, tags, project_id, milestone_id,
@@ -89,6 +198,27 @@ async def get_task(task_id: int, project_id: int = 0) -> dict:
     kind=plan tasks, the response also includes the project's applicable_rules
     and project_rules (new plans are milestones — use get_milestone for those).
 
+    AND `work_log` — the entries add_task_log wrote, newest first, with
+    `total` for how many exist. READ IT BEFORE TRUSTING THE BODY. A body is
+    written once, at the start, when the least is known; the log is written
+    during the work and after it. A task whose body says "not started" and
+    whose log records a partial ship is not a contradiction to resolve — the
+    log is simply later. This block exists because its absence cost a session
+    a day of rebuilding work that had already shipped (#4241).
+
+    The NEWEST entry arrives whole (to 4000 characters); older ones are
+    shortened to a headline. Anything shortened is cut from the MIDDLE, so
+    the opening and the closing both survive — a log entry's conclusion is at
+    its end — and the gap states how many characters went. An entry that was
+    cut says so and carries its `full_length`, and the block as a whole says
+    when it is showing you less than the record holds.
+
+    Args:
+        log_limit: How many of the most recent entries to include (default 3).
+            0 returns every entry.
+        log_chars: Budget for the OLDER entries (default 800). 0 returns
+            every entry in full, the newest included.
+
     A task another user shared with you also carries `shared`, `owner` and
     `permission` — it's their work item, not one you took on.
 
@@ -127,6 +257,17 @@ async def get_task(task_id: int, project_id: int = 0) -> dict:
     await systems_tools.attach_systems(
         uid, getattr(note, "user_id", uid) or uid, data, note.id, note.project_id
     )
+    # Counted separately rather than inferred from the rows handed back: with a
+    # limit applied, len(entries) is the size of the window, not of the record,
+    # and "3 entries" and "the last 3 of 9" have to read differently.
+    log_rows = await task_logs_svc.logs_for_task(
+        uid, int(note.id), limit=max(0, log_limit)
+    )
+    log_total = (
+        len(log_rows) if log_limit <= 0
+        else await task_logs_svc.count_logs_for_task(uid, int(note.id))
+    )
+    data["work_log"] = work_log_payload(log_rows, log_total, max(0, log_chars))
     record_pulled(
         user_id=uid, note_id=int(note.id),
         source="mcp_get_task", project_id=project_id,
@@ -353,6 +494,13 @@ async def add_task_log(task_id: int, content: str) -> dict:
     without overwriting the task's main body. Each entry is stored separately
     and shown chronologically in the task view.
 
+    What you write here comes back from `get_task` as `work_log`, newest
+    first, and is counted on every row of `list_tasks` and every step of
+    `get_milestone` — so write for the session that opens this task next, not
+    for a reader who already knows what you were doing. What that reader
+    cannot get from the body or the diff is what you tried, what you ruled
+    out, and where it actually stands.
+
     The response shows the task's `systems` — or, if the task is an untagged
     project record, the `systems_hint` question: logging work IS working in
     some area, so answer it (update_task with system_ids, or create_system

src/scribe/services/notes.py

@@ -1096,7 +1096,20 @@ async def get_note_for_user(
 # A field most rows leave empty (a one-line description, a parent, a due date)
 # is attached only when set: a hundred rows of `null` are a hundred chances to
 # learn to skip the key (#2483), and the bytes are the thing being cut.
-def brief_row(note: Note, milestone_titles: dict[int, str] | None = None) -> dict:
+def brief_row(
+    note: Note,
+    milestone_titles: dict[int, str] | None = None,
+    log_counts: dict[int, int] | None = None,
+) -> dict:
+    """One task as a list row — what it is and where it stands, not its body.
+
+    `log_counts` is a whole page's work-log counts fetched in one aggregate
+    (task_logs.log_counts_for_tasks). Passed in rather than looked up here so
+    the N+1 stays impossible, and applied to tasks only: a note has no work
+    log. Zero-filled when the mapping is given, because a row that omits the
+    key says "no record" on every row rather than on the ones with none, and
+    knowing WHICH rows carry a record is how a reader decides what to open.
+    """
     row = {
         "id": note.id,
         "title": note.title,
@@ -1120,4 +1133,6 @@ def brief_row(note: Note, milestone_titles: dict[int, str] | None = None) -> dic
             row["parent_id"] = note.parent_id
         if note.due_date:
             row["due_date"] = iso(note.due_date)
+        if log_counts is not None:
+            row["log_count"] = log_counts.get(int(note.id), 0)
     return row

src/scribe/services/task_logs.py

@@ -2,11 +2,12 @@
 import logging
 from datetime import datetime, timezone
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 
 from scribe.models import async_session
 from scribe.models.task_log import TaskLog
 from scribe.models.note import Note
+from scribe.services.access import can_read_note, readable_notes_clause
 
 logger = logging.getLogger(__name__)
 
@@ -48,6 +49,81 @@ async def list_logs(user_id: int, task_id: int) -> list[TaskLog]:
         return list(result.scalars().all())
 
 
+async def logs_for_task(
+    user_id: int, task_id: int, limit: int = 0
+) -> list[TaskLog]:
+    """Every work log on a task, NEWEST FIRST, whoever wrote it.
+
+    Two differences from `list_logs`, both deliberate:
+
+    1. Scoped by who may read the TASK, not by who wrote each entry. A work
+       log belongs to the task, and on a shared task the owner's record is
+       exactly what a collaborator needs — `list_logs`' `TaskLog.user_id ==
+       user_id` returns them an empty list, which reads as "no work has been
+       done" rather than "not yours". The permission question is asked of the
+       note through `can_read_note` (rule #78), so this is not an unscoped
+       read with the check left to the caller.
+    2. Newest first, because the question a reader asks of a work log is
+       "where does this actually stand", and the answer is the last entry.
+       `list_logs` stays ascending — the web UI renders a narrative.
+
+    `limit` of 0 means all of them. An unreadable or missing task returns []
+    the same way an empty one does: a work log is not a channel for proving a
+    record exists.
+    """
+    if not await can_read_note(user_id, task_id):
+        return []
+    async with async_session() as session:
+        stmt = (
+            select(TaskLog)
+            .where(TaskLog.task_id == task_id)
+            .order_by(TaskLog.created_at.desc(), TaskLog.id.desc())
+        )
+        if limit > 0:
+            stmt = stmt.limit(limit)
+        result = await session.execute(stmt)
+        return list(result.scalars().all())
+
+
+async def count_logs_for_task(user_id: int, task_id: int) -> int:
+    """How many work logs a task carries. Same scoping as logs_for_task."""
+    if not await can_read_note(user_id, task_id):
+        return 0
+    async with async_session() as session:
+        result = await session.execute(
+            select(func.count(TaskLog.id)).where(TaskLog.task_id == task_id)
+        )
+        return int(result.scalar() or 0)
+
+
+async def log_counts_for_tasks(
+    user_id: int, task_ids: list[int]
+) -> dict[int, int]:
+    """Log counts for a whole page of tasks in ONE query.
+
+    A per-row lookup would be N+1 by construction — and so would a per-row
+    `can_read_note` — which is the reason a list surface would otherwise keep
+    omitting this and leave a reader to guess which rows carry a record. So
+    the permission is expressed as set membership with
+    `readable_notes_clause` and folded into the same statement, the pattern
+    that function exists for.
+
+    Tasks with no logs are absent from the mapping; callers zero-fill, so
+    "none" reads as a count rather than a missing key.
+    """
+    ids = [int(t) for t in task_ids if t]
+    if not ids:
+        return {}
+    async with async_session() as session:
+        result = await session.execute(
+            select(TaskLog.task_id, func.count(TaskLog.id))
+            .join(Note, Note.id == TaskLog.task_id)
+            .where(TaskLog.task_id.in_(ids), readable_notes_clause(user_id))
+            .group_by(TaskLog.task_id)
+        )
+        return {int(tid): int(n) for tid, n in result.all()}
+
+
 async def update_log(
     user_id: int,
     log_id: int,

tests/conftest.py

@@ -102,6 +102,32 @@ def _no_supersession():
         yield
 
 
+@pytest.fixture(autouse=True)
+def _no_task_log_arm():
+    """Stub the task-log read arm that get_task / list_tasks / get_milestone
+    grew in #4241.
+
+    Autouse for the reason _no_rule_arm is: those three tools now read work
+    logs, and the reads go through the access layer to Postgres. Every unit
+    test that opens a task — and most of them do, because a task is what this
+    codebase is mostly about — would otherwise try to reach the fake
+    DATABASE_URL this file sets, to learn that a fake task has no logs.
+
+    The arm's own behaviour is covered where it belongs: the payload shape and
+    the tool wiring in tests/test_task_work_log_surface.py, which re-patches
+    these explicitly, and the ACL scoping against real Postgres in
+    tests/test_integration_task_work_log.py. A test that wants the arm live
+    re-patches it, same as the rules arm.
+    """
+    with patch("scribe.services.task_logs.logs_for_task",
+               AsyncMock(return_value=[])), \
+         patch("scribe.services.task_logs.count_logs_for_task",
+               AsyncMock(return_value=0)), \
+         patch("scribe.services.task_logs.log_counts_for_tasks",
+               AsyncMock(return_value={})):
+        yield
+
+
 @pytest.fixture(autouse=True)
 def _no_rule_arm():
     """Stub the write-path hint's standing-RULES arm (milestone 307).

tests/test_task_work_log_surface.py

@@ -0,0 +1,375 @@
+"""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
+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."""
+    from scribe.services import task_logs as svc
+
+    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 svc.logs_for_task(7, 42)
+    stmt = str(session.execute.await_args.args[0])
+    assert "task_logs.task_id" in stmt
+    assert "task_logs.user_id" not in stmt
+
+
+@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."""
+    from scribe.services import task_logs as svc
+
+    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 svc.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():
+    from scribe.services import task_logs as svc
+
+    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 svc.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."""
+    from scribe.services import task_logs as svc
+
+    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 svc.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 svc.log_counts_for_tasks(7, []) == {}
+    assert opened.call_count == 0