diff --git a/openquake/server/db/actions.py b/openquake/server/db/actions.py
index 8fec90f7480..36fc592f5b6 100644
--- a/openquake/server/db/actions.py
+++ b/openquake/server/db/actions.py
@@ -578,17 +578,7 @@ def get_calcs(db, request_get_dict, allowed_users, user_acl_on=False, id=None):
     order_dir = request_get_dict.get('order_dir', 'DESC').upper()
     if order_dir not in ('ASC', 'DESC'):
         order_dir = 'DESC'
-
-    tags_query = """
-GROUP_CONCAT(
-    CASE
-        WHEN t.is_preferred = 1 THEN t.tag || '★'
-        ELSE t.tag
-    END,
-    ', '
-) AS tags
-    """
-
+    tags_query = "GROUP_CONCAT(tag_value, ', ') AS tags"
     where_clause = f"?A AND ({users_filter} AND {user_name_like_filter}"
     if include_shared:
         where_clause += " OR j.status == 'shared'"
@@ -598,7 +588,12 @@ def get_calcs(db, request_get_dict, allowed_users, user_acl_on=False, id=None):
             " AND j.id IN (SELECT job_id FROM job_tag WHERE is_preferred = 1)")
     if filter_by_tag and filter_by_tag != '0':
         where_clause += (
-            " AND j.id IN (SELECT job_id FROM job_tag WHERE tag = ?x)")
+            " AND j.id IN ("
+            "SELECT jt.job_id "
+            "FROM job_tag jt "
+            "JOIN tag tg ON tg.id = jt.tag_id "
+            "WHERE tg.name = ?x"
+            ")")
         query_params.append(filter_by_tag)
 
     # NOTE: GROUP BY j.id returns one row per job (identified by j.id), even if that
@@ -607,7 +602,17 @@ def get_calcs(db, request_get_dict, allowed_users, user_acl_on=False, id=None):
     query = f"""
 SELECT j.*, {tags_query}
 FROM job AS j
-LEFT JOIN job_tag AS t ON j.id = t.job_id
+LEFT JOIN (
+    SELECT
+        jt.job_id,
+        CASE
+            WHEN jt.is_preferred THEN t.name || '★'
+            ELSE t.name
+        END AS tag_value
+    FROM job_tag jt
+    JOIN tag t ON t.id = jt.tag_id
+    ORDER BY jt.is_preferred DESC, t.name
+) jt ON jt.job_id = j.id
 WHERE {where_clause}
 GROUP BY j.id
 ORDER BY {order_by} {order_dir}
@@ -669,10 +674,34 @@ def share_job(db, job_id, share):
                        f' from "{initial_status}" to "{new_status}"'}
 
 
+def _get_or_create_tag_id(db, tag_name):
+    rows = db("SELECT id FROM tag WHERE name = ?x", tag_name)
+    if rows:
+        return rows[0].id
+
+    db("INSERT INTO tag (name) VALUES (?x)", tag_name)
+    rows = db("SELECT id FROM tag WHERE name = ?x", tag_name)
+    return rows[0].id
+
+
+def _get_tag_id(db, tag_name):
+    """
+    Resolve an existing tag name to tag_id.
+    Raises KeyError if the tag does not exist.
+    """
+    rows = db("SELECT id FROM tag WHERE name = ?x", tag_name)
+    if not rows:
+        raise KeyError(f"Tag '{tag_name}' does not exist")
+    return rows[0].id
+
+
 def add_tag_to_job(db, job_id, tag_name):
     try:
-        db("INSERT INTO job_tag (job_id, tag, is_preferred) VALUES (?x, ?x, 0)",
-           job_id, tag_name)
+        tag_id = _get_or_create_tag_id(db, tag_name)
+        db("""
+INSERT INTO job_tag (job_id, tag_id, is_preferred)
+VALUES (?x, ?x, 0)
+           """, job_id, tag_id)
     except Exception as exc:
         return {'error': str(exc)}
     else:
@@ -681,8 +710,14 @@ def add_tag_to_job(db, job_id, tag_name):
 
 def remove_tag_from_job(db, job_id, tag_name):
     try:
-        db("DELETE FROM job_tag WHERE job_id = ?x AND tag = ?x",
-           job_id, tag_name)
+        tag_id = _get_tag_id(db, tag_name)
+    except KeyError:
+        return {'success': f'Tag {tag_name} was not associated with job {job_id}'}
+    try:
+        db("""
+DELETE FROM job_tag
+WHERE job_id = ?x AND tag_id = ?x
+        """, job_id, tag_id)
     except Exception as exc:
         return {'error': str(exc)}
     else:
@@ -691,18 +726,25 @@ def remove_tag_from_job(db, job_id, tag_name):
 
 def set_preferred_job_for_tag(db, job_id, tag_name):
     try:
+        tag_id = _get_or_create_tag_id(db, tag_name)
+
         db("BEGIN")
+
         db("""
-UPDATE job_tag SET is_preferred = 0
-WHERE tag = ?x AND is_preferred = 1
-           """, tag_name)
+UPDATE job_tag
+SET is_preferred = 0
+WHERE tag_id = ?x AND is_preferred = 1
+        """, tag_id)
+
         db("""
-INSERT INTO job_tag (job_id, tag, is_preferred)
+INSERT INTO job_tag (job_id, tag_id, is_preferred)
 VALUES (?x, ?x, 1)
-ON CONFLICT(job_id, tag) DO UPDATE
+ON CONFLICT(job_id, tag_id) DO UPDATE
 SET is_preferred = 1
-           """, job_id, tag_name)
+        """, job_id, tag_id)
+
         db("COMMIT")
+
     except Exception as exc:
         return {'error': str(exc)}
     else:
@@ -711,10 +753,18 @@ def set_preferred_job_for_tag(db, job_id, tag_name):
 
 def unset_preferred_job_for_tag(db, tag_name):
     try:
+        rows = db("SELECT id FROM tag WHERE name = ?x", tag_name)
+        if not rows:
+            return {'success': f'Tag {tag_name} has no preferred job now'}
+
+        tag_id = rows[0].id
+
         db("""
-UPDATE job_tag SET is_preferred = 0
-WHERE tag = ?x AND is_preferred = 1
-           """, tag_name)
+UPDATE job_tag
+SET is_preferred = 0
+WHERE tag_id = ?x AND is_preferred = 1
+        """, tag_id)
+
     except Exception as exc:
         return {'error': str(exc)}
     else:
@@ -727,8 +777,10 @@ def get_preferred_job_for_tag(db, tag_name):
 SELECT j.*
 FROM job AS j
 JOIN job_tag jt ON j.id = jt.job_id
-WHERE jt.tag = ?x AND jt.is_preferred = 1
+JOIN tag t ON t.id = jt.tag_id
+WHERE t.name = ?x AND jt.is_preferred = 1
         """, tag_name)
+
     except Exception as exc:
         return {'error': str(exc)}
     else:
@@ -740,9 +792,40 @@ def get_preferred_job_for_tag(db, tag_name):
             return {'error': f'Unexpected multiple preferred jobs for tag {tag_name}'}
 
 
+def create_tag(db, name):
+    try:
+        rows = db("SELECT id FROM tag WHERE name = ?x", name)
+        if rows:
+            return {'success': f'Tag {name} already exists'}
+        db("INSERT INTO tag (name) VALUES (?x)", name)
+    except Exception as exc:
+        return {'error': str(exc)}
+    else:
+        return {'success': f'Tag {name} was created'}
+
+
+def delete_tag(db, name):
+    try:
+        rows = db("SELECT id FROM tag WHERE name = ?x", name)
+        if not rows:
+            return {'success': f'Tag {name} does not exist'}
+
+        tag_id = rows[0].id
+
+        db("""
+DELETE from tag
+WHERE id = ?x
+        """, tag_id)
+
+    except Exception as exc:
+        return {'error': str(exc)}
+    else:
+        return {'success': f'Tag {name} was deleted'}
+
+
 def list_tags(db):
-    rows = db("SELECT DISTINCT tag FROM job_tag ORDER BY tag")
-    tags = [row.tag for row in rows]
+    rows = db("SELECT name FROM tag ORDER BY name")
+    tags = [row.name for row in rows]
     return {'success': 'ok', 'tags': tags}
 
 
diff --git a/openquake/server/db/models.py b/openquake/server/db/models.py
index cc0de4a708a..d6dc7a9ccd5 100644
--- a/openquake/server/db/models.py
+++ b/openquake/server/db/models.py
@@ -16,7 +16,7 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with OpenQuake. If not, see <http://www.gnu.org/licenses/>.
 
-from django.db import models, connection
+from django.db import models
 
 
 class Job(models.Model):
@@ -28,44 +28,61 @@ class Job(models.Model):
     class Meta:
         managed = False  # the schema is not managed by Django
         db_table = 'job'
+        ordering = ['-id']  # descending by id
 
     def __str__(self):
         return f"{self.id} – {self.description[:80]}"  # show first 80 chars
 
 
+class Tag(models.Model):
+    id = models.AutoField(primary_key=True)
+    name = models.CharField(max_length=255, unique=True)
+
+    class Meta:
+        managed = False
+        db_table = "tag"
+        verbose_name = "Tag"
+        verbose_name_plural = "Tags"
+        ordering = ['name']
+
+    def __str__(self):
+        return self.name
+
+
 class JobTag(models.Model):
+    id = models.AutoField(primary_key=True)
     job = models.ForeignKey(
         Job,
         on_delete=models.CASCADE,
         db_column="job_id",
+        related_name="job_tags",
+    )
+    tag = models.ForeignKey(
+        Tag,
+        on_delete=models.CASCADE,
+        db_column="tag_id",
+        related_name="job_tags",
     )
-    tag = models.CharField(max_length=255)
     is_preferred = models.BooleanField(default=False)
 
     class Meta:
+        managed = False
         db_table = 'job_tag'
-        managed = False  # the schema is not managed by Django
         unique_together = ("job", "tag")
+        ordering = ['job_id', 'tag_id']
         indexes = [
             models.Index(
-                fields=['tag'],
-                name='uq_preferred_per_tag',
+                fields=["tag_id"],
+                name="uq_preferred_per_tag",
                 condition=models.Q(is_preferred=True),
             )
         ]
+
         verbose_name = "Job Tag"
         verbose_name_plural = "Job Tags"
 
     def __str__(self):
-        return (f"{self.tag} (job_id={self.job_id},"
-                f" {'preferred' if self.is_preferred else 'not preferred'})")
-
-    @property
-    def job_description(self):
-        """Return the job description (queried directly from the job table)."""
-        with connection.cursor() as cursor:
-            cursor.execute(
-                "SELECT description FROM job WHERE id = %s", [self.job_id]
-            )
-            row = cursor.fetchone()
-        return row[0] if row else "(unknown)"
+        return (
+            f"{self.tag.name} (job_id={self.job_id}, "
+            f"{'preferred' if self.is_preferred else 'not preferred'})"
+        )
diff --git a/openquake/server/db/schema/upgrades/0010-job-tag.sql b/openquake/server/db/schema/upgrades/0010-job-tag.sql
new file mode 100644
index 00000000000..bfa3dc5afe8
--- /dev/null
+++ b/openquake/server/db/schema/upgrades/0010-job-tag.sql
@@ -0,0 +1,54 @@
+-- WARNING
+-- This migration RESTRUCTURES the job/tag schema.
+-- It preserves existing data by migrating it from the old
+-- job_tag(job_id, tag, is_preferred) table into a normalized
+-- schema with:
+--   - tag(id, name)
+--   - job_tag(job_id, tag_id, is_preferred)
+-- Apply only once.
+
+-- Remove old constraint/index if present
+DROP INDEX IF EXISTS uq_preferred_per_tag;
+
+-- Rename old table (do NOT drop yet)
+ALTER TABLE job_tag RENAME TO job_tag_old;
+
+-- Create new tables
+CREATE TABLE tag (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    name TEXT NOT NULL UNIQUE CHECK (LENGTH(name) > 0)
+);
+
+CREATE TABLE job_tag (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    job_id INTEGER NOT NULL,
+    tag_id INTEGER NOT NULL,
+    is_preferred INTEGER NOT NULL DEFAULT 0
+        CHECK (is_preferred IN (0, 1)),
+    UNIQUE (job_id, tag_id),
+    FOREIGN KEY (job_id) REFERENCES job(id) ON DELETE CASCADE,
+    FOREIGN KEY (tag_id) REFERENCES tag(id) ON DELETE CASCADE
+);
+
+-- Migrate tag values
+INSERT INTO tag (name)
+SELECT DISTINCT tag
+FROM job_tag_old;
+
+-- Migrate job-tag relations
+INSERT INTO job_tag (job_id, tag_id, is_preferred)
+SELECT
+    jto.job_id,
+    t.id,
+    jto.is_preferred
+FROM job_tag_old AS jto
+JOIN tag AS t
+  ON t.name = jto.tag;
+
+-- Drop old table
+DROP TABLE job_tag_old;
+
+-- Recreate constraint, consistent with the new schema
+CREATE UNIQUE INDEX uq_preferred_per_tag
+ON job_tag(tag_id)
+WHERE is_preferred = 1;
diff --git a/openquake/server/db/tag_admin.py b/openquake/server/db/tag_admin.py
index 3229878ecb8..0211235b2b2 100644
--- a/openquake/server/db/tag_admin.py
+++ b/openquake/server/db/tag_admin.py
@@ -16,28 +16,30 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with OpenQuake. If not, see <http://www.gnu.org/licenses/>.
 
-from django.contrib import admin
+from django.contrib import admin, messages
+from django.db import IntegrityError, transaction
 from django.apps import apps
+from django.db.models import Q
 from openquake.server.db.tag_site import tag_admin_site
 
 
 def get_models():
     """
     Lazy model getter that only resolves models once Django apps are ready.
-    Returns (Job, JobTag).
+    Returns (Job, Tag, JobTag).
     """
     try:
         Job = apps.get_model("db", "Job")
+        Tag = apps.get_model("db", "Tag")
         JobTag = apps.get_model("db", "JobTag")
-        return Job, JobTag
+        return Job, Tag, JobTag
     except (LookupError, Exception):
-        return None, None
+        return None, None, None
 
 
-Job, JobTag = get_models()
+Job, Tag, JobTag = get_models()
 
-
-if Job and JobTag:
+if Job and JobTag and Tag:
 
     @admin.register(Job, site=tag_admin_site)
     class JobAdmin(admin.ModelAdmin):
@@ -50,10 +52,10 @@ class JobAdmin(admin.ModelAdmin):
 
         def has_view_permission(self, request, obj=None):
             user = request.user
-            if user.is_active and user.is_authenticated and (
-                    user.is_superuser or user.level >= 2):
-                return True
-            return False
+            return (
+                user.is_active and user.is_authenticated and
+                (user.is_superuser or user.level >= 2)
+            )
 
         def has_add_permission(self, request):
             return False
@@ -64,54 +66,59 @@ def has_change_permission(self, request, obj=None):
         def has_delete_permission(self, request, obj=None):
             return False
 
+    @admin.register(Tag, site=tag_admin_site)
+    class TagAdmin(admin.ModelAdmin):
+        list_display = ("name",)
+        search_fields = ("name",)
+        ordering = ("name",)
+
     @admin.register(JobTag, site=tag_admin_site)
     class JobTagAdmin(admin.ModelAdmin):
-        list_display = ('job_display', 'tag', 'is_preferred')
-        list_filter = ('is_preferred',)
-        search_fields = ('tag',)
-        autocomplete_fields = ["job"]
+        list_display = ("job_display", "tag", "is_preferred")
+        list_filter = ("is_preferred", "tag__name")
+        search_fields = ("tag__name",)
+        autocomplete_fields = ("job", "tag")
 
         def job_display(self, obj):
             return str(obj.job)
         job_display.short_description = "Job"
 
-        def job_description(self, obj):
-            return obj.job_description
-        job_description.admin_order_field = "job_id"
-        job_description.short_description = "Job Description"
+        def tag_display(self, obj):
+            return str(obj.obg)
+        tag_display.short_description = "Tag"
 
         def get_search_results(self, request, queryset, search_term):
             """
             Extend the default search to include both job descriptions and job IDs.
             """
             queryset, use_distinct = super().get_search_results(
-                request, queryset, search_term)
+                request, queryset, search_term
+            )
 
             if search_term:
-                from django.db.models import Q
-
-                # If the term is numeric, try matching job ID directly
                 job_filter = Q(description__icontains=search_term)
                 if search_term.isdigit():
                     job_filter |= Q(id=int(search_term))
 
-                # Find matching Job IDs
-                Job = self.model._meta.get_field("job").remote_field.model
-                job_ids = list(Job.objects.filter(job_filter).values_list(
-                    "id", flat=True))
+                JobModel = self.model._meta.get_field("job").remote_field.model
+                job_ids = list(
+                    JobModel.objects
+                    .filter(job_filter)
+                    .values_list("id", flat=True)
+                )
 
                 if job_ids:
                     queryset |= self.model.objects.filter(job_id__in=job_ids)
+                    use_distinct = True
 
             return queryset, use_distinct
 
         def has_module_permission(self, request):
             user = request.user
-            user = request.user
-            if user.is_active and user.is_authenticated and (
-                    user.is_superuser or user.level >= 2):
-                return True
-            return False
+            return (
+                user.is_active and user.is_authenticated and
+                (user.is_superuser or user.level >= 2)
+            )
 
         def has_view_permission(self, request, obj=None):
             return self.has_module_permission(request)
@@ -124,3 +131,35 @@ def has_change_permission(self, request, obj=None):
 
         def has_delete_permission(self, request, obj=None):
             return self.has_module_permission(request)
+
+        def save_model(self, request, obj, form, change):
+            """
+            Ensure that uniqueness constraints are respected and show friendly messages
+            """
+            try:
+                with transaction.atomic():
+                    # If marking as preferred, unset any other
+                    # preferred job for this tag
+                    if obj.is_preferred:
+                        JobTag.objects.filter(
+                            tag=obj.tag,
+                            is_preferred=True
+                        ).exclude(pk=obj.pk).update(is_preferred=False)
+                    super().save_model(request, obj, form, change)
+            except IntegrityError as exc:
+                self.message_user(
+                    request,
+                    f"Cannot save JobTag: {exc}",
+                    level=messages.ERROR
+                )
+
+        def delete_model(self, request, obj):
+            try:
+                with transaction.atomic():
+                    super().delete_model(request, obj)
+            except IntegrityError as exc:
+                self.message_user(
+                    request,
+                    f"Cannot delete JobTag: {exc}",
+                    level=messages.ERROR
+                )
diff --git a/openquake/server/db/tag_site.py b/openquake/server/db/tag_site.py
index 302f2d076f0..6df91829b6f 100644
--- a/openquake/server/db/tag_site.py
+++ b/openquake/server/db/tag_site.py
@@ -16,10 +16,12 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with OpenQuake. If not, see <http://www.gnu.org/licenses/>.
 
+from django.conf import settings
 from django.contrib.admin import AdminSite
 
 
 class TagAdminSite(AdminSite):
+    site_url = f"{settings.WEBUI_PATHPREFIX}/engine/"
     site_header = "Job Tags Management"
     site_title = "Job Tags Management"
     index_title = "Manage Job Tags"
diff --git a/openquake/server/tests/test_restricted_mode.py b/openquake/server/tests/test_restricted_mode.py
index 741a5941436..097e921058c 100644
--- a/openquake/server/tests/test_restricted_mode.py
+++ b/openquake/server/tests/test_restricted_mode.py
@@ -177,7 +177,7 @@ def test_tagging_job(self):
         self.assertEqual(ret.status_code, 404)
         ret = logs.dbcmd('add_tag_to_job', jobs[0].calc_id, tag_name)
         self.assertIn('error', ret)
-        self.assertIn("CHECK constraint failed: LENGTH(tag) > 0", ret['error'])
+        self.assertIn("CHECK constraint failed: LENGTH(name) > 0", ret['error'])
 
         # generate random tag names, 10 characters long
         first_tag = random_string(10)
@@ -257,6 +257,24 @@ def test_tagging_job(self):
         for job in jobs:
             self.remove_calc(job.calc_id)
 
+        # delete the tags
+        for tag in [first_tag, second_tag]:
+            ret = self.get(f'delete_tag/{tag}')
+            self.assertEqual(ret.status_code, 200)
+            self.assertIn('success', ret.json())
+            self.assertIn(f'Tag {tag} was deleted', ret.json()['success'])
+
+        # create and delete a tag
+        tag_name = random_string(10)
+        ret = self.get(f'create_tag/{tag_name}')
+        self.assertEqual(ret.status_code, 200)
+        self.assertIn('success', ret.json())
+        self.assertIn(f'Tag {tag_name} was created', ret.json()['success'])
+        ret = self.get(f'delete_tag/{tag_name}')
+        self.assertEqual(ret.status_code, 200)
+        self.assertIn('success', ret.json())
+        self.assertIn(f'Tag {tag_name} was deleted', ret.json()['success'])
+
     def test_calc_list(self):
         """
         Create jobs with different parameters and test that /v1/calc/list
diff --git a/openquake/server/v1/calc_urls.py b/openquake/server/v1/calc_urls.py
index 7baec662bb5..50f82986fdc 100644
--- a/openquake/server/v1/calc_urls.py
+++ b/openquake/server/v1/calc_urls.py
@@ -40,6 +40,8 @@
     re_path(r'^(\d+)/unshare$', views.calc_unshare),
     # Tagging
     re_path(r'^list_tags$', views.calc_list_tags),
+    path('create_tag/<str:tag_name>', views.calc_create_tag),
+    path('delete_tag/<str:tag_name>', views.calc_delete_tag),
     path('<int:calc_id>/add_tag/<str:tag_name>', views.calc_add_tag),
     path('<int:calc_id>/remove_tag/<str:tag_name>', views.calc_remove_tag),
     path('<int:calc_id>/set_preferred_job_for_tag/<str:tag_name>',
diff --git a/openquake/server/views.py b/openquake/server/views.py
index 06b872bb819..6ac44679127 100644
--- a/openquake/server/views.py
+++ b/openquake/server/views.py
@@ -693,6 +693,26 @@ def check_db_response(resp):
         return JsonResponse({'error': f'Unexpected response: {resp}'}, status=500)
 
 
+def create_tag(user_level, tag_name):
+    if user_level < 2:
+        return HttpResponseForbidden()
+    try:
+        resp = logs.dbcmd('create_tag', tag_name)
+    except dbapi.NotFound:
+        return HttpResponseNotFound()
+    return check_db_response(resp)
+
+
+def delete_tag(user_level, tag_name):
+    if user_level < 2:
+        return HttpResponseForbidden()
+    try:
+        resp = logs.dbcmd('delete_tag', tag_name)
+    except dbapi.NotFound:
+        return HttpResponseNotFound()
+    return check_db_response(resp)
+
+
 def add_tag_to_job(user_level, calc_id, tag_name):
     if user_level < 2:
         return HttpResponseForbidden()
@@ -2307,6 +2327,28 @@ def on_same_fs(request):
     return JsonResponse({'success': False}, status=200)
 
 
+@csrf_exempt
+@cross_domain_ajax
+@require_http_methods(['GET'])
+def calc_create_tag(request, tag_name):
+    """
+    Create a tag named `tag_name`
+    """
+    user_level = get_user_level(request)
+    return create_tag(user_level, tag_name)
+
+
+@csrf_exempt
+@cross_domain_ajax
+@require_http_methods(['GET'])
+def calc_delete_tag(request, tag_name):
+    """
+    Delete a tag named `tag_name`
+    """
+    user_level = get_user_level(request)
+    return delete_tag(user_level, tag_name)
+
+
 @csrf_exempt
 @cross_domain_ajax
 @require_http_methods(['GET'])