diff --git a/CHANGELOG.md b/CHANGELOG.md
index 73591c093..95193207c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,14 @@ Changed `wf10_batch.robot` to be less likely to create files, and
 `plan_generation_graph.robot` to give more test answers.
 
 ## [Unreleased]
+- Added team collaboration features: multi-user connection handling with
+  user identity tracking (TeamMember with owner/admin/member/viewer roles),
+  role-based access control (TeamPermission with read/write/admin/manage_members),
+  concurrent session support (SessionRegistry with thread-safe locking),
+  and optimistic-locking conflict resolution (VersionStamp with last-writer-wins,
+  reject, and merge strategies). Includes TeamCollaborationService orchestrating
+  all collaboration operations. Behave BDD tests and Robot Framework integration
+  tests included. (#863)
 - Added FastAPI-based ASGI server endpoint served by uvicorn for the
   CleverAgents server mode. Includes health check endpoint (`/health`),
   A2A Agent Card discovery (`/.well-known/agent.json`), A2A JSON-RPC 2.0
diff --git a/features/steps/team_collab_steps.py b/features/steps/team_collab_steps.py
new file mode 100644
index 000000000..90983aacc
--- /dev/null
+++ b/features/steps/team_collab_steps.py
@@ -0,0 +1,675 @@
+"""Step definitions for team collaboration feature tests.
+
+Covers team roles, permissions, user sessions, session registry,
+version stamps, conflict resolution, and the TeamCollaborationService.
+
+Based on issue #863 — feat(server): team collaboration features.
+"""
+
+from __future__ import annotations
+
+import re
+from time import sleep
+
+from behave import given, then, when
+from behave.runner import Context
+from pydantic import ValidationError
+
+from cleveragents.application.services.team_collab_service import (
+    TeamCollaborationService,
+)
+from cleveragents.domain.models.core.team_collab import (
+    TEAM_ROLE_PERMISSIONS,
+    ConflictDetectionResult,
+    ConflictResolutionStrategy,
+    SessionRegistry,
+    TeamCollaborationError,
+    TeamMember,
+    TeamMemberExistsError,
+    TeamMemberNotFoundError,
+    TeamPermission,
+    TeamRole,
+    UserSession,
+    VersionConflictError,
+    VersionStamp,
+)
+
+ULID_RE = re.compile(r"^[0-9A-HJKMNP-TV-Z]{26}$")
+
+
+# ── TeamRole enum ──────────────────────────────────────────────────
+
+
+@given("the TeamRole enum")
+def step_given_team_role_enum(context: Context) -> None:
+    context.enum_cls = TeamRole
+
+
+@then('the TeamRole enum should contain "{value}"')
+def step_then_team_role_contains(context: Context, value: str) -> None:
+    members = [m.value for m in TeamRole]
+    assert value in members, f"{value} not in TeamRole: {members}"
+
+
+# ── TeamPermission enum ────────────────────────────────────────────
+
+
+@given("the TeamPermission enum")
+def step_given_team_permission_enum(context: Context) -> None:
+    context.enum_cls = TeamPermission
+
+
+@then('the TeamPermission enum should contain "{value}"')
+def step_then_team_permission_contains(context: Context, value: str) -> None:
+    members = [m.value for m in TeamPermission]
+    assert value in members, f"{value} not in TeamPermission: {members}"
+
+
+# ── ConflictResolutionStrategy enum ────────────────────────────────
+
+
+@given("the ConflictResolutionStrategy enum")
+def step_given_conflict_strategy_enum(context: Context) -> None:
+    context.enum_cls = ConflictResolutionStrategy
+
+
+@then('the ConflictResolutionStrategy enum should contain "{value}"')
+def step_then_conflict_strategy_contains(context: Context, value: str) -> None:
+    members = [m.value for m in ConflictResolutionStrategy]
+    assert value in members, f"{value} not in ConflictResolutionStrategy: {members}"
+
+
+# ── TEAM_ROLE_PERMISSIONS constant ─────────────────────────────────
+
+
+@given("the TEAM_ROLE_PERMISSIONS constant")
+def step_given_team_role_permissions(context: Context) -> None:
+    context.role_permissions = TEAM_ROLE_PERMISSIONS
+
+
+@then('the "{role}" team role should have {count:d} allowed permissions')
+def step_then_team_role_has_n_permissions(
+    context: Context, role: str, count: int
+) -> None:
+    team_role = TeamRole(role)
+    perms = TEAM_ROLE_PERMISSIONS[team_role]
+    assert len(perms) == count, f"{role} has {len(perms)} permissions, expected {count}"
+
+
+# ── TeamMember model ───────────────────────────────────────────────
+
+
+@given('a team member "{user_id}" with role "{role}" and email "{email}"')
+def step_given_team_member(
+    context: Context, user_id: str, role: str, email: str
+) -> None:
+    context.team_member = TeamMember(
+        user_id=user_id,
+        display_name=user_id.title(),
+        email=email,
+        role=TeamRole(role),
+    )
+
+
+@then('the team member user_id should be "{expected}"')
+def step_then_member_user_id(context: Context, expected: str) -> None:
+    assert context.team_member.user_id == expected
+
+
+@then('the team member role should be "{expected}"')
+def step_then_member_role(context: Context, expected: str) -> None:
+    assert context.team_member.role == TeamRole(expected)
+
+
+@then('the team member email should be "{expected}"')
+def step_then_member_email(context: Context, expected: str) -> None:
+    assert context.team_member.email == expected
+
+
+@when('I try to create a team member with email "{email}"')
+def step_when_create_member_bad_email(context: Context, email: str) -> None:
+    context.validation_error = None
+    try:
+        TeamMember(
+            user_id="test",
+            display_name="Test",
+            email=email,
+            role=TeamRole.MEMBER,
+        )
+    except ValidationError as exc:
+        context.validation_error = exc
+
+
+@then("a team member validation error should be raised")
+def step_then_team_member_validation_error_raised(context: Context) -> None:
+    assert context.validation_error is not None, "Expected ValidationError"
+
+
+@then('the team member should have "{permission}" permission')
+def step_then_member_has_permission(context: Context, permission: str) -> None:
+    perm = TeamPermission(permission)
+    assert context.team_member.has_permission(perm), (
+        f"Expected member to have {permission}"
+    )
+
+
+@then('the team member should not have "{permission}" permission')
+def step_then_member_no_permission(context: Context, permission: str) -> None:
+    perm = TeamPermission(permission)
+    assert not context.team_member.has_permission(perm), (
+        f"Expected member NOT to have {permission}"
+    )
+
+
+# ── UserSession model ──────────────────────────────────────────────
+
+
+@given('a user session for user "{user_id}" on project "{project_id}"')
+def step_given_user_session(context: Context, user_id: str, project_id: str) -> None:
+    context.user_session = UserSession(
+        session_id=str(SessionRegistry().generate_session_id()),
+        user_id=user_id,
+        display_name=user_id.title(),
+        project_id=project_id,
+    )
+
+
+@then('the user session user_id should be "{expected}"')
+def step_then_session_user_id(context: Context, expected: str) -> None:
+    assert context.user_session.user_id == expected
+
+
+@then('the user session project_id should be "{expected}"')
+def step_then_session_project_id(context: Context, expected: str) -> None:
+    assert context.user_session.project_id == expected
+
+
+@then("the user session should be active")
+def step_then_session_active(context: Context) -> None:
+    assert context.user_session.is_active is True
+
+
+@when("the session is deactivated")
+def step_when_session_deactivated(context: Context) -> None:
+    context.user_session.deactivate()
+
+
+@then("the user session should not be active")
+def step_then_session_not_active(context: Context) -> None:
+    assert context.user_session.is_active is False
+
+
+@when("the session is touched")
+def step_when_session_touched(context: Context) -> None:
+    context.old_last_active = context.user_session.last_active
+    sleep(0.01)
+    context.user_session.touch()
+
+
+@then("the last_active timestamp should be updated")
+def step_then_last_active_updated(context: Context) -> None:
+    assert context.user_session.last_active >= context.old_last_active
+
+
+# ── VersionStamp model ─────────────────────────────────────────────
+
+
+@given('a version stamp for resource "{resource_id}" by user "{user_id}"')
+def step_given_version_stamp(context: Context, resource_id: str, user_id: str) -> None:
+    context.version_stamp = VersionStamp(
+        resource_id=resource_id,
+        updated_by=user_id,
+    )
+
+
+@then("the version stamp version should be {expected:d}")
+def step_then_stamp_version(context: Context, expected: int) -> None:
+    assert context.version_stamp.version == expected
+
+
+@then('the version stamp updated_by should be "{expected}"')
+def step_then_stamp_updated_by(context: Context, expected: str) -> None:
+    assert context.version_stamp.updated_by == expected
+
+
+@when('the version stamp is incremented by "{user_id}"')
+def step_when_stamp_incremented(context: Context, user_id: str) -> None:
+    context.new_version_stamp = context.version_stamp.increment(user_id)
+
+
+@then("the new version stamp version should be {expected:d}")
+def step_then_new_stamp_version(context: Context, expected: int) -> None:
+    assert context.new_version_stamp.version == expected
+
+
+@then('the new version stamp updated_by should be "{expected}"')
+def step_then_new_stamp_updated_by(context: Context, expected: str) -> None:
+    assert context.new_version_stamp.updated_by == expected
+
+
+@then("incrementing with empty user should raise ValueError")
+def step_then_increment_empty_raises(context: Context) -> None:
+    raised = False
+    try:
+        context.version_stamp.increment("")
+    except ValueError:
+        raised = True
+    assert raised, "Expected ValueError"
+
+
+# ── ConflictDetectionResult ────────────────────────────────────────
+
+
+@given('a conflict detection result with no conflict for "{resource_id}"')
+def step_given_no_conflict_result(context: Context, resource_id: str) -> None:
+    context.conflict_result = ConflictDetectionResult(
+        detected=False,
+        resource_id=resource_id,
+        resolved=True,
+        details="No conflict",
+    )
+
+
+@then("the result detected should be False")
+def step_then_result_not_detected(context: Context) -> None:
+    assert context.conflict_result.detected is False
+
+
+@then('the result resource_id should be "{expected}"')
+def step_then_result_resource_id(context: Context, expected: str) -> None:
+    assert context.conflict_result.resource_id == expected
+
+
+# ── VersionConflictError ───────────────────────────────────────────
+
+
+@when(
+    'a VersionConflictError is raised for "{resource_id}" '
+    "expected {expected:d} actual {actual:d}"
+)
+def step_when_version_conflict_error(
+    context: Context, resource_id: str, expected: int, actual: int
+) -> None:
+    context.conflict_error = VersionConflictError(
+        resource_id=resource_id,
+        expected_version=expected,
+        actual_version=actual,
+    )
+
+
+@then('the error resource_id should be "{expected}"')
+def step_then_error_resource_id(context: Context, expected: str) -> None:
+    assert context.conflict_error.resource_id == expected
+
+
+@then("the error expected_version should be {expected:d}")
+def step_then_error_expected_version(context: Context, expected: int) -> None:
+    assert context.conflict_error.expected_version == expected
+
+
+@then("the error actual_version should be {expected:d}")
+def step_then_error_actual_version(context: Context, expected: int) -> None:
+    assert context.conflict_error.actual_version == expected
+
+
+@then("creating VersionConflictError with empty resource_id raises ValueError")
+def step_then_conflict_error_empty_raises(context: Context) -> None:
+    raised = False
+    try:
+        VersionConflictError(resource_id="", expected_version=1, actual_version=2)
+    except ValueError:
+        raised = True
+    assert raised, "Expected ValueError"
+
+
+# ── SessionRegistry ────────────────────────────────────────────────
+
+
+@given("a session registry")
+def step_given_session_registry(context: Context) -> None:
+    context.session_registry = SessionRegistry()
+    context.sessions = {}
+
+
+@given('a user session "{session_id}" for user "{user_id}" on project "{project_id}"')
+def step_given_named_session(
+    context: Context, session_id: str, user_id: str, project_id: str
+) -> None:
+    session = UserSession(
+        session_id=session_id,
+        user_id=user_id,
+        display_name=user_id.title(),
+        project_id=project_id,
+    )
+    if not hasattr(context, "sessions") or context.sessions is None:
+        context.sessions = {}
+    context.sessions[session_id] = session
+
+
+@when("the session is registered")
+def step_when_session_registered(context: Context) -> None:
+    for session in context.sessions.values():
+        context.session_registry.register(session)
+
+
+@when("all sessions are registered")
+def step_when_all_sessions_registered(context: Context) -> None:
+    for session in context.sessions.values():
+        context.session_registry.register(session)
+
+
+@then('the registry should contain the session "{session_id}"')
+def step_then_registry_contains(context: Context, session_id: str) -> None:
+    result = context.session_registry.get(session_id)
+    assert result is not None, f"Session '{session_id}' not found"
+
+
+@then('the registry should not contain the session "{session_id}"')
+def step_then_registry_not_contains(context: Context, session_id: str) -> None:
+    result = context.session_registry.get(session_id)
+    assert result is None, f"Session '{session_id}' should not be found"
+
+
+@then("the registry active count should be {expected:d}")
+def step_then_registry_active_count(context: Context, expected: int) -> None:
+    assert context.session_registry.active_count == expected, (
+        f"Expected {expected}, got {context.session_registry.active_count}"
+    )
+
+
+@when('the session "{session_id}" is unregistered')
+def step_when_session_unregistered(context: Context, session_id: str) -> None:
+    context.session_registry.unregister(session_id)
+
+
+@then('the registry should have {count:d} sessions for user "{user_id}"')
+def step_then_registry_user_sessions(
+    context: Context, count: int, user_id: str
+) -> None:
+    sessions = context.session_registry.get_sessions_for_user(user_id)
+    assert len(sessions) == count, (
+        f"Expected {count} sessions for user '{user_id}', got {len(sessions)}"
+    )
+
+
+@then('the registry should have {count:d} sessions for project "{project_id}"')
+def step_then_registry_project_sessions(
+    context: Context, count: int, project_id: str
+) -> None:
+    sessions = context.session_registry.get_sessions_for_project(project_id)
+    assert len(sessions) == count, (
+        f"Expected {count} sessions for project '{project_id}', got {len(sessions)}"
+    )
+
+
+@when("the registry is cleared")
+def step_when_registry_cleared(context: Context) -> None:
+    context.session_registry.clear()
+
+
+@then("registering the same session again raises ValueError")
+def step_then_duplicate_registration_raises(context: Context) -> None:
+    session = next(iter(context.sessions.values()))
+    raised = False
+    try:
+        context.session_registry.register(session)
+    except ValueError:
+        raised = True
+    assert raised, "Expected ValueError for duplicate registration"
+
+
+@then("the generated session ID should be a valid ULID")
+def step_then_generated_id_valid(context: Context) -> None:
+    sid = context.session_registry.generate_session_id()
+    assert ULID_RE.match(sid), f"'{sid}' is not a valid ULID"
+
+
+# ── TeamCollaborationService — member management ───────────────────
+
+
+@given("a team collaboration service")
+def step_given_collab_service(context: Context) -> None:
+    context.collab_service = TeamCollaborationService()
+    context.last_session = None
+    context.last_conflict_result = None
+
+
+@when('I add member "{user_id}" with role "{role}" and email "{email}"')
+def step_when_add_member(context: Context, user_id: str, role: str, email: str) -> None:
+    context.collab_service.add_member(
+        user_id=user_id,
+        display_name=user_id.title(),
+        email=email,
+        role=TeamRole(role),
+    )
+
+
+@then("the service should have {count:d} members")
+def step_then_service_member_count(context: Context, count: int) -> None:
+    members = context.collab_service.list_members()
+    assert len(members) == count, f"Expected {count} members, got {len(members)}"
+
+
+@then('getting member "{user_id}" should return the member')
+def step_then_get_member(context: Context, user_id: str) -> None:
+    member = context.collab_service.get_member(user_id)
+    assert member.user_id == user_id
+
+
+@then('adding member "{user_id}" again raises TeamMemberExistsError')
+def step_then_add_duplicate_raises(context: Context, user_id: str) -> None:
+    raised = False
+    try:
+        context.collab_service.add_member(
+            user_id=user_id,
+            display_name="Dup",
+            email="dup@example.com",
+            role=TeamRole.MEMBER,
+        )
+    except TeamMemberExistsError:
+        raised = True
+    assert raised, "Expected TeamMemberExistsError"
+
+
+@when('I remove member "{user_id}"')
+def step_when_remove_member(context: Context, user_id: str) -> None:
+    context.collab_service.remove_member(user_id)
+
+
+@then('removing member "{user_id}" raises TeamMemberNotFoundError')
+def step_then_remove_not_found_raises(context: Context, user_id: str) -> None:
+    raised = False
+    try:
+        context.collab_service.remove_member(user_id)
+    except TeamMemberNotFoundError:
+        raised = True
+    assert raised, "Expected TeamMemberNotFoundError"
+
+
+@when('I update member "{user_id}" role to "{role}"')
+def step_when_update_role(context: Context, user_id: str, role: str) -> None:
+    context.collab_service.update_member_role(user_id, TeamRole(role))
+
+
+@then('member "{user_id}" should have role "{role}"')
+def step_then_member_has_role(context: Context, user_id: str, role: str) -> None:
+    member = context.collab_service.get_member(user_id)
+    assert member.role == TeamRole(role), f"Expected role {role}, got {member.role}"
+
+
+# ── TeamCollaborationService — permission checks ──────────────────
+
+
+@then('member "{user_id}" should have "{permission}" team permission')
+def step_then_service_member_has_perm(
+    context: Context, user_id: str, permission: str
+) -> None:
+    result = context.collab_service.check_permission(
+        user_id, TeamPermission(permission)
+    )
+    assert result is True, f"Expected {user_id} to have {permission}"
+
+
+@then('member "{user_id}" should not have "{permission}" team permission')
+def step_then_service_member_no_perm(
+    context: Context, user_id: str, permission: str
+) -> None:
+    result = context.collab_service.check_permission(
+        user_id, TeamPermission(permission)
+    )
+    assert result is False, f"Expected {user_id} NOT to have {permission}"
+
+
+@then('enforcing "{permission}" on "{user_id}" raises PermissionError')
+def step_then_enforce_raises(context: Context, permission: str, user_id: str) -> None:
+    raised = False
+    try:
+        context.collab_service.enforce_permission(user_id, TeamPermission(permission))
+    except PermissionError:
+        raised = True
+    assert raised, "Expected PermissionError"
+
+
+# ── TeamCollaborationService — session management ─────────────────
+
+
+@when('I register a session for "{user_id}" on project "{project_id}"')
+def step_when_register_service_session(
+    context: Context, user_id: str, project_id: str
+) -> None:
+    session = context.collab_service.register_session(
+        user_id=user_id,
+        project_id=project_id,
+    )
+    context.last_session = session
+
+
+@then('the project "{project_id}" should have {count:d} active sessions')
+def step_then_project_active_sessions(
+    context: Context, project_id: str, count: int
+) -> None:
+    sessions = context.collab_service.get_active_sessions_for_project(project_id)
+    assert len(sessions) == count, (
+        f"Expected {count} active sessions for '{project_id}', got {len(sessions)}"
+    )
+
+
+@when("I unregister the last session")
+def step_when_unregister_last_session(context: Context) -> None:
+    context.collab_service.unregister_session(context.last_session.session_id)
+
+
+# ── TeamCollaborationService — conflict resolution ─────────────────
+
+
+@when('I create a version stamp for "{resource_id}" by "{user_id}"')
+def step_when_create_stamp(context: Context, resource_id: str, user_id: str) -> None:
+    context.collab_service.create_version_stamp(resource_id, user_id)
+
+
+@then('the version stamp for "{resource_id}" should have version {version:d}')
+def step_then_stamp_has_version(
+    context: Context, resource_id: str, version: int
+) -> None:
+    stamp = context.collab_service.get_version_stamp(resource_id)
+    assert stamp is not None, f"No stamp for '{resource_id}'"
+    assert stamp.version == version, f"Expected version {version}, got {stamp.version}"
+
+
+@when('I update version for "{resource_id}" expecting {version:d} by "{user_id}"')
+def step_when_update_version(
+    context: Context, resource_id: str, version: int, user_id: str
+) -> None:
+    context.last_conflict_result = context.collab_service.update_version(
+        resource_id=resource_id,
+        expected_version=version,
+        user_id=user_id,
+    )
+
+
+@then(
+    'updating "{resource_id}" expecting {version:d} by "{user_id}" '
+    "with reject raises VersionConflictError"
+)
+def step_then_update_reject_raises(
+    context: Context, resource_id: str, version: int, user_id: str
+) -> None:
+    raised = False
+    try:
+        context.collab_service.update_version(
+            resource_id=resource_id,
+            expected_version=version,
+            user_id=user_id,
+            strategy=ConflictResolutionStrategy.REJECT,
+        )
+    except VersionConflictError:
+        raised = True
+    assert raised, "Expected VersionConflictError"
+
+
+@when(
+    'I update "{resource_id}" expecting {version:d} by "{user_id}" '
+    "with last_writer_wins"
+)
+def step_when_update_lww(
+    context: Context, resource_id: str, version: int, user_id: str
+) -> None:
+    context.last_conflict_result = context.collab_service.update_version(
+        resource_id=resource_id,
+        expected_version=version,
+        user_id=user_id,
+        strategy=ConflictResolutionStrategy.LAST_WRITER_WINS,
+    )
+
+
+@when('I update "{resource_id}" expecting {version:d} by "{user_id}" with merge')
+def step_when_update_merge(
+    context: Context, resource_id: str, version: int, user_id: str
+) -> None:
+    context.last_conflict_result = context.collab_service.update_version(
+        resource_id=resource_id,
+        expected_version=version,
+        user_id=user_id,
+        strategy=ConflictResolutionStrategy.MERGE,
+    )
+
+
+@then("the conflict result should be detected and resolved")
+def step_then_conflict_detected_resolved(context: Context) -> None:
+    result = context.last_conflict_result
+    assert result.detected is True, "Expected conflict detected"
+    assert result.resolved is True, "Expected conflict resolved"
+
+
+@then('detecting conflict on "{resource_id}" expecting {version:d} should return True')
+def step_then_detect_conflict_true(
+    context: Context, resource_id: str, version: int
+) -> None:
+    result = context.collab_service.detect_conflict(resource_id, version)
+    assert result is True, "Expected conflict detected"
+
+
+@then('detecting conflict on "{resource_id}" expecting {version:d} should return False')
+def step_then_detect_conflict_false(
+    context: Context, resource_id: str, version: int
+) -> None:
+    result = context.collab_service.detect_conflict(resource_id, version)
+    assert result is False, "Expected no conflict"
+
+
+# ── Error hierarchy ────────────────────────────────────────────────
+
+
+@then("TeamMemberNotFoundError should be a subclass of TeamCollaborationError")
+def step_then_not_found_is_subclass(context: Context) -> None:
+    assert issubclass(TeamMemberNotFoundError, TeamCollaborationError)
+
+
+@then("TeamMemberExistsError should be a subclass of TeamCollaborationError")
+def step_then_exists_is_subclass(context: Context) -> None:
+    assert issubclass(TeamMemberExistsError, TeamCollaborationError)
+
+
+@then("VersionConflictError should be a subclass of TeamCollaborationError")
+def step_then_conflict_is_subclass(context: Context) -> None:
+    assert issubclass(VersionConflictError, TeamCollaborationError)
diff --git a/features/team_collab.feature b/features/team_collab.feature
new file mode 100644
index 000000000..e679b681b
--- /dev/null
+++ b/features/team_collab.feature
@@ -0,0 +1,308 @@
+Feature: Team collaboration — multi-user sessions, RBAC, and conflict resolution
+  As a CleverAgents team lead
+  I want team collaboration features with role-based access control
+  So that multiple users can work concurrently on shared projects
+
+  # ── TeamRole enum ────────────────────────────────────────────────
+
+  Scenario: TeamRole enum contains expected members
+    Given the TeamRole enum
+    Then the TeamRole enum should contain "owner"
+    And the TeamRole enum should contain "admin"
+    And the TeamRole enum should contain "member"
+    And the TeamRole enum should contain "viewer"
+
+  # ── TeamPermission enum ──────────────────────────────────────────
+
+  Scenario: TeamPermission enum contains expected members
+    Given the TeamPermission enum
+    Then the TeamPermission enum should contain "read"
+    And the TeamPermission enum should contain "write"
+    And the TeamPermission enum should contain "admin"
+    And the TeamPermission enum should contain "manage_members"
+
+  # ── ConflictResolutionStrategy enum ──────────────────────────────
+
+  Scenario: ConflictResolutionStrategy enum contains expected members
+    Given the ConflictResolutionStrategy enum
+    Then the ConflictResolutionStrategy enum should contain "last_writer_wins"
+    And the ConflictResolutionStrategy enum should contain "reject"
+    And the ConflictResolutionStrategy enum should contain "merge"
+
+  # ── TEAM_ROLE_PERMISSIONS constant ───────────────────────────────
+
+  Scenario: OWNER role has all 4 permissions
+    Given the TEAM_ROLE_PERMISSIONS constant
+    Then the "owner" team role should have 4 allowed permissions
+
+  Scenario: ADMIN role has all 4 permissions
+    Given the TEAM_ROLE_PERMISSIONS constant
+    Then the "admin" team role should have 4 allowed permissions
+
+  Scenario: MEMBER role has read and write
+    Given the TEAM_ROLE_PERMISSIONS constant
+    Then the "member" team role should have 2 allowed permissions
+
+  Scenario: VIEWER role has only read
+    Given the TEAM_ROLE_PERMISSIONS constant
+    Then the "viewer" team role should have 1 allowed permissions
+
+  # ── TeamMember model ─────────────────────────────────────────────
+
+  Scenario: Create a valid TeamMember
+    Given a team member "alice" with role "owner" and email "alice@example.com"
+    Then the team member user_id should be "alice"
+    And the team member role should be "owner"
+    And the team member email should be "alice@example.com"
+
+  Scenario: TeamMember email validation rejects missing @
+    When I try to create a team member with email "invalid-email"
+    Then a team member validation error should be raised
+
+  Scenario: TeamMember has_permission returns True for allowed action
+    Given a team member "bob" with role "member" and email "bob@example.com"
+    Then the team member should have "read" permission
+    And the team member should have "write" permission
+
+  Scenario: TeamMember has_permission returns False for denied action
+    Given a team member "carol" with role "viewer" and email "carol@example.com"
+    Then the team member should not have "write" permission
+    And the team member should not have "admin" permission
+
+  # ── UserSession model ────────────────────────────────────────────
+
+  Scenario: Create a valid UserSession
+    Given a user session for user "alice" on project "proj-1"
+    Then the user session user_id should be "alice"
+    And the user session project_id should be "proj-1"
+    And the user session should be active
+
+  Scenario: Deactivate a UserSession
+    Given a user session for user "alice" on project "proj-1"
+    When the session is deactivated
+    Then the user session should not be active
+
+  Scenario: Touch updates last_active timestamp
+    Given a user session for user "alice" on project "proj-1"
+    When the session is touched
+    Then the last_active timestamp should be updated
+
+  # ── VersionStamp model ───────────────────────────────────────────
+
+  Scenario: Create a valid VersionStamp
+    Given a version stamp for resource "res-1" by user "alice"
+    Then the version stamp version should be 1
+    And the version stamp updated_by should be "alice"
+
+  Scenario: Increment a VersionStamp
+    Given a version stamp for resource "res-1" by user "alice"
+    When the version stamp is incremented by "bob"
+    Then the new version stamp version should be 2
+    And the new version stamp updated_by should be "bob"
+
+  Scenario: VersionStamp increment rejects empty user
+    Given a version stamp for resource "res-1" by user "alice"
+    Then incrementing with empty user should raise ValueError
+
+  # ── ConflictDetectionResult model ────────────────────────────────
+
+  Scenario: Create a no-conflict result
+    Given a conflict detection result with no conflict for "res-1"
+    Then the result detected should be False
+    And the result resource_id should be "res-1"
+
+  # ── VersionConflictError ─────────────────────────────────────────
+
+  Scenario: VersionConflictError carries version info
+    When a VersionConflictError is raised for "res-1" expected 1 actual 2
+    Then the error resource_id should be "res-1"
+    And the error expected_version should be 1
+    And the error actual_version should be 2
+
+  Scenario: VersionConflictError rejects empty resource_id
+    Then creating VersionConflictError with empty resource_id raises ValueError
+
+  # ── SessionRegistry ──────────────────────────────────────────────
+
+  Scenario: Register and retrieve a session
+    Given a session registry
+    And a user session "sess-1" for user "alice" on project "proj-1"
+    When the session is registered
+    Then the registry should contain the session "sess-1"
+    And the registry active count should be 1
+
+  Scenario: Unregister a session
+    Given a session registry
+    And a user session "sess-1" for user "alice" on project "proj-1"
+    When the session is registered
+    And the session "sess-1" is unregistered
+    Then the registry should not contain the session "sess-1"
+    And the registry active count should be 0
+
+  Scenario: Get sessions for a specific user
+    Given a session registry
+    And a user session "sess-1" for user "alice" on project "proj-1"
+    And a user session "sess-2" for user "alice" on project "proj-2"
+    And a user session "sess-3" for user "bob" on project "proj-1"
+    When all sessions are registered
+    Then the registry should have 2 sessions for user "alice"
+    And the registry should have 1 sessions for user "bob"
+
+  Scenario: Get sessions for a specific project
+    Given a session registry
+    And a user session "sess-1" for user "alice" on project "proj-1"
+    And a user session "sess-2" for user "bob" on project "proj-1"
+    And a user session "sess-3" for user "carol" on project "proj-2"
+    When all sessions are registered
+    Then the registry should have 2 sessions for project "proj-1"
+    And the registry should have 1 sessions for project "proj-2"
+
+  Scenario: Clear all sessions
+    Given a session registry
+    And a user session "sess-1" for user "alice" on project "proj-1"
+    And a user session "sess-2" for user "bob" on project "proj-2"
+    When all sessions are registered
+    And the registry is cleared
+    Then the registry active count should be 0
+
+  Scenario: Duplicate session registration raises ValueError
+    Given a session registry
+    And a user session "sess-1" for user "alice" on project "proj-1"
+    When the session is registered
+    Then registering the same session again raises ValueError
+
+  Scenario: Session registry generate_session_id returns valid ULID
+    Given a session registry
+    Then the generated session ID should be a valid ULID
+
+  # ── TeamCollaborationService — member management ─────────────────
+
+  Scenario: Add and retrieve a team member via service
+    Given a team collaboration service
+    When I add member "alice" with role "owner" and email "alice@example.com"
+    Then the service should have 1 members
+    And getting member "alice" should return the member
+
+  Scenario: Adding duplicate member raises TeamMemberExistsError
+    Given a team collaboration service
+    When I add member "alice" with role "owner" and email "alice@example.com"
+    Then adding member "alice" again raises TeamMemberExistsError
+
+  Scenario: Remove a team member via service
+    Given a team collaboration service
+    When I add member "alice" with role "owner" and email "alice@example.com"
+    And I remove member "alice"
+    Then the service should have 0 members
+
+  Scenario: Removing non-existent member raises TeamMemberNotFoundError
+    Given a team collaboration service
+    Then removing member "ghost" raises TeamMemberNotFoundError
+
+  Scenario: Update a member role
+    Given a team collaboration service
+    When I add member "alice" with role "member" and email "alice@example.com"
+    And I update member "alice" role to "admin"
+    Then member "alice" should have role "admin"
+
+  # ── TeamCollaborationService — permission checks ─────────────────
+
+  Scenario: Owner has all permissions
+    Given a team collaboration service
+    When I add member "alice" with role "owner" and email "alice@example.com"
+    Then member "alice" should have "read" team permission
+    And member "alice" should have "write" team permission
+    And member "alice" should have "admin" team permission
+    And member "alice" should have "manage_members" team permission
+
+  Scenario: Viewer has only read permission
+    Given a team collaboration service
+    When I add member "viewer-user" with role "viewer" and email "view@example.com"
+    Then member "viewer-user" should have "read" team permission
+    And member "viewer-user" should not have "write" team permission
+
+  Scenario: Enforce permission raises PermissionError for viewer writing
+    Given a team collaboration service
+    When I add member "viewer-user" with role "viewer" and email "view@example.com"
+    Then enforcing "write" on "viewer-user" raises PermissionError
+
+  # ── TeamCollaborationService — session management ────────────────
+
+  Scenario: Register a session for a team member
+    Given a team collaboration service
+    When I add member "alice" with role "owner" and email "alice@example.com"
+    And I register a session for "alice" on project "proj-1"
+    Then the project "proj-1" should have 1 active sessions
+
+  Scenario: Multiple users have concurrent sessions on same project
+    Given a team collaboration service
+    When I add member "alice" with role "owner" and email "alice@example.com"
+    And I add member "bob" with role "member" and email "bob@example.com"
+    And I register a session for "alice" on project "proj-1"
+    And I register a session for "bob" on project "proj-1"
+    Then the project "proj-1" should have 2 active sessions
+
+  Scenario: Unregister a session reduces active count
+    Given a team collaboration service
+    When I add member "alice" with role "owner" and email "alice@example.com"
+    And I register a session for "alice" on project "proj-1"
+    And I unregister the last session
+    Then the project "proj-1" should have 0 active sessions
+
+  Scenario: Removing a member cleans up their sessions
+    Given a team collaboration service
+    When I add member "alice" with role "owner" and email "alice@example.com"
+    And I register a session for "alice" on project "proj-1"
+    And I remove member "alice"
+    Then the project "proj-1" should have 0 active sessions
+
+  # ── TeamCollaborationService — conflict resolution ───────────────
+
+  Scenario: Create and retrieve a version stamp
+    Given a team collaboration service
+    When I create a version stamp for "res-1" by "alice"
+    Then the version stamp for "res-1" should have version 1
+
+  Scenario: Update version with matching expected version succeeds
+    Given a team collaboration service
+    When I create a version stamp for "res-1" by "alice"
+    And I update version for "res-1" expecting 1 by "bob"
+    Then the version stamp for "res-1" should have version 2
+
+  Scenario: Update version with mismatched version using reject strategy raises error
+    Given a team collaboration service
+    When I create a version stamp for "res-1" by "alice"
+    And I update version for "res-1" expecting 1 by "bob"
+    Then updating "res-1" expecting 1 by "carol" with reject raises VersionConflictError
+
+  Scenario: Update version with last-writer-wins resolves conflict
+    Given a team collaboration service
+    When I create a version stamp for "res-1" by "alice"
+    And I update version for "res-1" expecting 1 by "bob"
+    And I update "res-1" expecting 1 by "carol" with last_writer_wins
+    Then the conflict result should be detected and resolved
+    And the version stamp for "res-1" should have version 3
+
+  Scenario: Detect conflict without resolving
+    Given a team collaboration service
+    When I create a version stamp for "res-1" by "alice"
+    And I update version for "res-1" expecting 1 by "bob"
+    Then detecting conflict on "res-1" expecting 1 should return True
+    And detecting conflict on "res-1" expecting 2 should return False
+
+  Scenario: Update version with merge strategy resolves conflict
+    Given a team collaboration service
+    When I create a version stamp for "res-1" by "alice"
+    And I update version for "res-1" expecting 1 by "bob"
+    And I update "res-1" expecting 1 by "carol" with merge
+    Then the conflict result should be detected and resolved
+
+  # ── Error hierarchy ──────────────────────────────────────────────
+
+  Scenario: TeamMemberNotFoundError is a TeamCollaborationError
+    Then TeamMemberNotFoundError should be a subclass of TeamCollaborationError
+
+  Scenario: TeamMemberExistsError is a TeamCollaborationError
+    Then TeamMemberExistsError should be a subclass of TeamCollaborationError
+
+  Scenario: VersionConflictError is a TeamCollaborationError
+    Then VersionConflictError should be a subclass of TeamCollaborationError
diff --git a/robot/team_collab.robot b/robot/team_collab.robot
new file mode 100644
index 000000000..7e07b952c
--- /dev/null
+++ b/robot/team_collab.robot
@@ -0,0 +1,140 @@
+*** Settings ***
+Documentation     Integration tests for team collaboration features.
+...               Verifies multi-user session management, role-based access
+...               control, concurrent sessions, and conflict resolution.
+...               Based on issue #863.
+Library           Process
+Library           OperatingSystem
+Library           Collections
+
+Suite Setup       Set Suite Variable    ${PYTHON}    python
+
+*** Variables ***
+${PYTHON}         python
+
+*** Keywords ***
+Run Python Script
+    [Documentation]    Run a Python snippet and return stdout.
+    [Arguments]        ${script}
+    ${result}=    Run Process    ${PYTHON}    -c    ${script}
+    ...    env:PYTHONPATH=src
+    RETURN    ${result}
+
+*** Test Cases ***
+TeamRole Enum Contains Expected Members
+    [Documentation]    Verify TeamRole enum has owner, admin, member, viewer.
+    ${result}=    Run Python Script
+    ...    from cleveragents.domain.models.core.team_collab import TeamRole; roles = sorted([r.value for r in TeamRole]); print(','.join(roles))
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    admin
+    Should Contain    ${result.stdout}    member
+    Should Contain    ${result.stdout}    owner
+    Should Contain    ${result.stdout}    viewer
+
+TeamPermission Enum Contains Expected Members
+    [Documentation]    Verify TeamPermission enum has read, write, admin, manage_members.
+    ${result}=    Run Python Script
+    ...    from cleveragents.domain.models.core.team_collab import TeamPermission; perms = sorted([p.value for p in TeamPermission]); print(','.join(perms))
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    read
+    Should Contain    ${result.stdout}    write
+    Should Contain    ${result.stdout}    admin
+    Should Contain    ${result.stdout}    manage_members
+
+Team Role Permissions Mapping Is Correct
+    [Documentation]    Verify role-to-permission mapping counts.
+    ${result}=    Run Python Script
+    ...    from cleveragents.domain.models.core.team_collab import TEAM_ROLE_PERMISSIONS, TeamRole; print(len(TEAM_ROLE_PERMISSIONS[TeamRole.OWNER]), len(TEAM_ROLE_PERMISSIONS[TeamRole.ADMIN]), len(TEAM_ROLE_PERMISSIONS[TeamRole.MEMBER]), len(TEAM_ROLE_PERMISSIONS[TeamRole.VIEWER]))
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    4 4 2 1
+
+TeamMember Model Creation And Validation
+    [Documentation]    Create a TeamMember and verify fields.
+    ${result}=    Run Python Script
+    ...    from cleveragents.domain.models.core.team_collab import TeamMember, TeamRole; m = TeamMember(user_id='alice', display_name='Alice', email='alice@example.com', role=TeamRole.OWNER); print(f'{m.user_id},{m.role},{m.email}')
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    alice,owner,alice@example.com
+
+TeamMember Email Validation Rejects Invalid
+    [Documentation]    Creating a TeamMember with invalid email fails.
+    ${result}=    Run Python Script
+    ...    from cleveragents.domain.models.core.team_collab import TeamMember, TeamRole;\ntry:\n\ \ \ \ TeamMember(user_id='x', display_name='X', email='bad', role=TeamRole.MEMBER)\n\ \ \ \ print('NO_ERROR')\nexcept Exception as e:\n\ \ \ \ print('ERROR')
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    ERROR
+
+TeamMember Permission Check
+    [Documentation]    Verify has_permission for different roles.
+    ${result}=    Run Python Script
+    ...    from cleveragents.domain.models.core.team_collab import TeamMember, TeamRole, TeamPermission; m = TeamMember(user_id='v', display_name='V', email='v@x.com', role=TeamRole.VIEWER); print(m.has_permission(TeamPermission.READ), m.has_permission(TeamPermission.WRITE))
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    True False
+
+UserSession Creation And Deactivation
+    [Documentation]    Create a session, verify active, then deactivate.
+    ${result}=    Run Python Script
+    ...    from cleveragents.domain.models.core.team_collab import UserSession; s = UserSession(session_id='s1', user_id='alice', display_name='Alice', project_id='p1'); print(s.is_active); s.deactivate(); print(s.is_active)
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    True
+    Should Contain    ${result.stdout}    False
+
+VersionStamp Creation And Increment
+    [Documentation]    Create a version stamp and increment it.
+    ${result}=    Run Python Script
+    ...    from cleveragents.domain.models.core.team_collab import VersionStamp; v = VersionStamp(resource_id='r1', updated_by='alice'); v2 = v.increment('bob'); print(f'{v.version},{v2.version},{v2.updated_by}')
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    1,2,bob
+
+SessionRegistry Concurrent Session Management
+    [Documentation]    Register multiple sessions and query by user/project.
+    ${result}=    Run Python Script
+    ...    from cleveragents.domain.models.core.team_collab import SessionRegistry, UserSession; r = SessionRegistry(); r.register(UserSession(session_id='s1', user_id='alice', display_name='Alice', project_id='p1')); r.register(UserSession(session_id='s2', user_id='bob', display_name='Bob', project_id='p1')); r.register(UserSession(session_id='s3', user_id='alice', display_name='Alice', project_id='p2')); print(f'active={r.active_count},alice={len(r.get_sessions_for_user("alice"))},p1={len(r.get_sessions_for_project("p1"))}')
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    active=3,alice=2,p1=2
+
+TeamCollaborationService End To End Workflow
+    [Documentation]    Full workflow: add members, register sessions, check perms, handle conflicts.
+    ${result}=    Run Python Script
+    ...    from cleveragents.application.services.team_collab_service import TeamCollaborationService; from cleveragents.domain.models.core.team_collab import TeamRole, TeamPermission, ConflictResolutionStrategy, VersionConflictError; svc = TeamCollaborationService(); svc.add_member('alice', 'Alice', 'alice@x.com', TeamRole.OWNER); svc.add_member('bob', 'Bob', 'bob@x.com', TeamRole.MEMBER); svc.add_member('carol', 'Carol', 'carol@x.com', TeamRole.VIEWER); s1 = svc.register_session('alice', 'p1'); s2 = svc.register_session('bob', 'p1'); sessions = svc.get_active_sessions_for_project('p1'); print(f'sessions={len(sessions)}'); print(f'alice_write={svc.check_permission("alice", TeamPermission.WRITE)}'); print(f'carol_write={svc.check_permission("carol", TeamPermission.WRITE)}'); svc.create_version_stamp('r1', 'alice'); svc.update_version('r1', 1, 'bob'); conflict = svc.detect_conflict('r1', 1); print(f'conflict={conflict}'); result = svc.update_version('r1', 1, 'carol', ConflictResolutionStrategy.LAST_WRITER_WINS); print(f'resolved={result.resolved}'); stamp = svc.get_version_stamp('r1'); print(f'version={stamp.version}')
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    sessions=2
+    Should Contain    ${result.stdout}    alice_write=True
+    Should Contain    ${result.stdout}    carol_write=False
+    Should Contain    ${result.stdout}    conflict=True
+    Should Contain    ${result.stdout}    resolved=True
+    Should Contain    ${result.stdout}    version=3
+
+VersionConflictError With Reject Strategy
+    [Documentation]    Verify reject strategy raises VersionConflictError on conflict.
+    ${result}=    Run Python Script
+    ...    from cleveragents.application.services.team_collab_service import TeamCollaborationService; from cleveragents.domain.models.core.team_collab import TeamRole, ConflictResolutionStrategy, VersionConflictError; svc = TeamCollaborationService(); svc.add_member('alice', 'Alice', 'alice@x.com', TeamRole.OWNER); svc.create_version_stamp('r1', 'alice'); svc.update_version('r1', 1, 'alice');\ntry:\n\ \ \ \ svc.update_version('r1', 1, 'alice', ConflictResolutionStrategy.REJECT)\n\ \ \ \ print('NO_ERROR')\nexcept VersionConflictError:\n\ \ \ \ print('CONFLICT_ERROR')
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    CONFLICT_ERROR
+
+Error Hierarchy Verification
+    [Documentation]    Verify error class hierarchy.
+    ${result}=    Run Python Script
+    ...    from cleveragents.domain.models.core.team_collab import TeamCollaborationError, TeamMemberNotFoundError, TeamMemberExistsError, VersionConflictError; print(issubclass(TeamMemberNotFoundError, TeamCollaborationError), issubclass(TeamMemberExistsError, TeamCollaborationError), issubclass(VersionConflictError, TeamCollaborationError))
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    True True True
+
+Remove Member Cleans Up Sessions
+    [Documentation]    Verify removing a member cleans up their sessions.
+    ${result}=    Run Python Script
+    ...    from cleveragents.application.services.team_collab_service import TeamCollaborationService; from cleveragents.domain.models.core.team_collab import TeamRole; svc = TeamCollaborationService(); svc.add_member('alice', 'Alice', 'alice@x.com', TeamRole.OWNER); svc.register_session('alice', 'p1'); print(f'before={len(svc.get_active_sessions_for_project("p1"))}'); svc.remove_member('alice'); print(f'after={len(svc.get_active_sessions_for_project("p1"))}')
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    before=1
+    Should Contain    ${result.stdout}    after=0
+
+Enforce Permission Raises PermissionError
+    [Documentation]    Verify enforce_permission raises PermissionError for insufficient role.
+    ${result}=    Run Python Script
+    ...    from cleveragents.application.services.team_collab_service import TeamCollaborationService; from cleveragents.domain.models.core.team_collab import TeamRole, TeamPermission;\nsvc = TeamCollaborationService()\nsvc.add_member('viewer', 'Viewer', 'v@x.com', TeamRole.VIEWER)\ntry:\n\ \ \ \ svc.enforce_permission('viewer', TeamPermission.WRITE)\n\ \ \ \ print('NO_ERROR')\nexcept PermissionError:\n\ \ \ \ print('PERM_ERROR')
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    PERM_ERROR
+
+Merge Strategy Resolves Conflict
+    [Documentation]    Verify merge strategy resolves version conflicts.
+    ${result}=    Run Python Script
+    ...    from cleveragents.application.services.team_collab_service import TeamCollaborationService; from cleveragents.domain.models.core.team_collab import TeamRole, ConflictResolutionStrategy; svc = TeamCollaborationService(); svc.add_member('alice', 'Alice', 'alice@x.com', TeamRole.OWNER); svc.create_version_stamp('r1', 'alice'); svc.update_version('r1', 1, 'alice'); result = svc.update_version('r1', 1, 'alice', ConflictResolutionStrategy.MERGE); print(f'detected={result.detected},resolved={result.resolved},strategy={result.strategy}')
+    Should Be Equal As Strings    ${result.rc}    0
+    Should Contain    ${result.stdout}    detected=True,resolved=True,strategy=merge
diff --git a/src/cleveragents/application/services/team_collab_service.py b/src/cleveragents/application/services/team_collab_service.py
new file mode 100644
index 000000000..eede989ff
--- /dev/null
+++ b/src/cleveragents/application/services/team_collab_service.py
@@ -0,0 +1,507 @@
+"""Team collaboration service for CleverAgents.
+
+Orchestrates multi-user connection handling, role-based access control,
+concurrent session management, and optimistic-locking conflict resolution
+for shared project access.
+
+The ``TeamCollaborationService`` is the primary entry point for all
+team collaboration operations.  It delegates to the domain models in
+``cleveragents.domain.models.core.team_collab`` and coordinates
+session registration, permission checks, and version-stamp management.
+
+Based on issue #863 — feat(server): team collaboration features.
+"""
+
+from __future__ import annotations
+
+import structlog
+
+from cleveragents.domain.models.core.team_collab import (
+    ConflictDetectionResult,
+    ConflictResolutionStrategy,
+    SessionRegistry,
+    TeamMember,
+    TeamMemberExistsError,
+    TeamMemberNotFoundError,
+    TeamPermission,
+    TeamRole,
+    UserSession,
+    VersionConflictError,
+    VersionStamp,
+)
+
+__all__ = [
+    "TeamCollaborationService",
+]
+
+_logger = structlog.get_logger(__name__)
+
+
+class TeamCollaborationService:
+    """Central service for team collaboration features.
+
+    Manages team membership, concurrent user sessions, permission
+    enforcement, and optimistic-locking conflict resolution.
+
+    All operations are designed for concurrent access.  The underlying
+    ``SessionRegistry`` uses thread-safe locking; the version-stamp
+    registry uses an internal dict guarded by the same concurrency
+    guarantees.
+    """
+
+    def __init__(self) -> None:
+        """Initialise the service with empty registries."""
+        self._members: dict[str, TeamMember] = {}
+        self._session_registry = SessionRegistry()
+        self._version_stamps: dict[str, VersionStamp] = {}
+        self._logger = _logger.bind(service="team_collab")
+
+    # ------------------------------------------------------------------
+    # Properties
+    # ------------------------------------------------------------------
+
+    @property
+    def session_registry(self) -> SessionRegistry:
+        """Return the underlying session registry."""
+        return self._session_registry
+
+    # ------------------------------------------------------------------
+    # Team member management
+    # ------------------------------------------------------------------
+
+    def add_member(
+        self,
+        user_id: str,
+        display_name: str,
+        email: str,
+        role: TeamRole,
+    ) -> TeamMember:
+        """Add a new member to the team.
+
+        Args:
+            user_id: Unique user identifier.
+            display_name: Human-readable name.
+            email: Email address.
+            role: Team role for the new member.
+
+        Returns:
+            The newly created ``TeamMember``.
+
+        Raises:
+            ValueError: If any argument is empty.
+            TeamMemberExistsError: If the user is already a member.
+        """
+        if not user_id or not user_id.strip():
+            raise ValueError("user_id must be a non-empty string")
+        if not display_name or not display_name.strip():
+            raise ValueError("display_name must be a non-empty string")
+        if not email or not email.strip():
+            raise ValueError("email must be a non-empty string")
+
+        if user_id in self._members:
+            raise TeamMemberExistsError(f"User '{user_id}' is already a team member")
+
+        member = TeamMember(
+            user_id=user_id,
+            display_name=display_name,
+            email=email,
+            role=role,
+        )
+        self._members[user_id] = member
+        self._logger.info(
+            "member_added",
+            user_id=user_id,
+            role=str(role),
+        )
+        return member
+
+    def remove_member(self, user_id: str) -> bool:
+        """Remove a member from the team.
+
+        Also unregisters all active sessions for the removed user.
+
+        Args:
+            user_id: The user to remove.
+
+        Returns:
+            ``True`` if the member was found and removed.
+
+        Raises:
+            ValueError: If *user_id* is empty.
+            TeamMemberNotFoundError: If the user is not a member.
+        """
+        if not user_id or not user_id.strip():
+            raise ValueError("user_id must be a non-empty string")
+
+        if user_id not in self._members:
+            raise TeamMemberNotFoundError(f"User '{user_id}' is not a team member")
+
+        # Clean up active sessions for the removed user
+        user_sessions = self._session_registry.get_sessions_for_user(user_id)
+        for session in user_sessions:
+            self._session_registry.unregister(session.session_id)
+
+        del self._members[user_id]
+        self._logger.info("member_removed", user_id=user_id)
+        return True
+
+    def get_member(self, user_id: str) -> TeamMember:
+        """Retrieve a team member by user ID.
+
+        Args:
+            user_id: The user to look up.
+
+        Returns:
+            The ``TeamMember`` instance.
+
+        Raises:
+            ValueError: If *user_id* is empty.
+            TeamMemberNotFoundError: If the user is not a member.
+        """
+        if not user_id or not user_id.strip():
+            raise ValueError("user_id must be a non-empty string")
+
+        member = self._members.get(user_id)
+        if member is None:
+            raise TeamMemberNotFoundError(f"User '{user_id}' is not a team member")
+        return member
+
+    def list_members(self) -> list[TeamMember]:
+        """Return all team members.
+
+        Returns:
+            A list of all ``TeamMember`` instances.
+        """
+        return list(self._members.values())
+
+    def update_member_role(self, user_id: str, new_role: TeamRole) -> TeamMember:
+        """Change a member's role.
+
+        Args:
+            user_id: The user whose role to change.
+            new_role: The new role to assign.
+
+        Returns:
+            The updated ``TeamMember``.
+
+        Raises:
+            ValueError: If *user_id* is empty.
+            TeamMemberNotFoundError: If the user is not a member.
+        """
+        if not user_id or not user_id.strip():
+            raise ValueError("user_id must be a non-empty string")
+
+        member = self.get_member(user_id)
+        member.role = new_role
+        self._logger.info(
+            "member_role_updated",
+            user_id=user_id,
+            new_role=str(new_role),
+        )
+        return member
+
+    # ------------------------------------------------------------------
+    # Permission checks
+    # ------------------------------------------------------------------
+
+    def check_permission(
+        self,
+        user_id: str,
+        permission: TeamPermission,
+    ) -> bool:
+        """Check whether a user has a specific permission.
+
+        Args:
+            user_id: The user to check.
+            permission: The permission to verify.
+
+        Returns:
+            ``True`` if the user's role grants the permission.
+
+        Raises:
+            ValueError: If *user_id* is empty.
+            TeamMemberNotFoundError: If the user is not a member.
+        """
+        member = self.get_member(user_id)
+        return member.has_permission(permission)
+
+    def enforce_permission(
+        self,
+        user_id: str,
+        permission: TeamPermission,
+    ) -> None:
+        """Enforce that a user has a specific permission.
+
+        Args:
+            user_id: The user to check.
+            permission: The required permission.
+
+        Raises:
+            ValueError: If *user_id* is empty.
+            TeamMemberNotFoundError: If the user is not a member.
+            PermissionError: If the user lacks the permission.
+        """
+        if not self.check_permission(user_id, permission):
+            member = self.get_member(user_id)
+            raise PermissionError(
+                f"User '{user_id}' (role={member.role}) lacks '{permission}' permission"
+            )
+
+    # ------------------------------------------------------------------
+    # Session management
+    # ------------------------------------------------------------------
+
+    def register_session(
+        self,
+        user_id: str,
+        project_id: str,
+        metadata: dict[str, object] | None = None,
+    ) -> UserSession:
+        """Register a new session for a team member.
+
+        Args:
+            user_id: The user establishing the session.
+            project_id: The project the session is scoped to.
+            metadata: Optional session metadata.
+
+        Returns:
+            The newly created ``UserSession``.
+
+        Raises:
+            ValueError: If arguments are empty.
+            TeamMemberNotFoundError: If the user is not a member.
+        """
+        if not user_id or not user_id.strip():
+            raise ValueError("user_id must be a non-empty string")
+        if not project_id or not project_id.strip():
+            raise ValueError("project_id must be a non-empty string")
+
+        member = self.get_member(user_id)
+
+        session = UserSession(
+            session_id=self._session_registry.generate_session_id(),
+            user_id=user_id,
+            display_name=member.display_name,
+            project_id=project_id,
+            metadata=dict(metadata) if metadata else {},
+        )
+        self._session_registry.register(session)
+        self._logger.info(
+            "session_registered",
+            session_id=session.session_id,
+            user_id=user_id,
+            project_id=project_id,
+        )
+        return session
+
+    def unregister_session(self, session_id: str) -> bool:
+        """Unregister (disconnect) a session.
+
+        Args:
+            session_id: The session to remove.
+
+        Returns:
+            ``True`` if the session was found and removed.
+        """
+        if not session_id or not session_id.strip():
+            raise ValueError("session_id must be a non-empty string")
+        removed = self._session_registry.unregister(session_id)
+        if removed:
+            self._logger.info("session_unregistered", session_id=session_id)
+        return removed
+
+    def get_active_sessions_for_project(
+        self,
+        project_id: str,
+    ) -> list[UserSession]:
+        """Return all active sessions for a project.
+
+        Args:
+            project_id: The project to query.
+
+        Returns:
+            A list of active sessions.
+        """
+        if not project_id or not project_id.strip():
+            raise ValueError("project_id must be a non-empty string")
+        return self._session_registry.get_sessions_for_project(project_id)
+
+    # ------------------------------------------------------------------
+    # Version stamps and conflict resolution
+    # ------------------------------------------------------------------
+
+    def create_version_stamp(
+        self,
+        resource_id: str,
+        user_id: str,
+    ) -> VersionStamp:
+        """Create an initial version stamp for a resource.
+
+        Args:
+            resource_id: The resource to track.
+            user_id: The user creating the resource.
+
+        Returns:
+            A new ``VersionStamp`` at version 1.
+
+        Raises:
+            ValueError: If arguments are empty.
+        """
+        if not resource_id or not resource_id.strip():
+            raise ValueError("resource_id must be a non-empty string")
+        if not user_id or not user_id.strip():
+            raise ValueError("user_id must be a non-empty string")
+
+        stamp = VersionStamp(
+            resource_id=resource_id,
+            version=1,
+            updated_by=user_id,
+        )
+        self._version_stamps[resource_id] = stamp
+        self._logger.info(
+            "version_stamp_created",
+            resource_id=resource_id,
+            version=1,
+        )
+        return stamp
+
+    def get_version_stamp(self, resource_id: str) -> VersionStamp | None:
+        """Retrieve the current version stamp for a resource.
+
+        Args:
+            resource_id: The resource to look up.
+
+        Returns:
+            The current ``VersionStamp``, or ``None`` if untracked.
+        """
+        if not resource_id or not resource_id.strip():
+            raise ValueError("resource_id must be a non-empty string")
+        return self._version_stamps.get(resource_id)
+
+    def update_version(
+        self,
+        resource_id: str,
+        expected_version: int,
+        user_id: str,
+        strategy: ConflictResolutionStrategy = ConflictResolutionStrategy.REJECT,
+    ) -> ConflictDetectionResult:
+        """Attempt to update a resource's version stamp.
+
+        Implements optimistic locking: the caller provides the version
+        they last read (``expected_version``).  If the current version
+        matches, the stamp is incremented atomically.  If it does not
+        match, the configured ``strategy`` determines the outcome.
+
+        Args:
+            resource_id: The resource to update.
+            expected_version: The version the caller expects.
+            user_id: The user performing the update.
+            strategy: How to handle conflicts.
+
+        Returns:
+            A ``ConflictDetectionResult`` describing the outcome.
+
+        Raises:
+            ValueError: If arguments are invalid.
+            VersionConflictError: If strategy is ``reject`` and conflict found.
+        """
+        if not resource_id or not resource_id.strip():
+            raise ValueError("resource_id must be a non-empty string")
+        if not user_id or not user_id.strip():
+            raise ValueError("user_id must be a non-empty string")
+        if expected_version < 1:
+            raise ValueError("expected_version must be >= 1")
+
+        current = self._version_stamps.get(resource_id)
+        if current is None:
+            # Resource not tracked yet — create initial stamp
+            stamp = self.create_version_stamp(resource_id, user_id)
+            return ConflictDetectionResult(
+                detected=False,
+                resource_id=resource_id,
+                resolved=True,
+                details=f"New resource created at version {stamp.version}",
+            )
+
+        if current.version == expected_version:
+            # No conflict — increment
+            new_stamp = current.increment(user_id)
+            self._version_stamps[resource_id] = new_stamp
+            return ConflictDetectionResult(
+                detected=False,
+                resource_id=resource_id,
+                resolved=True,
+                details=f"Updated to version {new_stamp.version}",
+            )
+
+        # Conflict detected
+        self._logger.warning(
+            "version_conflict",
+            resource_id=resource_id,
+            expected=expected_version,
+            actual=current.version,
+            strategy=str(strategy),
+        )
+
+        if strategy == ConflictResolutionStrategy.LAST_WRITER_WINS:
+            new_stamp = current.increment(user_id)
+            self._version_stamps[resource_id] = new_stamp
+            return ConflictDetectionResult(
+                detected=True,
+                resource_id=resource_id,
+                strategy=strategy,
+                resolved=True,
+                details=(
+                    f"Conflict resolved (last-writer-wins): "
+                    f"version {current.version} -> {new_stamp.version}"
+                ),
+            )
+
+        if strategy == ConflictResolutionStrategy.MERGE:
+            # Merge strategy: increment but mark as merge-resolved
+            new_stamp = current.increment(user_id)
+            self._version_stamps[resource_id] = new_stamp
+            return ConflictDetectionResult(
+                detected=True,
+                resource_id=resource_id,
+                strategy=strategy,
+                resolved=True,
+                details=(
+                    f"Conflict resolved (merge): "
+                    f"version {current.version} -> {new_stamp.version}"
+                ),
+            )
+
+        # strategy == REJECT
+        raise VersionConflictError(
+            resource_id=resource_id,
+            expected_version=expected_version,
+            actual_version=current.version,
+        )
+
+    def detect_conflict(
+        self,
+        resource_id: str,
+        expected_version: int,
+    ) -> bool:
+        """Check whether a conflict exists without resolving it.
+
+        Args:
+            resource_id: The resource to check.
+            expected_version: The version the caller expects.
+
+        Returns:
+            ``True`` if a conflict is detected.
+
+        Raises:
+            ValueError: If arguments are invalid.
+        """
+        if not resource_id or not resource_id.strip():
+            raise ValueError("resource_id must be a non-empty string")
+        if expected_version < 1:
+            raise ValueError("expected_version must be >= 1")
+
+        current = self._version_stamps.get(resource_id)
+        if current is None:
+            return False
+        return current.version != expected_version
diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py
index b7460c461..f6c3eb552 100644
--- a/src/cleveragents/domain/models/core/__init__.py
+++ b/src/cleveragents/domain/models/core/__init__.py
@@ -301,6 +301,21 @@ from cleveragents.domain.models.core.skill import (
     SkillResolver,
     SkillToolRef,
 )
+from cleveragents.domain.models.core.team_collab import (
+    TEAM_ROLE_PERMISSIONS,
+    ConflictDetectionResult,
+    ConflictResolutionStrategy,
+    SessionRegistry,
+    TeamCollaborationError,
+    TeamMember,
+    TeamMemberExistsError,
+    TeamMemberNotFoundError,
+    TeamPermission,
+    TeamRole,
+    UserSession,
+    VersionConflictError,
+    VersionStamp,
+)
 from cleveragents.domain.models.core.tool import (
     BindingMode,
     CheckpointScope,
@@ -335,6 +350,7 @@ __all__ = [
     "DEFAULT_PROVIDER_RETRY",
     "DEFAULT_SAFETY_PROFILE",
     "ROLE_PERMISSIONS",
+    "TEAM_ROLE_PERMISSIONS",
     "VALID_JOB_TRANSITIONS",
     "VALID_PHASES",
     "ActionState",
@@ -365,6 +381,8 @@ __all__ = [
     "CircuitBreakerConfig",
     "CloudBillingFields",
     "ConfidenceFactors",
+    "ConflictDetectionResult",
+    "ConflictResolutionStrategy",
     "Context",
     "ContextBudget",
     "ContextConfig",
@@ -513,6 +531,7 @@ __all__ = [
     "SessionImportError",
     "SessionMessage",
     "SessionNotFoundError",
+    "SessionRegistry",
     "SessionService",
     "SessionServiceError",
     "SessionTokenUsage",
@@ -527,6 +546,12 @@ __all__ = [
     "SkillToolRef",
     "SpecChangeSet",
     "SummaryForUpdateContextParams",
+    "TeamCollaborationError",
+    "TeamMember",
+    "TeamMemberExistsError",
+    "TeamMemberNotFoundError",
+    "TeamPermission",
+    "TeamRole",
     "TemporalScope",
     "TextMatchEvaluator",
     "ThreadSafeOrgCostAccumulator",
@@ -541,8 +566,11 @@ __all__ = [
     "UKOPrefix",
     "UKOVersion",
     "User",
+    "UserSession",
     "Validation",
     "ValidationMode",
+    "VersionConflictError",
+    "VersionStamp",
     "build_provenance_map",
     "can_transition",
     "can_transition_job",
diff --git a/src/cleveragents/domain/models/core/team_collab.py b/src/cleveragents/domain/models/core/team_collab.py
new file mode 100644
index 000000000..f32d2e846
--- /dev/null
+++ b/src/cleveragents/domain/models/core/team_collab.py
@@ -0,0 +1,538 @@
+"""Team collaboration domain models for CleverAgents.
+
+Multi-user session management, role-based access control, concurrent
+session tracking, and optimistic-locking conflict resolution for
+shared project access.
+
+## Collaboration Model
+
+Teams are collections of users who share access to projects.  Each
+member has a ``TeamRole`` that determines which ``TeamPermission``
+actions they may perform.  Active sessions are tracked in a
+``SessionRegistry`` so that concurrent edits can be detected and
+resolved via configurable ``ConflictResolutionStrategy`` policies.
+
+## Role Hierarchy
+
+| Role    | Permissions                           |
+|---------|---------------------------------------|
+| owner   | read, write, admin, manage_members    |
+| admin   | read, write, admin, manage_members    |
+| member  | read, write                           |
+| viewer  | read                                  |
+
+Based on issue #863 — feat(server): team collaboration features.
+"""
+
+from __future__ import annotations
+
+import threading
+from datetime import datetime
+from enum import StrEnum
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+from ulid import ULID
+
+__all__ = [
+    "TEAM_ROLE_PERMISSIONS",
+    "ConflictDetectionResult",
+    "ConflictResolutionStrategy",
+    "SessionRegistry",
+    "TeamCollaborationError",
+    "TeamMember",
+    "TeamMemberExistsError",
+    "TeamMemberNotFoundError",
+    "TeamPermission",
+    "TeamRole",
+    "UserSession",
+    "VersionConflictError",
+    "VersionStamp",
+]
+
+
+# ---------------------------------------------------------------------------
+# Enums
+# ---------------------------------------------------------------------------
+
+
+class TeamRole(StrEnum):
+    """Role of a user within a collaboration team.
+
+    Determines the set of ``TeamPermission`` actions the user may
+    perform on shared resources.
+    """
+
+    OWNER = "owner"
+    ADMIN = "admin"
+    MEMBER = "member"
+    VIEWER = "viewer"
+
+
+class TeamPermission(StrEnum):
+    """Actions gated by team role membership."""
+
+    READ = "read"
+    WRITE = "write"
+    ADMIN = "admin"
+    MANAGE_MEMBERS = "manage_members"
+
+
+class ConflictResolutionStrategy(StrEnum):
+    """Strategy for resolving concurrent edit conflicts.
+
+    - ``last_writer_wins``: The most recent write silently overwrites.
+    - ``reject``: The conflicting write is rejected with an error.
+    - ``merge``: Both changes are merged (requires caller support).
+    """
+
+    LAST_WRITER_WINS = "last_writer_wins"
+    REJECT = "reject"
+    MERGE = "merge"
+
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+TEAM_ROLE_PERMISSIONS: dict[TeamRole, frozenset[TeamPermission]] = {
+    TeamRole.OWNER: frozenset(TeamPermission),
+    TeamRole.ADMIN: frozenset(
+        {
+            TeamPermission.READ,
+            TeamPermission.WRITE,
+            TeamPermission.ADMIN,
+            TeamPermission.MANAGE_MEMBERS,
+        }
+    ),
+    TeamRole.MEMBER: frozenset(
+        {
+            TeamPermission.READ,
+            TeamPermission.WRITE,
+        }
+    ),
+    TeamRole.VIEWER: frozenset(
+        {
+            TeamPermission.READ,
+        }
+    ),
+}
+"""Mapping of each team role to its set of allowed permissions.
+
+OWNER and ADMIN currently grant identical permissions.  The semantic
+distinction mirrors the ``PermissionRole`` design --- OWNER denotes the
+team/project creator while ADMIN is a delegated role.
+"""
+
+
+# ---------------------------------------------------------------------------
+# Errors
+# ---------------------------------------------------------------------------
+
+
+class TeamCollaborationError(Exception):
+    """Base error for team collaboration operations."""
+
+
+class TeamMemberNotFoundError(TeamCollaborationError):
+    """Raised when a team member is not found."""
+
+
+class TeamMemberExistsError(TeamCollaborationError):
+    """Raised when attempting to add a member who already exists."""
+
+
+class VersionConflictError(TeamCollaborationError):
+    """Raised when an optimistic-locking version conflict is detected.
+
+    Attributes:
+        resource_id: The resource that was modified concurrently.
+        expected_version: The version the caller expected.
+        actual_version: The current version in the registry.
+    """
+
+    def __init__(
+        self,
+        resource_id: str,
+        expected_version: int,
+        actual_version: int,
+    ) -> None:
+        if not resource_id or not resource_id.strip():
+            raise ValueError("resource_id must be a non-empty string")
+        if expected_version < 0:
+            raise ValueError("expected_version must be >= 0")
+        if actual_version < 0:
+            raise ValueError("actual_version must be >= 0")
+        self.resource_id = resource_id
+        self.expected_version = expected_version
+        self.actual_version = actual_version
+        super().__init__(
+            f"Version conflict on resource '{resource_id}': "
+            f"expected version {expected_version}, "
+            f"actual version {actual_version}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# TeamMember
+# ---------------------------------------------------------------------------
+
+
+class TeamMember(BaseModel):
+    """A user participating in a collaboration team.
+
+    Each member has a unique ``user_id``, a display name, an email
+    address, and a ``TeamRole`` controlling their permissions.
+    """
+
+    user_id: str = Field(
+        ...,
+        min_length=1,
+        description="Unique identifier for the team member.",
+    )
+    display_name: str = Field(
+        ...,
+        min_length=1,
+        description="Human-readable display name.",
+    )
+    email: str = Field(
+        ...,
+        min_length=3,
+        description="Email address of the team member.",
+    )
+    role: TeamRole = Field(
+        ...,
+        description="Team role determining permissions.",
+    )
+    joined_at: datetime = Field(
+        default_factory=datetime.now,
+        description="When the member joined the team.",
+    )
+
+    @field_validator("email")
+    @classmethod
+    def validate_email_format(cls: type[TeamMember], v: str) -> str:
+        """Basic email format validation (must contain '@')."""
+        if "@" not in v:
+            raise ValueError("email must contain '@'")
+        return v
+
+    def has_permission(self, permission: TeamPermission) -> bool:
+        """Check whether this member's role grants *permission*.
+
+        Args:
+            permission: The permission to check.
+
+        Returns:
+            ``True`` if the member's role includes the permission.
+        """
+        allowed = TEAM_ROLE_PERMISSIONS.get(self.role, frozenset())
+        return permission in allowed
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# UserSession
+# ---------------------------------------------------------------------------
+
+
+class UserSession(BaseModel):
+    """An active user session for concurrent project access.
+
+    Tracks which user is connected, which project they are working on,
+    and session lifecycle timestamps.
+    """
+
+    session_id: str = Field(
+        ...,
+        min_length=1,
+        description="Unique session identifier (ULID).",
+    )
+    user_id: str = Field(
+        ...,
+        min_length=1,
+        description="Identifier of the connected user.",
+    )
+    display_name: str = Field(
+        ...,
+        min_length=1,
+        description="Display name of the connected user.",
+    )
+    project_id: str = Field(
+        ...,
+        min_length=1,
+        description="Project the session is scoped to.",
+    )
+    started_at: datetime = Field(
+        default_factory=datetime.now,
+        description="When the session was established.",
+    )
+    last_active: datetime = Field(
+        default_factory=datetime.now,
+        description="When the session was last active.",
+    )
+    is_active: bool = Field(
+        default=True,
+        description="Whether the session is currently active.",
+    )
+    metadata: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Optional session metadata.",
+    )
+
+    def deactivate(self) -> None:
+        """Mark the session as inactive."""
+        self.is_active = False
+
+    def touch(self) -> None:
+        """Update the last-active timestamp to now."""
+        self.last_active = datetime.now()
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# VersionStamp
+# ---------------------------------------------------------------------------
+
+
+class VersionStamp(BaseModel):
+    """Optimistic-locking version stamp for a shared resource.
+
+    Each write increments the ``version`` counter.  Callers present
+    their last-known version when writing; a mismatch indicates a
+    concurrent modification.
+    """
+
+    resource_id: str = Field(
+        ...,
+        min_length=1,
+        description="Identifier of the versioned resource.",
+    )
+    version: int = Field(
+        default=1,
+        ge=1,
+        description="Monotonically increasing version counter.",
+    )
+    updated_by: str = Field(
+        ...,
+        min_length=1,
+        description="User who last modified the resource.",
+    )
+    updated_at: datetime = Field(
+        default_factory=datetime.now,
+        description="When the resource was last modified.",
+    )
+
+    def increment(self, updated_by: str) -> VersionStamp:
+        """Return a new stamp with an incremented version.
+
+        Args:
+            updated_by: The user performing the update.
+
+        Returns:
+            A new ``VersionStamp`` with ``version + 1``.
+
+        Raises:
+            ValueError: If *updated_by* is empty.
+        """
+        if not updated_by or not updated_by.strip():
+            raise ValueError("updated_by must be a non-empty string")
+        return VersionStamp(
+            resource_id=self.resource_id,
+            version=self.version + 1,
+            updated_by=updated_by,
+            updated_at=datetime.now(),
+        )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# ConflictDetectionResult
+# ---------------------------------------------------------------------------
+
+
+class ConflictDetectionResult(BaseModel):
+    """Result of a conflict detection check.
+
+    Encapsulates whether a conflict was detected and, if resolved,
+    which strategy was applied and the outcome.
+    """
+
+    detected: bool = Field(
+        ...,
+        description="Whether a version conflict was detected.",
+    )
+    resource_id: str = Field(
+        ...,
+        min_length=1,
+        description="Identifier of the checked resource.",
+    )
+    strategy: ConflictResolutionStrategy | None = Field(
+        default=None,
+        description="Strategy applied (None if no conflict).",
+    )
+    resolved: bool = Field(
+        default=False,
+        description="Whether the conflict was successfully resolved.",
+    )
+    details: str = Field(
+        default="",
+        description="Human-readable description of the outcome.",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# SessionRegistry
+# ---------------------------------------------------------------------------
+
+
+class SessionRegistry:
+    """Thread-safe registry of active user sessions.
+
+    Tracks concurrent sessions and provides queries by user, project,
+    or session ID.  All mutating operations are guarded by a
+    ``threading.Lock`` for safe concurrent access.
+    """
+
+    def __init__(self) -> None:
+        self._sessions: dict[str, UserSession] = {}
+        self._lock = threading.Lock()
+
+    def register(self, session: UserSession) -> None:
+        """Register a new user session.
+
+        Args:
+            session: The session to register.
+
+        Raises:
+            ValueError: If a session with the same ID already exists.
+        """
+        if not session.session_id or not session.session_id.strip():
+            raise ValueError("session_id must be a non-empty string")
+        with self._lock:
+            if session.session_id in self._sessions:
+                raise ValueError(
+                    f"Session '{session.session_id}' is already registered"
+                )
+            self._sessions[session.session_id] = session
+
+    def unregister(self, session_id: str) -> bool:
+        """Remove a session from the registry.
+
+        Args:
+            session_id: The session to remove.
+
+        Returns:
+            ``True`` if the session was found and removed.
+        """
+        if not session_id or not session_id.strip():
+            raise ValueError("session_id must be a non-empty string")
+        with self._lock:
+            if session_id in self._sessions:
+                self._sessions[session_id].deactivate()
+                del self._sessions[session_id]
+                return True
+            return False
+
+    def get(self, session_id: str) -> UserSession | None:
+        """Retrieve a session by ID.
+
+        Args:
+            session_id: The session to look up.
+
+        Returns:
+            The session, or ``None`` if not found.
+        """
+        if not session_id or not session_id.strip():
+            raise ValueError("session_id must be a non-empty string")
+        with self._lock:
+            return self._sessions.get(session_id)
+
+    def get_active_sessions(self) -> list[UserSession]:
+        """Return all currently active sessions.
+
+        Returns:
+            A list of active ``UserSession`` instances.
+        """
+        with self._lock:
+            return [s for s in self._sessions.values() if s.is_active]
+
+    def get_sessions_for_user(self, user_id: str) -> list[UserSession]:
+        """Return all sessions belonging to a specific user.
+
+        Args:
+            user_id: The user whose sessions to retrieve.
+
+        Returns:
+            A list of sessions for the given user.
+        """
+        if not user_id or not user_id.strip():
+            raise ValueError("user_id must be a non-empty string")
+        with self._lock:
+            return [
+                s
+                for s in self._sessions.values()
+                if s.user_id == user_id and s.is_active
+            ]
+
+    def get_sessions_for_project(self, project_id: str) -> list[UserSession]:
+        """Return all active sessions scoped to a project.
+
+        Args:
+            project_id: The project to query.
+
+        Returns:
+            A list of active sessions for the given project.
+        """
+        if not project_id or not project_id.strip():
+            raise ValueError("project_id must be a non-empty string")
+        with self._lock:
+            return [
+                s
+                for s in self._sessions.values()
+                if s.project_id == project_id and s.is_active
+            ]
+
+    @property
+    def active_count(self) -> int:
+        """Return the number of active sessions."""
+        with self._lock:
+            return sum(1 for s in self._sessions.values() if s.is_active)
+
+    def clear(self) -> int:
+        """Remove all sessions.
+
+        Returns:
+            The number of sessions that were removed.
+        """
+        with self._lock:
+            count = len(self._sessions)
+            for session in self._sessions.values():
+                session.deactivate()
+            self._sessions.clear()
+            return count
+
+    def generate_session_id(self) -> str:
+        """Generate a new unique session ID (ULID).
+
+        Returns:
+            A ULID string suitable for ``UserSession.session_id``.
+        """
+        return str(ULID())
diff --git a/vulture_whitelist.py b/vulture_whitelist.py
index 0a8dfe0b9..86428ec19 100644
--- a/vulture_whitelist.py
+++ b/vulture_whitelist.py
@@ -1224,3 +1224,23 @@ health_endpoint  # noqa: B018, F821
 a2a_endpoint  # noqa: B018, F821
 server_start  # noqa: B018, F821
 run_server  # noqa: B018, F821
+
+# Team collaboration module — public API entry points
+TeamRole  # noqa: B018, F821
+TeamPermission  # noqa: B018, F821
+ConflictResolutionStrategy  # noqa: B018, F821
+TEAM_ROLE_PERMISSIONS  # noqa: B018, F821
+TeamMember  # noqa: B018, F821
+TeamMemberExistsError  # noqa: B018, F821
+TeamMemberNotFoundError  # noqa: B018, F821
+TeamCollaborationError  # noqa: B018, F821
+VersionConflictError  # noqa: B018, F821
+VersionStamp  # noqa: B018, F821
+UserSession  # noqa: B018, F821
+SessionRegistry  # noqa: B018, F821
+ConflictDetectionResult  # noqa: B018, F821
+TeamCollaborationService  # noqa: B018, F821
+generate_session_id  # noqa: B018, F821
+
+# Pydantic @classmethod validator parameter required by decorator protocol
+cls  # noqa: B018, F821