bvandeusenandClaude Opus 5
7985f8c7d7
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 8s
CI & Build / integration (push) Successful in 29s
CI & Build / TypeScript typecheck (push) Successful in 38s
CI & Build / Python tests (push) Successful in 1m15s
CI & Build / Build & push image (push) Successful in 36s
CI caught the naming work reaching for Postgres from the unit lane, and the connection error was the symptom of a real design fault rather than a test gap: reading the title BEFORE the delete put a live query on the delete path, so a lookup that failed would have stopped the delete happening. That is a decoration breaking its payload — the same mistake just fixed in rules_etag, made again two commits later. Every title lookup now fails open: delete_task, delete_note, delete_milestone, delete_snippet and rule_history lose the name, never the operation. The five unit tests mock the lookup rather than reaching for a database, and delete_note gains one asserting the delete still happens when the lookup raises — the behaviour, not just the absence of a crash. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
397 lines
17 KiB
Python
397 lines
17 KiB
Python
"""Task CRUD MCP tools.
|
|
|
|
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.
|
|
|
|
There is no delete_task — matches the existing fable-mcp surface.
|
|
Cancel by updating status to "cancelled".
|
|
|
|
Sentinels (preserved from existing fable-mcp):
|
|
- status="" / priority="" / title="" / body="" → "leave unchanged" on update
|
|
- status="todo" is the default on create (creates a task; non-null status is
|
|
what makes a Note a Task)
|
|
- priority="none" sets explicit no-priority; priority="" is "leave unchanged"
|
|
- project_id=0 / milestone_id=0 / parent_id=0 → "no association" on create,
|
|
"leave unchanged" on update; on update, -1 clears the FK (sets it NULL)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
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 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
|
|
# avoid the database would otherwise stub the validation too — turning a
|
|
# guard into a MagicMock that approves anything.
|
|
from scribe.services.notes import minted_kind
|
|
from scribe.services import planning as planning_svc
|
|
from scribe.services import rulebooks as rulebooks_svc
|
|
from scribe.services import systems as systems_svc
|
|
from scribe.services import task_logs as task_logs_svc
|
|
from scribe.services import trash as trash_svc
|
|
from scribe.services.note_usage import record_pulled
|
|
|
|
|
|
async def list_tasks(
|
|
limit: int = 20,
|
|
offset: int = 0,
|
|
status: str = "",
|
|
project_id: int = 0,
|
|
kind: str = "",
|
|
) -> dict:
|
|
"""List tasks in Scribe.
|
|
|
|
Args:
|
|
status: Filter by status — one of: todo, in_progress, done, cancelled. Omit for all.
|
|
project_id: Filter to a specific project. PASS THE ACTIVE PROJECT'S ID
|
|
whenever a project is in scope so you list that project's tasks, not
|
|
every project's. 0 = no filter (all projects — use only for a
|
|
deliberate cross-project view).
|
|
kind: Filter by task kind — 'work', 'issue', 'spike' (or the retired
|
|
'plan'). Omit (empty) for all kinds.
|
|
|
|
Results are ordered by last-updated descending.
|
|
"""
|
|
uid = current_user_id()
|
|
rows, total = await notes_svc.list_notes(
|
|
uid,
|
|
is_task=True,
|
|
status=status or None,
|
|
project_id=project_id or None,
|
|
task_kind=kind or None,
|
|
limit=max(1, min(limit, 100)),
|
|
offset=max(0, offset),
|
|
)
|
|
return {"tasks": [n.to_dict() for n in rows], "total": total}
|
|
|
|
|
|
async def get_task(task_id: int) -> dict:
|
|
"""Fetch a single Scribe task by ID.
|
|
|
|
Returns id, title, body, status, priority, tags, project_id, milestone_id,
|
|
parent_id, parent_title, due_date, created_at, updated_at — plus `systems`
|
|
(the areas this task is filed under; read a subsystem's whole pile with
|
|
list_system_records) or, for an untagged project task, the `systems_hint`
|
|
question. For legacy
|
|
kind=plan tasks, the response also includes applicable_rules +
|
|
subscribed_rulebooks from the task's project's rulebook subscriptions (new
|
|
plans are milestones — use get_milestone for those).
|
|
|
|
A task another user shared with you also carries `shared`, `owner` and
|
|
`permission` — it's their work item, not one you took on.
|
|
"""
|
|
uid = current_user_id()
|
|
loaded = await notes_svc.get_note_for_user(uid, task_id)
|
|
note = loaded[0] if loaded else None
|
|
if note is None or note.deleted_at is not None:
|
|
raise ValueError(f"task {task_id} not found")
|
|
data = note.to_dict()
|
|
parent_title = None
|
|
if note.parent_id:
|
|
parent_loaded = await notes_svc.get_note_for_user(uid, note.parent_id)
|
|
if parent_loaded is not None:
|
|
parent_title = parent_loaded[0].title
|
|
data["parent_title"] = parent_title
|
|
|
|
# Legacy kind=plan tasks predate milestone-as-plan; still surface their
|
|
# project's rules on read so the historical plans stay useful.
|
|
if data.get("task_kind") == "plan" and note.project_id:
|
|
applicable = await rulebooks_svc.get_applicable_rules(
|
|
project_id=note.project_id, user_id=uid,
|
|
)
|
|
data.update(rulebooks_svc.rules_payload(applicable))
|
|
data.update(await access_svc.describe_provenance(uid, note))
|
|
# Same reasoning as get_note's record_pulled, and this is the tool where it
|
|
# matters MOST: auto-inject ranks kind-blind over a corpus that is
|
|
# overwhelmingly tasks and issues, so tasks dominate what it surfaces. Without
|
|
# this the surfaced→pulled loop was open exactly where the volume is — every
|
|
# surfaced task counted as never-pulled because the tool that opens one didn't
|
|
# say so, driving auto-inject's measured pull-through toward zero for its own
|
|
# dominant kind. #1038 and #2085 are explicitly gated on that number (#2245).
|
|
await systems_tools.attach_systems(
|
|
uid, getattr(note, "user_id", uid) or uid, data, note.id, note.project_id
|
|
)
|
|
record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_task")
|
|
return data
|
|
|
|
|
|
async def create_task(
|
|
title: str,
|
|
body: str = "",
|
|
status: str = "todo",
|
|
priority: str = "",
|
|
project_id: int = 0,
|
|
milestone_id: int = 0,
|
|
parent_id: int = 0,
|
|
tags: list[str] | None = None,
|
|
kind: str = "work",
|
|
system_ids: list[int] | None = None,
|
|
arose_from_id: int = 0,
|
|
force: bool = False,
|
|
) -> dict:
|
|
"""Create a new task in Scribe.
|
|
|
|
IS ANYTHING ACTUALLY OWED? A task carries a status and someone is on the
|
|
hook to move it. If nothing is owed — you are recording what you learned,
|
|
decided or observed — that is a note (create_note), and filing it here
|
|
leaves a to-do nobody will ever close. If the work is an ARC of several
|
|
steps toward one goal, start_planning makes the milestone that holds
|
|
them; a task is one step, not the plan.
|
|
|
|
Args:
|
|
title: Task title (required).
|
|
body: Markdown description / notes for the task.
|
|
status: Initial status — one of: todo (default), in_progress, done, cancelled.
|
|
priority: One of: low, medium, high, or 'none'. Omit (empty string) to leave unset.
|
|
project_id: Associate with a project (0 = no project).
|
|
milestone_id: Place within a project milestone (0 = no milestone).
|
|
parent_id: Make this a sub-task of another task (0 = top-level).
|
|
tags: List of plain-string tags without # prefix.
|
|
kind: 'work' (default), 'issue', or 'spike'.
|
|
An ISSUE is corrective work — a problem you fixed or are fixing;
|
|
record symptom → root cause → fix in the body.
|
|
A SPIKE is time-boxed and its output is KNOWLEDGE rather than a
|
|
change: "find out whether the runner can be given a bash shell",
|
|
"work out why the index is not used". It succeeds by producing an
|
|
answer, so nothing ships at the end of it — which is why filing
|
|
one as `work` makes a finished investigation look like an
|
|
abandoned change. Reach for it when the honest deliverable is a
|
|
finding, and say in the body what would close the box: a time, or
|
|
the question being answered well enough to act on.
|
|
(Plans are milestones now — call start_planning to begin a plan;
|
|
'plan' is not a valid kind here.)
|
|
system_ids: Ids of the project's Systems (reusable subsystem/area
|
|
objects; see list_systems / create_system) to associate this task with.
|
|
arose_from_id: For an issue, the id of the task/feature it arose from;
|
|
for a spike, the record that raised the question — including a
|
|
standing rule whose check just failed. 0 = none.
|
|
force: Bypass the near-duplicate gate. By default, if a title- or
|
|
meaning-similar task already exists in the same project, creation is
|
|
BLOCKED and the existing task's id is returned so you update it
|
|
instead. Set true only for a genuinely distinct task.
|
|
|
|
Returns the created task, OR — when a near-duplicate is found and force is
|
|
false — {"duplicate": true, "existing_id": ..., "message": ...} (nothing
|
|
created). A tagged record shows its `systems`; created untagged in a
|
|
project, the response carries the `systems_hint` question instead —
|
|
answer it: tag the record, create the missing System, or deliberately
|
|
leave it untagged.
|
|
"""
|
|
uid = current_user_id()
|
|
if kind == "plan":
|
|
raise ValueError(
|
|
"kind=plan is retired — a plan is now a milestone. Call "
|
|
"start_planning(project_id, title) to begin a plan (it creates the "
|
|
"milestone + seeds the design), then create each step as its own "
|
|
"task with create_task(milestone_id=<that milestone>)."
|
|
)
|
|
if not force:
|
|
dup = await dedup_svc.find_duplicate_note(
|
|
uid, title, body, project_id=project_id or None,
|
|
is_task=True, note_type="note",
|
|
)
|
|
if dup is not None:
|
|
return dedup_svc.duplicate_response(dup, "task")
|
|
note = await notes_svc.create_note(
|
|
uid,
|
|
title=title,
|
|
body=body,
|
|
status=status,
|
|
priority=priority or None,
|
|
project_id=project_id or None,
|
|
milestone_id=milestone_id or None,
|
|
parent_id=parent_id or None,
|
|
tags=tags,
|
|
task_kind=minted_kind(kind),
|
|
arose_from_id=arose_from_id or None,
|
|
)
|
|
if system_ids:
|
|
await systems_svc.set_record_systems(uid, note.id, system_ids)
|
|
data = note.to_dict()
|
|
await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
|
|
return data
|
|
|
|
|
|
async def update_task(
|
|
task_id: int,
|
|
title: str = "",
|
|
body: str = "",
|
|
status: str = "",
|
|
priority: str = "",
|
|
project_id: int = 0,
|
|
milestone_id: int = 0,
|
|
system_ids: list[int] | None = None,
|
|
arose_from_id: int = 0,
|
|
kind: str = "",
|
|
) -> dict:
|
|
"""Update an existing Scribe task. Only explicitly provided fields are changed.
|
|
|
|
Args:
|
|
task_id: ID of the task to update.
|
|
title: New title, or omit to leave unchanged.
|
|
body: New markdown body, or omit to leave unchanged.
|
|
status: New status — one of: todo, in_progress, done, cancelled. Drive
|
|
the lifecycle: set in_progress when you start, done when complete —
|
|
don't leave finished work at todo.
|
|
priority: New priority — one of: none, low, medium, high.
|
|
project_id: New project. 0 = leave unchanged, -1 = clear (remove from
|
|
its project; also clears the milestone), positive = set.
|
|
milestone_id: New milestone. 0 = leave unchanged, -1 = clear (remove
|
|
from its milestone), positive = set.
|
|
system_ids: Replace this task's System associations with these ids
|
|
(set-semantics). None = leave unchanged; [] = clear all.
|
|
arose_from_id: Provenance (issue → originating task). 0 = leave unchanged,
|
|
-1 = clear, positive = set.
|
|
kind: Re-file this task as 'work', 'issue' or 'spike'. Omit (empty) to
|
|
leave unchanged. Correcting a kind is ordinary — what a task turns
|
|
out to BE is often clear only once it is under way, and a piece of
|
|
work that becomes an investigation should say so. 'plan' is
|
|
refused: plans are milestones (start_planning), and the value
|
|
survives only so historical plan-tasks stay writable.
|
|
"""
|
|
uid = current_user_id()
|
|
fields: dict = {}
|
|
if title:
|
|
fields["title"] = title
|
|
if body:
|
|
fields["body"] = body
|
|
if status:
|
|
fields["status"] = status
|
|
if priority:
|
|
fields["priority"] = priority
|
|
# Optional FKs: 0 = leave unchanged, -1 = clear (set NULL), positive = set.
|
|
if project_id == -1:
|
|
fields["project_id"] = None
|
|
fields["milestone_id"] = None # a milestone can't outlive its project
|
|
elif project_id:
|
|
fields["project_id"] = project_id
|
|
if milestone_id == -1:
|
|
fields["milestone_id"] = None
|
|
elif milestone_id:
|
|
fields["milestone_id"] = milestone_id
|
|
if arose_from_id == -1:
|
|
fields["arose_from_id"] = None
|
|
elif arose_from_id:
|
|
fields["arose_from_id"] = arose_from_id
|
|
if kind:
|
|
fields["task_kind"] = minted_kind(kind)
|
|
note = await notes_svc.update_note(uid, task_id, **fields)
|
|
if note is None:
|
|
raise ValueError(f"task {task_id} not found")
|
|
if system_ids is not None:
|
|
await systems_svc.set_record_systems(uid, task_id, system_ids)
|
|
data = note.to_dict()
|
|
await systems_tools.attach_systems(
|
|
uid, getattr(note, "user_id", uid) or uid, data, task_id, note.project_id
|
|
)
|
|
return data
|
|
|
|
|
|
async def add_task_log(task_id: int, content: str) -> dict:
|
|
"""Append a timestamped progress log entry to a Scribe task.
|
|
|
|
Use this to record work sessions, decisions, or status updates over time
|
|
without overwriting the task's main body. Each entry is stored separately
|
|
and shown chronologically in the task view.
|
|
|
|
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
|
|
the missing area) rather than logging past it.
|
|
"""
|
|
uid = current_user_id()
|
|
log = await task_logs_svc.create_log(uid, task_id, content)
|
|
data = log.to_dict() if hasattr(log, "to_dict") else {
|
|
"id": log.id, "task_id": log.task_id, "content": log.content,
|
|
"created_at": log.created_at.isoformat() if log.created_at else None,
|
|
}
|
|
# A work-log is work happening on the task NOW — the strongest moment to
|
|
# ask which System's territory that work is in. Fail-open decoration.
|
|
try:
|
|
loaded = await notes_svc.get_note_for_user(uid, task_id)
|
|
if loaded:
|
|
task = loaded[0]
|
|
await systems_tools.attach_systems(
|
|
uid, getattr(task, "user_id", uid) or uid,
|
|
data, task_id, task.project_id,
|
|
)
|
|
except Exception:
|
|
pass
|
|
return data
|
|
|
|
|
|
async def start_planning(project_id: int, title: str) -> dict:
|
|
"""Begin a plan in Scribe (the preferred home for plans — not a local .md file).
|
|
|
|
Reach for this when the work has an ARC — several steps toward one goal,
|
|
worth tracking as a unit. Work without one (a fix, a one-file change, a
|
|
question answered) is a task, not a plan: create_task, drive its status, and
|
|
record progress with add_task_log. A design or decision you are RECORDING
|
|
rather than executing is a note (create_note) — a plan nobody is going to
|
|
work through is a document filed in the place reserved for open work. A milestone holding a single step is
|
|
ceremony, and it leaves the project with a plan that never meant anything.
|
|
|
|
Creates a MILESTONE that IS the plan: its `body` is seeded with a design
|
|
template (Goal/Approach/Verification) under the given project, and the call
|
|
returns it together with the project's applicable Rulebook rules and brief
|
|
context. The milestone is the plan container — the individual steps live as
|
|
first-class child tasks under it, not as checkboxes in the body.
|
|
|
|
Afterwards:
|
|
- Edit the plan/design with update_milestone(milestone_id, body=...).
|
|
- Create each step as its own task with create_task(milestone_id=<this id>);
|
|
track it with status + add_task_log. Do NOT put steps as checkboxes in the
|
|
milestone body.
|
|
|
|
(kind=plan tasks are retired — use this instead. Existing historical
|
|
plan-tasks remain readable but new planning goes through milestones.)
|
|
|
|
Args:
|
|
project_id: The project this plan is for.
|
|
title: A short title for the plan/milestone.
|
|
"""
|
|
uid = current_user_id()
|
|
return await planning_svc.start_planning(
|
|
user_id=uid, project_id=project_id, title=title,
|
|
)
|
|
|
|
|
|
async def delete_task(task_id: int) -> dict:
|
|
"""Move a Scribe task (or plan) to the trash (recoverable). Sub-tasks go with it.
|
|
Restore via restore(batch_id)."""
|
|
uid = current_user_id()
|
|
# Read the title BEFORE the delete: afterwards the row is trashed and the
|
|
# confirmation could only echo the number back. A deletion the operator
|
|
# cannot recognise is one they cannot tell was the wrong one.
|
|
# Fail-open: the title is a COURTESY on top of the delete, so a lookup
|
|
# that errors must not stop the delete happening. Same posture the
|
|
# staleness marker takes — a decoration may never break its payload.
|
|
try:
|
|
loaded = await notes_svc.get_note_for_user(uid, task_id)
|
|
title = getattr(loaded[0], "title", "") if loaded else ""
|
|
except Exception:
|
|
title = ""
|
|
batch = await trash_svc.delete(uid, "task", task_id)
|
|
if batch is None:
|
|
raise ValueError(f"task {task_id} not found")
|
|
return {"deleted": task_id, "title": title, "deleted_batch_id": batch,
|
|
"message": f'Task {task_id} ("{title}") moved to trash. '
|
|
f"Restore with restore('{batch}')."}
|
|
|
|
|
|
def register(mcp) -> None:
|
|
for fn in (
|
|
list_tasks,
|
|
get_task,
|
|
create_task,
|
|
update_task,
|
|
add_task_log,
|
|
start_planning,
|
|
delete_task,
|
|
):
|
|
mcp.tool(name=fn.__name__)(fn)
|