@@ -0,0 +1,161 @@
+"""Reconcile proposed User items against the ones already stored.
+
+Pure logic: given proposed users and an index of existing items, classify each
+into NEW, GAP_FILL, CONFLICT or UNCHANGED and record the per-field differences.
+The interactive layer consumes this plan; nothing here does IO.
+"""
+
+from __future__ import annotations
+
+from collections import Counter
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+from .existing import ExistingUsers
+from .mapping import ProposedUser
+
+NEW = "new"
+GAP_FILL = "gap_fill"
+CONFLICT = "conflict"
+UNCHANGED = "unchanged"
+
+# username is the match key and is never reconciled as a value.
+SCALAR_FIELDS = ("label", "first_name", "surname", "orcid")
+SET_FIELDS = ("emails", "websites", "organizations")
+URL_FIELDS = ("orcid", "websites")
+RECONCILED_FIELDS = SCALAR_FIELDS + SET_FIELDS
+
+
+def _is_empty(value: Any) -> bool:
+    if value is None or value == "":
+        return True
+    return isinstance(value, (list, set, tuple)) and len(value) == 0
+
+
+def _norm_url(value: Any) -> Optional[str]:
+    return None if value is None else str(value).rstrip("/")
+
+
+def _norm_scalar(name: str, value: Any) -> Optional[str]:
+    if value is None:
+        return None
+    return _norm_url(value) if name in URL_FIELDS else str(value)
+
+
+def _norm_set(name: str, value: Any) -> set:
+    if not value:
+        return set()
+    if name in URL_FIELDS:
+        return {_norm_url(x) for x in value}
+    return {str(x) for x in value}
+
+
+def _existing_label(entity: Any) -> Optional[str]:
+    for label in getattr(entity, "label", None) or []:
+        text = getattr(label, "text", None)
+        if text:
+            return text
+    return None
+
+
+def existing_fields(entity: Any) -> Dict[str, Any]:
+    """Normalized comparable field view of a loaded User entity."""
+    return {
+        "label": _existing_label(entity),
+        "first_name": getattr(entity, "first_name", None),
+        "surname": getattr(entity, "surname", None),
+        "orcid": _norm_scalar("orcid", getattr(entity, "orcid", None)),
+        "emails": _norm_set("emails", getattr(entity, "email", None)),
+        "websites": _norm_set("websites", getattr(entity, "website", None)),
+        "organizations": _norm_set(
+            "organizations", getattr(entity, "organization", None)
+        ),
+    }
+
+
+def proposed_fields(proposed: ProposedUser) -> Dict[str, Any]:
+    """Normalized comparable field view of a proposed user."""
+    return {
+        "label": proposed.label,
+        "first_name": proposed.first_name,
+        "surname": proposed.surname,
+        "orcid": _norm_scalar("orcid", proposed.orcid),
+        "emails": _norm_set("emails", proposed.emails),
+        "websites": _norm_set("websites", proposed.websites),
+        "organizations": _norm_set("organizations", proposed.organizations),
+    }
+
+
+@dataclass
+class FieldDiff:
+    """A single field that would change on an existing item."""
+
+    name: str
+    existing: Any
+    proposed: Any
+    status: str  # GAP_FILL or CONFLICT
+
+
+@dataclass
+class UserChange:
+    """The reconciliation outcome for one proposed user."""
+
+    proposed: ProposedUser
+    existing: Optional[Any]
+    category: str
+    diffs: List[FieldDiff] = field(default_factory=list)
+
+
+def _compare(existing_val: Any, proposed_val: Any) -> Optional[str]:
+    """Per-field status, or None when there is nothing to propose."""
+    if _is_empty(proposed_val):
+        return None
+    if _is_empty(existing_val):
+        return GAP_FILL
+    return UNCHANGED if existing_val == proposed_val else CONFLICT
+
+
+def reconcile_user(proposed: ProposedUser, existing: Optional[Any]) -> UserChange:
+    if existing is None:
+        return UserChange(proposed=proposed, existing=None, category=NEW)
+    ef = existing_fields(existing)
+    pf = proposed_fields(proposed)
+    diffs: List[FieldDiff] = []
+    has_conflict = has_gap = False
+    for name in RECONCILED_FIELDS:
+        status = _compare(ef[name], pf[name])
+        if status in (GAP_FILL, CONFLICT):
+            diffs.append(FieldDiff(name, ef[name], pf[name], status))
+            has_conflict = has_conflict or status == CONFLICT
+            has_gap = has_gap or status == GAP_FILL
+    category = CONFLICT if has_conflict else GAP_FILL if has_gap else UNCHANGED
+    return UserChange(
+        proposed=proposed, existing=existing, category=category, diffs=diffs
+    )
+
+
+@dataclass
+class ReconcilePlan:
+    """All per-user reconciliation outcomes for a run."""
+
+    changes: List[UserChange] = field(default_factory=list)
+
+    def by_category(self, category: str) -> List[UserChange]:
+        return [c for c in self.changes if c.category == category]
+
+    @property
+    def conflicts(self) -> List[UserChange]:
+        return self.by_category(CONFLICT)
+
+    def counts(self) -> Dict[str, int]:
+        counter = Counter(c.category for c in self.changes)
+        return {
+            cat: counter.get(cat, 0) for cat in (NEW, GAP_FILL, CONFLICT, UNCHANGED)
+        }
+
+
+def reconcile(
+    proposed_users: List[ProposedUser], existing: ExistingUsers
+) -> ReconcilePlan:
+    changes = [reconcile_user(p, existing.get(p.username)) for p in proposed_users]
+    return ReconcilePlan(changes=changes)
@@ -0,0 +1,137 @@
+"""Unit tests for the reconciliation engine."""
+
+import uuid
+from types import SimpleNamespace
+
+from osw.tools.user_sync.existing import ExistingUsers
+from osw.tools.user_sync.mapping import ProposedUser
+from osw.tools.user_sync.reconcile import (
+    CONFLICT,
+    GAP_FILL,
+    NEW,
+    UNCHANGED,
+    reconcile,
+    reconcile_user,
+)
+
+ORCID = "0000-0002-6374-9831"
+
+
+def _proposed(**kw):
+    base = dict(
+        uuid=uuid.uuid4(),
+        username=ORCID,
+        first_name="Lukas",
+        surname="Koschmieder",
+        label="Lukas Koschmieder",
+        orcid=f"https://orcid.org/{ORCID}",
+        emails=[],
+        websites=[],
+        organizations=[],
+    )
+    base.update(kw)
+    return ProposedUser(**base)
+
+
+def _entity(**kw):
+    base = dict(
+        label="Lukas Koschmieder",
+        first_name="Lukas",
+        surname="Koschmieder",
+        orcid=f"https://orcid.org/{ORCID}",
+        email=None,
+        website=None,
+        organization=None,
+    )
+    base.update(kw)
+    return SimpleNamespace(
+        label=[SimpleNamespace(text=base["label"])] if base["label"] else [],
+        first_name=base["first_name"],
+        surname=base["surname"],
+        orcid=base["orcid"],
+        email=set(base["email"] or []),
+        website=set(base["website"] or []),
+        organization=set(base["organization"] or []),
+    )
+
+
+def test_new_when_no_existing():
+    change = reconcile_user(_proposed(), None)
+    assert change.category == NEW
+    assert change.diffs == []
+
+
+def test_unchanged_when_equal():
+    change = reconcile_user(_proposed(), _entity())
+    assert change.category == UNCHANGED
+    assert change.diffs == []
+
+
+def test_gap_fill_for_missing_field():
+    proposed = _proposed(emails=["l@example.org"])
+    change = reconcile_user(proposed, _entity(email=None))
+    assert change.category == GAP_FILL
+    assert [d.name for d in change.diffs] == ["emails"]
+    assert change.diffs[0].status == GAP_FILL
+
+
+def test_conflict_on_differing_value():
+    change = reconcile_user(_proposed(surname="Koschmieder"), _entity(surname="Kosch"))
+    assert change.category == CONFLICT
+    assert change.diffs[0].name == "surname"
+    assert change.diffs[0].existing == "Kosch"
+    assert change.diffs[0].proposed == "Koschmieder"
+
+
+def test_conflict_dominates_gap_fill():
+    proposed = _proposed(surname="Koschmieder", emails=["l@example.org"])
+    change = reconcile_user(proposed, _entity(surname="Kosch", email=None))
+    assert change.category == CONFLICT
+    names = {d.name for d in change.diffs}
+    assert names == {"surname", "emails"}
+
+
+def test_empty_proposal_does_not_clear_existing():
+    change = reconcile_user(_proposed(emails=[]), _entity(email=["keep@example.org"]))
+    assert change.category == UNCHANGED
+    assert change.diffs == []
+
+
+def test_orcid_trailing_slash_not_a_conflict():
+    change = reconcile_user(
+        _proposed(orcid=f"https://orcid.org/{ORCID}"),
+        _entity(orcid=f"https://orcid.org/{ORCID}/"),
+    )
+    assert change.category == UNCHANGED
+
+
+def test_set_order_independent():
+    change = reconcile_user(
+        _proposed(organizations=["Item:OSWb", "Item:OSWa"]),
+        _entity(organization=["Item:OSWa", "Item:OSWb"]),
+    )
+    assert change.category == UNCHANGED
+
+
+def test_missing_label_is_gap_fill():
+    change = reconcile_user(_proposed(label="Lukas Koschmieder"), _entity(label=None))
+    assert change.category == GAP_FILL
+    assert change.diffs[0].name == "label"
+
+
+def test_reconcile_plan_counts():
+    proposed = [
+        _proposed(username="new-user"),
+        _proposed(username=ORCID),
+        _proposed(username="conflict-user", surname="New"),
+    ]
+    existing = ExistingUsers(
+        by_username={
+            ORCID: _entity(),
+            "conflict-user": _entity(surname="Old"),
+        }
+    )
+    plan = reconcile(proposed, existing)
+    counts = plan.counts()
+    assert counts == {NEW: 1, GAP_FILL: 0, CONFLICT: 1, UNCHANGED: 1}
+    assert [c.proposed.username for c in plan.conflicts] == ["conflict-user"]