fix(mcp): list tools return rows that say what a record is, not what it says (#4061)
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 8s
CI & Build / integration (push) Successful in 49s
CI & Build / TypeScript typecheck (push) Successful in 53s
CI & Build / Python tests (push) Successful in 1m31s
CI & Build / Build & push image (push) Successful in 22s

list_tasks returned every row's to_dict(), body included: a project's todo list
came to 93-165k characters, past what an MCP client accepts inline, so the list
arrived as a file to page through (the #4045 failure, one call over).

- notes.brief_row: id, title, type, project, tags, updated_at; for tasks,
  status, kind, priority, milestone id and title; description, parent and due
  date only when set.
- milestones.titles_for: one query for the milestone titles a page of rows
  names.
- Brief rows on list_tasks, list_notes, get_milestone's steps, get_system and
  list_system_records. get_task / get_note / get_snippet read a record in full,
  and each docstring says so.
- tests/test_list_rows_brief.py pins the ceiling: 100 rows of ~5k-character
  step plans stay under 40k characters.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy
This commit is contained in:
2026-09-15 16:07:00 -04:00
co-authored by Claude Opus 5
parent 6eb5ef5c73
commit 5e4fd017ae
10 changed files with 188 additions and 17 deletions
+5 -2
View File
@@ -44,7 +44,10 @@ async def get_milestone(milestone_id: int) -> dict:
rules surface again on recall.
Returns: milestone (incl. body), progress, steps (its tasks ordered by
status then update), and applicable_rules / project_rules.
status then update), and applicable_rules / project_rules. Each step says
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.
"""
uid = current_user_id()
milestone = await milestones_svc.get_milestone(uid, milestone_id)
@@ -61,7 +64,7 @@ async def get_milestone(milestone_id: int) -> dict:
out.update(progress)
return {
"milestone": out,
"steps": [t.to_dict() for t in steps],
"steps": [notes_svc.brief_row(t, {milestone.id: milestone.title}) for t in steps],
**rulebooks_svc.rules_payload(applicable, user_id=uid, source="get_milestone"),
}
+4 -1
View File
@@ -44,6 +44,9 @@ async def list_notes(
Prefer `search` for a pure lookup — it returns scores and reaches
everything readable; this adds the lifecycle filters on top.
Each row says what the note is — id, title, description, type, project,
tags, updated_at — and not what it says: read one in full with get_note(id).
Args:
project_id: Scope to one project. PASS THE ACTIVE PROJECT'S ID whenever a
project is in scope so you list that project's notes, not every
@@ -60,7 +63,7 @@ async def list_notes(
limit=max(1, min(limit, 100)),
offset=max(0, offset),
)
return {"notes": [n.to_dict() for n in rows], "total": total}
return {"notes": [notes_svc.brief_row(n) for n in rows], "total": total}
+10 -3
View File
@@ -14,6 +14,7 @@ from __future__ import annotations
from scribe.mcp._context import current_user_id
from scribe.services import canonical_systems as canonical_systems_svc
from scribe.services import milestones as milestones_svc
from scribe.services import notes as notes_svc
from scribe.services import systems as systems_svc
@@ -267,16 +268,18 @@ async def get_system(system_id: int) -> dict:
"""Fetch a System plus the records associated with it.
Returns the system, plus its associated records split into `issues`,
`tasks` (work/plan), and `notes`.
`tasks` (work/plan), and `notes` — each saying what the record is and where
it sits, not what it says (open one with get_note / get_task).
"""
uid = current_user_id()
system = await systems_svc.get_system(uid, system_id)
if system is None:
raise ValueError(f"system {system_id} not found")
records = await systems_svc.list_records_for_system(uid, system_id)
titles = await milestones_svc.titles_for({r.milestone_id for r in records})
issues, tasks, notes = [], [], []
for r in records:
d = r.to_dict()
d = notes_svc.brief_row(r, titles)
if r.status is None:
notes.append(d)
elif r.task_kind == "issue":
@@ -339,12 +342,16 @@ async def list_system_records(
kind: filter by task_kind — 'issue', 'work', 'spike' (or the retired
'plan'). Omit for all.
open_only: limit to tasks not done/cancelled (e.g. open issues only).
Each record says what it is and where it sits, not what it says — open one
with get_note / get_task / get_snippet.
"""
uid = current_user_id()
rows = await systems_svc.list_records_for_system(
uid, system_id, kind=kind or None, open_only=open_only,
)
return {"records": [r.to_dict() for r in rows]}
titles = await milestones_svc.titles_for({r.milestone_id for r in rows})
return {"records": [notes_svc.brief_row(r, titles) for r in rows]}
async def delete_system(system_id: int) -> dict:
+7 -2
View File
@@ -22,6 +22,7 @@ from scribe.mcp._context import current_user_id
from scribe.mcp.tools import systems as systems_tools
from scribe.services import access as access_svc
from scribe.services import dedup as dedup_svc
from scribe.services import milestones as milestones_svc
from scribe.services import notes as notes_svc
# Imported by NAME, not reached through notes_svc: minted_kind is pure
# validation, not a service call, and a test that stubs the service module to
@@ -58,7 +59,10 @@ async def list_tasks(
kind: Filter by task kind — 'work', 'issue', 'spike' (or the retired
'plan'). Omit (empty) for all kinds.
Results are ordered by last-updated descending.
Results are ordered by last-updated descending. Each row says what the task
is and where it sits — id, title, status, kind, priority, milestone (id and
title), tags, updated_at, plus description, parent and due date when set —
and not what it says: read one in full with get_task(id).
"""
uid = current_user_id()
rows, total = await notes_svc.list_notes(
@@ -70,7 +74,8 @@ async def list_tasks(
limit=max(1, min(limit, 100)),
offset=max(0, offset),
)
return {"tasks": [n.to_dict() for n in rows], "total": total}
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}
async def get_task(task_id: int) -> dict:
+20
View File
@@ -72,6 +72,26 @@ async def get_milestone(user_id: int, milestone_id: int) -> Milestone | None:
return result.scalars().first()
async def titles_for(milestone_ids: set[int]) -> dict[int, str]:
"""{id: title} for the given milestones, for rows that name where a record sits.
No ownership filter, deliberately: the callers are listings whose rows the
caller could already read, and a milestone title is part of "where does
this record sit". Filtering here would blank the placement of a shared
task in someone else's plan while still showing the task.
"""
ids = {i for i in milestone_ids if i}
if not ids:
return {}
async with async_session() as session:
rows = (await session.execute(
select(Milestone.id, Milestone.title).where(
Milestone.id.in_(ids), Milestone.deleted_at.is_(None),
)
)).all()
return {mid: title for mid, title in rows}
async def get_milestone_in_project(project_id: int, milestone_id: int) -> Milestone | None:
"""Fetch a milestone by id within a project, without a user_id ownership check.
Callers must verify project access separately before using this."""
+38
View File
@@ -7,6 +7,7 @@ from sqlalchemy import func, or_, select, text
from scribe.models import async_session
from scribe.models.note import Note, TaskKind, TaskPriority, TaskStatus
from scribe.models.base import iso
logger = logging.getLogger(__name__)
@@ -1057,3 +1058,40 @@ async def get_note_for_user(
async with async_session() as session:
note = await session.get(Note, note_id)
return (note, perm) if note else None
# What a LISTING row carries (#4061): what the record is and where it sits,
# never what it says. list_tasks once returned every row's to_dict(), body
# included, and a project's todo list came to 93-165k characters, past what an
# MCP client accepts inline, so the list arrived as a file to page through. The
# same failure #4045 fixed for enter_project, one call over. get_task / get_note
# read a record in full; a list is for choosing which one to open.
#
# 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:
row = {
"id": note.id,
"title": note.title,
"note_type": note.note_type or "note",
"project_id": note.project_id,
"tags": note.tags or [],
"updated_at": iso(note.updated_at),
}
if note.description:
row["description"] = note.description
if note.is_task:
row.update({
"status": note.status,
"task_kind": note.task_kind,
"priority": note.priority,
"milestone_id": note.milestone_id,
})
if milestone_titles is not None and note.milestone_id:
row["milestone_title"] = milestone_titles.get(note.milestone_id)
if note.parent_id:
row["parent_id"] = note.parent_id
if note.due_date:
row["due_date"] = iso(note.due_date)
return row