diff --git a/api/cohorts/migrations/0002_cohort_deletion_requested_at.py b/api/cohorts/migrations/0002_cohort_deletion_requested_at.py
new file mode 100644
index 000000000000..4de64bb770b4
--- /dev/null
+++ b/api/cohorts/migrations/0002_cohort_deletion_requested_at.py
@@ -0,0 +1,18 @@
+# Generated by Django 5.2.16 on 2026-08-06 07:03
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cohorts", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="cohort",
+            name="deletion_requested_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+    ]
diff --git a/api/cohorts/models.py b/api/cohorts/models.py
index cb2e9f910dd8..605b67be0526 100644
--- a/api/cohorts/models.py
+++ b/api/cohorts/models.py
@@ -26,6 +26,9 @@ class Cohort(SoftDeleteExportableModel):
     )
     version = models.PositiveIntegerField(default=0)
     created_at = models.DateTimeField(auto_now_add=True)
+    # Deletion drains memberships from the identity store first; the cohort is
+    # only soft-deleted once drained. This marks it as awaiting that final step.
+    deletion_requested_at = models.DateTimeField(null=True, blank=True)
 
     @property
     def system_trait_key(self) -> str:
diff --git a/api/cohorts/permissions.py b/api/cohorts/permissions.py
new file mode 100644
index 000000000000..b4d3aacfc9c2
--- /dev/null
+++ b/api/cohorts/permissions.py
@@ -0,0 +1,48 @@
+from common.environments.permissions import VIEW_ENVIRONMENT
+from common.projects.permissions import MANAGE_SEGMENTS
+from rest_framework.permissions import BasePermission
+from rest_framework.request import Request
+from rest_framework.views import APIView
+
+from environments.models import Environment
+from organisations.subscriptions.constants import SubscriptionPlanFamily
+from organisations.subscriptions.permissions import require_minimum_plan
+from users.models import FFAdminUser
+
+_READ_ACTIONS = ("list", "retrieve")
+
+_MinimumStartupPlan = require_minimum_plan(SubscriptionPlanFamily.START_UP)
+
+
+class CohortPlanPermission(_MinimumStartupPlan):  # type: ignore[misc,valid-type]
+    def has_permission(self, request: Request, view: APIView) -> bool:
+        try:
+            environment = Environment.objects.get(
+                api_key=view.kwargs.get("environment_api_key")
+            )
+        except Environment.DoesNotExist:
+            return False
+        # The base class reads the organisation from an `organisation` request
+        # param our URLs don't carry; the project provides it instead.
+        return bool(super().has_object_permission(request, view, environment.project))
+
+    def has_object_permission(
+        self, request: Request, view: APIView, obj: object
+    ) -> bool:
+        # DRF hands us a Cohort here, which doesn't carry an organisation;
+        # re-run the environment-based check instead.
+        return self.has_permission(request, view)
+
+
+class CohortPermission(BasePermission):
+    def has_permission(self, request: Request, view: APIView) -> bool:
+        try:
+            environment = Environment.objects.get(
+                api_key=view.kwargs.get("environment_api_key")
+            )
+        except Environment.DoesNotExist:
+            return False
+        user: FFAdminUser = request.user  # type: ignore[assignment]
+        if getattr(view, "action", None) in _READ_ACTIONS:
+            return user.has_environment_permission(VIEW_ENVIRONMENT, environment)
+        return user.has_project_permission(MANAGE_SEGMENTS, environment.project)
diff --git a/api/cohorts/serializers.py b/api/cohorts/serializers.py
new file mode 100644
index 000000000000..780216bb9f4a
--- /dev/null
+++ b/api/cohorts/serializers.py
@@ -0,0 +1,35 @@
+import typing
+
+from rest_framework import serializers
+
+from cohorts.models import Cohort
+from cohorts.services import create_cohort
+
+
+class CohortSerializer(serializers.ModelSerializer[Cohort]):
+    name = serializers.CharField(max_length=2000, source="segment.name")
+    description = serializers.CharField(
+        source="segment.description", required=False, allow_null=True
+    )
+
+    class Meta:
+        model = Cohort
+        fields = (
+            "id",
+            "uuid",
+            "name",
+            "description",
+            "segment",
+            "source_type",
+            "version",
+            "created_at",
+        )
+        read_only_fields = ("segment", "source_type", "version", "created_at")
+
+    def create(self, validated_data: dict[str, typing.Any]) -> Cohort:
+        segment_data = validated_data["segment"]
+        return create_cohort(
+            environment=validated_data["environment"],
+            name=segment_data["name"],
+            description=segment_data.get("description"),
+        )
diff --git a/api/cohorts/services.py b/api/cohorts/services.py
index 91e9ba3bcc98..c2ba9f3bd111 100644
--- a/api/cohorts/services.py
+++ b/api/cohorts/services.py
@@ -1,11 +1,22 @@
+import typing
+
 import structlog
+from django.db import transaction
 from django.db.models import QuerySet
 from django.utils import timezone
+from flag_engine.segments.constants import IS_SET
 
 from cohorts.constants import COHORT_MEMBERSHIP_APPLY_BATCH_SIZE
 from cohorts.metrics import flagsmith_cohorts_membership_deltas_applied_total
 from cohorts.models import Cohort, CohortMembership, CohortMembershipState
+from core.dataclasses import AuthorData
 from environments.dynamodb import DynamoIdentityWrapper
+from segments.models import Condition, Segment, SegmentManagedBy, SegmentRule
+from segments.services import delete_segment
+
+if typing.TYPE_CHECKING:
+    from environments.models import Environment
+    from projects.models import Project
 
 logger = structlog.get_logger("cohorts")
 
@@ -66,3 +77,68 @@ def apply_pending_memberships(cohort: Cohort) -> bool:
             removes__count=removed_count,
         )
     return pending_memberships(cohort).exists()
+
+
+def create_cohort(
+    *,
+    environment: "Environment",
+    name: str,
+    description: str | None = None,
+) -> Cohort:
+    with transaction.atomic():
+        segment = Segment.objects.create(
+            name=name,
+            project=environment.project,
+            description=description,
+            managed_by=SegmentManagedBy.COHORT,
+        )
+        rule = SegmentRule.objects.create(segment=segment, type=SegmentRule.ALL_RULE)
+        cohort: Cohort = Cohort.objects.create(environment=environment, segment=segment)
+        Condition.objects.create(
+            rule=rule,
+            operator=IS_SET,
+            property=cohort.system_trait_key,
+            created_with_segment=True,
+        )
+    logger.info(
+        "cohort.created",
+        cohort__id=cohort.id,
+        segment__id=segment.id,
+        environment__id=environment.id,
+        project__id=environment.project_id,
+        organisation__id=environment.project.organisation_id,
+    )
+    return cohort
+
+
+def edge_sync_enabled(project: "Project") -> bool:
+    return bool(project.enable_dynamo_db and DynamoIdentityWrapper().is_enabled)
+
+
+def delete_cohort(cohort: Cohort) -> None:
+    from cohorts.tasks import apply_cohort_membership_deltas
+
+    with transaction.atomic():
+        cohort.deletion_requested_at = timezone.now()
+        cohort.save(update_fields=["deletion_requested_at"])
+        logger.info(
+            "cohort.deletion_requested",
+            cohort__id=cohort.id,
+            environment__id=cohort.environment_id,
+        )
+        CohortMembership.objects.filter(cohort=cohort).update(
+            state=CohortMembershipState.PENDING_REMOVE, updated_at=timezone.now()
+        )
+        apply_cohort_membership_deltas.delay(kwargs={"cohort_id": cohort.id})
+
+
+def finalise_cohort_deletion(cohort: Cohort) -> None:
+    segment = cohort.segment
+    with transaction.atomic():
+        cohort.delete()
+        delete_segment(segment, AuthorData())
+    logger.info(
+        "cohort.deleted",
+        cohort__id=cohort.id,
+        environment__id=cohort.environment_id,
+    )
diff --git a/api/cohorts/tasks.py b/api/cohorts/tasks.py
index b992672619df..6b98d6eb6cb8 100644
--- a/api/cohorts/tasks.py
+++ b/api/cohorts/tasks.py
@@ -11,7 +11,6 @@
     DYNAMODB_THROTTLING_ERROR_CODES,
 )
 from cohorts.models import Cohort
-from environments.dynamodb import DynamoIdentityWrapper
 
 logger = structlog.get_logger("cohorts")
 
@@ -22,15 +21,14 @@ def apply_cohort_membership_deltas(cohort_id: int) -> None:
     if (cohort := Cohort.objects.filter(id=cohort_id).first()) is None:
         log.info("membership.apply.skipped", reason="cohort_missing")
         return
-    if not (
-        cohort.environment.project.enable_dynamo_db
-        and DynamoIdentityWrapper().is_enabled
-    ):
+    if not services.edge_sync_enabled(cohort.environment.project):
         log.info("membership.apply.skipped", reason="not_edge")
         return
     try:
         for _ in range(COHORT_MEMBERSHIP_APPLY_MAX_BATCHES_PER_RUN):
             if not services.apply_pending_memberships(cohort):
+                if cohort.deletion_requested_at is not None:
+                    services.finalise_cohort_deletion(cohort)
                 return
     except ClientError as exc:
         if exc.response["Error"]["Code"] in DYNAMODB_THROTTLING_ERROR_CODES:
diff --git a/api/cohorts/urls.py b/api/cohorts/urls.py
new file mode 100644
index 000000000000..1cb2c65460ae
--- /dev/null
+++ b/api/cohorts/urls.py
@@ -0,0 +1,10 @@
+from rest_framework.routers import DefaultRouter
+
+from cohorts.views import CohortViewSet
+
+app_name = "cohorts"
+
+router = DefaultRouter()
+router.register(r"", CohortViewSet, basename="cohorts")
+
+urlpatterns = router.urls
diff --git a/api/cohorts/views.py b/api/cohorts/views.py
new file mode 100644
index 000000000000..eda7c209a3d9
--- /dev/null
+++ b/api/cohorts/views.py
@@ -0,0 +1,64 @@
+from django.db.models import QuerySet
+from drf_spectacular.utils import extend_schema, extend_schema_view
+from rest_framework import mixins, status
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from cohorts import services
+from cohorts.models import Cohort
+from cohorts.permissions import CohortPermission, CohortPlanPermission
+from cohorts.serializers import CohortSerializer
+from environments.views import NestedEnvironmentViewSet
+from projects.exceptions import DynamoNotEnabledError
+
+
+@extend_schema_view(
+    list=extend_schema(description="List the environment's cohorts."),
+    create=extend_schema(
+        description="Create a cohort and the managed segment that targets it."
+    ),
+    retrieve=extend_schema(description="Retrieve a cohort."),
+    destroy=extend_schema(
+        description=(
+            "Request cohort deletion. Memberships are drained from identity "
+            "data first; the cohort and its segment are deleted once drained."
+        ),
+        responses={202: None},
+    ),
+)
+class CohortViewSet(
+    NestedEnvironmentViewSet[Cohort],
+    mixins.ListModelMixin,
+    mixins.CreateModelMixin,
+    mixins.RetrieveModelMixin,
+    mixins.DestroyModelMixin,
+):
+    serializer_class = CohortSerializer
+    pagination_class = None
+    permission_classes = [IsAuthenticated, CohortPlanPermission, CohortPermission]
+    model_class = Cohort
+    lookup_field = "id"
+    lookup_url_kwarg = "cohort_id"
+
+    def initial(self, request: Request, *args: object, **kwargs: object) -> None:
+        super().initial(request, *args, **kwargs)
+        # Cohorts only sync to edge identities for now; core (Postgres
+        # identities) support comes later.
+        if not services.edge_sync_enabled(self._get_environment().project):
+            raise DynamoNotEnabledError()
+
+    def get_queryset(self) -> QuerySet[Cohort]:
+        # A cohort awaiting drain-then-delete is already gone from the
+        # user's point of view.
+        return (
+            super()
+            .get_queryset()
+            .filter(deletion_requested_at__isnull=True)
+            .select_related("segment")
+            .order_by("id")
+        )
+
+    def destroy(self, request: Request, *args: object, **kwargs: object) -> Response:
+        services.delete_cohort(self.get_object())
+        return Response(status=status.HTTP_202_ACCEPTED)
diff --git a/api/environments/urls.py b/api/environments/urls.py
index e0f66d90052b..ad0bebebf622 100644
--- a/api/environments/urls.py
+++ b/api/environments/urls.py
@@ -177,6 +177,10 @@
         "<str:environment_api_key>/warehouse-connections/",
         include("experimentation.urls"),
     ),
+    path(
+        "<str:environment_api_key>/cohorts/",
+        include("cohorts.urls"),
+    ),
     path(
         "<str:environment_api_key>/experiments/",
         include("experimentation.experiment_urls"),
diff --git a/api/segments/migrations/0031_segment_managed_by.py b/api/segments/migrations/0031_segment_managed_by.py
new file mode 100644
index 000000000000..7e413a253566
--- /dev/null
+++ b/api/segments/migrations/0031_segment_managed_by.py
@@ -0,0 +1,27 @@
+# Generated by Django 5.2.16 on 2026-08-11 07:43
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("segments", "0030_add_default_to_segment_version"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="historicalsegment",
+            name="managed_by",
+            field=models.CharField(
+                blank=True, choices=[("cohort", "Cohort")], default="", max_length=50
+            ),
+        ),
+        migrations.AddField(
+            model_name="segment",
+            name="managed_by",
+            field=models.CharField(
+                blank=True, choices=[("cohort", "Cohort")], default="", max_length=50
+            ),
+        ),
+    ]
diff --git a/api/segments/models.py b/api/segments/models.py
index 5d0ddeb0636f..aa771c6e1e81 100644
--- a/api/segments/models.py
+++ b/api/segments/models.py
@@ -77,6 +77,10 @@ class SegmentConditionManager(ConfiguredOrderManager["Condition"]):
     setting_name = "SEGMENT_CONDITIONS_EXPLICIT_ORDERING_ENABLED"
 
 
+class SegmentManagedBy(models.TextChoices):
+    COHORT = "cohort", "Cohort"
+
+
 class Segment(
     LifecycleModelMixin,  # type: ignore[misc]
     SoftDeleteExportableModel,
@@ -121,6 +125,12 @@ class Segment(
     created_at = models.DateTimeField(null=True, auto_now_add=True)
     updated_at = models.DateTimeField(null=True, auto_now=True)
     is_system_segment = models.BooleanField(default=False)
+    # A managed segment is created and maintained by another feature (e.g. a
+    # cohort). Unlike system segments it stays visible in the API, but the
+    # dashboard renders it differently and cannot edit it.
+    managed_by = models.CharField(
+        max_length=50, choices=SegmentManagedBy.choices, default="", blank=True
+    )
 
     objects = SegmentManager()  # type: ignore[misc]
 
diff --git a/api/segments/serializers.py b/api/segments/serializers.py
index 7bb06eaa0f85..eb0caee4ed92 100644
--- a/api/segments/serializers.py
+++ b/api/segments/serializers.py
@@ -126,8 +126,9 @@ class Meta:
             "rules",
             "metadata",
             "membership_counts",
+            "managed_by",
         ]
-        read_only_fields = ["membership_counts"]
+        read_only_fields = ["membership_counts", "managed_by"]
 
     def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
         attrs = super().validate(attrs)
diff --git a/api/segments/views.py b/api/segments/views.py
index b1dbafa4ed33..51a80de92eb7 100644
--- a/api/segments/views.py
+++ b/api/segments/views.py
@@ -214,6 +214,17 @@ def members(self, request: Request, *args: Any, **kwargs: Any) -> Response:
         next_cursor = members[-1]["identifier"] if has_more else None
         return Response({"results": members, "next_cursor": next_cursor})
 
+    def check_object_permissions(self, request: Request, obj: Segment) -> None:
+        super().check_object_permissions(request, obj)
+        if (
+            self.action in ("update", "partial_update", "destroy", "clone")
+            and obj.cohorts.exists()
+        ):
+            raise PermissionDenied(
+                "This segment is managed by a cohort and cannot be edited "
+                "or cloned directly."
+            )
+
     def destroy(self, request: Request, *args: Any, **kwargs: Any) -> Response:
         segment = self.get_object()
         author = AuthorData.from_request(request)
diff --git a/api/tests/unit/cohorts/test_permissions.py b/api/tests/unit/cohorts/test_permissions.py
new file mode 100644
index 000000000000..b1cc9a50748e
--- /dev/null
+++ b/api/tests/unit/cohorts/test_permissions.py
@@ -0,0 +1,15 @@
+from unittest.mock import MagicMock
+
+from cohorts.permissions import CohortPermission
+
+
+def test_cohort_permission__unknown_environment__returns_false(db: None) -> None:
+    # Given
+    permission = CohortPermission()
+    view = MagicMock(kwargs={"environment_api_key": "missing"})
+
+    # When
+    result = permission.has_permission(MagicMock(), view)
+
+    # Then
+    assert result is False
diff --git a/api/tests/unit/cohorts/test_services.py b/api/tests/unit/cohorts/test_services.py
index 1737caea6465..5bf7eb51b3cf 100644
--- a/api/tests/unit/cohorts/test_services.py
+++ b/api/tests/unit/cohorts/test_services.py
@@ -1,9 +1,16 @@
+from flag_engine.segments.constants import IS_SET
 from pytest_mock import MockerFixture
 from pytest_structlog import StructuredLogCapture
 
 from cohorts.models import Cohort, CohortMembership, CohortMembershipState
-from cohorts.services import apply_pending_memberships
+from cohorts.services import (
+    apply_pending_memberships,
+    create_cohort,
+    delete_cohort,
+)
 from environments.dynamodb import DynamoIdentityWrapper
+from environments.models import Environment
+from segments.models import SegmentManagedBy, SegmentRule
 
 
 def test_apply_pending_memberships__no_pending_rows__returns_false(
@@ -108,3 +115,68 @@ def transition_row(**kwargs: str) -> None:
     assert membership.state == CohortMembershipState.PENDING_REMOVE
     assert result is True
     assert not log.has("membership.applied")
+
+
+def test_create_cohort__valid_name__creates_segment_with_is_set_condition(
+    environment: Environment,
+) -> None:
+    # Given / When
+    cohort = create_cohort(environment=environment, name="Beta users")
+
+    # Then
+    segment = cohort.segment
+    assert segment.name == "Beta users"
+    assert segment.project == environment.project
+    assert segment.managed_by == SegmentManagedBy.COHORT
+    rule = segment.rules.get()
+    assert rule.type == SegmentRule.ALL_RULE
+    condition = rule.conditions.get()
+    assert condition.operator == IS_SET
+    assert condition.property == cohort.system_trait_key
+    assert condition.created_with_segment is True
+
+
+def test_create_cohort__valid_name__logs_created_event(
+    environment: Environment,
+    log: StructuredLogCapture,
+) -> None:
+    # Given / When
+    cohort = create_cohort(environment=environment, name="Beta users")
+
+    # Then
+    assert log.has(
+        "cohort.created",
+        cohort__id=cohort.id,
+        segment__id=cohort.segment_id,
+        environment__id=environment.id,
+    )
+
+
+def test_delete_cohort__edge__drains_traits_then_deletes(
+    edge_cohort: Cohort,
+    dynamodb_identity_wrapper: DynamoIdentityWrapper,
+    log: StructuredLogCapture,
+) -> None:
+    # Given: one member already applied (trait present), one still pending add
+    api_key = edge_cohort.environment.api_key
+    dynamodb_identity_wrapper.set_system_trait(
+        environment_api_key=api_key,
+        identifier="applied",
+        trait_key=edge_cohort.system_trait_key,
+    )
+    CohortMembership.objects.create(
+        cohort=edge_cohort, identifier="applied", state=CohortMembershipState.APPLIED
+    )
+    CohortMembership.objects.create(cohort=edge_cohort, identifier="pending")
+
+    # When (synchronous task runner executes the enqueued applier inline)
+    delete_cohort(edge_cohort)
+
+    # Then
+    document = dynamodb_identity_wrapper.get_item(f"{api_key}_applied")
+    assert document is not None
+    assert document["system_traits"] == {}
+    assert dynamodb_identity_wrapper.get_item(f"{api_key}_pending") is None
+    assert not Cohort.objects.filter(id=edge_cohort.id).exists()
+    assert not CohortMembership.objects.filter(cohort_id=edge_cohort.id).exists()
+    assert log.has("cohort.deletion_requested", cohort__id=edge_cohort.id)
diff --git a/api/tests/unit/cohorts/test_tasks.py b/api/tests/unit/cohorts/test_tasks.py
index 9b00be6b7a81..b195d8a04e38 100644
--- a/api/tests/unit/cohorts/test_tasks.py
+++ b/api/tests/unit/cohorts/test_tasks.py
@@ -1,5 +1,6 @@
 import pytest
 from botocore.exceptions import ClientError
+from django.utils import timezone
 from prometheus_client import REGISTRY
 from pytest_mock import MockerFixture
 from pytest_structlog import StructuredLogCapture
@@ -9,6 +10,7 @@
 from cohorts.models import Cohort, CohortMembership, CohortMembershipState
 from cohorts.tasks import apply_cohort_membership_deltas
 from environments.dynamodb import DynamoIdentityWrapper
+from segments.models import Segment
 
 
 def test_apply_cohort_membership_deltas__pending_adds__applies_to_documents(
@@ -83,7 +85,9 @@ def test_apply_cohort_membership_deltas__non_edge_project__skips(
     log: StructuredLogCapture,
 ) -> None:
     # Given
-    mocker.patch("cohorts.tasks.DynamoIdentityWrapper").return_value.is_enabled = True
+    mocker.patch(
+        "cohorts.services.DynamoIdentityWrapper"
+    ).return_value.is_enabled = True
     membership = CohortMembership.objects.create(cohort=cohort, identifier="user-1")
 
     # When
@@ -129,7 +133,9 @@ def test_apply_cohort_membership_deltas__dynamo_throttled__raises_backoff(
     log: StructuredLogCapture,
 ) -> None:
     # Given
-    mocker.patch("cohorts.tasks.DynamoIdentityWrapper").return_value.is_enabled = True
+    mocker.patch(
+        "cohorts.services.DynamoIdentityWrapper"
+    ).return_value.is_enabled = True
     mocker.patch.object(
         services,
         "apply_pending_memberships",
@@ -150,7 +156,9 @@ def test_apply_cohort_membership_deltas__other_client_error__reraises(
     log: StructuredLogCapture,
 ) -> None:
     # Given
-    mocker.patch("cohorts.tasks.DynamoIdentityWrapper").return_value.is_enabled = True
+    mocker.patch(
+        "cohorts.services.DynamoIdentityWrapper"
+    ).return_value.is_enabled = True
     mocker.patch.object(
         services,
         "apply_pending_memberships",
@@ -203,3 +211,35 @@ def test_apply_cohort_membership_deltas__deltas_applied__increments_metric(
 
     # Then
     assert REGISTRY.get_sample_value(metric, {"operation": "add"}) == before + 1
+
+
+def test_apply_cohort_membership_deltas__deletion_requested__finalises_after_drain(
+    edge_cohort: Cohort,
+    dynamodb_identity_wrapper: DynamoIdentityWrapper,
+    log: StructuredLogCapture,
+) -> None:
+    # Given
+    api_key = edge_cohort.environment.api_key
+    dynamodb_identity_wrapper.set_system_trait(
+        environment_api_key=api_key,
+        identifier="member",
+        trait_key=edge_cohort.system_trait_key,
+    )
+    CohortMembership.objects.create(
+        cohort=edge_cohort,
+        identifier="member",
+        state=CohortMembershipState.PENDING_REMOVE,
+    )
+    edge_cohort.deletion_requested_at = timezone.now()
+    edge_cohort.save()
+
+    # When
+    apply_cohort_membership_deltas(cohort_id=edge_cohort.id)
+
+    # Then
+    document = dynamodb_identity_wrapper.get_item(f"{api_key}_member")
+    assert document is not None
+    assert document["system_traits"] == {}
+    assert not Cohort.objects.filter(id=edge_cohort.id).exists()
+    assert not Segment.objects.filter(id=edge_cohort.segment_id).exists()
+    assert log.has("cohort.deleted", cohort__id=edge_cohort.id)
diff --git a/api/tests/unit/cohorts/test_views.py b/api/tests/unit/cohorts/test_views.py
new file mode 100644
index 000000000000..fb56c55d64ae
--- /dev/null
+++ b/api/tests/unit/cohorts/test_views.py
@@ -0,0 +1,204 @@
+import pytest
+from common.environments.permissions import VIEW_ENVIRONMENT
+from common.projects.permissions import MANAGE_SEGMENTS
+from django.urls import reverse
+from django.utils import timezone
+from pytest_mock import MockerFixture
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from cohorts.models import Cohort
+from environments.dynamodb import DynamoIdentityWrapper
+from environments.models import Environment
+from organisations.models import Subscription
+from projects.models import Project
+from segments.models import Segment
+from tests.types import (
+    WithEnvironmentPermissionsCallable,
+    WithProjectPermissionsCallable,
+)
+
+
+def test_create_cohort__staff_with_manage_segments__returns_201(
+    staff_client: APIClient,
+    dynamo_enabled_project: Project,
+    dynamo_enabled_project_environment_one: Environment,
+    dynamodb_identity_wrapper: DynamoIdentityWrapper,
+    with_project_permissions: WithProjectPermissionsCallable,
+) -> None:
+    # Given
+    with_project_permissions(  # type: ignore[call-arg]
+        [MANAGE_SEGMENTS], project_id=dynamo_enabled_project.id
+    )
+    url = reverse(
+        "api-v1:environments:cohorts:cohorts-list",
+        args=[dynamo_enabled_project_environment_one.api_key],
+    )
+
+    # When
+    response = staff_client.post(
+        url,
+        data={"name": "Beta users", "description": "Early access group"},
+        format="json",
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_201_CREATED
+    cohort = Cohort.objects.get(id=response.json()["id"])
+    assert response.json()["name"] == "Beta users"
+    assert response.json()["description"] == "Early access group"
+    assert cohort.segment.description == "Early access group"
+    assert response.json()["segment"] == cohort.segment_id
+    assert cohort.environment == dynamo_enabled_project_environment_one
+
+
+def test_create_cohort__staff_without_permission__returns_403(
+    staff_client: APIClient,
+    environment: Environment,
+) -> None:
+    # Given
+    url = reverse(
+        "api-v1:environments:cohorts:cohorts-list", args=[environment.api_key]
+    )
+
+    # When
+    response = staff_client.post(url, data={"name": "Beta users"}, format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+def test_create_cohort__unknown_environment__returns_403(
+    staff_client: APIClient,
+) -> None:
+    # Given
+    url = reverse("api-v1:environments:cohorts:cohorts-list", args=["missing-key"])
+
+    # When
+    response = staff_client.post(url, data={"name": "Beta users"}, format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+def test_list_cohorts__deletion_requested_cohort__excluded(
+    staff_client: APIClient,
+    edge_cohort: Cohort,
+    dynamodb_identity_wrapper: DynamoIdentityWrapper,
+    with_environment_permissions: WithEnvironmentPermissionsCallable,
+) -> None:
+    # Given
+    environment = edge_cohort.environment
+    with_environment_permissions(  # type: ignore[call-arg]
+        [VIEW_ENVIRONMENT], environment_id=environment.id
+    )
+    deleting_segment = Segment.objects.create(
+        name="going away", project=environment.project
+    )
+    Cohort.objects.create(
+        environment=environment,
+        segment=deleting_segment,
+        deletion_requested_at=timezone.now(),
+    )
+    url = reverse(
+        "api-v1:environments:cohorts:cohorts-list", args=[environment.api_key]
+    )
+
+    # When
+    response = staff_client.get(url)
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    assert [row["id"] for row in response.json()] == [edge_cohort.id]
+    assert response.json()[0]["name"] == edge_cohort.segment.name
+
+
+def test_delete_cohort__staff_with_manage_segments__returns_202(
+    staff_client: APIClient,
+    dynamo_enabled_project: Project,
+    edge_cohort: Cohort,
+    dynamodb_identity_wrapper: DynamoIdentityWrapper,
+    with_project_permissions: WithProjectPermissionsCallable,
+) -> None:
+    # Given
+    with_project_permissions(  # type: ignore[call-arg]
+        [MANAGE_SEGMENTS], project_id=dynamo_enabled_project.id
+    )
+    url = reverse(
+        "api-v1:environments:cohorts:cohorts-detail",
+        args=[edge_cohort.environment.api_key, edge_cohort.id],
+    )
+
+    # When
+    response = staff_client.delete(url)
+
+    # Then
+    assert response.status_code == status.HTTP_202_ACCEPTED
+    assert not Cohort.objects.filter(id=edge_cohort.id).exists()
+
+
+@pytest.mark.saas_mode
+def test_create_cohort__saas_free_plan__returns_403(
+    staff_client: APIClient,
+    environment: Environment,
+    with_project_permissions: WithProjectPermissionsCallable,
+) -> None:
+    # Given
+    with_project_permissions([MANAGE_SEGMENTS])  # type: ignore[call-arg]
+    url = reverse(
+        "api-v1:environments:cohorts:cohorts-list", args=[environment.api_key]
+    )
+
+    # When
+    response = staff_client.post(url, data={"name": "Beta users"}, format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert response.json()["detail"] == (
+        "This resource requires a START_UP plan or above."
+    )
+
+
+def test_create_cohort__saas_startup_plan__returns_201(
+    staff_client: APIClient,
+    dynamo_enabled_project: Project,
+    dynamo_enabled_project_environment_one: Environment,
+    dynamodb_identity_wrapper: DynamoIdentityWrapper,
+    with_project_permissions: WithProjectPermissionsCallable,
+    startup_subscription: Subscription,
+    mocker: MockerFixture,
+) -> None:
+    # Given (saas_mode's fake filesystem breaks moto, so patch is_saas instead)
+    mocker.patch("organisations.subscriptions.permissions.is_saas", return_value=True)
+    with_project_permissions(  # type: ignore[call-arg]
+        [MANAGE_SEGMENTS], project_id=dynamo_enabled_project.id
+    )
+    url = reverse(
+        "api-v1:environments:cohorts:cohorts-list",
+        args=[dynamo_enabled_project_environment_one.api_key],
+    )
+
+    # When
+    response = staff_client.post(url, data={"name": "Beta users"}, format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_201_CREATED
+
+
+def test_create_cohort__non_edge_project__returns_400(
+    staff_client: APIClient,
+    environment: Environment,
+    with_project_permissions: WithProjectPermissionsCallable,
+) -> None:
+    # Given
+    with_project_permissions([MANAGE_SEGMENTS])  # type: ignore[call-arg]
+    url = reverse(
+        "api-v1:environments:cohorts:cohorts-list", args=[environment.api_key]
+    )
+
+    # When
+    response = staff_client.post(url, data={"name": "Beta users"}, format="json")
+
+    # Then
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json()["detail"] == "Dynamo DB is not enabled for this project"
diff --git a/api/tests/unit/segments/test_unit_segments_views.py b/api/tests/unit/segments/test_unit_segments_views.py
index c015d2e348f1..111e8f2a6d98 100644
--- a/api/tests/unit/segments/test_unit_segments_views.py
+++ b/api/tests/unit/segments/test_unit_segments_views.py
@@ -21,6 +21,7 @@
 from audit.constants import SEGMENT_DELETED_MESSAGE
 from audit.models import AuditLog
 from audit.related_object_type import RelatedObjectType
+from cohorts.models import Cohort
 from environments.models import Environment
 from features.models import Feature, FeatureSegment, FeatureState
 from features.versioning.models import EnvironmentFeatureVersion
@@ -32,7 +33,13 @@
 )
 from organisations.models import Organisation
 from projects.models import Project
-from segments.models import Condition, Segment, SegmentRule, WhitelistedSegment
+from segments.models import (
+    Condition,
+    Segment,
+    SegmentManagedBy,
+    SegmentRule,
+    WhitelistedSegment,
+)
 from tests.types import WithProjectPermissionsCallable
 from util.mappers import map_identity_to_identity_document
 
@@ -1967,3 +1974,95 @@ def test_create_segment__body_project_differs_from_url__does_not_create_in_other
     assert response.status_code == status.HTTP_201_CREATED
     assert response.json()["project"] == project.id
     assert not Segment.objects.filter(project=other_project).exists()
+
+
+def test_update_segment__cohort_managed__returns_403(
+    admin_client: APIClient,
+    project: Project,
+    segment: Segment,
+    environment: Environment,
+) -> None:
+    # Given
+    Cohort.objects.create(environment=environment, segment=segment)
+    url = reverse(
+        "api-v1:projects:project-segments-detail", args=[project.id, segment.id]
+    )
+    data = {
+        "name": "New segment name",
+        "project": project.id,
+        "rules": [{"type": "ALL", "rules": [], "conditions": []}],
+    }
+
+    # When
+    response = admin_client.put(
+        url, data=json.dumps(data), content_type="application/json"
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+def test_delete_segment__cohort_managed__returns_403(
+    admin_client: APIClient,
+    project: Project,
+    segment: Segment,
+    environment: Environment,
+) -> None:
+    # Given
+    Cohort.objects.create(environment=environment, segment=segment)
+    url = reverse(
+        "api-v1:projects:project-segments-detail", args=[project.id, segment.id]
+    )
+
+    # When
+    response = admin_client.delete(url)
+
+    # Then
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert Segment.objects.filter(id=segment.id).exists()
+
+
+def test_create_segment__managed_by_in_payload__ignored(
+    admin_client: APIClient,
+    project: Project,
+) -> None:
+    # Given
+    url = reverse("api-v1:projects:project-segments-list", args=[project.id])
+    data = {
+        "name": "New segment",
+        "project": project.id,
+        "managed_by": SegmentManagedBy.COHORT,
+        "rules": [{"type": "ALL", "rules": [], "conditions": []}],
+    }
+
+    # When
+    response = admin_client.post(
+        url, data=json.dumps(data), content_type="application/json"
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_201_CREATED
+    assert response.json()["managed_by"] == ""
+    assert Segment.objects.get(id=response.json()["id"]).managed_by == ""
+
+
+def test_clone_segment__cohort_managed__returns_403(
+    admin_client: APIClient,
+    project: Project,
+    segment: Segment,
+    environment: Environment,
+) -> None:
+    # Given
+    Cohort.objects.create(environment=environment, segment=segment)
+    url = reverse(
+        "api-v1:projects:project-segments-clone", args=[project.id, segment.id]
+    )
+
+    # When
+    response = admin_client.post(
+        url, data=json.dumps({"name": "my copy"}), content_type="application/json"
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert Segment.objects.count() == 1
diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md
index af4f83009db0..f1dc15ec4238 100644
--- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md
+++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md
@@ -71,10 +71,40 @@ Attributes:
  - `feature.count`
  - `organisation.id`
 
+### `cohorts.cohort.created`
+
+Logged at `info` from:
+ - `api/cohorts/services.py:103`
+
+Attributes:
+ - `cohort.id`
+ - `environment.id`
+ - `organisation.id`
+ - `project.id`
+ - `segment.id`
+
+### `cohorts.cohort.deleted`
+
+Logged at `info` from:
+ - `api/cohorts/services.py:140`
+
+Attributes:
+ - `cohort.id`
+ - `environment.id`
+
+### `cohorts.cohort.deletion_requested`
+
+Logged at `info` from:
+ - `api/cohorts/services.py:124`
+
+Attributes:
+ - `cohort.id`
+ - `environment.id`
+
 ### `cohorts.membership.applied`
 
 Logged at `info` from:
- - `api/cohorts/services.py:61`
+ - `api/cohorts/services.py:72`
 
 Attributes:
  - `adds.count`
@@ -85,8 +115,8 @@ Attributes:
 ### `cohorts.membership.apply.skipped`
 
 Logged at `info` from:
- - `api/cohorts/tasks.py:23`
- - `api/cohorts/tasks.py:29`
+ - `api/cohorts/tasks.py:22`
+ - `api/cohorts/tasks.py:25`
 
 Attributes:
  - `cohort.id`
@@ -95,7 +125,7 @@ Attributes:
 ### `cohorts.membership.apply.throttled`
 
 Logged at `warning` from:
- - `api/cohorts/tasks.py:37`
+ - `api/cohorts/tasks.py:35`
 
 Attributes:
  - `cohort.id`
@@ -577,7 +607,7 @@ Attributes:
 ### `segments.serializers.segment_revision_created`
 
 Logged at `info` from:
- - `api/segments/serializers.py:158`
+ - `api/segments/serializers.py:159`
 
 Attributes:
  - `revision_id`
diff --git a/mcp/src/flagsmith_mcp/openapi.json b/mcp/src/flagsmith_mcp/openapi.json
index 330b2f25c097..d9baec716a50 100644
--- a/mcp/src/flagsmith_mcp/openapi.json
+++ b/mcp/src/flagsmith_mcp/openapi.json
@@ -5460,6 +5460,13 @@
           "last_name"
         ]
       },
+      "ManagedByEnum": {
+        "description": "* `cohort` - Cohort",
+        "type": "string",
+        "enum": [
+          "cohort"
+        ]
+      },
       "Metadata": {
         "type": "object",
         "properties": {
@@ -6897,6 +6904,14 @@
               "$ref": "#/components/schemas/SegmentMembershipCount"
             },
             "readOnly": true
+          },
+          "managed_by": {
+            "allOf": [
+              {
+                "$ref": "#/components/schemas/ManagedByEnum"
+              }
+            ],
+            "readOnly": true
           }
         },
         "required": [
diff --git a/openapi.yaml b/openapi.yaml
index 86cd0495cb33..18c5b9007d3d 100644
--- a/openapi.yaml
+++ b/openapi.yaml
@@ -2037,6 +2037,118 @@ paths:
         - Master API Key: []
       tags:
         - Environments
+  '/api/v1/environments/{environment_api_key}/cohorts/':
+    get:
+      operationId: api_v1_environments_cohorts_list
+      description: List the environment's cohorts.
+      parameters:
+        - name: environment_api_key
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        '200':
+          description: ''
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/Cohort'
+      security:
+        - tokenAuth: []
+        - Master API Key: []
+      tags:
+        - Environments
+      x-flagsmith-minimum-plan: START_UP
+    post:
+      operationId: api_v1_environments_cohorts_create
+      description: Create a cohort and the managed segment that targets it.
+      parameters:
+        - name: environment_api_key
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/Cohort'
+          application/x-www-form-urlencoded:
+            schema:
+              $ref: '#/components/schemas/Cohort'
+          multipart/form-data:
+            schema:
+              $ref: '#/components/schemas/Cohort'
+      responses:
+        '201':
+          description: ''
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Cohort'
+      security:
+        - tokenAuth: []
+        - Master API Key: []
+      tags:
+        - Environments
+      x-flagsmith-minimum-plan: START_UP
+  '/api/v1/environments/{environment_api_key}/cohorts/{cohort_id}/':
+    get:
+      operationId: api_v1_environments_cohorts_retrieve
+      description: Retrieve a cohort.
+      parameters:
+        - name: cohort_id
+          in: path
+          description: A unique integer value identifying this cohort.
+          required: true
+          schema:
+            type: integer
+        - name: environment_api_key
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        '200':
+          description: ''
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Cohort'
+      security:
+        - tokenAuth: []
+        - Master API Key: []
+      tags:
+        - Environments
+      x-flagsmith-minimum-plan: START_UP
+    delete:
+      operationId: api_v1_environments_cohorts_destroy
+      description: Request cohort deletion. Memberships are drained from identity data first; the cohort and its segment are deleted once drained.
+      parameters:
+        - name: cohort_id
+          in: path
+          description: A unique integer value identifying this cohort.
+          required: true
+          schema:
+            type: integer
+        - name: environment_api_key
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        '202':
+          description: No response body
+      security:
+        - tokenAuth: []
+        - Master API Key: []
+      tags:
+        - Environments
+      x-flagsmith-minimum-plan: START_UP
   '/api/v1/environments/{environment_api_key}/create-change-request/':
     post:
       operationId: create_environment_feature_change_request
@@ -18558,6 +18670,10 @@ components:
           items:
             $ref: '#/components/schemas/SegmentMembershipCount'
           readOnly: true
+        managed_by:
+          allOf:
+            - $ref: '#/components/schemas/ManagedByEnum'
+          readOnly: true
         change_request:
           type:
             - integer
@@ -18681,6 +18797,39 @@ components:
           maxLength: 2000
       required:
         - name
+    Cohort:
+      type: object
+      properties:
+        id:
+          type: integer
+          readOnly: true
+        uuid:
+          type: string
+          format: uuid
+          readOnly: true
+        name:
+          type: string
+          maxLength: 2000
+        description:
+          type:
+            - string
+            - 'null'
+        segment:
+          type: integer
+          readOnly: true
+        source_type:
+          allOf:
+            - $ref: '#/components/schemas/SourceTypeEnum'
+          readOnly: true
+        version:
+          type: integer
+          readOnly: true
+        created_at:
+          type: string
+          format: date-time
+          readOnly: true
+      required:
+        - name
     Condition:
       type: object
       properties:
@@ -21734,6 +21883,11 @@ components:
           $ref: '#/components/schemas/UserList'
       required:
         - user
+    ManagedByEnum:
+      description: '* `cohort` - Cohort'
+      type: string
+      enum:
+        - cohort
     MasterAPIKey:
       type: object
       properties:
@@ -24605,6 +24759,10 @@ components:
           items:
             $ref: '#/components/schemas/SegmentMembershipCount'
           readOnly: true
+        managed_by:
+          allOf:
+            - $ref: '#/components/schemas/ManagedByEnum'
+          readOnly: true
     PatchedSegmentConfiguration:
       type: object
       properties:
@@ -26280,6 +26438,10 @@ components:
           items:
             $ref: '#/components/schemas/SegmentMembershipCount'
           readOnly: true
+        managed_by:
+          allOf:
+            - $ref: '#/components/schemas/ManagedByEnum'
+          readOnly: true
       required:
         - name
         - project
@@ -26518,6 +26680,11 @@ components:
           type: boolean
       required:
         - channel_id
+    SourceTypeEnum:
+      description: '* `csv` - CSV'
+      type: string
+      enum:
+        - csv
     StageAction:
       type: object
       properties: