diff --git a/alembic_osm/versions/9221408912dd_add_user_role_table.py b/alembic_osm/versions/9221408912dd_add_user_role_table.py
index 7d7fa57..04beaf7 100644
--- a/alembic_osm/versions/9221408912dd_add_user_role_table.py
+++ b/alembic_osm/versions/9221408912dd_add_user_role_table.py
@@ -36,7 +36,9 @@ def upgrade() -> None:
             "users",
             # `id` matches the Rails `users.id` numeric PK so the FK
             # from `team_user.user_id` in the next migration can attach.
-            sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+            sa.Column(
+                "id", sa.BigInteger(), autoincrement=True, nullable=False
+            ),
             sa.Column("auth_uid", sa.String(), nullable=False),
             sa.Column("email", sa.String(), nullable=True),
             sa.Column("display_name", sa.String(), nullable=True),
@@ -84,25 +86,6 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
-    bind = op.get_bind()
-    assert bind is not None
-    insp = inspect(bind)
-
-    if insp.has_table("user_workspace_roles"):
-        op.drop_table("user_workspace_roles")
-
-    # Drop the enum type
-    result = bind.execute(
-        text("SELECT 1 FROM pg_type WHERE typname = 'workspace_role'")
-    )
-    if result.scalar():
-        workspace_role = sa.Enum(
-            "lead", "validator", "contributor", name="workspace_role"
-        )
-        workspace_role.drop(bind)
-
-    constraint_exists = bind.execute(
-        text("SELECT 1 FROM pg_constraint WHERE conname = 'auth_uid_unique'")
-    ).scalar()
-    if constraint_exists:
-        op.drop_constraint("auth_uid_unique", "users", type_="unique")
+    op.drop_table("user_workspace_roles")
+    op.execute(text("DROP TYPE workspace_role"))
+    op.drop_constraint("auth_uid_unique", "users", type_="unique")
diff --git a/api/src/tasking/projects/repository.py b/api/src/tasking/projects/repository.py
index eaa81eb..3a73f66 100644
--- a/api/src/tasking/projects/repository.py
+++ b/api/src/tasking/projects/repository.py
@@ -878,7 +878,9 @@ async def upload_aoi(
     # violations on `user_auth_uid` are caught with a preflight so the
     # caller gets a 422 listing the missing user id.
 
-    async def _is_project_lead(self, project_id: int, user_uuid: UUID) -> bool:
+    async def _is_project_lead(
+        self, project_id: int, user_uuid: UUID
+    ) -> bool:
         """True if the user holds a `lead` role on the given project."""
         from sqlalchemy import text
 
@@ -911,7 +913,8 @@ async def assert_can_manage_roles(
         if await self._is_project_lead(project_id, current_user.user_uuid):
             return
         raise ForbiddenException(
-            "Only a workspace lead or project lead can manage roles " "on this project."
+            "Only a workspace lead or project lead can manage roles "
+            "on this project."
         )
 
     async def _lead_count(self, project_id: int) -> int:
@@ -1252,7 +1255,9 @@ async def remove_role(
         )
         await self.session.commit()
 
-    async def _get_role(self, project_id: int, user_id: UUID) -> ProjectRoleItem:
+    async def _get_role(
+        self, project_id: int, user_id: UUID
+    ) -> ProjectRoleItem:
         item = await self._get_role_or_none(project_id, user_id)
         if item is None:  # pragma: no cover — only called after insert/update
             raise NotFoundException(
@@ -1285,9 +1290,7 @@ async def _get_role_or_none(
             updated_at=updated,
         )
 
-    async def delete_aoi(
-        self, workspace_id: int, project_id: int, current_user: UserInfo
-    ) -> None:
+    async def delete_aoi(self, workspace_id: int, project_id: int) -> None:
         project = await self._get_active(workspace_id, project_id)
         if project.status != ProjectStatus.DRAFT:
             raise HTTPException(
diff --git a/api/src/tasking/projects/routes.py b/api/src/tasking/projects/routes.py
index 7389246..a6fa719 100644
--- a/api/src/tasking/projects/routes.py
+++ b/api/src/tasking/projects/routes.py
@@ -22,6 +22,7 @@
 )
 from api.src.tasking.projects.repository import TaskingProjectRepository
 from api.src.tasking.projects.schemas import AoiInput, ProjectStatus
+from uuid import UUID
 from api.src.workspaces.repository import WorkspaceRepository
 
 router = APIRouter(
@@ -355,6 +356,134 @@ async def remove_project_role(
     await project_repo.remove_role(workspace_id, project_id, user_id)
 
 
+# ---------------------------------------------------------------------------
+# Project role management
+#
+# Writes require either workspace LEAD or project LEAD (delegated to the
+# repository so the project-LEAD check can hit `tasking_project_roles`).
+# Reads are open to any caller who passes the workspace tenancy gate.
+# ---------------------------------------------------------------------------
+
+
+@router.get(
+    "/{project_id}/roles",
+    response_model=ProjectRoleListResponse,
+)
+async def list_project_roles(
+    workspace_id: int,
+    project_id: int,
+    current_user: UserInfo = Depends(validate_token),
+    workspace_repo: WorkspaceRepository = Depends(get_workspace_repo),
+    project_repo: TaskingProjectRepository = Depends(get_project_repo),
+):
+    await assert_workspace_visible(workspace_id, current_user, workspace_repo)
+    return await project_repo.list_roles(workspace_id, project_id)
+
+
+@router.post(
+    "/{project_id}/roles",
+    response_model=ProjectRoleItem,
+    status_code=status.HTTP_201_CREATED,
+)
+async def add_project_role(
+    workspace_id: int,
+    project_id: int,
+    body: ProjectRoleAddRequest,
+    current_user: UserInfo = Depends(validate_token),
+    workspace_repo: WorkspaceRepository = Depends(get_workspace_repo),
+    project_repo: TaskingProjectRepository = Depends(get_project_repo),
+):
+    await assert_workspace_visible(workspace_id, current_user, workspace_repo)
+    await project_repo.assert_can_manage_roles(
+        workspace_id, project_id, current_user
+    )
+    return await project_repo.add_role(workspace_id, project_id, body)
+
+
+@router.get(
+    "/{project_id}/roles/{user_id}",
+    response_model=ProjectRoleItem,
+)
+async def get_project_role(
+    workspace_id: int,
+    project_id: int,
+    user_id: UUID,
+    current_user: UserInfo = Depends(validate_token),
+    workspace_repo: WorkspaceRepository = Depends(get_workspace_repo),
+    project_repo: TaskingProjectRepository = Depends(get_project_repo),
+):
+    await assert_workspace_visible(workspace_id, current_user, workspace_repo)
+    return await project_repo.get_role(workspace_id, project_id, user_id)
+
+
+@router.put(
+    "/{project_id}/roles/{user_id}",
+    response_model=ProjectRoleItem,
+)
+async def put_project_role(
+    workspace_id: int,
+    project_id: int,
+    user_id: UUID,
+    body: ProjectRoleUpdateRequest,
+    response: Response,
+    current_user: UserInfo = Depends(validate_token),
+    workspace_repo: WorkspaceRepository = Depends(get_workspace_repo),
+    project_repo: TaskingProjectRepository = Depends(get_project_repo),
+):
+    """Idempotent upsert. 201 on insert, 200 on update. Last-LEAD guarded."""
+    await assert_workspace_visible(workspace_id, current_user, workspace_repo)
+    await project_repo.assert_can_manage_roles(
+        workspace_id, project_id, current_user
+    )
+    item, created = await project_repo.upsert_role(
+        workspace_id, project_id, user_id, body
+    )
+    if created:
+        response.status_code = status.HTTP_201_CREATED
+    return item
+
+
+@router.patch(
+    "/{project_id}/roles/{user_id}",
+    response_model=ProjectRoleItem,
+)
+async def update_project_role(
+    workspace_id: int,
+    project_id: int,
+    user_id: UUID,
+    body: ProjectRoleUpdateRequest,
+    current_user: UserInfo = Depends(validate_token),
+    workspace_repo: WorkspaceRepository = Depends(get_workspace_repo),
+    project_repo: TaskingProjectRepository = Depends(get_project_repo),
+):
+    await assert_workspace_visible(workspace_id, current_user, workspace_repo)
+    await project_repo.assert_can_manage_roles(
+        workspace_id, project_id, current_user
+    )
+    return await project_repo.update_role(
+        workspace_id, project_id, user_id, body
+    )
+
+
+@router.delete(
+    "/{project_id}/roles/{user_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+)
+async def remove_project_role(
+    workspace_id: int,
+    project_id: int,
+    user_id: UUID,
+    current_user: UserInfo = Depends(validate_token),
+    workspace_repo: WorkspaceRepository = Depends(get_workspace_repo),
+    project_repo: TaskingProjectRepository = Depends(get_project_repo),
+):
+    await assert_workspace_visible(workspace_id, current_user, workspace_repo)
+    await project_repo.assert_can_manage_roles(
+        workspace_id, project_id, current_user
+    )
+    await project_repo.remove_role(workspace_id, project_id, user_id)
+
+
 @router.delete("/{project_id}/aoi", status_code=status.HTTP_204_NO_CONTENT)
 async def delete_project_aoi(
     workspace_id: int,
@@ -365,7 +494,7 @@ async def delete_project_aoi(
 ):
     await assert_workspace_visible(workspace_id, current_user, workspace_repo)
     assert_workspace_lead(workspace_id, current_user)
-    await project_repo.delete_aoi(workspace_id, project_id, current_user)
+    await project_repo.delete_aoi(workspace_id, project_id)
 
 
 # ---------------------------------------------------------------------------
@@ -394,4 +523,6 @@ async def list_self_project_roles(
     Single round-trip for the project-list page.
     """
     await assert_workspace_visible(workspace_id, current_user, workspace_repo)
-    return await project_repo.list_self_project_roles(workspace_id, current_user)
+    return await project_repo.list_self_project_roles(
+        workspace_id, current_user
+    )
diff --git a/tests/conftest.py b/tests/conftest.py
index 309c430..ed1fb9d 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -71,8 +71,9 @@ def pytest_html_results_table_header(cells):
 
     def pytest_html_results_table_row(report, cells):
         """Inject the docstring as the matching cell on each row."""
-        cells.insert(2, f"<td>{getattr(report, 'description', '') or '—'}</td>")
-
+        cells.insert(
+            2, f"<td>{getattr(report, 'description', '') or '—'}</td>"
+        )
 except ImportError:
     pass
 
diff --git a/tests/integration/test_projects_flow.py b/tests/integration/test_projects_flow.py
index c1575de..2ba120d 100644
--- a/tests/integration/test_projects_flow.py
+++ b/tests/integration/test_projects_flow.py
@@ -485,6 +485,154 @@ async def test_update_unknown_user_returns_404(
         )
         assert r.status_code == 404
 
+
+# ---------------------------------------------------------------------------
+# Workflow 5 — project role management
+#   - LEAD can list/add/update/remove role assignments
+#   - workspace-LEAD passes the manage-roles gate even without a project row
+#   - contributor cannot manage roles (403)
+#   - 422 mapping for unknown user_id, duplicate (409), missing assignment (404)
+#   - last-LEAD guard blocks the demote / delete that would orphan the project
+# ---------------------------------------------------------------------------
+
+
+class TestProjectRoles:
+    async def test_list_includes_creator_auto_lead(
+        self, client, as_lead, seeded_workspace_id
+    ):
+        """Project creator is auto-seeded as LEAD and appears in GET /roles."""
+        r = await client.post(
+            API.format(wid=seeded_workspace_id),
+            json={"name": "roles-list-1"},
+        )
+        pid = r.json()["id"]
+        r = await client.get(
+            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles"
+        )
+        assert r.status_code == 200, r.text
+        rows = r.json()["results"]
+        assert len(rows) == 1
+        assert rows[0]["role"] == "lead"
+        assert rows[0]["user_id"] == str(as_lead.user_uuid)
+
+    async def test_add_role_round_trip(
+        self,
+        client,
+        as_lead,
+        seeded_workspace_id,
+        extra_user_factory,
+    ):
+        """POST /roles inserts a row; GET reflects it."""
+        contrib = await extra_user_factory("contributor")
+        r = await client.post(
+            API.format(wid=seeded_workspace_id),
+            json={"name": "roles-add"},
+        )
+        pid = r.json()["id"]
+
+        r = await client.post(
+            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles",
+            json={"user_id": str(contrib.user_uuid), "role": "contributor"},
+        )
+        assert r.status_code == 201, r.text
+        body = r.json()
+        assert body["user_id"] == str(contrib.user_uuid)
+        assert body["role"] == "contributor"
+
+        r = await client.get(
+            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles"
+        )
+        ids = {row["user_id"] for row in r.json()["results"]}
+        assert str(contrib.user_uuid) in ids
+
+    async def test_add_duplicate_returns_409(
+        self,
+        client,
+        as_lead,
+        seeded_workspace_id,
+        extra_user_factory,
+    ):
+        """Re-adding the same user returns 409 with an actionable hint."""
+        contrib = await extra_user_factory("contributor")
+        r = await client.post(
+            API.format(wid=seeded_workspace_id),
+            json={"name": "roles-dup"},
+        )
+        pid = r.json()["id"]
+
+        path = f"{API.format(wid=seeded_workspace_id)}/{pid}/roles"
+        await client.post(
+            path,
+            json={"user_id": str(contrib.user_uuid), "role": "contributor"},
+        )
+        r2 = await client.post(
+            path,
+            json={"user_id": str(contrib.user_uuid), "role": "validator"},
+        )
+        assert r2.status_code == 409, r2.text
+        assert "patch" in r2.json()["detail"].lower()
+
+    async def test_add_unknown_user_returns_422(
+        self, client, as_lead, seeded_workspace_id
+    ):
+        """An unknown user_id surfaces as 422 + missing_user_ids list."""
+        bogus = "ffffffff-ffff-ffff-ffff-ffffffffffff"
+        r = await client.post(
+            API.format(wid=seeded_workspace_id),
+            json={"name": "roles-unknown"},
+        )
+        pid = r.json()["id"]
+
+        r = await client.post(
+            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles",
+            json={"user_id": bogus, "role": "contributor"},
+        )
+        assert r.status_code == 422, r.text
+        assert bogus in r.json()["detail"]["missing_user_ids"]
+
+    async def test_update_role_promotes_contributor_to_validator(
+        self,
+        client,
+        as_lead,
+        seeded_workspace_id,
+        extra_user_factory,
+    ):
+        """PATCH /roles/{uid} changes the stored role."""
+        contrib = await extra_user_factory("contributor")
+        r = await client.post(
+            API.format(wid=seeded_workspace_id),
+            json={"name": "roles-patch"},
+        )
+        pid = r.json()["id"]
+        await client.post(
+            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles",
+            json={"user_id": str(contrib.user_uuid), "role": "contributor"},
+        )
+
+        r = await client.patch(
+            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles/"
+            f"{contrib.user_uuid}",
+            json={"role": "validator"},
+        )
+        assert r.status_code == 200, r.text
+        assert r.json()["role"] == "validator"
+
+    async def test_update_unknown_user_returns_404(
+        self, client, as_lead, seeded_workspace_id
+    ):
+        """PATCH on a user without a role row returns 404."""
+        r = await client.post(
+            API.format(wid=seeded_workspace_id),
+            json={"name": "roles-patch-404"},
+        )
+        pid = r.json()["id"]
+        absent = "deadbeef-dead-dead-dead-deadbeefdead"
+        r = await client.patch(
+            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles/{absent}",
+            json={"role": "validator"},
+        )
+        assert r.status_code == 404
+
     async def test_remove_role_round_trip(
         self,
         client,
@@ -505,7 +653,8 @@ async def test_remove_role_round_trip(
         )
 
         r = await client.delete(
-            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles/" f"{contrib.user_uuid}"
+            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles/"
+            f"{contrib.user_uuid}"
         )
         assert r.status_code == 204
 
@@ -517,7 +666,9 @@ async def test_remove_role_round_trip(
         )
         assert r.status_code == 404
 
-    async def test_last_lead_demote_blocked(self, client, as_lead, seeded_workspace_id):
+    async def test_last_lead_demote_blocked(
+        self, client, as_lead, seeded_workspace_id
+    ):
         """Cannot demote the only LEAD — projects must always have one."""
         r = await client.post(
             API.format(wid=seeded_workspace_id),
@@ -533,7 +684,9 @@ async def test_last_lead_demote_blocked(self, client, as_lead, seeded_workspace_
         assert r.status_code == 422, r.text
         assert "last lead" in r.json()["detail"].lower()
 
-    async def test_last_lead_delete_blocked(self, client, as_lead, seeded_workspace_id):
+    async def test_last_lead_delete_blocked(
+        self, client, as_lead, seeded_workspace_id
+    ):
         """Cannot delete the only LEAD — would orphan the project."""
         r = await client.post(
             API.format(wid=seeded_workspace_id),
@@ -569,7 +722,8 @@ async def test_demote_lead_works_when_two_leads_exist(
 
         # Now demote the second lead — first lead is still there.
         r = await client.patch(
-            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles/" f"{lead2.user_uuid}",
+            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles/"
+            f"{lead2.user_uuid}",
             json={"role": "contributor"},
         )
         assert r.status_code == 200, r.text
@@ -595,7 +749,8 @@ async def test_get_single_role(
         )
 
         r = await client.get(
-            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles/" f"{contrib.user_uuid}"
+            f"{API.format(wid=seeded_workspace_id)}/{pid}/roles/"
+            f"{contrib.user_uuid}"
         )
         assert r.status_code == 200, r.text
         body = r.json()