"""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 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". 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 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 # 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 from scribe.services.text import elide # 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 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, 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. 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( 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), ) titles = await milestones_svc.titles_for({n.milestone_id for n in rows}) # 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, 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, 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 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. `project_id` is the project you are WORKING IN, not this record's own. Passing the active project is what makes "opened away from where it was written" answerable; 0 leaves it unreported and the pull still counts. """ 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 ) # 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, ) 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). When the work belongs to an active plan — a milestone in enter_project's lists or found by search(content_type= "milestone") — pass its id, so the plan shows all of its work. 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. 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 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). This is how steps are added to a plan that already exists, including one start_planning handed back as `existing_milestone`. 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 both duplicate gates. By default nothing is created — milestone included — when an ACTIVE plan in the project already has this title or reads as the same plan, or when a step near-duplicates an existing task. Pass it once you have read the match and know this is separate work. 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 with nothing created: `existing_milestone` when a plan already covers this (add your steps to it with create_records( milestone_id=...)), or `record` naming the step that matched a task. """ 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 not force: # The plan before its steps: when a plan already covers this, the # answer is to add these steps THERE, and a step-level match would # only name one symptom of that. match = await dedup_svc.plan_gate( uid, project_id, title, dedup_svc.plan_candidate_text( body=body, steps=[(i.title, i.body) for i in items], ), ) if match is not None: return match 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)