@@ -0,0 +1,65 @@
+"""Build pydantic User/Organization entities from proposals.
+
+Must run within an open OSW connection (the models resolve linked entities on
+construction). Organization references are passed as a list of page-title IRIs,
+which oold stores in the entity's ``__iris__`` map; the reference attribute is
+never read back here, as that would trigger a network resolve.
+"""
+
+from __future__ import annotations
+
+from typing import Any, List, Set
+
+from opensemantic.base.v1 import Organization, User
+
+from .mapping import ProposedOrganization, ProposedUser
+
+# The concrete DisplayName type used by the label field.
+_DISPLAY_NAME = User.__fields__["label"].type_
+
+
+def _labels(text: str) -> List[Any]:
+    return [_DISPLAY_NAME(text=text)]
+
+
+def build_organization(proposed: ProposedOrganization) -> Organization:
+    return Organization(uuid=proposed.uuid, label=_labels(proposed.name))
+
+
+def build_user(proposed: ProposedUser) -> User:
+    """Build a full User entity for creation."""
+    args: dict = {
+        "uuid": proposed.uuid,
+        "username": proposed.username,
+        "first_name": proposed.first_name,
+        "surname": proposed.surname,
+        "label": _labels(proposed.label),
+    }
+    if proposed.orcid:
+        args["orcid"] = proposed.orcid
+    if proposed.emails:
+        args["email"] = set(proposed.emails)
+    if proposed.websites:
+        args["website"] = set(proposed.websites)
+    if proposed.organizations:
+        args["organization"] = list(proposed.organizations)
+    return User(**args)
+
+
+def apply_update(entity: Any, proposed: ProposedUser, apply_fields: Set[str]) -> Any:
+    """Apply the accepted fields onto a loaded existing entity, in place."""
+    if "label" in apply_fields:
+        entity.label = _labels(proposed.label)
+    if "first_name" in apply_fields:
+        entity.first_name = proposed.first_name
+    if "surname" in apply_fields:
+        entity.surname = proposed.surname
+    if "orcid" in apply_fields:
+        entity.orcid = proposed.orcid
+    if "emails" in apply_fields:
+        entity.email = set(proposed.emails)
+    if "websites" in apply_fields:
+        entity.website = set(proposed.websites)
+    if "organizations" in apply_fields:
+        entity.__iris__["organization"] = list(proposed.organizations)
+    return entity
@@ -0,0 +1,35 @@
+"""Create ``User:<username>`` redirect pages pointing at the user item.
+
+An existing user page with real content is never overwritten; only missing
+pages or pages that are already redirects are (re)written.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from osw.wtsite import WtPage
+
+
+def redirect_wikitext(item_title: str) -> str:
+    return f"#REDIRECT [[{item_title}]]"
+
+
+def is_redirect(content: str) -> bool:
+    return content.lstrip().upper().startswith("#REDIRECT")
+
+
+def ensure_redirect(osw: Any, username: str, item_title: str) -> Optional[str]:
+    """Create or refresh ``User:<username>`` -> item redirect.
+
+    Returns the page title if written, or None if skipped (real page exists).
+    """
+    title = f"User:{username}"
+    page = WtPage(wtSite=osw.site, title=title)
+    if getattr(page, "exists", False):
+        content = page.get_content() or ""
+        if content.strip() and not is_redirect(content):
+            return None
+    page.set_content(redirect_wikitext(item_title))
+    page.edit(comment="user-sync: redirect to user item")
+    return title
@@ -1,18 +1,30 @@
 """Orchestrator for the user-item sync tool.
 
-Later phases implement enumeration, ORCID enrichment, reconciliation, the
-interactive layer and the store step. This module currently defines the report
-type and the public entry point.
+Ties the sources, mapping, reconciliation, interactive and store steps into one
+run: enumerate MediaWiki users, enrich ORCID users, reconcile against existing
+items, preview and confirm, then store items, redirects and organizations and
+verify.
 """
 
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 from osw.core import OSW
 
+from .build import apply_update, build_organization, build_user
 from .config import SyncConfig
+from .existing import load_existing_users
+from .interactive import Prompter, Resolution, resolve_plan
+from .mapping import ProposedOrganization, map_user
+from .reconcile import reconcile
+from .redirects import ensure_redirect
+from .sources import (
+    OrcidRateLimitError,
+    enumerate_mw_users,
+    fetch_orcid_record,
+)
 
 
 @dataclass
@@ -25,22 +37,154 @@ class SyncReport:
     failed: Dict[str, str] = field(default_factory=dict)
     redirects_created: List[str] = field(default_factory=list)
     organizations: List[str] = field(default_factory=list)
+    accepted: bool = False
 
     def summary(self) -> str:
-        """One-line human-readable summary of the run."""
+        state = "applied" if self.accepted else "no changes written"
         return (
-            f"created={len(self.created)} updated={len(self.updated)} "
+            f"{state}: created={len(self.created)} updated={len(self.updated)} "
             f"skipped={len(self.skipped)} failed={len(self.failed)} "
             f"redirects={len(self.redirects_created)} orgs={len(self.organizations)}"
         )
 
 
-def run_user_sync(config: SyncConfig, osw: Optional[OSW] = None) -> SyncReport:
-    """Create or update OSW User items from MediaWiki accounts and ORCID data.
+def _build_proposals(config: SyncConfig, osw: Any, session: Any):
+    """Enumerate MediaWiki users, enrich ORCID users and map to proposals."""
+    excluded = tuple(config.excluded_groups) if config.exclude_bot_group else ()
+    mw_users = enumerate_mw_users(
+        osw.mw_site,
+        excluded_groups=excluded,
+        include_non_orcid=config.include_non_orcid,
+        limit=config.limit,
+    )
+    cache: Dict[str, Any] = {}
+    proposed_users = []
+    org_map: Dict[Any, ProposedOrganization] = {}
+    for mw_user in mw_users:
+        profile = None
+        if mw_user.is_orcid:
+            try:
+                profile = fetch_orcid_record(
+                    mw_user.orcid_id,
+                    session=session,
+                    base=config.orcid_api_base,
+                    cache=cache,
+                )
+            except OrcidRateLimitError:
+                profile = None
+        proposed, orgs = map_user(
+            mw_user, profile, link_organizations=config.link_organizations
+        )
+        proposed_users.append(proposed)
+        for org in orgs:
+            org_map.setdefault(org.uuid, org)
+    return proposed_users, org_map
+
+
+def _store_organizations(osw: Any, org_map, report: SyncReport) -> None:
+    if not org_map:
+        return
+    entities = [build_organization(org) for org in org_map.values()]
+    try:
+        osw.store_entity(
+            OSW.StoreEntityParam(entities=entities, edit_comment="user-sync: orgs")
+        )
+        report.organizations = [org.full_page_title for org in org_map.values()]
+    except Exception as exc:  # pragma: no cover - network failure path
+        report.failed["organizations"] = str(exc)
+
+
+def _store_users(osw: Any, resolution: Resolution, report: SyncReport) -> None:
+    entities = []
+    for resolved in resolution.resolved:
+        proposed = resolved.change.proposed
+        if resolved.action == "create":
+            entities.append(build_user(proposed))
+            report.created.append(proposed.full_page_title)
+        elif resolved.action == "update":
+            entities.append(
+                apply_update(resolved.change.existing, proposed, resolved.apply_fields)
+            )
+            report.updated.append(proposed.full_page_title)
+        else:
+            report.skipped.append(proposed.full_page_title)
+    if entities:
+        osw.store_entity(
+            OSW.StoreEntityParam(
+                entities=entities, overwrite=True, edit_comment="user-sync"
+            )
+        )
+
+
+def _create_redirects(osw: Any, resolution: Resolution, report: SyncReport) -> None:
+    for resolved in resolution.resolved:
+        if resolved.action not in ("create", "update"):
+            continue
+        proposed = resolved.change.proposed
+        try:
+            written = ensure_redirect(osw, proposed.username, proposed.full_page_title)
+            if written:
+                report.redirects_created.append(written)
+        except Exception as exc:  # pragma: no cover - network failure path
+            report.failed[f"redirect:{proposed.username}"] = str(exc)
+
+
+def _verify(osw: Any, report: SyncReport) -> None:
+    titles = report.created + report.updated
+    if not titles:
+        return
+    try:
+        loaded = osw.load_entity(titles)
+        loaded_list = loaded if isinstance(loaded, list) else [loaded]
+        found = {getattr(e, "username", None) for e in loaded_list}
+        missing = [
+            t
+            for t, e in zip(titles, loaded_list)
+            if getattr(e, "username", None) is None
+        ]
+        if missing or None in found:
+            report.failed["verify"] = f"missing username on {missing}"
+    except Exception as exc:  # pragma: no cover - network failure path
+        report.failed["verify"] = str(exc)
+
+
+def run_user_sync(
+    config: SyncConfig,
+    osw: Optional[OSW] = None,
+    prompter: Optional[Prompter] = None,
+    session: Any = None,
+) -> SyncReport:
+    """Create or update OSW User items from MediaWiki accounts and ORCID data."""
+    if osw is None:
+        raise ValueError("run_user_sync requires an authenticated OSW connection")
+    prompter = prompter or Prompter()
+
+    proposed_users, org_map = _build_proposals(config, osw, session)
+    existing = load_existing_users(osw)
+    plan = reconcile(proposed_users, existing)
+
+    summary_lines = []
+    if config.link_organizations and org_map:
+        summary_lines.append(f"Organizations to ensure: {len(org_map)}")
+    if config.create_redirects:
+        summary_lines.append("Redirects: User:<username> -> item for created/updated")
+
+    resolution = resolve_plan(
+        plan,
+        prompter,
+        dry_run=config.dry_run,
+        assume_yes=config.assume_yes,
+        summary_lines=summary_lines,
+    )
+
+    report = SyncReport(accepted=resolution.accepted)
+    if not resolution.accepted:
+        return report
 
-    Args:
-        config: Runtime options for the run.
-        osw: An authenticated OSW/OswExpress connection. Required for any wiki
-            access; the example wrapper builds one from ``config``.
-    """
-    raise NotImplementedError("Implemented incrementally in phases 1 to 6.")
+    if config.link_organizations:
+        _store_organizations(osw, org_map, report)
+    _store_users(osw, resolution, report)
+    if config.create_redirects:
+        _create_redirects(osw, resolution, report)
+    _verify(osw, report)
+    return report
@@ -41,8 +41,8 @@ def test_report_summary():
     assert "updated=1" in report.summary()
 
 
-def test_run_user_sync_is_stub():
+def test_run_user_sync_requires_connection():
     import pytest
 
-    with pytest.raises(NotImplementedError):
+    with pytest.raises(ValueError, match="connection"):
         run_user_sync(SyncConfig())
@@ -0,0 +1,69 @@
+"""Unit tests for redirect page handling."""
+
+from osw.tools.user_sync import redirects
+from osw.tools.user_sync.redirects import (
+    ensure_redirect,
+    is_redirect,
+    redirect_wikitext,
+)
+
+
+def test_redirect_wikitext():
+    assert redirect_wikitext("Item:OSW123") == "#REDIRECT [[Item:OSW123]]"
+
+
+def test_is_redirect():
+    assert is_redirect("#REDIRECT [[Item:OSW1]]")
+    assert is_redirect("  #redirect [[x]]")
+    assert not is_redirect("Some real user page content")
+
+
+class FakeWtPage:
+    exists_map: dict = {}
+    content_map: dict = {}
+
+    def __init__(self, wtSite=None, title=None, do_init=True):
+        self.title = title
+        self.exists = self.exists_map.get(title, False)
+        self._content = self.content_map.get(title, "")
+        self.written = None
+        self.edited = False
+
+    def get_content(self):
+        return self._content
+
+    def set_content(self, content):
+        self.written = content
+
+    def edit(self, comment=None):
+        self.edited = True
+
+
+class FakeOsw:
+    site = object()
+
+
+def _patch(monkeypatch, exists_map, content_map):
+    FakeWtPage.exists_map = exists_map
+    FakeWtPage.content_map = content_map
+    monkeypatch.setattr(redirects, "WtPage", FakeWtPage)
+
+
+def test_ensure_redirect_creates_missing(monkeypatch):
+    _patch(monkeypatch, {}, {})
+    written = ensure_redirect(FakeOsw(), "0000-0002-6374-9831", "Item:OSW1")
+    assert written == "User:0000-0002-6374-9831"
+
+
+def test_ensure_redirect_overwrites_existing_redirect(monkeypatch):
+    title = "User:alice"
+    _patch(monkeypatch, {title: True}, {title: "#REDIRECT [[Item:OSWold]]"})
+    written = ensure_redirect(FakeOsw(), "alice", "Item:OSWnew")
+    assert written == title
+
+
+def test_ensure_redirect_skips_real_page(monkeypatch):
+    title = "User:bob"
+    _patch(monkeypatch, {title: True}, {title: "This is my real user page."})
+    written = ensure_redirect(FakeOsw(), "bob", "Item:OSW1")
+    assert written is None