"""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 placement as placement_svc from scribe.services import planning as planning_svc from scribe.services import record_batch as batch_svc from scribe.services import reply_preferences as reply_prefs_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 from scribe.services.record_refs import refuse_guessed_ids 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, user_id=uid, source="get_task")) 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. AN ID EXISTS ONLY ONCE A CREATE RETURNS IT. Never write the id you expect a record to get: every session and user draws from one sequence, so the number is taken by whoever creates next. A body citing a `#N` that has not been assigned yet is refused. Creating several records that cite each other? Use create_records (or start_planning(steps=...)) and write {{ref:N}} — the real ids are filled in as they are created. Returns the created task, OR — when a near-duplicate is found and force is false — {"duplicate": true, "existing_id": ..., "message": ...} (nothing created). A task in a project carries `placement` (see update_task). 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=)." ) await refuse_guessed_ids(title, body) 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 await placement_svc.attach_placement(uid, data, note) 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. The response carries `placement` for a task in a project: its `project`, and for a task in a milestone its `milestone`, `position` ({step, of}), `progress` ({completed, total, pct}) and `next` (the next open step, or null). These are the facts to use when telling the operator where the work sits and what comes next — read them from here rather than reconstructing them, because a remembered milestone title or "next step" reads exactly like a real one when it is wrong. Closing a task (done or cancelled) also returns `report_back`: a one-line reminder of what the reply to the operator should cover. When the operator has preferences for how a completion report is written, they come back as `reply_preferences` ({id, title, statement, kind}) — found by their `when_to_apply`, so a preference whose trigger is writing the report after finishing a task is the one that arrives here. Where one differs from the default shape, the preference is what the operator asked for. """ 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) await refuse_guessed_ids(title, body) 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 ) await placement_svc.attach_placement(uid, data, note) if status in _CLOSING_STATUSES: data["report_back"] = REPORT_BACK_CUE # The operator's own adjustments to the completion report, retrieved # at the one moment a server can see that report coming (milestone # 409 step 4). Omitted rather than sent empty, like every decoration. prefs = await reply_prefs_svc.completion_preferences( uid, project_id=getattr(note, "project_id", None)) if prefs: data["reply_preferences"] = prefs data["report_back"] = REPORT_BACK_CUE + " " + REPLY_PREFERENCES_CUE 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 # The in-band half of milestone 409 step 3. The reporting-back skill and the # static context carry the full shapes, but both live only in the Claude Code # plugin; a tool response reaches every MCP client, at the moment a piece of # work closes, which is exactly when the report is about to be written. One # line on purpose: a template here would be read as the reply itself. _CLOSING_STATUSES = ("done", "cancelled") REPORT_BACK_CUE = ( "Reporting this to the operator? Say where it sits (from `placement`), " "what now works, what needs them, and what comes next." ) # Appended only when `reply_preferences` is present, so the key never arrives # unexplained and a session with no preferences reads exactly what it did. REPLY_PREFERENCES_CUE = ( "The operator has preferences for how this report is written — " "follow `reply_preferences` over the default shape where they differ." ) _ITEM_KEYS = {"title", "body", "type", "status", "priority", "kind", "tags", "system_ids"} def _batch_items(records: list[dict], *, what: str = "record") -> list[batch_svc.BatchItem]: """Parse the door's plain dicts into BatchItems, refusing unknown keys. Strict for the same reason StrictArgsFastMCP is (#2709): a misspelt key silently dropped — `milestone` for a per-record milestone, `desc` for body — creates a record that looks right and is missing what the caller sent. """ items: list[batch_svc.BatchItem] = [] for i, raw in enumerate(records or [], start=1): if not isinstance(raw, dict): raise ValueError(f"{what} {i} must be an object with at least a title") unknown = set(raw) - _ITEM_KEYS if unknown: raise ValueError( f"{what} {i} has unknown field(s) {sorted(unknown)}; " f"allowed: {sorted(_ITEM_KEYS)}" ) rtype = raw.get("type") or "task" if rtype not in ("task", "note"): raise ValueError(f"{what} {i}: type must be 'task' or 'note', got {rtype!r}") items.append(batch_svc.BatchItem( title=raw.get("title") or "", body=raw.get("body") or "", is_task=rtype == "task", status=raw.get("status") or "todo", priority=raw.get("priority") or None, task_kind=raw.get("kind") or "work", tags=list(raw.get("tags") or []), system_ids=list(raw.get("system_ids") or []), )) return items async def _first_duplicate(uid: int, items: list, project_id: int | None) -> dict | None: """The duplicate gate over a whole batch — the first hit blocks all of it.""" for i, item in enumerate(items, start=1): dup = await dedup_svc.find_duplicate_note( uid, item.title, item.body, project_id=project_id, is_task=item.is_task, note_type="note", ) if dup is not None: payload = dedup_svc.duplicate_response(dup, "task" if item.is_task else "note") payload["record"] = i payload["message"] = f"Record {i} of the batch: {payload['message']} Nothing in the batch was created." return payload return None async def create_records( records: list[dict], project_id: int = 0, milestone_id: int = 0, force: bool = False, ) -> dict: """Create several tasks and/or notes in ONE call, so they can cite each other. Reach for this whenever records you are about to create need to reference one another — a set of tasks that name their siblings, a reference note listing the tasks it indexes. Writing the ids you EXPECT them to get is always wrong: every session and user draws from one sequence, so another create takes those numbers, and a body citing an unassigned `#N` is refused. A plan (a milestone plus its steps) is start_planning(steps=...), which uses the same mechanism. One record with nothing to cite is create_task or create_note. In any record's body write {{ref:N}} where the Nth record's id belongs (1-based, in the order listed). It is replaced with `# ""` once the ids exist. Everything is created in one transaction: a bad placeholder, an invalid field or a duplicate creates NOTHING. Args: records: The records, in order. Each is an object with `title` (required) and optionally `body`, `type` ('task', the default, or 'note'), and for tasks `status`, `priority` and `kind` ('work' | 'issue' | 'spike'); plus `tags` and `system_ids`. project_id: The project every record belongs to (0 = none, or taken from milestone_id). milestone_id: File every record under this existing milestone (0 = none). force: Bypass the near-duplicate gate for the whole batch. By default the first record that near-duplicates an existing one BLOCKS the batch, and its existing id comes back so you can update it instead. Returns {"ids": [...], "records": [...]} in the order given — the ids cite-able from here on — OR a duplicate payload naming which `record` matched, with nothing created. The ids are the real ones and are not necessarily consecutive: other sessions keep creating meanwhile, and nothing here depends on the numbers being adjacent. """ uid = current_user_id() items = _batch_items(records) await refuse_guessed_ids(*[t for item in items for t in (item.title, item.body)]) if not force: dup = await _first_duplicate(uid, items, project_id or None) if dup is not None: return dup _ms, notes = await batch_svc.create_batch( uid, items, project_id=project_id or None, milestone_id=milestone_id or None, ) return {"ids": [n.id for n in notes], "records": [n.to_dict() for n in notes]} async def start_planning( project_id: int, title: str, body: str = "", steps: list[dict] | None = None, force: bool = False, ) -> 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. PASS THE STEPS HERE when you already know them. The milestone and every step are then created in one transaction, and the plan body and the steps can cite each other with placeholders that become real ids: {{ref:N}} is the Nth step (1-based) and {{ref:milestone}} is this milestone. Never write the ids you expect records to get — other sessions and users draw from the same sequence, and a body citing an unassigned `#N` is refused. Without steps: edit the design afterwards with update_milestone( milestone_id, body=...), and add steps with create_records(milestone_id= <this id>, ...) — or create_task for a single one. Track each 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. body: The plan's design (markdown). Omit to seed the Goal/Approach/ Verification template. steps: The plan's step-tasks, in order — each an object with `title` (required) and optionally `body`, `status`, `priority`, `kind` ('work' | 'issue' | 'spike'), `tags`, `system_ids`. force: Bypass the near-duplicate gate on the steps. By default a step that near-duplicates an existing task blocks the whole plan, and nothing — milestone included — is created. Returns the milestone, the project's applicable rules and brief context, plus `steps` (the created tasks, in order) when steps were given — OR a duplicate payload naming the `record` that matched, with nothing created. """ uid = current_user_id() items = _batch_items(steps or [], what="step") await refuse_guessed_ids(body, *[t for item in items for t in (item.title, item.body)]) if items and not force: dup = await _first_duplicate(uid, items, project_id or None) if dup is not None: return dup return await planning_svc.start_planning( user_id=uid, project_id=project_id, title=title, body=body or None, steps=items or None, ) 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, create_records, update_task, add_task_log, start_planning, delete_task, ): mcp.tool(name=fn.__name__)(fn)