diff --git a/ghost/core/core/server/models/tag.js b/ghost/core/core/server/models/tag.js
index 31177ad0a80..276cec66157 100644
--- a/ghost/core/core/server/models/tag.js
+++ b/ghost/core/core/server/models/tag.js
@@ -176,7 +176,7 @@ Tag = ghostBookshelf.Model.extend({
         return {
             posts(modelOrCollection, options) {
                 modelOrCollection.query('columns', 'tags.*', (qb) => {
-                    qb.count('posts.id')
+                    qb.countDistinct('posts.id')
                         .from('posts')
                         .leftOuterJoin('posts_tags', 'posts.id', 'posts_tags.post_id')
                         .whereRaw('posts_tags.tag_id = tags.id')
diff --git a/ghost/core/core/server/services/posts/posts-service.js b/ghost/core/core/server/services/posts/posts-service.js
index 2558d235247..0f4e32109dc 100644
--- a/ghost/core/core/server/services/posts/posts-service.js
+++ b/ghost/core/core/server/services/posts/posts-service.js
@@ -175,8 +175,12 @@ class PostsService {
                     message: tpl(messages.invalidTags)
                 });
             }
+            // null and undefined mean the field was not supplied, anything else
+            // that is not a string is a malformed value rather than an absent one
+            const isMalformed = value => value !== undefined && value !== null && typeof value !== 'string';
+
             for (const tag of data.meta.tags) {
-                if (typeof tag !== 'object') {
+                if (!tag || typeof tag !== 'object') {
                     throw new errors.IncorrectUsageError({
                         message: tpl(messages.invalidTags)
                     });
@@ -186,6 +190,11 @@ class PostsService {
                         message: tpl(messages.invalidTags)
                     });
                 }
+                if (isMalformed(tag.id) || isMalformed(tag.name)) {
+                    throw new errors.IncorrectUsageError({
+                        message: tpl(messages.invalidTags)
+                    });
+                }
             }
 
             const bulkResult = await this.#bulkAddTags({tags: data.meta.tags}, {filter: options.filter, context: options.context});
@@ -217,28 +226,76 @@ class PostsService {
             });
         }
 
+        // The same tag can be passed more than once in a single request. Dedupe
+        // before creating, otherwise a repeated name-only tag is created twice
+        const seen = new Set();
+        const tags = data.tags.filter((tag) => {
+            const key = tag.id ? `id:${tag.id}` : `name:${tag.name.toLocaleLowerCase()}`;
+
+            if (seen.has(key)) {
+                return false;
+            }
+
+            seen.add(key);
+            return true;
+        });
+
         // Create tags that don't exist
-        for (const tag of data.tags) {
-            if (!tag.id) {
-                const createdTag = await this.models.Tag.add(tag, {transacting: options.transacting, context: options.context});
-                tag.id = createdTag.id;
+        for (const tag of tags) {
+            if (tag.id) {
+                continue;
             }
+
+            // The tag may already exist. Match on the slug it generates rather
+            // than the name, because the slug is unique and case insensitive on
+            // every supported database. Derive it the way Tag.onSaving does, so
+            // a supplied slug is looked up rather than silently ignored
+            const slug = await this.models.Base.Model.generateSlug(this.models.Tag, tag.slug || tag.name, {skipDuplicateChecks: true});
+            const existingTag = await this.models.Tag.findOne({slug}, {transacting: options.transacting});
+
+            if (existingTag) {
+                tag.id = existingTag.id;
+                continue;
+            }
+
+            const createdTag = await this.models.Tag.add(tag, {transacting: options.transacting, context: options.context});
+            tag.id = createdTag.id;
         }
 
         const postRows = await this.#getFilteredBulkPostQuery(options).select('posts.id');
+        const postIds = postRows.map(post => post.id);
+
+        const tagIds = [...new Set(tags.map(tag => tag.id))];
+
+        // Posts in the selection may already carry the tag, filter those pairs out
+        const existingRows = tagIds.length && postIds.length ?
+            await options.transacting('posts_tags')
+                .whereIn('tag_id', tagIds)
+                .whereIn('post_id', postIds)
+                .select('post_id', 'tag_id') :
+            [];
+        const existing = new Set(existingRows.map(row => `${row.post_id}:${row.tag_id}`));
+
+        const postTags = [];
+        for (const tagId of tagIds) {
+            for (const postId of postIds) {
+                if (existing.has(`${postId}:${tagId}`)) {
+                    continue;
+                }
 
-        const postTags = data.tags.reduce((pt, tag) => {
-            return pt.concat(postRows.map((post) => {
-                return {
+                postTags.push({
                     id: (new ObjectId()).toHexString(),
-                    post_id: post.id,
-                    tag_id: tag.id,
+                    post_id: postId,
+                    tag_id: tagId,
                     sort_order: 0
-                };
-            }));
-        }, []);
+                });
+            }
+        }
+
+        if (postTags.length) {
+            await options.transacting('posts_tags').insert(postTags);
+        }
 
-        await options.transacting('posts_tags').insert(postTags);
         await this.models.Post.addActions('edited', postRows.map(p => p.id), options);
 
         return {
diff --git a/ghost/core/test/e2e-api/admin/posts-bulk.test.js b/ghost/core/test/e2e-api/admin/posts-bulk.test.js
index 9488bf5ed08..59d9f6bd0ea 100644
--- a/ghost/core/test/e2e-api/admin/posts-bulk.test.js
+++ b/ghost/core/test/e2e-api/admin/posts-bulk.test.js
@@ -233,6 +233,162 @@ describe('Posts Bulk API', function () {
             }
         });
 
+        it('Does not add a tag to posts that already have it', async function () {
+            const filter = 'status:[published]';
+            const tag = await models.Tag.findOne({slug: fixtureManager.get('tags', 0).slug});
+            assert(tag);
+
+            const addTag = async () => {
+                await agent
+                    .put('/posts/bulk/?filter=' + encodeURIComponent(filter))
+                    .body({bulk: {action: 'addTag', meta: {tags: [{id: tag.id}]}}})
+                    .expectStatus(200);
+            };
+
+            // Apply the tag, then apply the same tag again - which the bulk action in Admin allows
+            await addTag();
+            await addTag();
+
+            const duplicates = await models.Base.knex('posts_tags')
+                .where('tag_id', tag.id)
+                .select('post_id')
+                .count('* as rows')
+                .groupBy('post_id')
+                .having('rows', '>', 1);
+
+            assert.equal(duplicates.length, 0, `Expect no duplicate posts_tags rows, got ${JSON.stringify(duplicates)}`);
+        });
+
+        it('Does not add the same tag twice when it is passed twice', async function () {
+            const filter = 'status:[draft]';
+            const tag = await models.Tag.findOne({slug: fixtureManager.get('tags', 2).slug});
+            assert(tag);
+
+            await agent
+                .put('/posts/bulk/?filter=' + encodeURIComponent(filter))
+                .body({bulk: {action: 'addTag', meta: {tags: [{id: tag.id}, {id: tag.id}]}}})
+                .expectStatus(200);
+
+            const duplicates = await models.Base.knex('posts_tags')
+                .where('tag_id', tag.id)
+                .select('post_id')
+                .count('* as rows')
+                .groupBy('post_id')
+                .having('rows', '>', 1);
+
+            assert.equal(duplicates.length, 0, `Expect no duplicate posts_tags rows, got ${JSON.stringify(duplicates)}`);
+        });
+
+        it('Only creates one tag when the same new tag name is passed twice', async function () {
+            const filter = 'status:[draft]';
+            const name = 'Repeated new tag';
+
+            await agent
+                .put('/posts/bulk/?filter=' + encodeURIComponent(filter))
+                .body({bulk: {action: 'addTag', meta: {tags: [{name}, {name}]}}})
+                .expectStatus(200);
+
+            const tags = await models.Tag.findAll({filter: `name:'${name}'`});
+            assert.equal(tags.length, 1, `Expect a single tag to be created, got ${JSON.stringify(tags.models.map(t => t.get('slug')))}`);
+
+            const duplicates = await models.Base.knex('posts_tags')
+                .where('tag_id', tags.models[0].id)
+                .select('post_id')
+                .count('* as rows')
+                .groupBy('post_id')
+                .having('rows', '>', 1);
+
+            assert.equal(duplicates.length, 0, `Expect no duplicate posts_tags rows, got ${JSON.stringify(duplicates)}`);
+        });
+
+        it('Reuses an existing tag when it is added by name', async function () {
+            const filter = 'status:[draft]';
+            const existing = await models.Tag.add({name: 'Already here'}, {context: {internal: true}});
+
+            const countTagsLike = async () => {
+                const rows = await models.Base.knex('tags')
+                    .whereRaw('LOWER(name) = ?', ['already here'])
+                    .select('id');
+                return rows.length;
+            };
+
+            // Same name, and the same name in a different case, both belong to
+            // the tag that is already there
+            for (const name of ['Already here', 'ALREADY HERE']) {
+                await agent
+                    .put('/posts/bulk/?filter=' + encodeURIComponent(filter))
+                    .body({bulk: {action: 'addTag', meta: {tags: [{name}]}}})
+                    .expectStatus(200);
+
+                assert.equal(await countTagsLike(), 1, `Expect no new tag to be created for "${name}"`);
+            }
+
+            const posts = await models.Post.findAll({filter, status: 'all', withRelated: ['tags']});
+            for (const post of posts) {
+                assert(post.related('tags').find(t => t.id === existing.id), `Expect post ${post.id} to have the existing tag`);
+            }
+        });
+
+        it('Reuses an existing tag when it is added by slug', async function () {
+            const filter = 'status:[draft]';
+            const existing = await models.Tag.add({name: 'Slug owner', slug: 'slug-owner'}, {context: {internal: true}});
+
+            // Tag.add honours a supplied slug, so the lookup has to use it too
+            await agent
+                .put('/posts/bulk/?filter=' + encodeURIComponent(filter))
+                .body({bulk: {action: 'addTag', meta: {tags: [{name: 'A different name', slug: 'slug-owner'}]}}})
+                .expectStatus(200);
+
+            const owners = await models.Base.knex('tags').where('slug', 'like', 'slug-owner%').select('slug');
+            assert.deepEqual(owners.map(t => t.slug), ['slug-owner'], 'Expect no second tag to be created');
+
+            const posts = await models.Post.findAll({filter, status: 'all', withRelated: ['tags']});
+            for (const post of posts) {
+                assert(post.related('tags').find(t => t.id === existing.id), `Expect post ${post.id} to have the existing tag`);
+            }
+        });
+
+        it('Does not confuse a tag id with another tag of the same name', async function () {
+            const filter = 'status:[draft]';
+            const existing = await models.Tag.findOne({slug: fixtureManager.get('tags', 1).slug});
+            assert(existing);
+
+            // A tag named after another tag's id - contrived, but the two are
+            // deduplicated against each other if they share a key space
+            await agent
+                .put('/posts/bulk/?filter=' + encodeURIComponent(filter))
+                .body({bulk: {action: 'addTag', meta: {tags: [{id: existing.id}, {name: existing.id}]}}})
+                .expectStatus(200);
+
+            const named = await models.Tag.findAll({filter: `name:'${existing.id}'`});
+            assert.equal(named.length, 1, 'Expect the tag named after the id to still be created');
+        });
+
+        it('Rejects tags that are not usable objects', async function () {
+            const validTag = await models.Tag.findOne({slug: fixtureManager.get('tags', 0).slug});
+            const invalidTags = [
+                [null],
+                ['a-tag-id'],
+                [1],
+                [true],
+                [{id: 1}],
+                [{name: 1}],
+                [{name: {}}],
+                // Falsy non-strings are supplied values too, not absent ones
+                [{id: 0, name: 'Falsy id'}],
+                [{id: false, name: 'Falsy id'}],
+                [{name: 0, id: validTag.id}],
+                [{name: false, id: validTag.id}]
+            ];
+
+            for (const tags of invalidTags) {
+                await agent
+                    .put('/posts/bulk/?filter=' + encodeURIComponent('status:[draft]'))
+                    .body({bulk: {action: 'addTag', meta: {tags}}})
+                    .expectStatus(400);
+            }
+        });
+
         it('Can add multiple tags to posts and create new tags', async function () {
             const filter = 'status:[draft]';
             const tag = await models.Tag.findOne({id: fixtureManager.get('tags', 1).id});
diff --git a/ghost/core/test/e2e-api/admin/tags.test.js b/ghost/core/test/e2e-api/admin/tags.test.js
index 180081851ae..4bacf8bc417 100644
--- a/ghost/core/test/e2e-api/admin/tags.test.js
+++ b/ghost/core/test/e2e-api/admin/tags.test.js
@@ -1,10 +1,12 @@
 const assert = require('node:assert/strict');
+const ObjectId = require('bson-objectid').default;
 const {assertExists} = require('../../utils/assertions');
 const sinon = require('sinon');
 const supertest = require('supertest');
 const testUtils = require('../../utils');
 const config = require('../../../core/shared/config');
 const localUtils = require('./utils');
+const models = require('../../../core/server/models');
 const urlUtilsHelper = require('../../utils/url-utils');
 
 describe('Tag API', function () {
@@ -89,6 +91,39 @@ describe('Tag API', function () {
         assert.equal(new URL(jsonResponse.tags[0].url).pathname, '/tag/getting-started/');
     });
 
+    it('Counts posts, not posts_tags rows', async function () {
+        const tagId = testUtils.getExistingData().tags[0].id;
+
+        const readCount = async () => {
+            const res = await request
+                .get(localUtils.API.getApiQuery(`tags/${tagId}/?include=count.posts`))
+                .set('Origin', config.get('url'))
+                .expect(200);
+
+            return res.body.tags[0].count.posts;
+        };
+
+        const before = await readCount();
+        assert(before > 0, 'Expect the tag to have posts for this test to work');
+
+        const existing = await models.Base.knex('posts_tags')
+            .where('tag_id', tagId)
+            .select('post_id');
+        const redundant = existing.map(row => ({
+            id: new ObjectId().toHexString(),
+            post_id: row.post_id,
+            tag_id: tagId,
+            sort_order: 0
+        }));
+        await models.Base.knex('posts_tags').insert(redundant);
+
+        try {
+            assert.equal(await readCount(), before, 'Expect duplicate posts_tags rows not to inflate count.posts');
+        } finally {
+            await models.Base.knex('posts_tags').whereIn('id', redundant.map(r => r.id)).del();
+        }
+    });
+
     it('Can add a tag', async function () {
         const tag = testUtils.DataGenerator.forKnex.createTag();
 
diff --git a/ghost/core/test/unit/server/models/tag.test.js b/ghost/core/test/unit/server/models/tag.test.js
index ce7dd4baa75..2dd3436bafb 100644
--- a/ghost/core/test/unit/server/models/tag.test.js
+++ b/ghost/core/test/unit/server/models/tag.test.js
@@ -42,7 +42,7 @@ describe('Unit: models/tag', function () {
             }).then(() => {
                 assert.equal(queries.length, 1);
 
-                assert.equal(queries[0].sql, 'select `tags`.*, (select count(`posts`.`id`) from `posts` left outer join `posts_tags` on `posts`.`id` = `posts_tags`.`post_id` where posts_tags.tag_id = tags.id) as `count__posts` from `tags` where `count`.`posts` >= ? order by `count__posts` DESC');
+                assert.equal(queries[0].sql, 'select `tags`.*, (select count(distinct `posts`.`id`) from `posts` left outer join `posts_tags` on `posts`.`id` = `posts_tags`.`post_id` where posts_tags.tag_id = tags.id) as `count__posts` from `tags` where `count`.`posts` >= ? order by `count__posts` DESC');
                 assert.deepEqual(queries[0].bindings, [
                     1
                 ]);