feat: add animal tagging projection and service

Implement AnimalTagged and AnimalTagEnded event handling:
- Add migration for tag_suggestions table
- Create TagProjection for tag intervals and suggestions
- Update EventAnimalsProjection to handle tag events
- Add add_tag() and end_tag() to AnimalService

Key behaviors:
- No-op idempotence (adding active tag or ending inactive tag)
- Updates live_animals_by_location.tags JSON array
- Tracks tag usage statistics in tag_suggestions

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

migrations/0005-tag-suggestions.sql

@@ -0,0 +1,15 @@
+-- ABOUTME: Migration to create tag_suggestions table.
+-- ABOUTME: Tracks tag usage for autocomplete in the UI.
+
+-- Tag suggestions for autocomplete
+-- Updated synchronously during tagging operations
+CREATE TABLE tag_suggestions (
+    tag TEXT PRIMARY KEY,
+    total_assignments INTEGER NOT NULL DEFAULT 0,
+    active_animals INTEGER NOT NULL DEFAULT 0,
+    last_used_utc INTEGER,
+    updated_at_utc INTEGER NOT NULL
+);
+
+-- Index for sorting by popularity
+CREATE INDEX idx_tag_suggestions_popularity ON tag_suggestions(active_animals DESC, total_assignments DESC);

src/animaltrack/projections/event_animals.py

@@ -7,6 +7,8 @@ from animaltrack.events.types import (
     ANIMAL_ATTRIBUTES_UPDATED,
     ANIMAL_COHORT_CREATED,
     ANIMAL_MOVED,
+    ANIMAL_TAG_ENDED,
+    ANIMAL_TAGGED,
 )
 from animaltrack.models.events import Event
 from animaltrack.projections.base import Projection
@@ -30,7 +32,13 @@ class EventAnimalsProjection(Projection):
 
     def get_event_types(self) -> list[str]:
         """Return the event types this projection handles."""
-        return [ANIMAL_COHORT_CREATED, ANIMAL_MOVED, ANIMAL_ATTRIBUTES_UPDATED]
+        return [
+            ANIMAL_COHORT_CREATED,
+            ANIMAL_MOVED,
+            ANIMAL_ATTRIBUTES_UPDATED,
+            ANIMAL_TAGGED,
+            ANIMAL_TAG_ENDED,
+        ]
 
     def apply(self, event: Event) -> None:
         """Link event to affected animals."""

src/animaltrack/projections/tags.py

@@ -0,0 +1,235 @@
+# ABOUTME: Projection for animal tag intervals and tag suggestions.
+# ABOUTME: Handles AnimalTagged and AnimalTagEnded events.
+
+import json
+from typing import Any
+
+from animaltrack.events.types import ANIMAL_TAG_ENDED, ANIMAL_TAGGED
+from animaltrack.models.events import Event
+from animaltrack.projections.base import Projection
+
+
+class TagProjection(Projection):
+    """Maintains tag intervals and tag suggestions.
+
+    This projection handles tag add/end events, maintaining:
+    - animal_tag_intervals: Historical record of when animals had tags
+    - live_animals_by_location.tags: Current tags JSON array
+    - tag_suggestions: Usage statistics for autocomplete
+    """
+
+    def __init__(self, db: Any) -> None:
+        """Initialize the projection with a database connection.
+
+        Args:
+            db: A fastlite database connection.
+        """
+        super().__init__(db)
+
+    def get_event_types(self) -> list[str]:
+        """Return the event types this projection handles."""
+        return [ANIMAL_TAGGED, ANIMAL_TAG_ENDED]
+
+    def apply(self, event: Event) -> None:
+        """Apply tag event to update projections."""
+        if event.type == ANIMAL_TAGGED:
+            self._apply_animal_tagged(event)
+        elif event.type == ANIMAL_TAG_ENDED:
+            self._apply_animal_tag_ended(event)
+
+    def revert(self, event: Event) -> None:
+        """Revert tag event from projections."""
+        if event.type == ANIMAL_TAGGED:
+            self._revert_animal_tagged(event)
+        elif event.type == ANIMAL_TAG_ENDED:
+            self._revert_animal_tag_ended(event)
+
+    def _apply_animal_tagged(self, event: Event) -> None:
+        """Add tag to animals.
+
+        For each animal that doesn't already have this tag active:
+        - Create open interval in animal_tag_intervals
+        - Add tag to live_animals_by_location.tags JSON
+        - Update tag_suggestions counters
+        """
+        tag = event.entity_refs.get("tag")
+        ts_utc = event.ts_utc
+        actually_tagged = event.entity_refs.get("actually_tagged", [])
+
+        for animal_id in actually_tagged:
+            # Create new open interval
+            self.db.execute(
+                """
+                INSERT INTO animal_tag_intervals (animal_id, tag, start_utc, end_utc)
+                VALUES (?, ?, ?, NULL)
+                """,
+                (animal_id, tag, ts_utc),
+            )
+
+            # Update tags JSON in live_animals_by_location
+            self._add_tag_to_live_animals(animal_id, tag)
+
+        # Update tag suggestions
+        if actually_tagged:
+            self._update_tag_suggestions_on_add(tag, len(actually_tagged), ts_utc)
+
+    def _revert_animal_tagged(self, event: Event) -> None:
+        """Revert tag add by removing intervals and updating counts."""
+        tag = event.entity_refs.get("tag")
+        ts_utc = event.ts_utc
+        actually_tagged = event.entity_refs.get("actually_tagged", [])
+
+        for animal_id in actually_tagged:
+            # Delete the interval created by this event
+            self.db.execute(
+                """
+                DELETE FROM animal_tag_intervals
+                WHERE animal_id = ? AND tag = ? AND start_utc = ?
+                """,
+                (animal_id, tag, ts_utc),
+            )
+
+            # Remove tag from live_animals_by_location
+            self._remove_tag_from_live_animals(animal_id, tag)
+
+        # Revert tag suggestions
+        if actually_tagged:
+            self._revert_tag_suggestions_on_add(tag, len(actually_tagged))
+
+    def _apply_animal_tag_ended(self, event: Event) -> None:
+        """End tag for animals.
+
+        For each animal that has this tag active:
+        - Close the interval by setting end_utc
+        - Remove tag from live_animals_by_location.tags JSON
+        - Update tag_suggestions active_animals count
+        """
+        tag = event.entity_refs.get("tag")
+        ts_utc = event.ts_utc
+        actually_ended = event.entity_refs.get("actually_ended", [])
+
+        for animal_id in actually_ended:
+            # Close the open interval
+            self.db.execute(
+                """
+                UPDATE animal_tag_intervals
+                SET end_utc = ?
+                WHERE animal_id = ? AND tag = ? AND end_utc IS NULL
+                """,
+                (ts_utc, animal_id, tag),
+            )
+
+            # Remove tag from live_animals_by_location
+            self._remove_tag_from_live_animals(animal_id, tag)
+
+        # Update tag suggestions
+        if actually_ended:
+            self._update_tag_suggestions_on_end(tag, len(actually_ended), ts_utc)
+
+    def _revert_animal_tag_ended(self, event: Event) -> None:
+        """Revert tag end by reopening intervals."""
+        tag = event.entity_refs.get("tag")
+        ts_utc = event.ts_utc
+        actually_ended = event.entity_refs.get("actually_ended", [])
+
+        for animal_id in actually_ended:
+            # Reopen the interval
+            self.db.execute(
+                """
+                UPDATE animal_tag_intervals
+                SET end_utc = NULL
+                WHERE animal_id = ? AND tag = ? AND end_utc = ?
+                """,
+                (animal_id, tag, ts_utc),
+            )
+
+            # Add tag back to live_animals_by_location
+            self._add_tag_to_live_animals(animal_id, tag)
+
+        # Revert tag suggestions
+        if actually_ended:
+            self._revert_tag_suggestions_on_end(tag, len(actually_ended))
+
+    def _add_tag_to_live_animals(self, animal_id: str, tag: str) -> None:
+        """Add tag to the tags JSON array in live_animals_by_location."""
+        # Get current tags
+        row = self.db.execute(
+            "SELECT tags FROM live_animals_by_location WHERE animal_id = ?",
+            (animal_id,),
+        ).fetchone()
+
+        if row:
+            tags = json.loads(row[0])
+            if tag not in tags:
+                tags.append(tag)
+                self.db.execute(
+                    "UPDATE live_animals_by_location SET tags = ? WHERE animal_id = ?",
+                    (json.dumps(tags), animal_id),
+                )
+
+    def _remove_tag_from_live_animals(self, animal_id: str, tag: str) -> None:
+        """Remove tag from the tags JSON array in live_animals_by_location."""
+        row = self.db.execute(
+            "SELECT tags FROM live_animals_by_location WHERE animal_id = ?",
+            (animal_id,),
+        ).fetchone()
+
+        if row:
+            tags = json.loads(row[0])
+            if tag in tags:
+                tags.remove(tag)
+                self.db.execute(
+                    "UPDATE live_animals_by_location SET tags = ? WHERE animal_id = ?",
+                    (json.dumps(tags), animal_id),
+                )
+
+    def _update_tag_suggestions_on_add(self, tag: str, count: int, ts_utc: int) -> None:
+        """Update tag_suggestions when tags are added."""
+        self.db.execute(
+            """
+            INSERT INTO tag_suggestions (tag, total_assignments, active_animals, last_used_utc, updated_at_utc)
+            VALUES (?, ?, ?, ?, ?)
+            ON CONFLICT(tag) DO UPDATE SET
+                total_assignments = total_assignments + excluded.total_assignments,
+                active_animals = active_animals + excluded.active_animals,
+                last_used_utc = excluded.last_used_utc,
+                updated_at_utc = excluded.updated_at_utc
+            """,
+            (tag, count, count, ts_utc, ts_utc),
+        )
+
+    def _revert_tag_suggestions_on_add(self, tag: str, count: int) -> None:
+        """Revert tag_suggestions when add is reverted."""
+        self.db.execute(
+            """
+            UPDATE tag_suggestions
+            SET total_assignments = total_assignments - ?,
+                active_animals = active_animals - ?
+            WHERE tag = ?
+            """,
+            (count, count, tag),
+        )
+
+    def _update_tag_suggestions_on_end(self, tag: str, count: int, ts_utc: int) -> None:
+        """Update tag_suggestions when tags are ended."""
+        self.db.execute(
+            """
+            UPDATE tag_suggestions
+            SET active_animals = active_animals - ?,
+                last_used_utc = ?,
+                updated_at_utc = ?
+            WHERE tag = ?
+            """,
+            (count, ts_utc, ts_utc, tag),
+        )
+
+    def _revert_tag_suggestions_on_end(self, tag: str, count: int) -> None:
+        """Revert tag_suggestions when end is reverted."""
+        self.db.execute(
+            """
+            UPDATE tag_suggestions
+            SET active_animals = active_animals + ?
+            WHERE tag = ?
+            """,
+            (count, tag),
+        )

src/animaltrack/services/animal.py

@@ -8,6 +8,8 @@ from animaltrack.events.payloads import (
     AnimalAttributesUpdatedPayload,
     AnimalCohortCreatedPayload,
     AnimalMovedPayload,
+    AnimalTagEndedPayload,
+    AnimalTaggedPayload,
 )
 from animaltrack.events.processor import process_event
 from animaltrack.events.store import EventStore
@@ -15,6 +17,8 @@ from animaltrack.events.types import (
     ANIMAL_ATTRIBUTES_UPDATED,
     ANIMAL_COHORT_CREATED,
     ANIMAL_MOVED,
+    ANIMAL_TAG_ENDED,
+    ANIMAL_TAGGED,
 )
 from animaltrack.id_gen import generate_id
 from animaltrack.models.events import Event
@@ -370,3 +374,178 @@ class AnimalService:
                 changed_attrs[animal_id] = animal_changes
 
         return changed_attrs
+
+    def add_tag(
+        self,
+        payload: AnimalTaggedPayload,
+        ts_utc: int,
+        actor: str,
+        nonce: str | None = None,
+        route: str | None = None,
+    ) -> Event:
+        """Add a tag to animals.
+
+        Creates an AnimalTagged event and processes it through
+        all registered projections. Animals that already have this
+        tag active will be skipped (no-op behavior).
+
+        Args:
+            payload: Validated tag payload with resolved_ids and tag.
+            ts_utc: Timestamp in milliseconds since epoch.
+            actor: The user performing the action.
+            nonce: Optional idempotency nonce.
+            route: Required if nonce provided.
+
+        Returns:
+            The created event.
+
+        Raises:
+            ValidationError: If validation fails.
+        """
+        # Validate all animals exist
+        self._validate_animals_exist(payload.resolved_ids)
+
+        # Determine which animals don't already have this tag active
+        actually_tagged = self._find_animals_without_active_tag(payload.resolved_ids, payload.tag)
+
+        # Build entity_refs
+        entity_refs = {
+            "animal_ids": payload.resolved_ids,
+            "tag": payload.tag,
+            "actually_tagged": actually_tagged,
+        }
+
+        with transaction(self.db):
+            event = self.event_store.append_event(
+                event_type=ANIMAL_TAGGED,
+                ts_utc=ts_utc,
+                actor=actor,
+                entity_refs=entity_refs,
+                payload=payload.model_dump(),
+                nonce=nonce,
+                route=route,
+            )
+
+            process_event(event, self.registry)
+
+        return event
+
+    def end_tag(
+        self,
+        payload: AnimalTagEndedPayload,
+        ts_utc: int,
+        actor: str,
+        nonce: str | None = None,
+        route: str | None = None,
+    ) -> Event:
+        """End a tag for animals.
+
+        Creates an AnimalTagEnded event and processes it through
+        all registered projections. Animals that don't have this
+        tag active will be skipped (no-op behavior).
+
+        Args:
+            payload: Validated tag end payload with resolved_ids and tag.
+            ts_utc: Timestamp in milliseconds since epoch.
+            actor: The user performing the action.
+            nonce: Optional idempotency nonce.
+            route: Required if nonce provided.
+
+        Returns:
+            The created event.
+
+        Raises:
+            ValidationError: If validation fails.
+        """
+        # Validate all animals exist
+        self._validate_animals_exist(payload.resolved_ids)
+
+        # Determine which animals actually have this tag active
+        actually_ended = self._find_animals_with_active_tag(payload.resolved_ids, payload.tag)
+
+        # Build entity_refs
+        entity_refs = {
+            "animal_ids": payload.resolved_ids,
+            "tag": payload.tag,
+            "actually_ended": actually_ended,
+        }
+
+        with transaction(self.db):
+            event = self.event_store.append_event(
+                event_type=ANIMAL_TAG_ENDED,
+                ts_utc=ts_utc,
+                actor=actor,
+                entity_refs=entity_refs,
+                payload=payload.model_dump(),
+                nonce=nonce,
+                route=route,
+            )
+
+            process_event(event, self.registry)
+
+        return event
+
+    def _validate_animals_exist(self, animal_ids: list[str]) -> None:
+        """Validate all animals exist.
+
+        Args:
+            animal_ids: List of animal IDs to validate.
+
+        Raises:
+            ValidationError: If any animal doesn't exist.
+        """
+        for animal_id in animal_ids:
+            row = self.db.execute(
+                "SELECT 1 FROM animal_registry WHERE animal_id = ?",
+                (animal_id,),
+            ).fetchone()
+
+            if row is None:
+                msg = f"Animal {animal_id} not found"
+                raise ValidationError(msg)
+
+    def _find_animals_without_active_tag(self, animal_ids: list[str], tag: str) -> list[str]:
+        """Find animals that don't have the tag active.
+
+        Args:
+            animal_ids: List of animal IDs to check.
+            tag: The tag to look for.
+
+        Returns:
+            List of animal IDs that don't have an open interval for this tag.
+        """
+        result = []
+        for animal_id in animal_ids:
+            row = self.db.execute(
+                """SELECT 1 FROM animal_tag_intervals
+                   WHERE animal_id = ? AND tag = ? AND end_utc IS NULL""",
+                (animal_id, tag),
+            ).fetchone()
+
+            if row is None:
+                result.append(animal_id)
+
+        return result
+
+    def _find_animals_with_active_tag(self, animal_ids: list[str], tag: str) -> list[str]:
+        """Find animals that have the tag active.
+
+        Args:
+            animal_ids: List of animal IDs to check.
+            tag: The tag to look for.
+
+        Returns:
+            List of animal IDs that have an open interval for this tag.
+        """
+        result = []
+        for animal_id in animal_ids:
+            row = self.db.execute(
+                """SELECT 1 FROM animal_tag_intervals
+                   WHERE animal_id = ? AND tag = ? AND end_utc IS NULL""",
+                (animal_id, tag),
+            ).fetchone()
+
+            if row is not None:
+                result.append(animal_id)
+
+        return result

tests/test_service_animal_tagging.py

@@ -0,0 +1,455 @@
+# ABOUTME: Tests for AnimalService tagging operations.
+# ABOUTME: Tests add_tag and end_tag with interval and suggestion tracking.
+
+import json
+import time
+
+import pytest
+
+from animaltrack.events.payloads import (
+    AnimalCohortCreatedPayload,
+    AnimalTagEndedPayload,
+    AnimalTaggedPayload,
+)
+from animaltrack.events.store import EventStore
+from animaltrack.events.types import ANIMAL_TAG_ENDED, ANIMAL_TAGGED
+from animaltrack.projections import ProjectionRegistry
+from animaltrack.projections.animal_registry import AnimalRegistryProjection
+from animaltrack.projections.event_animals import EventAnimalsProjection
+from animaltrack.projections.intervals import IntervalProjection
+from animaltrack.services.animal import AnimalService, ValidationError
+
+
+@pytest.fixture
+def event_store(seeded_db):
+    """Create an EventStore for testing."""
+    return EventStore(seeded_db)
+
+
+@pytest.fixture
+def projection_registry(seeded_db):
+    """Create a ProjectionRegistry with all cohort projections registered."""
+    from animaltrack.projections.tags import TagProjection
+
+    registry = ProjectionRegistry()
+    registry.register(AnimalRegistryProjection(seeded_db))
+    registry.register(EventAnimalsProjection(seeded_db))
+    registry.register(IntervalProjection(seeded_db))
+    registry.register(TagProjection(seeded_db))
+    return registry
+
+
+@pytest.fixture
+def animal_service(seeded_db, event_store, projection_registry):
+    """Create an AnimalService for testing."""
+    return AnimalService(seeded_db, event_store, projection_registry)
+
+
+@pytest.fixture
+def valid_location_id(seeded_db):
+    """Get a valid location ID from seeds."""
+    row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()
+    return row[0]
+
+
+def make_cohort_payload(
+    location_id: str,
+    count: int = 1,
+    species: str = "duck",
+    life_stage: str = "adult",
+    sex: str = "unknown",
+    origin: str = "purchased",
+) -> AnimalCohortCreatedPayload:
+    """Create a cohort payload for testing."""
+    return AnimalCohortCreatedPayload(
+        species=species,
+        count=count,
+        life_stage=life_stage,
+        sex=sex,
+        location_id=location_id,
+        origin=origin,
+    )
+
+
+def make_tag_payload(resolved_ids: list[str], tag: str) -> AnimalTaggedPayload:
+    """Create a tag payload for testing."""
+    return AnimalTaggedPayload(resolved_ids=resolved_ids, tag=tag)
+
+
+def make_tag_end_payload(resolved_ids: list[str], tag: str) -> AnimalTagEndedPayload:
+    """Create a tag end payload for testing."""
+    return AnimalTagEndedPayload(resolved_ids=resolved_ids, tag=tag)
+
+
+# =============================================================================
+# add_tag Tests
+# =============================================================================
+
+
+class TestAnimalServiceAddTag:
+    """Tests for add_tag()."""
+
+    def test_creates_animal_tagged_event(self, seeded_db, animal_service, valid_location_id):
+        """add_tag creates an AnimalTagged event."""
+        # Create a cohort first
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_ids = cohort_event.entity_refs["animal_ids"]
+
+        # Add a tag
+        tag_payload = make_tag_payload(animal_ids, "red-band")
+        tag_ts = ts_utc + 1000
+        tag_event = animal_service.add_tag(tag_payload, tag_ts, "test_user")
+
+        assert tag_event.type == ANIMAL_TAGGED
+        assert tag_event.actor == "test_user"
+        assert tag_event.ts_utc == tag_ts
+
+    def test_event_has_animal_ids_in_entity_refs(
+        self, seeded_db, animal_service, valid_location_id
+    ):
+        """Event entity_refs contains animal_ids list."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=2)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_ids = cohort_event.entity_refs["animal_ids"]
+
+        tag_payload = make_tag_payload(animal_ids, "injured")
+        tag_event = animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+
+        assert "animal_ids" in tag_event.entity_refs
+        assert set(tag_event.entity_refs["animal_ids"]) == set(animal_ids)
+
+    def test_event_has_tag_in_entity_refs(self, seeded_db, animal_service, valid_location_id):
+        """Event entity_refs contains the tag."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_ids = cohort_event.entity_refs["animal_ids"]
+
+        tag_payload = make_tag_payload(animal_ids, "layer-1")
+        tag_event = animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+
+        assert tag_event.entity_refs["tag"] == "layer-1"
+
+    def test_creates_tag_interval(self, seeded_db, animal_service, valid_location_id):
+        """add_tag creates an open tag interval."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_id = cohort_event.entity_refs["animal_ids"][0]
+
+        tag_payload = make_tag_payload([animal_id], "blue-band")
+        animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+
+        # Check interval was created
+        row = seeded_db.execute(
+            """SELECT start_utc, end_utc FROM animal_tag_intervals
+               WHERE animal_id = ? AND tag = ?""",
+            (animal_id, "blue-band"),
+        ).fetchone()
+
+        assert row is not None
+        assert row[0] == ts_utc + 1000  # start_utc
+        assert row[1] is None  # end_utc is NULL (open interval)
+
+    def test_updates_live_animals_tags_json(self, seeded_db, animal_service, valid_location_id):
+        """add_tag updates tags JSON in live_animals_by_location."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_id = cohort_event.entity_refs["animal_ids"][0]
+
+        tag_payload = make_tag_payload([animal_id], "broody")
+        animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+
+        # Check tags JSON was updated
+        row = seeded_db.execute(
+            "SELECT tags FROM live_animals_by_location WHERE animal_id = ?",
+            (animal_id,),
+        ).fetchone()
+
+        tags = json.loads(row[0])
+        assert "broody" in tags
+
+    def test_creates_event_animal_links(self, seeded_db, animal_service, valid_location_id):
+        """add_tag creates event_animals links."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=3)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_ids = cohort_event.entity_refs["animal_ids"]
+
+        tag_payload = make_tag_payload(animal_ids, "layer")
+        tag_event = animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+
+        count = seeded_db.execute(
+            "SELECT COUNT(*) FROM event_animals WHERE event_id = ?",
+            (tag_event.id,),
+        ).fetchone()[0]
+        assert count == 3
+
+    def test_updates_tag_suggestions(self, seeded_db, animal_service, valid_location_id):
+        """add_tag creates/updates tag_suggestions entry."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=2)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_ids = cohort_event.entity_refs["animal_ids"]
+
+        tag_payload = make_tag_payload(animal_ids, "new-tag")
+        animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+
+        row = seeded_db.execute(
+            "SELECT total_assignments, active_animals FROM tag_suggestions WHERE tag = ?",
+            ("new-tag",),
+        ).fetchone()
+
+        assert row is not None
+        assert row[0] == 2  # total_assignments
+        assert row[1] == 2  # active_animals
+
+
+class TestAnimalServiceAddTagNoOp:
+    """Tests for add_tag() no-op behavior."""
+
+    def test_noop_when_tag_already_active(self, seeded_db, animal_service, valid_location_id):
+        """Adding same tag twice doesn't create duplicate interval."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_id = cohort_event.entity_refs["animal_ids"][0]
+
+        # Add tag twice
+        tag_payload = make_tag_payload([animal_id], "duplicate-test")
+        animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+        animal_service.add_tag(tag_payload, ts_utc + 2000, "test_user")
+
+        # Should only have one interval
+        count = seeded_db.execute(
+            """SELECT COUNT(*) FROM animal_tag_intervals
+               WHERE animal_id = ? AND tag = ?""",
+            (animal_id, "duplicate-test"),
+        ).fetchone()[0]
+        assert count == 1
+
+
+class TestAnimalServiceAddTagValidation:
+    """Tests for add_tag() validation."""
+
+    def test_rejects_nonexistent_animal(self, seeded_db, animal_service):
+        """Raises ValidationError for non-existent animal_id."""
+        fake_animal_id = "01ARZ3NDEKTSV4RRFFQ69G5XXX"
+        tag_payload = make_tag_payload([fake_animal_id], "test-tag")
+
+        with pytest.raises(ValidationError, match="not found"):
+            animal_service.add_tag(tag_payload, int(time.time() * 1000), "test_user")
+
+
+# =============================================================================
+# end_tag Tests
+# =============================================================================
+
+
+class TestAnimalServiceEndTag:
+    """Tests for end_tag()."""
+
+    def test_creates_animal_tag_ended_event(self, seeded_db, animal_service, valid_location_id):
+        """end_tag creates an AnimalTagEnded event."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_ids = cohort_event.entity_refs["animal_ids"]
+
+        # Add then end tag
+        tag_payload = make_tag_payload(animal_ids, "temp-tag")
+        animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+
+        end_payload = make_tag_end_payload(animal_ids, "temp-tag")
+        end_event = animal_service.end_tag(end_payload, ts_utc + 2000, "test_user")
+
+        assert end_event.type == ANIMAL_TAG_ENDED
+        assert end_event.actor == "test_user"
+        assert end_event.ts_utc == ts_utc + 2000
+
+    def test_closes_tag_interval(self, seeded_db, animal_service, valid_location_id):
+        """end_tag closes the open tag interval."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_id = cohort_event.entity_refs["animal_ids"][0]
+
+        # Add then end tag
+        tag_payload = make_tag_payload([animal_id], "closing-tag")
+        animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+
+        end_payload = make_tag_end_payload([animal_id], "closing-tag")
+        animal_service.end_tag(end_payload, ts_utc + 2000, "test_user")
+
+        # Check interval was closed
+        row = seeded_db.execute(
+            """SELECT start_utc, end_utc FROM animal_tag_intervals
+               WHERE animal_id = ? AND tag = ?""",
+            (animal_id, "closing-tag"),
+        ).fetchone()
+
+        assert row[0] == ts_utc + 1000  # start_utc
+        assert row[1] == ts_utc + 2000  # end_utc is set
+
+    def test_removes_tag_from_live_animals_json(self, seeded_db, animal_service, valid_location_id):
+        """end_tag removes tag from live_animals_by_location tags JSON."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_id = cohort_event.entity_refs["animal_ids"][0]
+
+        # Add then end tag
+        tag_payload = make_tag_payload([animal_id], "removing-tag")
+        animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+
+        end_payload = make_tag_end_payload([animal_id], "removing-tag")
+        animal_service.end_tag(end_payload, ts_utc + 2000, "test_user")
+
+        # Check tags JSON was updated
+        row = seeded_db.execute(
+            "SELECT tags FROM live_animals_by_location WHERE animal_id = ?",
+            (animal_id,),
+        ).fetchone()
+
+        tags = json.loads(row[0])
+        assert "removing-tag" not in tags
+
+    def test_decrements_tag_suggestions_active(self, seeded_db, animal_service, valid_location_id):
+        """end_tag decrements active_animals in tag_suggestions."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=2)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_ids = cohort_event.entity_refs["animal_ids"]
+
+        # Add tag to both
+        tag_payload = make_tag_payload(animal_ids, "decrement-test")
+        animal_service.add_tag(tag_payload, ts_utc + 1000, "test_user")
+
+        # End tag for one animal
+        end_payload = make_tag_end_payload([animal_ids[0]], "decrement-test")
+        animal_service.end_tag(end_payload, ts_utc + 2000, "test_user")
+
+        row = seeded_db.execute(
+            "SELECT total_assignments, active_animals FROM tag_suggestions WHERE tag = ?",
+            ("decrement-test",),
+        ).fetchone()
+
+        assert row[0] == 2  # total_assignments unchanged
+        assert row[1] == 1  # active_animals decremented
+
+
+class TestAnimalServiceEndTagNoOp:
+    """Tests for end_tag() no-op behavior."""
+
+    def test_noop_when_tag_not_active(self, seeded_db, animal_service, valid_location_id):
+        """Ending a tag that's not active is a no-op."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_id = cohort_event.entity_refs["animal_ids"][0]
+
+        # End a tag that was never added
+        end_payload = make_tag_end_payload([animal_id], "never-added")
+        end_event = animal_service.end_tag(end_payload, ts_utc + 1000, "test_user")
+
+        # Event should still be created
+        assert end_event.type == ANIMAL_TAG_ENDED
+
+        # But no intervals should exist
+        count = seeded_db.execute(
+            """SELECT COUNT(*) FROM animal_tag_intervals
+               WHERE animal_id = ? AND tag = ?""",
+            (animal_id, "never-added"),
+        ).fetchone()[0]
+        assert count == 0
+
+
+class TestAnimalServiceEndTagValidation:
+    """Tests for end_tag() validation."""
+
+    def test_rejects_nonexistent_animal(self, seeded_db, animal_service):
+        """Raises ValidationError for non-existent animal_id."""
+        fake_animal_id = "01ARZ3NDEKTSV4RRFFQ69G5XXX"
+        end_payload = make_tag_end_payload([fake_animal_id], "test-tag")
+
+        with pytest.raises(ValidationError, match="not found"):
+            animal_service.end_tag(end_payload, int(time.time() * 1000), "test_user")
+
+
+# =============================================================================
+# Multiple Tags Tests
+# =============================================================================
+
+
+class TestMultipleTags:
+    """Tests for multiple tags on the same animal."""
+
+    def test_multiple_tags_on_same_animal(self, seeded_db, animal_service, valid_location_id):
+        """An animal can have multiple active tags."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_id = cohort_event.entity_refs["animal_ids"][0]
+
+        # Add multiple tags
+        animal_service.add_tag(make_tag_payload([animal_id], "tag-a"), ts_utc + 1000, "test_user")
+        animal_service.add_tag(make_tag_payload([animal_id], "tag-b"), ts_utc + 2000, "test_user")
+        animal_service.add_tag(make_tag_payload([animal_id], "tag-c"), ts_utc + 3000, "test_user")
+
+        # Check all three intervals exist and are open
+        count = seeded_db.execute(
+            """SELECT COUNT(*) FROM animal_tag_intervals
+               WHERE animal_id = ? AND end_utc IS NULL""",
+            (animal_id,),
+        ).fetchone()[0]
+        assert count == 3
+
+        # Check tags JSON has all three
+        row = seeded_db.execute(
+            "SELECT tags FROM live_animals_by_location WHERE animal_id = ?",
+            (animal_id,),
+        ).fetchone()
+        tags = json.loads(row[0])
+        assert set(tags) == {"tag-a", "tag-b", "tag-c"}
+
+    def test_retagging_after_end(self, seeded_db, animal_service, valid_location_id):
+        """A tag can be added again after being ended."""
+        cohort_payload = make_cohort_payload(valid_location_id, count=1)
+        ts_utc = int(time.time() * 1000)
+        cohort_event = animal_service.create_cohort(cohort_payload, ts_utc, "test_user")
+        animal_id = cohort_event.entity_refs["animal_ids"][0]
+
+        # Add, end, re-add
+        animal_service.add_tag(make_tag_payload([animal_id], "retag"), ts_utc + 1000, "test_user")
+        animal_service.end_tag(
+            make_tag_end_payload([animal_id], "retag"), ts_utc + 2000, "test_user"
+        )
+        animal_service.add_tag(make_tag_payload([animal_id], "retag"), ts_utc + 3000, "test_user")
+
+        # Should have two intervals (one closed, one open)
+        count = seeded_db.execute(
+            """SELECT COUNT(*) FROM animal_tag_intervals
+               WHERE animal_id = ? AND tag = ?""",
+            (animal_id, "retag"),
+        ).fetchone()[0]
+        assert count == 2
+
+        # One should be closed
+        closed = seeded_db.execute(
+            """SELECT COUNT(*) FROM animal_tag_intervals
+               WHERE animal_id = ? AND tag = ? AND end_utc IS NOT NULL""",
+            (animal_id, "retag"),
+        ).fetchone()[0]
+        assert closed == 1
+
+        # One should be open
+        open_count = seeded_db.execute(
+            """SELECT COUNT(*) FROM animal_tag_intervals
+               WHERE animal_id = ? AND tag = ? AND end_utc IS NULL""",
+            (animal_id, "retag"),
+        ).fetchone()[0]
+        assert open_count == 1