diff --git a/src/scribe/mcp/tools/milestones.py b/src/scribe/mcp/tools/milestones.py
index 9e4b6f9..fec165e 100644
--- a/src/scribe/mcp/tools/milestones.py
+++ b/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"),
     }
 
diff --git a/src/scribe/mcp/tools/tasks.py b/src/scribe/mcp/tools/tasks.py
index 24a740f..9562d38 100644
--- a/src/scribe/mcp/tools/tasks.py
+++ b/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
diff --git a/src/scribe/services/notes.py b/src/scribe/services/notes.py
index 7a21123..165eaee 100644
--- a/src/scribe/services/notes.py
+++ b/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
diff --git a/src/scribe/services/task_logs.py b/src/scribe/services/task_logs.py
index 7a96165..e06975f 100644
--- a/src/scribe/services/task_logs.py
+++ b/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,
diff --git a/tests/conftest.py b/tests/conftest.py
index d562c8f..24db63d 100644
--- a/tests/conftest.py
+++ b/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).
diff --git a/tests/test_task_work_log_surface.py b/tests/test_task_work_log_surface.py
new file mode 100644
index 0000000..9376173
--- /dev/null
+++ b/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