diff --git a/src/hive/server/db.py b/src/hive/server/db.py
index 881c2aa..0a98e1b 100644
--- a/src/hive/server/db.py
+++ b/src/hive/server/db.py
@@ -412,6 +412,24 @@ def _ensure_postgres_migrations(conn) -> None:
     ).fetchone()
     if not row:
         conn.execute("ALTER TABLE forks ADD COLUMN branch_prefix TEXT")
+    # slug + owner on tasks — compat with prod schema applied externally from
+    # staging refactor 472d4b7. Additive + idempotent: each ALTER is gated on
+    # an information_schema check, slug is backfilled before SET NOT NULL,
+    # owner gets a default so existing rows satisfy NOT NULL immediately.
+    row = conn.execute(
+        "SELECT 1 FROM information_schema.columns"
+        " WHERE table_name = 'tasks' AND column_name = 'slug'"
+    ).fetchone()
+    if not row:
+        conn.execute("ALTER TABLE tasks ADD COLUMN slug TEXT")
+        conn.execute("UPDATE tasks SET slug = id WHERE slug IS NULL")
+        conn.execute("ALTER TABLE tasks ALTER COLUMN slug SET NOT NULL")
+    row = conn.execute(
+        "SELECT 1 FROM information_schema.columns"
+        " WHERE table_name = 'tasks' AND column_name = 'owner'"
+    ).fetchone()
+    if not row:
+        conn.execute("ALTER TABLE tasks ADD COLUMN owner TEXT NOT NULL DEFAULT 'hive'")
 
 
 # --- Async connection pool (one per worker process) ---
diff --git a/src/hive/server/main.py b/src/hive/server/main.py
index fde511e..6fee4fb 100644
--- a/src/hive/server/main.py
+++ b/src/hive/server/main.py
@@ -341,8 +341,9 @@ def _sync_tasks_from_github():
                     continue
                 desc = repo.get("description") or ""
                 conn.execute(
-                    "INSERT INTO tasks (id, name, description, repo_url, created_at) VALUES (%s, %s, %s, %s, %s)",
-                    (task_id, task_id, desc, repo["html_url"], now()),
+                    "INSERT INTO tasks (id, slug, owner, name, description, repo_url, created_at)"
+                    " VALUES (%s, %s, %s, %s, %s, %s, %s)",
+                    (task_id, task_id, "hive", task_id, desc, repo["html_url"], now()),
                 )
     except Exception:
         pass  # best-effort; server starts even if GitHub is unreachable
@@ -886,32 +887,37 @@ def _validate_task_description(description: str):
 @router.post("/tasks", status_code=201)
 async def create_task(
     archive: UploadFile = File(...),
-    id: str = Form(...),
+    id: str | None = Form(None),
+    slug: str | None = Form(None),
     name: str = Form(...),
     description: str = Form(...),
     config: str | None = Form(None),
     x_admin_key: str = Header(""), authorization: str = Header(""),
 ):
     await require_admin(x_admin_key, authorization)
-    _validate_task_id(id)
+    task_id = (id or slug or "").strip()
+    if not task_id:
+        raise HTTPException(400, "id (or slug) is required")
+    _validate_task_id(task_id)
     _validate_task_description(description)
     async with get_db() as conn:
-        if await (await conn.execute("SELECT id FROM tasks WHERE id = %s", (id,))).fetchone():
+        if await (await conn.execute("SELECT id FROM tasks WHERE id = %s", (task_id,))).fetchone():
             raise HTTPException(409, "A public or private task with this ID already exists. Try a different ID.")
     try:
         gh = get_github_app()
     except Exception as e:
         raise HTTPException(503, f"GitHub App not configured: {e}")
     try:
-        repo_url = await asyncio.to_thread(gh.create_task_repo, id, archive.file.read(), description)
+        repo_url = await asyncio.to_thread(gh.create_task_repo, task_id, archive.file.read(), description)
     except Exception as e:
         raise HTTPException(502, f"Failed to create GitHub repo: {e}")
     async with get_db() as conn:
         await conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, config, created_at) VALUES (%s, %s, %s, %s, %s, %s)",
-            (id, name, description, repo_url, config, now()),
+            "INSERT INTO tasks (id, slug, owner, name, description, repo_url, config, created_at)"
+            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s)",
+            (task_id, task_id, "hive", name, description, repo_url, config, now()),
         )
-    return JSONResponse({"id": id, "name": name, "repo_url": repo_url, "status": "active"}, status_code=201)
+    return JSONResponse({"id": task_id, "name": name, "repo_url": repo_url, "status": "active"}, status_code=201)
 
 
 @router.get("/tasks/mine")
@@ -992,9 +998,9 @@ def _validate_repo():
             except Exception:
                 pass  # best-effort
         await conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, task_type, owner_id, visibility, source_repo, installation_id, created_at)"
-            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
-            (task_id, task_name, description, repo_url, "private", user_id, "private", repo_full_name, installation_id, now()),
+            "INSERT INTO tasks (id, slug, owner, name, description, repo_url, task_type, owner_id, visibility, source_repo, installation_id, created_at)"
+            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
+            (task_id, task_id, str(user_id), task_name, description, repo_url, "private", user_id, "private", repo_full_name, installation_id, now()),
         )
     resp_body: dict[str, Any] = {
         "id": task_id, "name": task_name, "repo_url": repo_url,