diff --git a/routes/task_routes.py b/routes/task_routes.py
index fb1a7c746..d786c5730 100644
--- a/routes/task_routes.py
+++ b/routes/task_routes.py
@@ -11,10 +11,14 @@ from fastapi import APIRouter, HTTPException, Request
 from pydantic import BaseModel
 
 from core.database import SessionLocal, ScheduledTask, TaskRun
-from core.middleware import INTERNAL_TOOL_USER
 from core.constants import internal_api_base
 from src.auth_helpers import get_current_user
 from src.constants import DATA_DIR, EMAIL_URGENCY_CACHE_DIR
+from src.task_action_policy import (
+    ADMIN_ONLY_TASK_ACTIONS,
+    is_admin_only_task_action,
+    owner_has_admin_task_privileges,
+)
 from src.task_scheduler import compute_next_run, HOUSEKEEPING_DEFAULTS
 from routes.prefs_routes import _load_for_user, _save_for_user
 
@@ -417,28 +421,18 @@ def setup_task_routes(task_scheduler) -> APIRouter:
                 db.close()
         return {"ok": True, "opened": True, "enabled": bool(prefs.get("tasks_enabled")), "resumed": resumed}
 
-    # Actions that execute shell/SSH commands — restricted to admins.
+    # Actions that execute shell/SSH commands or cross into admin-only
+    # Cookbook serving surfaces — restricted to admins.
     # Non-admin users cannot create tasks with these action types via the
     # API. See review CRIT-C.
-    _ADMIN_ONLY_ACTIONS = {"run_local", "run_script", "ssh_command"}
+    _ADMIN_ONLY_ACTIONS = ADMIN_ONLY_TASK_ACTIONS
 
     def _is_admin(user: str | None) -> bool:
-        if not user:
-            return False
-        # In-process tool-loopback marker — AuthMiddleware validated
-        # the internal token + loopback client before stamping this,
-        # so treat as admin-equivalent.
-        if user == INTERNAL_TOOL_USER:
-            return True
-        try:
-            from core.auth import AuthManager
-            auth = AuthManager()
-            if not auth.is_configured:
-                # Unconfigured single-user deploy: trust the local owner.
-                return True
-            return bool(auth.is_admin(user))
-        except Exception:
-            return False
+        return owner_has_admin_task_privileges(user)
+
+    def _require_admin_for_task_action(user: str | None, task_type: str | None, action: str | None) -> None:
+        if is_admin_only_task_action(task_type, action) and not _is_admin(user):
+            raise HTTPException(403, f"Action '{action}' requires admin privileges")
 
     def _validate_then_task_id(db, then_task_id: Optional[str], user: Optional[str], current_task_id: Optional[str] = None) -> Optional[str]:
         target_id = (then_task_id or "").strip()
@@ -466,8 +460,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
         # Block shell-executing action types for non-admins. action_run_local
         # uses subprocess.run(shell=True) and ssh_command / run_script run
         # arbitrary commands.
-        if req.task_type == "action" and req.action in _ADMIN_ONLY_ACTIONS and not _is_admin(user):
-            raise HTTPException(403, f"Action '{req.action}' requires admin privileges")
+        _require_admin_for_task_action(user, req.task_type, req.action)
         if req.trigger_type == "schedule" and not req.schedule:
             raise HTTPException(400, "Schedule is required for schedule-triggered tasks")
         if req.trigger_type == "schedule" and req.schedule == "cron" and not req.cron_expression:
@@ -681,6 +674,10 @@ def setup_task_routes(task_scheduler) -> APIRouter:
             if user and task.owner != user:
                 raise HTTPException(403, "Access denied")
 
+            next_task_type = req.task_type if req.task_type is not None else task.task_type
+            next_action = req.action if req.action is not None else task.action
+            _require_admin_for_task_action(user, next_task_type, next_action)
+
             if req.name is not None:
                 task.name = req.name
             if req.prompt is not None:
@@ -688,9 +685,6 @@ def setup_task_routes(task_scheduler) -> APIRouter:
             if req.task_type is not None:
                 task.task_type = req.task_type
             if req.action is not None:
-                # Same admin-only gate as create — see CRIT-C.
-                if req.action in _ADMIN_ONLY_ACTIONS and not _is_admin(user):
-                    raise HTTPException(403, f"Action '{req.action}' requires admin privileges")
                 task.action = req.action
             if req.output_target is not None:
                 task.output_target = req.output_target
@@ -807,6 +801,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
                 raise HTTPException(404, "Task not found")
             if user and task.owner != user:
                 raise HTTPException(403, "Access denied")
+            _require_admin_for_task_action(user, task.task_type, task.action)
             task.status = "active"
             if (task.trigger_type or "schedule") == "schedule":
                 task.next_run = compute_next_run(
@@ -869,6 +864,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
                 raise HTTPException(404, "Task not found")
             if user and task.owner != user:
                 raise HTTPException(403, "Access denied")
+            _require_admin_for_task_action(user, task.task_type, task.action)
         finally:
             db.close()
         started = await task_scheduler.run_task_now(task_id, force=force)
@@ -1058,6 +1054,14 @@ def setup_task_routes(task_scheduler) -> APIRouter:
             ).first()
             if not task:
                 raise HTTPException(404, "Not found")
+            if (
+                is_admin_only_task_action(task.task_type, task.action)
+                and not owner_has_admin_task_privileges(task.owner)
+            ):
+                task.status = "paused"
+                task.next_run = None
+                db.commit()
+                raise HTTPException(403, f"Action '{task.action}' requires admin privileges")
         finally:
             db.close()
         started = await task_scheduler.run_task_now(task_id)
diff --git a/src/task_action_policy.py b/src/task_action_policy.py
new file mode 100644
index 000000000..76ae1337e
--- /dev/null
+++ b/src/task_action_policy.py
@@ -0,0 +1,47 @@
+"""Shared privilege policy for scheduled task actions."""
+
+from __future__ import annotations
+
+ADMIN_ONLY_TASK_ACTIONS = frozenset({
+    "run_local",
+    "run_script",
+    "ssh_command",
+    "cookbook_serve",
+})
+
+
+def is_admin_only_task_action(task_type: str | None, action: str | None) -> bool:
+    return (task_type or "llm") == "action" and (action or "") in ADMIN_ONLY_TASK_ACTIONS
+
+
+def owner_has_admin_task_privileges(owner: str | None) -> bool:
+    try:
+        from src.auth_helpers import _auth_disabled
+        if _auth_disabled():
+            return True
+    except Exception:
+        pass
+
+    if owner:
+        try:
+            from core.middleware import INTERNAL_TOOL_USER
+            if owner == INTERNAL_TOOL_USER:
+                return True
+        except Exception:
+            pass
+
+    try:
+        from core.auth import AuthManager
+        auth = AuthManager()
+        if not auth.is_configured:
+            return True
+        if not owner:
+            return False
+        return bool(auth.is_admin(owner))
+    except Exception:
+        pass
+
+    if not owner:
+        return False
+
+    return False
diff --git a/src/task_scheduler.py b/src/task_scheduler.py
index df515ebea..878edfbea 100644
--- a/src/task_scheduler.py
+++ b/src/task_scheduler.py
@@ -10,6 +10,10 @@ from datetime import datetime, timedelta, timezone
 from typing import Any, Awaitable, Callable, Dict, Tuple
 
 from core.auth import RESERVED_USERNAMES
+from src.task_action_policy import (
+    is_admin_only_task_action,
+    owner_has_admin_task_privileges,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -821,6 +825,28 @@ class TaskScheduler:
                     db.commit()
                 return
 
+            if (
+                is_admin_only_task_action(task.task_type, task.action)
+                and not owner_has_admin_task_privileges(task.owner)
+            ):
+                msg = f"Action '{task.action}' requires admin privileges"
+                blocked = db.query(TaskRun).filter(TaskRun.id == run_id).first()
+                if blocked:
+                    blocked.status = "error"
+                    blocked.result = msg
+                    blocked.error = msg
+                    blocked.finished_at = _utcnow()
+                task.status = "paused"
+                task.next_run = None
+                task.last_run = _utcnow()
+                logger.warning(
+                    "Paused admin-only task %s for non-admin owner %r",
+                    task_id,
+                    task.owner,
+                )
+                db.commit()
+                return
+
             if gate_foreground:
                 waiting = db.query(TaskRun).filter(TaskRun.id == run_id).first()
                 if waiting and waiting.status == "queued":
diff --git a/tests/test_auth_regressions.py b/tests/test_auth_regressions.py
index b16966e3a..62b479748 100644
--- a/tests/test_auth_regressions.py
+++ b/tests/test_auth_regressions.py
@@ -334,16 +334,11 @@ def test_pop_notifications_owner_filtered():
 def test_admin_only_actions_set_contains_shell_runners():
     """The constant defining shell-executing action types must include
     the three risky entries. Catches accidental removal."""
-    from routes import task_routes
-    # `_ADMIN_ONLY_ACTIONS` is a closure constant. Easiest pin: re-read
-    # the source and check for the three risky entries + the admin gate
-    # wording.
-    src = open(task_routes.__file__, encoding="utf-8").read()
-    assert '"run_local"' in src
-    assert '"run_script"' in src
-    assert '"ssh_command"' in src
-    # And the gate is wired into both create and update paths.
-    assert "Action '" in src and "requires admin privileges" in src
+    from src.task_action_policy import ADMIN_ONLY_TASK_ACTIONS
+
+    assert "run_local" in ADMIN_ONLY_TASK_ACTIONS
+    assert "run_script" in ADMIN_ONLY_TASK_ACTIONS
+    assert "ssh_command" in ADMIN_ONLY_TASK_ACTIONS
 
 
 def test_task_create_notification_default_allows_action_specific_defaults():
diff --git a/tests/test_task_cookbook_admin_gate.py b/tests/test_task_cookbook_admin_gate.py
new file mode 100644
index 000000000..d7e72f9ef
--- /dev/null
+++ b/tests/test_task_cookbook_admin_gate.py
@@ -0,0 +1,350 @@
+"""Task CRUD must not let non-admins schedule Cookbook serve actions."""
+
+import sys
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import NullPool
+
+from tests.helpers.import_state import clear_fake_database_modules
+
+clear_fake_database_modules()
+
+import core.auth as core_auth
+import core.database as cdb
+import routes.task_routes as task_routes
+from core.database import ScheduledTask
+from core.database import TaskRun
+from src.task_scheduler import TaskScheduler
+
+_REAL_DATABASE_ATTRS = {
+    "Base": cdb.Base,
+    "SessionLocal": cdb.SessionLocal,
+    "ScheduledTask": ScheduledTask,
+    "TaskRun": TaskRun,
+}
+if hasattr(cdb, "engine"):
+    _REAL_DATABASE_ATTRS["engine"] = cdb.engine
+
+
+def _restore_module_binding(monkeypatch, name, module):
+    monkeypatch.setitem(sys.modules, name, module)
+    parent_name, _, attr = name.rpartition(".")
+    parent = sys.modules.get(parent_name)
+    if parent is not None:
+        monkeypatch.setattr(parent, attr, module, raising=False)
+
+
+@pytest.fixture()
+def task_db(monkeypatch, tmp_path):
+    _restore_module_binding(monkeypatch, "core.database", cdb)
+    for attr, value in _REAL_DATABASE_ATTRS.items():
+        monkeypatch.setattr(cdb, attr, value, raising=False)
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'tasks.db'}",
+        connect_args={"check_same_thread": False},
+        poolclass=NullPool,
+    )
+    cdb.Base.metadata.create_all(engine)
+    testing_session = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+    monkeypatch.setattr(task_routes, "SessionLocal", testing_session)
+    monkeypatch.setattr(cdb, "SessionLocal", testing_session)
+    return testing_session
+
+
+@pytest.fixture()
+def configured_auth(monkeypatch):
+    _restore_module_binding(monkeypatch, "core.auth", core_auth)
+    monkeypatch.setenv("AUTH_ENABLED", "true")
+
+    class FakeAuthManager:
+        is_configured = True
+
+        def is_admin(self, user):
+            return user == "admin"
+
+    monkeypatch.setattr(core_auth, "AuthManager", FakeAuthManager)
+
+
+@pytest.fixture()
+def builtin_action_info(monkeypatch):
+    mod = sys.modules.get("src.builtin_actions")
+    if mod is None:
+        import src.builtin_actions as mod
+    monkeypatch.setattr(
+        mod,
+        "BUILTIN_ACTION_INFO",
+        {
+            "summarize_emails": "Summarize emails",
+            "cookbook_serve": "Serve Cookbook model",
+        },
+        raising=False,
+    )
+
+
+def _req(user):
+    return SimpleNamespace(state=SimpleNamespace(current_user=user))
+
+
+def _endpoint(method, path):
+    router = task_routes.setup_task_routes(MagicMock())
+    for route in router.routes:
+        if getattr(route, "path", None) == path and method in getattr(route, "methods", set()):
+            return route.endpoint
+    raise RuntimeError(f"{method} {path} not found")
+
+
+def _cookbook_create_req():
+    return task_routes.TaskCreate(
+        name="Serve test model",
+        prompt="{}",
+        task_type="action",
+        action="cookbook_serve",
+        trigger_type="webhook",
+    )
+
+
+def _seed_action_task(
+    session_factory,
+    task_id,
+    owner,
+    action="summarize_emails",
+    *,
+    task_type="action",
+    webhook_token=None,
+    next_run=None,
+):
+    db = session_factory()
+    try:
+        task = ScheduledTask(
+            id=task_id,
+            owner=owner,
+            name=task_id,
+            prompt="{}",
+            task_type=task_type,
+            action=action,
+            trigger_type="webhook",
+            status="active",
+            output_target="session",
+            webhook_token=webhook_token,
+            next_run=next_run,
+        )
+        db.add(task)
+        db.commit()
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_non_admin_cannot_create_cookbook_serve_task(task_db, configured_auth):
+    create_task = _endpoint("POST", "/api/tasks")
+
+    with pytest.raises(HTTPException) as exc:
+        await create_task(_req("alice"), _cookbook_create_req())
+
+    assert exc.value.status_code == 403
+    db = task_db()
+    try:
+        assert db.query(ScheduledTask).count() == 0
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_non_admin_cannot_update_task_to_cookbook_serve(task_db, configured_auth):
+    _seed_action_task(task_db, "alice-task", "alice")
+    update_task = _endpoint("PUT", "/api/tasks/{task_id}")
+
+    with pytest.raises(HTTPException) as exc:
+        await update_task(
+            _req("alice"),
+            "alice-task",
+            task_routes.TaskUpdate(action="cookbook_serve"),
+        )
+
+    assert exc.value.status_code == 403
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == "alice-task").first()
+        assert task.action == "summarize_emails"
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_non_admin_cannot_update_task_type_to_activate_existing_cookbook_serve(
+    task_db, configured_auth
+):
+    _seed_action_task(
+        task_db,
+        "alice-task",
+        "alice",
+        action="cookbook_serve",
+        task_type="llm",
+    )
+    update_task = _endpoint("PUT", "/api/tasks/{task_id}")
+
+    with pytest.raises(HTTPException) as exc:
+        await update_task(
+            _req("alice"),
+            "alice-task",
+            task_routes.TaskUpdate(task_type="action"),
+        )
+
+    assert exc.value.status_code == 403
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == "alice-task").first()
+        assert task.task_type == "llm"
+        assert task.action == "cookbook_serve"
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_non_admin_cannot_manually_run_existing_cookbook_serve_task(
+    task_db, configured_auth
+):
+    _seed_action_task(task_db, "alice-task", "alice", action="cookbook_serve")
+    scheduler = SimpleNamespace(run_task_now=MagicMock())
+    router = task_routes.setup_task_routes(scheduler)
+    for route in router.routes:
+        if getattr(route, "path", None) == "/api/tasks/{task_id}/run":
+            run_task = route.endpoint
+            break
+    else:
+        raise RuntimeError("POST /api/tasks/{task_id}/run not found")
+
+    with pytest.raises(HTTPException) as exc:
+        await run_task(_req("alice"), "alice-task")
+
+    assert exc.value.status_code == 403
+    scheduler.run_task_now.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_webhook_rejects_stale_non_admin_cookbook_serve_task(
+    task_db, configured_auth
+):
+    _seed_action_task(
+        task_db,
+        "alice-task",
+        "alice",
+        action="cookbook_serve",
+        webhook_token="secret",
+    )
+    webhook_trigger = _endpoint("POST", "/api/tasks/{task_id}/webhook/{token}")
+
+    with pytest.raises(HTTPException) as exc:
+        await webhook_trigger("alice-task", "secret")
+
+    assert exc.value.status_code == 403
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == "alice-task").first()
+        assert task.status == "paused"
+        assert task.next_run is None
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_scheduler_pauses_stale_non_admin_cookbook_serve_task(
+    task_db, configured_auth
+):
+    due = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(minutes=1)
+    _seed_action_task(
+        task_db,
+        "alice-task",
+        "alice",
+        action="cookbook_serve",
+        next_run=due,
+    )
+    db = task_db()
+    try:
+        db.add(TaskRun(id="run-1", task_id="alice-task", status="queued"))
+        db.commit()
+    finally:
+        db.close()
+
+    scheduler = TaskScheduler.__new__(TaskScheduler)
+    scheduler._task_handles = {}
+    await scheduler._execute_task_locked(
+        "alice-task",
+        "run-1",
+        gate_foreground=False,
+        release_executing=False,
+    )
+
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == "alice-task").first()
+        run = db.query(TaskRun).filter(TaskRun.id == "run-1").first()
+        assert task.status == "paused"
+        assert task.next_run is None
+        assert run.status == "error"
+        assert run.error == "Action 'cookbook_serve' requires admin privileges"
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_non_admin_action_metadata_hides_cookbook_serve(
+    configured_auth, builtin_action_info
+):
+    list_actions = _endpoint("GET", "/api/tasks/meta/actions")
+
+    out = await list_actions(_req("alice"))
+
+    action_names = {action["name"] for action in out["actions"]}
+    assert "cookbook_serve" not in action_names
+
+
+@pytest.mark.asyncio
+async def test_admin_can_create_cookbook_serve_task(task_db, configured_auth):
+    create_task = _endpoint("POST", "/api/tasks")
+
+    out = await create_task(_req("admin"), _cookbook_create_req())
+
+    assert out["action"] == "cookbook_serve"
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == out["id"]).first()
+        assert task.owner == "admin"
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_admin_action_metadata_includes_cookbook_serve(
+    configured_auth, builtin_action_info
+):
+    list_actions = _endpoint("GET", "/api/tasks/meta/actions")
+
+    out = await list_actions(_req("admin"))
+
+    action_names = {action["name"] for action in out["actions"]}
+    assert "cookbook_serve" in action_names
+
+
+@pytest.mark.asyncio
+async def test_auth_disabled_single_user_can_create_cookbook_serve_task(
+    monkeypatch, task_db
+):
+    monkeypatch.setenv("AUTH_ENABLED", "false")
+    create_task = _endpoint("POST", "/api/tasks")
+
+    out = await create_task(_req(None), _cookbook_create_req())
+
+    assert out["action"] == "cookbook_serve"
+    db = task_db()
+    try:
+        task = db.query(ScheduledTask).filter(ScheduledTask.id == out["id"]).first()
+        assert task.owner is None
+    finally:
+        db.close()