server: one credential check, one coerce_bool, one top-position query, and the shared response helpers

From the audit (#5179, server half).

- auth: login and device-login share _check_credentials (dummy hash for a
  missing account, throttle bookkeeping, the failure log line) and
  _bad_credentials.
- settings uses common.coerce_bool. Its private copy differed only in
  treating a non-string as its truthiness, which the shared one now does.
- notes: create and import share helpers.top_position.
- auth, settings_api, sync and client_dist return errors through
  responses.json_error / not_found, and parse ids with parse_uuid.
- sync: push replies are built by _result(id, entity, status, **extra).

Already merged by earlier steps, so nothing to do here: attachment storage
(store_attachment), the preview upsert (only unfurl_queue writes one now),
and _serialize_note (delegates to _serialize_notes).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/inkwell/auth.py

@@ -16,6 +16,7 @@ from .password_resets import INVALID as INVALID_RESET, claim as claim_reset, mai
 from .models.device_token import DeviceToken
 from .models.user import User
 from .proxy import client_address
+from .responses import json_error, not_found, parse_uuid
 from .ratelimit import (
     register_by_address,
     reset_mail_by_account,
@@ -81,11 +82,10 @@ def _session_user_id() -> uuid.UUID | None:
     raw = session.get(SESSION_KEY)
     if not raw:
         return None
-    try:
-        return uuid.UUID(raw)
-    except (ValueError, TypeError):
+    uid = parse_uuid(raw)
+    if uid is None:
         _sign_out()
-        return None
+    return uid
 
 
 def _bearer_token() -> str | None:
@@ -131,7 +131,7 @@ def login_required(fn):
         if uid is None:
             uid = await _user_id_from_bearer()
         if uid is None:
-            return jsonify({"error": "authentication required"}), 401
+            return json_error("authentication required", 401)
         g.user_id = uid
         return await fn(*args, **kwargs)
 
@@ -146,13 +146,13 @@ def require_admin(fn):
     async def wrapper(*args, **kwargs):
         uid = _session_user_id()
         if uid is None:
-            return jsonify({"error": "authentication required"}), 401
+            return json_error("authentication required", 401)
         async with session_scope() as db:
             user = await db.get(User, uid)
             if not _session_is_current(user.session_epoch if user else None):
-                return jsonify({"error": "authentication required"}), 401
+                return json_error("authentication required", 401)
             if not user.is_admin:
-                return jsonify({"error": "admin access required"}), 403
+                return json_error("admin access required", 403)
         g.user_id = uid
         return await fn(*args, **kwargs)
 
@@ -168,11 +168,7 @@ def _throttled(retry_after: int):
     genuinely needs.
     """
     logger.warning("throttled credential attempt from=%s retry_after=%ss", client_address(), retry_after)
-    return (
-        jsonify({"error": "too many attempts — try again shortly"}),
-        429,
-        {"Retry-After": str(retry_after)},
-    )
+    return (*json_error("too many attempts — try again shortly", 429), {"Retry-After": str(retry_after)})
 
 
 def _sign_in_block(email: str) -> int | None:
@@ -215,9 +211,9 @@ async def register():
     invite = (data.get("invite") or "").strip()
 
     if not email or "@" not in email:
-        return jsonify({"error": "a valid email is required"}), 400
+        return json_error("a valid email is required", 400)
     if len(password) < MIN_PASSWORD_LEN:
-        return jsonify({"error": f"password must be at least {MIN_PASSWORD_LEN} characters"}), 400
+        return json_error(f"password must be at least {MIN_PASSWORD_LEN} characters", 400)
     if not display_name:
         display_name = email.split("@", 1)[0]
 
@@ -242,15 +238,15 @@ async def register():
             invite_id = await redeem(db, invite, email)
             if invite_id is None:
                 logger.warning("registration refused (bad invite) email=%s from=%s", email, client_address())
-                return jsonify({"error": INVALID_INVITE}), 403
+                return json_error(INVALID_INVITE, 403)
         elif not is_first and not await get_setting(db, "allow_registration"):
             logger.warning("registration refused (closed) email=%s from=%s", email, client_address())
-            return jsonify({"error": "registration is closed"}), 403
+            return json_error("registration is closed", 403)
         # Returning without a commit rolls back the redemption above with it, so a
         # taken email leaves the invite usable.
         existing = await db.scalar(select(User).where(User.email == email))
         if existing is not None:
-            return jsonify({"error": "an account with that email already exists"}), 409
+            return json_error("an account with that email already exists", 409)
         user = User(
             email=email,
             password_hash=hash_password(password),
@@ -284,6 +280,32 @@ async def register():
         return jsonify(_serialize_user(user)), 201
 
 
+async def _check_credentials(db, email: str, password: str, route: str) -> User | None:
+    """The account these credentials open, or None. Counts the failure against the
+    throttle, logs it under `route`, and clears the account's history on success.
+
+    A missing account still pays for a hash. Without that, "no such account" returns
+    in microseconds while a wrong password takes bcrypt's deliberate ~100ms, and the
+    difference is a reliable oracle for which emails have accounts here.
+    """
+    user = await db.scalar(select(User).where(User.email == email))
+    if user is None or not user.password_hash:
+        dummy_verify(password)
+        reason = "no such account"
+    elif not verify_password(password, user.password_hash):
+        reason = "bad password"
+    else:
+        _sign_in_succeeded(email)
+        return user
+    _sign_in_failed(email)
+    logger.warning("%s failed (%s) email=%s from=%s", route, reason, email, client_address())
+    return None
+
+
+def _bad_credentials():
+    return json_error("invalid email or password", 401)
+
+
 @bp.post("/login")
 async def login():
     data = await request.get_json(silent=True) or {}
@@ -295,20 +317,9 @@ async def login():
         return _throttled(wait)
 
     async with session_scope() as db:
-        user = await db.scalar(select(User).where(User.email == email))
-        if user is None or not user.password_hash:
-            # Hash anyway. Without this, "no such account" returns in microseconds
-            # while a wrong password takes bcrypt's deliberate ~100ms, and the
-            # difference is a reliable oracle for which emails have accounts here.
-            dummy_verify(password)
-            _sign_in_failed(email)
-            logger.warning("sign-in failed (no such account) email=%s from=%s", email, client_address())
-            return jsonify({"error": "invalid email or password"}), 401
-        if not verify_password(password, user.password_hash):
-            _sign_in_failed(email)
-            logger.warning("sign-in failed (bad password) email=%s from=%s", email, client_address())
-            return jsonify({"error": "invalid email or password"}), 401
-        _sign_in_succeeded(email)
+        user = await _check_credentials(db, email, password, "sign-in")
+        if user is None:
+            return _bad_credentials()
         _sign_in(user)
         logger.info("sign-in ok email=%s from=%s", email, client_address())
         return jsonify(_serialize_user(user))
@@ -327,7 +338,7 @@ async def me():
         user = await db.get(User, g.user_id)
         if user is None:
             _sign_out()
-            return jsonify({"error": "authentication required"}), 401
+            return json_error("authentication required", 401)
         return jsonify(_serialize_user(user))
 
 
@@ -347,7 +358,7 @@ async def forgot_password():
     data = await request.get_json(silent=True) or {}
     email = (data.get("email") or "").strip().lower()
     if not email or "@" not in email:
-        return jsonify({"error": "a valid email is required"}), 400
+        return json_error("a valid email is required", 400)
 
     wait = _sign_in_block("")
     if wait is not None:
@@ -356,7 +367,7 @@ async def forgot_password():
 
     async with session_scope() as db:
         if not await mail_configured(db):
-            return jsonify({"error": "this server can't send email; ask your admin for a reset link"}), 400
+            return json_error("this server can't send email; ask your admin for a reset link", 400)
 
     if reset_mail_by_account.retry_after(email) is None:
         reset_mail_by_account.record(email)
@@ -379,7 +390,7 @@ async def reset_password():
     token = (data.get("token") or "").strip()
     password = data.get("password") or ""
     if len(password) < MIN_PASSWORD_LEN:
-        return jsonify({"error": f"password must be at least {MIN_PASSWORD_LEN} characters"}), 400
+        return json_error(f"password must be at least {MIN_PASSWORD_LEN} characters", 400)
 
     # The sign-in budget for this address. A token can't be guessed, but each attempt
     # is still a credential attempt, and the log should show a run of them.
@@ -392,7 +403,7 @@ async def reset_password():
         if user_id is None:
             _sign_in_failed("")
             logger.warning("password reset refused (bad link) from=%s", client_address())
-            return jsonify({"error": INVALID_RESET}), 403
+            return json_error(INVALID_RESET, 403)
         user = await db.get(User, user_id)
         user.password_hash = hash_password(password)
         user.session_epoch = User.session_epoch + 1
@@ -439,7 +450,7 @@ async def device_login():
     email = (data.get("email") or "").strip().lower()
     password = data.get("password") or ""
     if not email or not password:
-        return jsonify({"error": "email and password are required"}), 400
+        return json_error("email and password are required", 400)
 
     # Same budget as the web sign-in, and the SAME counters — this route hands out a
     # long-lived bearer token, so leaving it unthrottled would just move the guessing
@@ -449,17 +460,9 @@ async def device_login():
         return _throttled(wait)
 
     async with session_scope() as db:
-        user = await db.scalar(select(User).where(User.email == email))
-        if user is None or not user.password_hash:
-            dummy_verify(password)
-            _sign_in_failed(email)
-            logger.warning("device-login failed (no such account) email=%s from=%s", email, client_address())
-            return jsonify({"error": "invalid email or password"}), 401
-        if not verify_password(password, user.password_hash):
-            _sign_in_failed(email)
-            logger.warning("device-login failed (bad password) email=%s from=%s", email, client_address())
-            return jsonify({"error": "invalid email or password"}), 401
-        _sign_in_succeeded(email)
+        user = await _check_credentials(db, email, password, "device-login")
+        if user is None:
+            return _bad_credentials()
         row, token = await _issue_device_token(db, user.id, data.get("name") or "")
         # A device token outlives the session that made it, so its creation is the
         # most consequential thing on this blueprint.
@@ -513,7 +516,7 @@ async def revoke_own_device():
         # A session-cookie caller holds no device token, so "revoke the one I'm
         # using" is meaningless rather than merely unauthorized. The web app
         # revokes by id.
-        return jsonify({"error": "no device token was presented"}), 400
+        return json_error("no device token was presented", 400)
     async with session_scope() as db:
         row = await db.scalar(
             select(DeviceToken).where(
@@ -524,7 +527,7 @@ async def revoke_own_device():
             )
         )
         if row is None:
-            return jsonify({"error": "not found"}), 404
+            return not_found()
         await db.delete(row)
         await db.commit()
         return jsonify({"ok": True})
@@ -533,16 +536,15 @@ async def revoke_own_device():
 @bp.delete("/devices/<device_id>")
 @login_required
 async def revoke_device(device_id: str):
-    try:
-        did = uuid.UUID(device_id)
-    except (ValueError, TypeError):
-        return jsonify({"error": "not found"}), 404
+    did = parse_uuid(device_id)
+    if did is None:
+        return not_found()
     async with session_scope() as db:
         row = await db.scalar(
             select(DeviceToken).where(DeviceToken.id == did, DeviceToken.user_id == g.user_id)
         )
         if row is None:
-            return jsonify({"error": "not found"}), 404
+            return not_found()
         await db.delete(row)
         await db.commit()
         return jsonify({"ok": True})

src/inkwell/client_dist.py

@@ -80,6 +80,7 @@ from quart import Blueprint, jsonify, send_from_directory
 
 from .auth import login_required
 from .config import Config
+from .responses import json_error
 
 
 @dataclass(frozen=True)
@@ -323,7 +324,7 @@ async def client_metadata(platform_id: str):
     """
     found = release(platform_id)
     if found is None:
-        return jsonify({"error": f"this server has no {platform_id} client"}), 404
+        return json_error(f"this server has no {platform_id} client", 404)
     return jsonify(found)
 
 
@@ -338,10 +339,10 @@ async def client_download(platform_id: str):
     """
     platform = BY_ID.get(platform_id)
     if platform is None:
-        return jsonify({"error": f"unknown client platform '{platform_id}'"}), 404
+        return json_error(f"unknown client platform '{platform_id}'", 404)
     resolved = _resolve(platform)
     if resolved is None:
-        return jsonify({"error": f"this server has no {platform_id} client"}), 404
+        return json_error(f"this server has no {platform_id} client", 404)
     # From the SAME directory the metadata came from, or a drop-in appearing between
     # the two calls would serve bytes the metadata does not describe.
     root, _ = resolved

src/inkwell/common.py

@@ -32,10 +32,11 @@ def coerce_bool(raw: object) -> bool:
     """Truthy for the common flag spellings ('true'/'1'/'yes'/'on', or a real bool).
 
     Used for query-string booleans (?has_reminder=true) and DB-backed bool settings,
-    which arrive as strings.
+    which arrive as strings; anything else (a number from a JSON body, None) is its
+    ordinary truthiness.
     """
     if isinstance(raw, bool):
         return raw
     if isinstance(raw, str):
         return raw.strip().lower() in ("true", "1", "yes", "on")
-    return False
+    return bool(raw)

src/inkwell/notes/__init__.py

@@ -53,6 +53,7 @@ from .helpers import (
     derive_display_title,
     is_empty_note,
     store_attachment,
+    top_position,
     unlink_media,
     parse_list_items,
 )
@@ -331,12 +332,7 @@ async def import_notes():
         imported = 0
         skipped = 0
         async with session_scope() as db:
-            max_pos = await db.scalar(
-                select(func.coalesce(func.max(Note.position), 0)).where(
-                    Note.owner_id == g.user_id, Note.deleted_at.is_(None)
-                )
-            )
-            pos = int(max_pos)
+            pos = await top_position(db, g.user_id)
             for spec in specs:
                 if await _create_imported_note(db, g.user_id, spec, zf, pos + 1, budget):
                     pos += 1
@@ -417,17 +413,11 @@ async def create_note():
     if is_empty_note(body, item_texts):
         return json_error("note is empty", 400)
     async with session_scope() as db:
-        # New notes go to the top of the manual order.
-        max_pos = await db.scalar(
-            select(func.coalesce(func.max(Note.position), 0)).where(
-                Note.owner_id == g.user_id, Note.deleted_at.is_(None)
-            )
-        )
         # Items still arrive separately — a client holds a list, not a blob — but they
         # are folded into the body, which is where a checklist lives now (M304).
         for text in item_texts:
             body = append_item(body, text)
-        note = Note(owner_id=g.user_id, body="", display_title="", position=int(max_pos) + 1)
+        note = Note(owner_id=g.user_id, body="", display_title="", position=await top_position(db, g.user_id) + 1)
         db.add(note)
         # The FOLDED body: an item can carry a #tag too.
         changed = await write_body(db, note, body, snapshot="never")

src/inkwell/notes/helpers.py

@@ -13,7 +13,7 @@ import re
 import uuid
 
 from quart import g
-from sqlalchemy import select
+from sqlalchemy import func, select
 
 from ..acl import EDIT, visible_to_user
 from ..config import Config
@@ -88,6 +88,18 @@ def apply_filter(stmt, filter_name: str, archived):
     return stmt.where(Note.deleted_at.is_(None), archived.is_(False))
 
 
+async def top_position(db, owner_id) -> int:
+    """The highest manual position among `owner_id`'s live notes, 0 for none. A new
+    note goes one above it, to the top of the board."""
+    return int(
+        await db.scalar(
+            select(func.coalesce(func.max(Note.position), 0)).where(
+                Note.owner_id == owner_id, Note.deleted_at.is_(None)
+            )
+        )
+    )
+
+
 async def _get_owned(db, note_id: str) -> Note | None:
     """Fetch a note the current user OWNS, for every change only an owner may make:
     trash, delete, labels, reminders, attachments, previews, history and sharing

src/inkwell/settings.py

@@ -6,6 +6,7 @@ from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Any, Literal
 
+from .common import coerce_bool
 from .models.settings import Setting
 
 SettingType = Literal["string", "bool", "int"]
@@ -215,17 +216,9 @@ _BY_KEY: dict[str, SettingDef] = {d.key: d for d in REGISTRY}
 SECRET_KEY_SETTING = "secret_key"
 
 
-def _coerce_bool(raw: Any) -> bool:
-    if isinstance(raw, bool):
-        return raw
-    if isinstance(raw, str):
-        return raw.strip().lower() in ("1", "true", "yes", "on")
-    return bool(raw)
-
-
 def _coerce(defn: SettingDef, raw: Any) -> Any:
     if defn.type == "bool":
-        return _coerce_bool(raw)
+        return coerce_bool(raw)
     if defn.type == "int":
         try:
             return int(raw)
@@ -374,7 +367,7 @@ def validate_updates(updates: dict) -> tuple[dict, str | None]:
                 return {}, f"{defn.label} must be at most {defn.maximum}"
             clean[key] = n
         elif defn.type == "bool":
-            clean[key] = _coerce_bool(val)
+            clean[key] = coerce_bool(val)
         else:
             text = str(val if val is not None else "").strip()
             if defn.secret and not text:

src/inkwell/settings_api.py

@@ -9,6 +9,7 @@ from quart import Blueprint, current_app, g, jsonify, request
 from .auth import require_admin
 from .db import session_scope
 from .mailer import mail_settings, send
+from .responses import json_error
 from .models.user import User
 from .settings import get_admin_settings, get_setting, refresh_live, set_settings, validate_updates
 
@@ -32,11 +33,11 @@ async def update_settings():
     # Accept either {settings: {...}} or a bare {key: value} object.
     updates = data.get("settings") if isinstance(data.get("settings"), dict) else data
     if not isinstance(updates, dict):
-        return jsonify({"error": "expected an object of settings"}), 400
+        return json_error("expected an object of settings", 400)
 
     clean, error = validate_updates(updates)
     if error is not None:
-        return jsonify({"error": error}), 400
+        return json_error(error, 400)
 
     async with session_scope() as db:
         await set_settings(db, clean)
@@ -77,5 +78,5 @@ async def test_email():
         )
     except Exception as e:  # any SMTP, socket or TLS failure; the admin needs its text
         logger.warning("test email failed: %s", e)
-        return jsonify({"error": f"Couldn't send: {e}"}), 502
+        return json_error(f"Couldn't send: {e}", 502)
     return jsonify({"ok": True, "to": admin.email})

src/inkwell/sync.py

@@ -282,24 +282,25 @@ async def _apply_note_manual_labels(db, note: Note, ch: dict) -> None:
     raw = ch.get("label_ids")
     if not isinstance(raw, list):
         return
-    wanted: set = set()
-    for r in raw:
-        try:
-            wanted.add(uuid.UUID(str(r)))
-        except (ValueError, TypeError):
-            continue  # sync is lenient: skip a malformed id rather than reject the push
+    # Sync is lenient: a malformed id is skipped rather than rejecting the push.
+    wanted = {lid for lid in map(parse_uuid, raw) if lid is not None}
     owned = await resolve_owned_label_ids(db, wanted, g.user_id)
     await reconcile_manual_labels(db, note, owned)
 
 
+def _result(rid, entity: str, status: str, **extra) -> dict:
+    """One entry of a push reply: the change's id and entity, what became of it
+    (created, applied, kept, noop or rejected), plus `error` or `sync_revision`."""
+    return {"id": rid, "entity": entity, "status": status, **extra}
+
+
 async def _apply_note(db, ch: dict, previews: list[tuple[uuid.UUID, str]]) -> dict:
     """Apply one pushed note. A note whose text changed is appended to `previews` as
     (id, final body), for the caller to queue link previews once the batch commits."""
     raw_id = ch.get("id")
-    try:
-        nid = uuid.UUID(str(raw_id))
-    except (ValueError, TypeError):
-        return {"id": raw_id, "entity": "note", "status": "rejected", "error": "invalid id"}
+    nid = parse_uuid(raw_id)
+    if nid is None:
+        return _result(raw_id, "note", "rejected", error="invalid id")
     op = ch.get("op", "upsert")
     edited_at = parse_dt(ch.get("edited_at"))
 
@@ -309,13 +310,13 @@ async def _apply_note(db, ch: dict, previews: list[tuple[uuid.UUID, str]]) -> di
 
     if op == "delete":
         if note is None:
-            return {"id": str(nid), "entity": "note", "status": "noop"}
+            return _result(str(nid), "note", "noop")
         if not client_wins(edited_at, note.updated_at):
-            return {"id": str(nid), "entity": "note", "status": "kept", "sync_revision": note.sync_revision}
+            return _result(str(nid), "note", "kept", sync_revision=note.sync_revision)
         await purge_note(db, note, edited_at)
         await db.flush()
         await db.refresh(note, ["sync_revision"])
-        return {"id": str(nid), "entity": "note", "status": "applied", "sync_revision": note.sync_revision}
+        return _result(str(nid), "note", "applied", sync_revision=note.sync_revision)
 
     creating = note is None
     if creating:
@@ -325,7 +326,7 @@ async def _apply_note(db, ch: dict, previews: list[tuple[uuid.UUID, str]]) -> di
             note.created_at = created
         db.add(note)
     elif not client_wins(edited_at, note.updated_at):
-        return {"id": str(nid), "entity": "note", "status": "kept", "sync_revision": note.sync_revision}
+        return _result(str(nid), "note", "kept", sync_revision=note.sync_revision)
     elif note.purged_at is not None:
         note.purged_at = None  # client re-created/edited → clear the tombstone
 
@@ -348,12 +349,7 @@ async def _apply_note(db, ch: dict, previews: list[tuple[uuid.UUID, str]]) -> di
     await _apply_note_manual_labels(db, note, ch)
     await db.flush()
     await db.refresh(note, ["sync_revision"])
-    return {
-        "id": str(nid),
-        "entity": "note",
-        "status": "created" if creating else "applied",
-        "sync_revision": note.sync_revision,
-    }
+    return _result(str(nid), "note", "created" if creating else "applied", sync_revision=note.sync_revision)
 
 
 async def _apply_shared_note(db, note: Note, ch: dict, edited_at, previews: list[tuple[uuid.UUID, str]]) -> dict:
@@ -375,8 +371,8 @@ async def _apply_shared_note(db, note: Note, ch: dict, edited_at, previews: list
     """
     nid = str(note.id)
     if await _get_visible(db, nid) is None:
-        return {"id": nid, "entity": "note", "status": "rejected", "error": "cannot apply"}
-    owner_only = {"id": nid, "entity": "note", "status": "rejected", "error": "only its owner can change that"}
+        return _result(nid, "note", "rejected", error="cannot apply")
+    owner_only = _result(nid, "note", "rejected", error="only its owner can change that")
     if ch.get("op", "upsert") != "upsert":
         return owner_only
     applied = kept = False
@@ -409,35 +405,34 @@ async def _apply_shared_note(db, note: Note, ch: dict, edited_at, previews: list
 
 async def _apply_label(db, ch: dict) -> dict:
     raw_id = ch.get("id")
-    try:
-        lid = uuid.UUID(str(raw_id))
-    except (ValueError, TypeError):
-        return {"id": raw_id, "entity": "label", "status": "rejected", "error": "invalid id"}
+    lid = parse_uuid(raw_id)
+    if lid is None:
+        return _result(raw_id, "label", "rejected", error="invalid id")
     op = ch.get("op", "upsert")
     edited_at = parse_dt(ch.get("edited_at"))
 
     label = await db.scalar(select(Label).where(Label.id == lid))
     if label is not None and label.owner_id != g.user_id:
         # Generic rejection (see _apply_note): don't confirm a foreign-owned id exists.
-        return {"id": str(lid), "entity": "label", "status": "rejected", "error": "cannot apply"}
+        return _result(str(lid), "label", "rejected", error="cannot apply")
 
     if op == "delete":
         if label is None:
-            return {"id": str(lid), "entity": "label", "status": "noop"}
+            return _result(str(lid), "label", "noop")
         if not client_wins(edited_at, label.updated_at):
-            return {"id": str(lid), "entity": "label", "status": "kept", "sync_revision": label.sync_revision}
+            return _result(str(lid), "label", "kept", sync_revision=label.sync_revision)
         await db.execute(sa_delete(NoteLabel).where(NoteLabel.label_id == label.id))
         label.purged_at = datetime.now(timezone.utc)
         if edited_at is not None:
             label.updated_at = edited_at
         await db.flush()
         await db.refresh(label, ["sync_revision"])
-        return {"id": str(lid), "entity": "label", "status": "applied", "sync_revision": label.sync_revision}
+        return _result(str(lid), "label", "applied", sync_revision=label.sync_revision)
 
     name = (ch.get("name") or "").strip()
     creating = label is None
     if not creating and not client_wins(edited_at, label.updated_at):
-        return {"id": str(lid), "entity": "label", "status": "kept", "sync_revision": label.sync_revision}
+        return _result(str(lid), "label", "kept", sync_revision=label.sync_revision)
     # Names are unique per owner — a same-name clash on a DIFFERENT id can't be an insert.
     if name:
         clash = await db.scalar(
@@ -446,11 +441,11 @@ async def _apply_label(db, ch: dict) -> dict:
             )
         )
         if clash is not None:
-            return {"id": str(lid), "entity": "label", "status": "rejected", "error": "name in use"}
+            return _result(str(lid), "label", "rejected", error="name in use")
 
     if creating:
         if not name:
-            return {"id": str(lid), "entity": "label", "status": "rejected", "error": "name required"}
+            return _result(str(lid), "label", "rejected", error="name required")
         label = Label(id=lid, owner_id=g.user_id, name=name, color=normalize_color(ch.get("color")))
         db.add(label)
     else:
@@ -463,12 +458,7 @@ async def _apply_label(db, ch: dict) -> dict:
         label.updated_at = edited_at
     await db.flush()
     await db.refresh(label, ["sync_revision"])
-    return {
-        "id": str(lid),
-        "entity": "label",
-        "status": "created" if creating else "applied",
-        "sync_revision": label.sync_revision,
-    }
+    return _result(str(lid), "label", "created" if creating else "applied", sync_revision=label.sync_revision)
 
 
 # Entities a push may only DELETE. Each is a note's child, so deleting one bumps its
@@ -492,20 +482,20 @@ async def _apply_child_delete(db, ch: dict, removed_files: list[str]) -> dict:
     raw_id = ch.get("id")
     child_id = parse_uuid(str(raw_id)) if raw_id is not None else None
     if child_id is None:
-        return {"id": raw_id, "entity": entity, "status": "rejected", "error": "invalid id"}
+        return _result(raw_id, entity, "rejected", error="invalid id")
     if ch.get("op") != "delete":
-        return {"id": str(child_id), "entity": entity, "status": "rejected", "error": "only delete is supported"}
+        return _result(str(child_id), entity, "rejected", error="only delete is supported")
     model = _CHILD_ENTITIES[entity]
     row = await db.scalar(
         select(model).join(Note, Note.id == model.note_id).where(model.id == child_id, Note.owner_id == g.user_id)
     )
     if row is None:
-        return {"id": str(child_id), "entity": entity, "status": "noop"}
+        return _result(str(child_id), entity, "noop")
     if isinstance(row, NoteAttachment):
         removed_files.append(row.path)
     await db.delete(row)
     await db.flush()
-    return {"id": str(child_id), "entity": entity, "status": "applied"}
+    return _result(str(child_id), entity, "applied")
 
 
 @bp.post("/push")
@@ -516,9 +506,9 @@ async def push():
     body = await request.get_json(silent=True) or {}
     changes = body.get("changes")
     if not isinstance(changes, list):
-        return jsonify({"error": "changes must be a list"}), 400
+        return json_error("changes must be a list", 400)
     if len(changes) > MAX_PUSH:
-        return jsonify({"error": f"too many changes in one push (max {MAX_PUSH})"}), 400
+        return json_error(f"too many changes in one push (max {MAX_PUSH})", 400)
 
     results = []
     previews: list[tuple[uuid.UUID, str]] = []

tests/test_notes.py

@@ -375,6 +375,8 @@ def test_coerce_bool():
     assert not coerce_bool(None)
     assert not coerce_bool("")
     assert not coerce_bool(False)
+    # A bool setting can arrive as a JSON number.
+    assert coerce_bool(1) and not coerce_bool(0)
 
 
 def test_normalize_recurrence():