diff --git a/alembic/versions/0017_label_updated_at.py b/alembic/versions/0017_label_updated_at.py
new file mode 100644
index 0000000..75d241b
--- /dev/null
+++ b/alembic/versions/0017_label_updated_at.py
@@ -0,0 +1,29 @@
+"""labels.updated_at (M8 sync hub, step 4 — LWW field for the label catalog)
+
+Revision ID: 0017
+Revises: 0016
+Create Date: 2026-07-23
+
+Labels need a last-edit timestamp so sync push can resolve label rename/recolor
+conflicts by last-write-wins (client edit-time vs this). Backfilled from created_at.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0017"
+down_revision = "0016"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "labels",
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+    )
+    # Seed existing rows' updated_at from their created_at (best available edit time).
+    op.execute("UPDATE labels SET updated_at = created_at")
+
+
+def downgrade() -> None:
+    op.drop_column("labels", "updated_at")
diff --git a/src/thoughtsync/models/label.py b/src/thoughtsync/models/label.py
index de60a3b..f57d408 100644
--- a/src/thoughtsync/models/label.py
+++ b/src/thoughtsync/models/label.py
@@ -21,6 +21,11 @@ class Label(Base):
     name: Mapped[str] = mapped_column(Text(), nullable=False)
     color: Mapped[str] = mapped_column(Text(), nullable=False, default="default", server_default="default")
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
+    # Last-edit time — the LWW comparison field for sync (a rename/recolor bumps it).
+    # Sync-push overrides it with the winning client edit-time; web edits set server-now.
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )
     # Sync (M8): monotonic per-row revision (from sync_revision_seq via DB trigger) so a
     # label rename/recolor/merge/delete propagates to native clients independently of notes.
     sync_revision: Mapped[int | None] = mapped_column(BigInteger(), nullable=True)
diff --git a/src/thoughtsync/sync.py b/src/thoughtsync/sync.py
index cbdb0d5..eede8e2 100644
--- a/src/thoughtsync/sync.py
+++ b/src/thoughtsync/sync.py
@@ -11,19 +11,36 @@ from where it left off (since=0 = full initial sync).
 """
 from __future__ import annotations
 
+import uuid
+from datetime import datetime, timezone
+
 from quart import Blueprint, g, jsonify, request
-from sqlalchemy import select
+from sqlalchemy import delete as sa_delete
+from sqlalchemy import func, select
 
 from .auth import login_required
+from .config import Config
 from .db import session_scope
-from .models.label import Label
+from .models.label import Label, NoteLabel
 from .models.note import Note
-from .notes import _serialize_notes
+from .models.note_attachment import NoteAttachment
+from .models.note_item import NoteItem
+from .models.note_link import NoteLink
+from .models.note_revision import NoteRevision
+from .notes import (
+    _reconcile_tags,
+    _rename_inbound_links,
+    _rewrite_links,
+    _serialize_notes,
+    derive_display_title,
+    normalize_color,
+)
 
 bp = Blueprint("sync", __name__, url_prefix="/api/sync")
 
 DEFAULT_LIMIT = 500
 MAX_LIMIT = 1000
+MAX_PUSH = 1000  # per-batch change cap
 
 
 def _parse_since(raw: str | None) -> int:
@@ -120,3 +137,266 @@ async def changes():
                 "has_more": has_more,
             }
         )
+
+
+# --- Push: apply client mutations (LWW by client edit-time, non-destructive) ---
+
+
+def _parse_client_dt(raw: object) -> datetime | None:
+    if not isinstance(raw, str) or not raw:
+        return None
+    try:
+        return datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def client_wins(client_edited_at: datetime | None, server_edited_at: datetime | None) -> bool:
+    """Last-write-wins: the client's version is applied iff its edit-time is at least
+    the server's. A missing client time never overwrites a real server edit; a missing
+    server time (new/unknown row) always yields to a present client edit."""
+    if client_edited_at is None:
+        return server_edited_at is None
+    if server_edited_at is None:
+        return True
+    return client_edited_at >= server_edited_at
+
+
+def _assign_note_fields(note: Note, ch: dict) -> None:
+    """Overwrite a note's scalar fields from a client's FULL-state change (sync is
+    whole-note, not a partial patch — the client sends its authoritative version)."""
+    title = ch.get("title")
+    note.title = (title or "").strip() or None if isinstance(title, str) else None
+    note.body = ch["body"] if isinstance(ch.get("body"), str) else ""
+    note.color = normalize_color(ch.get("color"))
+    note.kind = ch["kind"] if ch.get("kind") in ("text", "list") else "text"
+    note.pinned = bool(ch.get("pinned"))
+    note.archived = bool(ch.get("archived"))
+    if ch.get("trashed"):
+        if note.deleted_at is None:
+            note.deleted_at = datetime.now(timezone.utc)
+    else:
+        note.deleted_at = None
+    note.remind_at = _parse_client_dt(ch.get("remind_at"))
+    if isinstance(ch.get("position"), int):
+        note.position = ch["position"]
+
+
+async def _apply_note_items(db, note: Note, ch: dict) -> None:
+    """Replace the note's checklist items with the client's (items sync inline)."""
+    if note.kind != "list":
+        await db.execute(sa_delete(NoteItem).where(NoteItem.note_id == note.id))
+        return
+    items = ch.get("items")
+    if not isinstance(items, list):
+        return
+    await db.execute(sa_delete(NoteItem).where(NoteItem.note_id == note.id))
+    for pos, it in enumerate(items):
+        if not isinstance(it, dict):
+            continue
+        text = (it.get("text") or "").strip()
+        if text:
+            db.add(NoteItem(note_id=note.id, text=text, checked=bool(it.get("checked")), position=pos))
+
+
+async def _apply_note_manual_labels(db, note: Note, ch: dict) -> None:
+    """Set the note's MANUAL (picker) label memberships from client label_ids, leaving
+    tag-sourced (via_tag) rows to _reconcile_tags. Only labels the caller owns count."""
+    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
+    owned: set = set()
+    if wanted:
+        owned = set(
+            (await db.scalars(select(Label.id).where(Label.owner_id == g.user_id, Label.id.in_(wanted)))).all()
+        )
+    existing = (await db.scalars(select(NoteLabel).where(NoteLabel.note_id == note.id))).all()
+    attached = {r.label_id for r in existing}
+    for r in existing:
+        if not r.via_tag and r.label_id not in owned:
+            await db.delete(r)
+    for lid in owned:
+        if lid not in attached:
+            db.add(NoteLabel(note_id=note.id, label_id=lid, via_tag=False))
+
+
+async def _purge_note(db, note: Note, edited_at: datetime | None) -> None:
+    """Turn a note into a content-less tombstone: delete children (+ attachment files),
+    clear content, set purged_at. Kept so offline clients learn it's gone."""
+    atts = (await db.scalars(select(NoteAttachment).where(NoteAttachment.note_id == note.id))).all()
+    for a in atts:
+        try:
+            (Config.media_root() / a.path).unlink(missing_ok=True)
+        except OSError:
+            pass
+    await db.execute(sa_delete(NoteAttachment).where(NoteAttachment.note_id == note.id))
+    await db.execute(sa_delete(NoteItem).where(NoteItem.note_id == note.id))
+    await db.execute(sa_delete(NoteLabel).where(NoteLabel.note_id == note.id))
+    await db.execute(sa_delete(NoteLink).where(NoteLink.source_id == note.id))
+    note.title = None
+    note.body = ""
+    note.display_title = ""
+    note.deleted_at = None
+    note.remind_at = None
+    note.purged_at = datetime.now(timezone.utc)
+    if edited_at is not None:
+        note.updated_at = edited_at
+
+
+async def _apply_note(db, ch: dict) -> dict:
+    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"}
+    op = ch.get("op", "upsert")
+    edited_at = _parse_client_dt(ch.get("edited_at"))
+
+    note = await db.scalar(select(Note).where(Note.id == nid))
+    if note is not None and note.owner_id != g.user_id:
+        return {"id": str(nid), "entity": "note", "status": "rejected", "error": "not yours"}
+
+    if op == "delete":
+        if note is None:
+            return {"id": str(nid), "entity": "note", "status": "noop"}
+        if not client_wins(edited_at, note.updated_at):
+            return {"id": str(nid), "entity": "note", "status": "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}
+
+    creating = note is None
+    if creating:
+        note = Note(id=nid, owner_id=g.user_id, body="", display_title="")
+        created = _parse_client_dt(ch.get("created_at"))
+        if created is not None:
+            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}
+    elif note.purged_at is not None:
+        note.purged_at = None  # client re-created/edited → clear the tombstone
+
+    old_title, old_body, old_display = note.title, note.body, note.display_title
+    _assign_note_fields(note, ch)
+    note.display_title = derive_display_title(note.title, note.body)
+    if edited_at is not None:
+        note.updated_at = edited_at
+    # Non-destructive LWW: snapshot the overwritten server title+body into history.
+    if not creating and (note.title != old_title or note.body != old_body):
+        db.add(NoteRevision(note_id=note.id, title=old_title, body=old_body))
+    await db.flush()  # assign note.id before items/labels/links
+    await _apply_note_items(db, note, ch)
+    await _rewrite_links(db, note)
+    await _reconcile_tags(db, note)
+    await _apply_note_manual_labels(db, note, ch)
+    new_display = note.display_title
+    if old_display and new_display and old_display.strip().lower() != new_display.strip().lower():
+        await _rename_inbound_links(db, note, old_display, new_display)
+    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,
+    }
+
+
+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"}
+    op = ch.get("op", "upsert")
+    edited_at = _parse_client_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:
+        return {"id": str(lid), "entity": "label", "status": "rejected", "error": "not yours"}
+
+    if op == "delete":
+        if label is None:
+            return {"id": str(lid), "entity": "label", "status": "noop"}
+        if not client_wins(edited_at, label.updated_at):
+            return {"id": str(lid), "entity": "label", "status": "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}
+
+    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}
+    # Names are unique per owner — a same-name clash on a DIFFERENT id can't be an insert.
+    if name:
+        clash = await db.scalar(
+            select(Label.id).where(
+                Label.owner_id == g.user_id, func.lower(Label.name) == name.lower(), Label.id != lid
+            )
+        )
+        if clash is not None:
+            return {"id": str(lid), "entity": "label", "status": "rejected", "error": "name in use"}
+
+    if creating:
+        if not name:
+            return {"id": str(lid), "entity": "label", "status": "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:
+        if label.purged_at is not None:
+            label.purged_at = None
+        if name:
+            label.name = name
+        label.color = normalize_color(ch.get("color"))
+    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": "created" if creating else "applied",
+        "sync_revision": label.sync_revision,
+    }
+
+
+@bp.post("/push")
+@login_required
+async def push():
+    """Apply a batch of client changes. Additive + owner-scoped; LWW by client
+    edit-time with a version-history snapshot on any overwrite (nothing is lost)."""
+    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
+    if len(changes) > MAX_PUSH:
+        return jsonify({"error": f"too many changes in one push (max {MAX_PUSH})"}), 400
+
+    results = []
+    async with session_scope() as db:
+        for ch in changes:
+            if not isinstance(ch, dict):
+                results.append({"status": "rejected", "error": "not an object"})
+                continue
+            entity = ch.get("entity")
+            if entity == "note":
+                results.append(await _apply_note(db, ch))
+            elif entity == "label":
+                results.append(await _apply_label(db, ch))
+            else:
+                results.append({"id": ch.get("id"), "status": "rejected", "error": "unknown entity"})
+        await db.commit()
+    return jsonify({"results": results})
diff --git a/tests/test_sync.py b/tests/test_sync.py
index 1e7b315..abf8153 100644
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -1,7 +1,17 @@
+from datetime import datetime, timezone
+
 import pytest
 
 from thoughtsync.app import create_app
-from thoughtsync.sync import DEFAULT_LIMIT, MAX_LIMIT, _clamp_limit, _page_cursor, _parse_since
+from thoughtsync.sync import (
+    DEFAULT_LIMIT,
+    MAX_LIMIT,
+    _clamp_limit,
+    _page_cursor,
+    _parse_client_dt,
+    _parse_since,
+    client_wins,
+)
 
 
 @pytest.fixture
@@ -15,6 +25,31 @@ async def test_changes_requires_auth(app):
     assert resp.status_code == 401
 
 
+async def test_push_requires_auth(app):
+    client = app.test_client()
+    resp = await client.post("/api/sync/push", json={"changes": []})
+    assert resp.status_code == 401
+
+
+def test_client_wins():
+    older = datetime(2026, 7, 20, tzinfo=timezone.utc)
+    newer = datetime(2026, 7, 22, tzinfo=timezone.utc)
+    assert client_wins(newer, older) is True  # newer client edit wins
+    assert client_wins(older, newer) is False  # older client edit loses (server kept)
+    assert client_wins(older, older) is True  # tie → client applies (idempotent)
+    assert client_wins(None, older) is False  # unknown client time can't overwrite a real edit
+    assert client_wins(older, None) is True  # new/unknown server side yields
+    assert client_wins(None, None) is True
+
+
+def test_parse_client_dt():
+    assert _parse_client_dt("2026-07-22T00:00:00Z").year == 2026
+    assert _parse_client_dt("2026-07-22T00:00:00+00:00").tzinfo is not None
+    assert _parse_client_dt("garbage") is None
+    assert _parse_client_dt(None) is None
+    assert _parse_client_dt(123) is None
+
+
 def test_parse_since():
     assert _parse_since(None) == 0
     assert _parse_since("42") == 42