bvandeusenandClaude Fable 5
64c641ce80
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / integration (push) Successful in 25s
CI & Build / TypeScript typecheck (push) Successful in 32s
CI & Build / Python tests (push) Failing after 37s
CI & Build / Build & push image (push) Skipped
Reading the 28 route modules against each other and against the MCP tools: - routes/notes.py carried a PUT and a PATCH handler that were the same function minus the supersedes contract on one of them — one handler now serves both verbs, so both carry it. - The two _attach_supersession copies (REST + MCP) become supersession_svc.attach_relations(uid, note_id, data, hint=) — the seam the two surfaces must agree through; only the agent surface adds the one-sentence reading hint. - Three local _uid() wrappers over g.user.id → scribe.auth.get_current_user_id like every other module; design_systems' private _not_found → routes.utils. not_found; the four "********" literals → settings_svc.SECRET_MASK with the read/write contract written once. - routes/plugin.py: the project_id/repo resolution block and the comma-separated id parse were copied into three endpoints — _project_scope() and _int_list() now. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
493 lines
18 KiB
Python
493 lines
18 KiB
Python
import logging
|
|
import re
|
|
|
|
|
|
from quart import Blueprint, jsonify, request
|
|
|
|
from scribe.auth import login_required, get_current_user_id
|
|
from scribe.routes.utils import not_found, parse_iso_date, parse_pagination
|
|
from scribe.services.access import can_write_note
|
|
from scribe.services.notes import (
|
|
build_note_graph,
|
|
convert_note_to_task,
|
|
convert_task_to_note,
|
|
create_note,
|
|
get_all_tags,
|
|
get_backlinks,
|
|
get_note,
|
|
get_note_by_title,
|
|
get_note_for_user,
|
|
get_or_create_note_by_title,
|
|
list_notes,
|
|
update_note,
|
|
)
|
|
from scribe.services.note_drafts import upsert_draft, get_draft, delete_draft
|
|
from scribe.services import dedup as dedup_svc
|
|
from scribe.services import supersession as supersession_svc
|
|
from scribe.services.note_usage import record_pulled
|
|
|
|
from scribe.services.note_versions import list_versions, get_version
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
notes_bp = Blueprint("notes", __name__, url_prefix="/api/notes")
|
|
|
|
|
|
@notes_bp.route("", methods=["GET"])
|
|
@login_required
|
|
async def list_notes_route():
|
|
uid = get_current_user_id()
|
|
q = request.args.get("q")
|
|
tag = request.args.getlist("tag")
|
|
sort = request.args.get("sort", "updated_at")
|
|
order = request.args.get("order", "desc")
|
|
limit, offset = parse_pagination()
|
|
|
|
# Default to non-task notes only; ?is_task=true for tasks, ?all=true for everything
|
|
is_task: bool | None = False
|
|
if request.args.get("all", "").lower() == "true":
|
|
is_task = None
|
|
elif request.args.get("is_task", "").lower() == "true":
|
|
is_task = True
|
|
|
|
project_id = request.args.get("project_id", type=int)
|
|
milestone_id = request.args.get("milestone_id", type=int)
|
|
parent_id = request.args.get("parent_id", type=int)
|
|
no_project = request.args.get("no_project", "").lower() == "true"
|
|
|
|
# type= shorthand used by web frontend (?type=task or ?type=note)
|
|
type_param = request.args.get("type")
|
|
if type_param == "task":
|
|
is_task = True
|
|
elif type_param == "note":
|
|
is_task = False
|
|
|
|
status = request.args.getlist("status") or None
|
|
priority = request.args.getlist("priority") or None
|
|
|
|
notes, total = await list_notes(
|
|
uid, q=q, tags=tag or None, is_task=is_task, sort=sort, order=order,
|
|
limit=limit, offset=offset,
|
|
project_id=project_id, milestone_id=milestone_id, parent_id=parent_id,
|
|
no_project=no_project, status=status, priority=priority,
|
|
)
|
|
return jsonify({"notes": [n.to_dict() for n in notes], "total": total})
|
|
|
|
|
|
@notes_bp.route("", methods=["POST"])
|
|
@login_required
|
|
async def create_note_route():
|
|
uid = get_current_user_id()
|
|
data = await request.get_json()
|
|
body = data.get("body", "")
|
|
tags = data.get("tags", [])
|
|
|
|
# Optional task fields
|
|
status = data.get("status")
|
|
priority = data.get("priority")
|
|
due_date = parse_iso_date(data.get("due_date"), "due_date")
|
|
if isinstance(due_date, tuple):
|
|
return due_date
|
|
|
|
project_id = data.get("project_id")
|
|
if project_id is None and data.get("project"):
|
|
from scribe.services.projects import get_project_by_title as _gpbt
|
|
proj = await _gpbt(uid, data["project"])
|
|
if proj:
|
|
project_id = proj.id
|
|
|
|
note_type = data.get("note_type", "note")
|
|
|
|
try:
|
|
note = await create_note(
|
|
uid,
|
|
title=data.get("title", ""),
|
|
body=body,
|
|
description=data.get("description"),
|
|
tags=tags,
|
|
parent_id=data.get("parent_id"),
|
|
project_id=project_id,
|
|
milestone_id=data.get("milestone_id"),
|
|
status=status,
|
|
priority=priority,
|
|
due_date=due_date,
|
|
note_type=note_type,
|
|
)
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
# Same capability as the MCP create path (#33). Without it the web UI would
|
|
# be the surface on which a supersession claim silently cannot be made.
|
|
if data.get("supersedes"):
|
|
try:
|
|
await supersession_svc.set_supersedes(uid, note.id, data["supersedes"])
|
|
except PermissionError as exc:
|
|
# 403, not 400: the request is well-formed and the caller simply
|
|
# may not write the target. The note itself was created.
|
|
return jsonify({"error": str(exc), "note": note.to_dict()}), 403
|
|
out = note.to_dict()
|
|
await supersession_svc.attach_relations(uid, note.id, out)
|
|
return jsonify(out), 201
|
|
|
|
|
|
@notes_bp.route("/tags", methods=["GET"])
|
|
@login_required
|
|
async def list_tags_route():
|
|
uid = get_current_user_id()
|
|
q = request.args.get("q")
|
|
tags = await get_all_tags(uid, q=q)
|
|
return jsonify({"tags": tags})
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>/append-tag", methods=["POST"])
|
|
@login_required
|
|
async def append_tag_route(note_id: int):
|
|
uid = get_current_user_id()
|
|
data = await request.get_json()
|
|
tag = data.get("tag", "").strip().replace(" ", "-")
|
|
if not tag:
|
|
return jsonify({"error": "tag is required"}), 400
|
|
|
|
note = await get_note(uid, note_id)
|
|
if note is None:
|
|
return not_found("Note")
|
|
|
|
existing = list(note.tags or [])
|
|
if tag not in existing:
|
|
existing.append(tag)
|
|
updated = await update_note(uid, note_id, tags=existing)
|
|
return jsonify(updated.to_dict())
|
|
|
|
|
|
@notes_bp.route("/by-title", methods=["GET"])
|
|
@login_required
|
|
async def get_note_by_title_route():
|
|
uid = get_current_user_id()
|
|
title = request.args.get("title", "")
|
|
if not title:
|
|
return jsonify({"error": "title parameter is required"}), 400
|
|
note = await get_note_by_title(uid, title)
|
|
if note is None:
|
|
return not_found("Note")
|
|
return jsonify(note.to_dict())
|
|
|
|
|
|
@notes_bp.route("/resolve-title", methods=["POST"])
|
|
@login_required
|
|
async def resolve_title_route():
|
|
uid = get_current_user_id()
|
|
data = await request.get_json()
|
|
title = data.get("title", "").strip()
|
|
if not title:
|
|
return jsonify({"error": "title is required"}), 400
|
|
note = await get_or_create_note_by_title(uid, title)
|
|
return jsonify(note.to_dict())
|
|
|
|
|
|
@notes_bp.route("/duplicates", methods=["GET"])
|
|
@login_required
|
|
async def find_duplicate_records_route():
|
|
"""Near-duplicate notes or tasks, grouped, with the per-kind suggestion.
|
|
|
|
?kind=note|task, ?threshold= (0 = configured setting). Registered above the
|
|
`/<int:note_id>` routes on purpose — the literal path first, same reasoning
|
|
as /api/snippets/duplicates. Mirrors the MCP tool (#33): the web UI has no
|
|
dedup gate by design, so this report is the UI's only window onto what that
|
|
decision admits.
|
|
"""
|
|
uid = get_current_user_id()
|
|
kind = request.args.get("kind", "note")
|
|
if kind not in ("note", "task"):
|
|
return jsonify({"error": 'kind must be "note" or "task"'}), 400
|
|
try:
|
|
threshold = float(request.args.get("threshold", 0) or 0)
|
|
except (TypeError, ValueError):
|
|
threshold = 0.0
|
|
return jsonify(await dedup_svc.find_duplicate_records(
|
|
uid, kind=kind, threshold=threshold if threshold > 0 else None,
|
|
))
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>", methods=["GET"])
|
|
@login_required
|
|
async def get_note_route(note_id: int):
|
|
uid = get_current_user_id()
|
|
result = await get_note_for_user(uid, note_id)
|
|
if result is None:
|
|
return not_found("Note")
|
|
note, permission = result
|
|
data = note.to_dict()
|
|
data["permission"] = permission
|
|
# Opening the detail view IS a pull — the operator chose to look. Tagged by
|
|
# SURFACE, not by the record's kind, matching rest_snippet: the kind is a
|
|
# join away, but which surface asked is not recoverable after the fact.
|
|
# Keeping rest_* apart from mcp_* is load-bearing, not tidiness — "was that
|
|
# injected line useful?" is answered by agent pulls alone, and a human
|
|
# clicking a link would inflate exactly the number #1038 and #2085 gate on.
|
|
record_pulled(user_id=uid, note_id=note_id, source="rest_note")
|
|
await supersession_svc.attach_relations(uid, note_id, data)
|
|
return jsonify(data)
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>", methods=["PUT", "PATCH"])
|
|
@login_required
|
|
async def update_note_route(note_id: int):
|
|
"""Partial update — only the keys present in the payload change. PUT and
|
|
PATCH are the same handler on purpose: the form sends the field set it
|
|
edited, and the two verbs used to be two near-identical copies of this
|
|
function that drifted (one carried the supersedes contract, one did not)."""
|
|
uid = get_current_user_id()
|
|
# Share-aware: resolve through the ACL and write as the OWNER, so a shared
|
|
# editor's save isn't rejected by the owner-scoped update service.
|
|
result = await get_note_for_user(uid, note_id)
|
|
if result is None:
|
|
return not_found("Note")
|
|
note_obj, _ = result
|
|
if not await can_write_note(uid, note_id):
|
|
return jsonify({"error": "Permission denied"}), 403
|
|
owner_uid = note_obj.user_id
|
|
data = await request.get_json()
|
|
fields = {}
|
|
for key in ("title", "body", "description", "parent_id", "project_id", "milestone_id", "status", "priority", "note_type"):
|
|
if key in data:
|
|
fields[key] = data[key]
|
|
if "due_date" in data:
|
|
if data["due_date"]:
|
|
result = parse_iso_date(data["due_date"], "due_date")
|
|
if isinstance(result, tuple):
|
|
return result
|
|
fields["due_date"] = result
|
|
else:
|
|
fields["due_date"] = None
|
|
|
|
if "tags" in data:
|
|
fields["tags"] = data["tags"]
|
|
try:
|
|
note = await update_note(owner_uid, note_id, **fields)
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
if note is None:
|
|
return not_found("Note")
|
|
# Set-semantics, matching MCP and the PATCH route: present-and-empty
|
|
# clears, absent leaves alone. Scoped by the CALLER, not owner_uid — an
|
|
# editor-share holder may edit this note and must not thereby inherit the
|
|
# owner's write access to whatever they name as superseded (#47).
|
|
if "supersedes" in data:
|
|
try:
|
|
await supersession_svc.set_supersedes(uid, note_id, data["supersedes"] or [])
|
|
except PermissionError as exc:
|
|
return jsonify({"error": str(exc)}), 403
|
|
out = note.to_dict()
|
|
await supersession_svc.attach_relations(uid, note_id, out)
|
|
return jsonify(out)
|
|
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>", methods=["DELETE"])
|
|
@login_required
|
|
async def delete_note_route(note_id: int):
|
|
uid = get_current_user_id()
|
|
result = await get_note_for_user(uid, note_id)
|
|
if result is None:
|
|
return not_found("Note")
|
|
note_obj, _ = result
|
|
if not await can_write_note(uid, note_id):
|
|
return jsonify({"error": "Permission denied"}), 403
|
|
from scribe.services.trash import delete as trash_delete
|
|
batch = await trash_delete(note_obj.user_id, "note", note_id)
|
|
if batch is None:
|
|
return not_found("Note")
|
|
return "", 204
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>/convert-to-task", methods=["POST"])
|
|
@login_required
|
|
async def convert_note_to_task_route(note_id: int):
|
|
uid = get_current_user_id()
|
|
try:
|
|
note = await convert_note_to_task(uid, note_id)
|
|
return jsonify(note.to_dict()), 200
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 404
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>/convert-to-note", methods=["POST"])
|
|
@login_required
|
|
async def convert_task_to_note_route(note_id: int):
|
|
uid = get_current_user_id()
|
|
try:
|
|
note = await convert_task_to_note(uid, note_id)
|
|
return jsonify(note.to_dict()), 200
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 404
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>/backlinks", methods=["GET"])
|
|
@login_required
|
|
async def get_backlinks_route(note_id: int):
|
|
uid = get_current_user_id()
|
|
links = await get_backlinks(uid, note_id)
|
|
return jsonify({"backlinks": links})
|
|
|
|
|
|
# ── Link suggestions ─────────────────────────────────────────────────────────
|
|
|
|
_WIKILINK_RE = re.compile(r'\[\[[^\]]+\]\]')
|
|
|
|
|
|
def _find_unlinked_terms(body: str, note_titles: list[tuple[int, str]]) -> list[dict]:
|
|
"""Return project note titles that appear in body as plain text (not inside [[...]])."""
|
|
linked_ranges = [(m.start(), m.end()) for m in _WIKILINK_RE.finditer(body)]
|
|
|
|
def in_wikilink(start: int, end: int) -> bool:
|
|
return any(ls <= start and end <= le for ls, le in linked_ranges)
|
|
|
|
suggestions = []
|
|
for note_id, title in note_titles:
|
|
title = title.strip()
|
|
if len(title) < 3:
|
|
continue
|
|
pattern = re.compile(r'\b' + re.escape(title) + r'\b', re.IGNORECASE)
|
|
count = sum(1 for m in pattern.finditer(body) if not in_wikilink(m.start(), m.end()))
|
|
if count > 0:
|
|
suggestions.append({"note_id": note_id, "title": title, "count": count})
|
|
|
|
suggestions.sort(key=lambda x: x["count"], reverse=True)
|
|
return suggestions
|
|
|
|
|
|
@notes_bp.route("/link-suggestions", methods=["POST"])
|
|
@login_required
|
|
async def link_suggestions_route():
|
|
"""Find project note titles that appear unlinked in the given body text."""
|
|
uid = get_current_user_id()
|
|
data = await request.get_json()
|
|
|
|
body_text = data.get("body", "")
|
|
project_id = data.get("project_id")
|
|
exclude_note_id = data.get("exclude_note_id")
|
|
|
|
if not project_id or not body_text:
|
|
return jsonify({"suggestions": []})
|
|
|
|
try:
|
|
project_notes, _ = await list_notes(
|
|
uid, project_id=int(project_id), sort="title", order="asc", limit=200
|
|
)
|
|
titles = [
|
|
(n.id, n.title)
|
|
for n in project_notes
|
|
if n.id != exclude_note_id and n.title
|
|
]
|
|
suggestions = _find_unlinked_terms(body_text, titles)
|
|
except Exception:
|
|
logger.warning("Failed to compute link suggestions", exc_info=True)
|
|
suggestions = []
|
|
|
|
return jsonify({"suggestions": suggestions})
|
|
|
|
|
|
# ── Draft routes ─────────────────────────────────────────────────────────────
|
|
|
|
@notes_bp.route("/<int:note_id>/draft", methods=["GET"])
|
|
@login_required
|
|
async def get_draft_route(note_id: int):
|
|
uid = get_current_user_id()
|
|
draft = await get_draft(uid, note_id)
|
|
if draft is None:
|
|
return jsonify({"error": "No draft found"}), 404
|
|
return jsonify(draft.to_dict())
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>/draft", methods=["PUT"])
|
|
@login_required
|
|
async def upsert_draft_route(note_id: int):
|
|
uid = get_current_user_id()
|
|
# Verify note ownership
|
|
note = await get_note(uid, note_id)
|
|
if note is None:
|
|
return not_found("Note")
|
|
data = await request.get_json()
|
|
draft = await upsert_draft(
|
|
user_id=uid,
|
|
note_id=note_id,
|
|
proposed_body=data.get("proposed_body", ""),
|
|
original_body=data.get("original_body", ""),
|
|
instruction=data.get("instruction", ""),
|
|
scope=data.get("scope", "document"),
|
|
)
|
|
return jsonify(draft.to_dict()), 200
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>/draft", methods=["DELETE"])
|
|
@login_required
|
|
async def delete_draft_route(note_id: int):
|
|
uid = get_current_user_id()
|
|
await delete_draft(uid, note_id)
|
|
return "", 204
|
|
|
|
|
|
# ── Version routes ────────────────────────────────────────────────────────────
|
|
|
|
@notes_bp.route("/<int:note_id>/versions", methods=["GET"])
|
|
@login_required
|
|
async def list_versions_route(note_id: int):
|
|
uid = get_current_user_id()
|
|
versions = await list_versions(uid, note_id)
|
|
return jsonify({"versions": [v.to_dict(include_body=False) for v in versions]})
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>/versions/<int:version_id>", methods=["GET"])
|
|
@login_required
|
|
async def get_version_route(note_id: int, version_id: int):
|
|
uid = get_current_user_id()
|
|
version = await get_version(uid, note_id, version_id)
|
|
if version is None:
|
|
return not_found("Version")
|
|
return jsonify(version.to_dict(include_body=True))
|
|
|
|
|
|
@notes_bp.route("/<int:note_id>/versions/<int:version_id>/pin", methods=["POST"])
|
|
@login_required
|
|
async def pin_version_route(note_id: int, version_id: int):
|
|
"""Mark a version as manually pinned. Body: {"label": str | null}."""
|
|
uid = get_current_user_id()
|
|
data = await request.get_json() or {}
|
|
label = data.get("label")
|
|
if label is not None and not isinstance(label, str):
|
|
return jsonify({"error": "label must be a string or null"}), 400
|
|
from scribe.services.version_pinning import pin_version
|
|
try:
|
|
version = await pin_version(uid, note_id, version_id, label=label)
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
if version is None:
|
|
return not_found("Version")
|
|
return jsonify(version.to_dict(include_body=False))
|
|
|
|
|
|
@notes_bp.route(
|
|
"/<int:note_id>/versions/<int:version_id>/pin", methods=["DELETE"],
|
|
)
|
|
@login_required
|
|
async def unpin_version_route(note_id: int, version_id: int):
|
|
"""Downgrade a manually-pinned version back to rolling."""
|
|
uid = get_current_user_id()
|
|
from scribe.services.version_pinning import unpin_version
|
|
version = await unpin_version(uid, note_id, version_id)
|
|
if version is None:
|
|
return not_found("Version")
|
|
return jsonify(version.to_dict(include_body=False))
|
|
|
|
|
|
# ── Graph route ────────────────────────────────────────────────────────────────
|
|
|
|
@notes_bp.route("/graph", methods=["GET"])
|
|
@login_required
|
|
async def graph_route():
|
|
uid = get_current_user_id()
|
|
project_id = request.args.get("project_id", type=int)
|
|
shared_tags = request.args.get("shared_tags", "false").lower() == "true"
|
|
graph = await build_note_graph(uid, project_id=project_id, include_shared_tags=shared_tags)
|
|
return jsonify(graph)
|