feat: defer item tagging to an external agent (phase 2)

backend/app/api/health.py

@@ -32,9 +32,9 @@ async def capabilities() -> dict[str, Any]:
             "vision": settings.effective_ai_vision_enabled,
             "text": settings.effective_ai_text_enabled,
         },
-        # Set to True in PR 2-3 when agent write-back endpoints exist.
+        # Set to True when the agent write-back endpoints for outfits exist.
         "features": {
-            "external_tagging": False,
+            "external_tagging": True,
             "external_suggestions": False,
             "external_pairings": False,
         },

backend/app/api/items.py

@@ -11,6 +11,7 @@
 
 from app.config import get_settings
 from app.database import get_db
+from app.models.item import ItemStatus, TaggedBy, TaggingStatus
 from app.models.user import User
 from app.schemas.item import (
     ArchiveRequest,
@@ -37,6 +38,8 @@
 from app.utils.auth import get_current_user
 from app.workers.settings import get_redis_settings
 
+TAG_WRITEBACK_FIELDS = {"type", "subtype", "tags", "colors", "primary_color"}
+
 logger = logging.getLogger(__name__)
 settings = get_settings()
 
@@ -53,6 +56,7 @@ async def list_items(
     subtype: str | None = None,
     colors: str | None = None,
     status: str | None = None,
+    tagging_status: str | None = None,
     favorite: bool | None = None,
     needs_wash: bool | None = None,
     is_archived: bool = False,
@@ -67,6 +71,7 @@ async def list_items(
         subtype=subtype,
         colors=color_list,
         status=status,
+        tagging_status=tagging_status,
         favorite=favorite,
         needs_wash=needs_wash,
         is_archived=is_archived,
@@ -105,6 +110,7 @@ async def create_item(
     colors: str | None = Form(None),
     primary_color: str | None = Form(None),
     favorite: bool = Form(False),
+    auto_tag: bool | None = Form(None),
 ) -> ItemResponse:
     # Validate and process image
     image_service = ImageService()
@@ -168,23 +174,27 @@ async def create_item(
         image_paths=image_paths,
     )
 
-    # Queue AI tagging job
-    try:
-        redis = await create_pool(get_redis_settings())
+    do_auto_tag = settings.effective_ai_vision_enabled and auto_tag is not False
+    if do_auto_tag:
         try:
-            full_image_path = f"{settings.storage_path}/{image_paths['image_path']}"
-            await redis.enqueue_job(
-                "tag_item_image",
-                str(item.id),
-                full_image_path,
-                _queue_name="arq:tagging",
-            )
-            logger.info(f"Queued AI tagging job for item {item.id}")
-        finally:
-            await redis.aclose()
-    except Exception as e:
-        # Don't fail the upload if queueing fails
-        logger.error(f"Failed to queue AI tagging job: {e}")
+            redis = await create_pool(get_redis_settings())
+            try:
+                full_image_path = f"{settings.storage_path}/{image_paths['image_path']}"
+                await redis.enqueue_job(
+                    "tag_item_image",
+                    str(item.id),
+                    full_image_path,
+                    _queue_name="arq:tagging",
+                )
+                logger.info(f"Queued AI tagging job for item {item.id}")
+            finally:
+                await redis.aclose()
+        except Exception as e:
+            # Don't fail the upload if queueing fails
+            logger.error(f"Failed to queue AI tagging job: {e}")
+    else:
+        item = await item_service.mark_pending(item, set_ready=True)
+        logger.info(f"Item {item.id} left pending for external tagging")
 
     return ItemResponse.model_validate(item)
 
@@ -214,11 +224,13 @@ async def bulk_create_items(
     failed = 0
 
     # Create Redis pool once for all jobs
+    do_auto_tag = settings.effective_ai_vision_enabled
     redis = None
-    try:
-        redis = await create_pool(get_redis_settings())
-    except Exception as e:
-        logger.error(f"Failed to connect to Redis for bulk upload: {e}")
+    if do_auto_tag:
+        try:
+            redis = await create_pool(get_redis_settings())
+        except Exception as e:
+            logger.error(f"Failed to connect to Redis for bulk upload: {e}")
 
     try:
         for upload_file in images:
@@ -276,7 +288,7 @@ async def bulk_create_items(
                 )
 
                 # Queue AI tagging job
-                if redis:
+                if do_auto_tag and redis:
                     try:
                         full_image_path = f"{settings.storage_path}/{image_paths['image_path']}"
                         await redis.enqueue_job(
@@ -288,6 +300,8 @@ async def bulk_create_items(
                         logger.info(f"Queued AI tagging for bulk item {item.id}")
                     except Exception as e:
                         logger.error(f"Failed to queue AI tagging for {item.id}: {e}")
+                elif not do_auto_tag:
+                    item = await item_service.mark_pending(item, set_ready=True)
 
                 results.append(
                     BulkUploadResult(
@@ -390,8 +404,6 @@ async def bulk_analyze_items(
     db: Annotated[AsyncSession, Depends(get_db)],
     current_user: Annotated[User, Depends(get_current_user)],
 ) -> BulkAnalyzeResponse:
-    from app.models.item import ItemStatus
-
     item_service = ItemService(db)
     queued = 0
     failed = 0
@@ -422,6 +434,15 @@ async def bulk_analyze_items(
             continue
         items_to_process.append(item)
 
+    if not settings.effective_ai_vision_enabled:
+        for item in items_to_process:
+            item.status = ItemStatus.ready
+            item.tagging_status = TaggingStatus.pending
+            item.tagged_by = None
+            item.tagged_at = None
+        await db.commit()
+        return BulkAnalyzeResponse(queued=0, failed=failed, errors=errors)
+
     # Set all items to processing status
     for item in items_to_process:
         item.status = ItemStatus.processing
@@ -520,6 +541,15 @@ async def update_item(
             detail="Item not found",
         )
 
+    update_data = item_data.model_dump(exclude_unset=True)
+    is_writeback = item.tagging_status == TaggingStatus.pending and any(
+        update_data.get(field) not in (None, "", [], {}) for field in TAG_WRITEBACK_FIELDS
+    )
+    if is_writeback:
+        item.tagging_status = TaggingStatus.tagged
+        item.tagged_by = TaggedBy.manual
+        item.tagged_at = datetime.now(UTC)
+
     item = await item_service.update(item, item_data)
     return ItemResponse.model_validate(item)
 
@@ -792,10 +822,16 @@ async def trigger_ai_analysis(
             detail="Item not found",
         )
 
+    if not settings.effective_ai_vision_enabled:
+        item.status = ItemStatus.ready
+        item.tagging_status = TaggingStatus.pending
+        item.tagged_by = None
+        item.tagged_at = None
+        await db.commit()
+        return {"status": "deferred", "detail": "Internal vision disabled; item marked pending"}
+
     try:
         # Set item status to processing so UI shows feedback
-        from app.models.item import ItemStatus
-
         item.status = ItemStatus.processing
         await db.commit()
 
@@ -820,6 +856,26 @@ async def trigger_ai_analysis(
         ) from None
 
 
+@router.post("/{item_id}/retag", response_model=ItemResponse)
+async def retag_item(
+    item_id: UUID,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(get_current_user)],
+) -> ItemResponse:
+    """Reset an item to the pending tagging queue and clear its tagging origin."""
+    item_service = ItemService(db)
+    item = await item_service.get_by_id(item_id, current_user.id)
+
+    if not item:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Item not found",
+        )
+
+    item = await item_service.mark_pending(item)
+    return ItemResponse.model_validate(item)
+
+
 @router.post("/{item_id}/rotate", response_model=ItemResponse)
 async def rotate_item_image(
     item_id: UUID,

backend/app/models/item.py

@@ -33,6 +33,16 @@ class ItemStatus(enum.StrEnum):
     archived = "archived"
 
 
+class TaggingStatus(enum.StrEnum):
+    pending = "pending"
+    tagged = "tagged"
+
+
+class TaggedBy(enum.StrEnum):
+    auto = "auto"
+    manual = "manual"
+
+
 class ClothingItem(Base):
     __tablename__ = "clothing_items"
 
@@ -69,6 +79,17 @@ class ClothingItem(Base):
     ai_confidence: Mapped[Decimal | None] = mapped_column(Numeric(3, 2))
     ai_raw_response: Mapped[dict | None] = mapped_column(JSONB)
 
+    # Tagging lifecycle
+    tagging_status: Mapped[TaggingStatus] = mapped_column(
+        Enum(TaggingStatus, name="tagging_status", create_type=False),
+        default=TaggingStatus.pending,
+        nullable=False,
+    )
+    tagged_by: Mapped[TaggedBy | None] = mapped_column(
+        Enum(TaggedBy, name="tagged_by", create_type=False)
+    )
+    tagged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
     # Usage tracking
     wear_count: Mapped[int] = mapped_column(Integer, default=0)
     last_worn_at: Mapped[date | None] = mapped_column(Date)

backend/app/schemas/item.py

@@ -88,6 +88,9 @@ class ItemResponse(ItemBase):
     formality: str | None = None
     season: list[str] = Field(default_factory=list)
     status: str
+    tagging_status: str = "pending"
+    tagged_by: str | None = None
+    tagged_at: datetime | None = None
     ai_processed: bool = False
     ai_confidence: Decimal | None = None
     ai_description: str | None = None
@@ -147,6 +150,7 @@ class ItemFilter(BaseModel):
     subtype: str | None = None
     colors: list[str] | None = None
     status: str | None = None
+    tagging_status: str | None = None
     favorite: bool | None = None
     needs_wash: bool | None = None
     is_archived: bool = False

backend/app/services/item_service.py

@@ -7,7 +7,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import attributes, selectinload
 
-from app.models.item import ClothingItem, ItemHistory, ItemStatus, WashHistory
+from app.models.item import ClothingItem, ItemHistory, ItemStatus, TaggingStatus, WashHistory
 from app.schemas.item import DEFAULT_WASH_INTERVALS, ItemCreate, ItemFilter, ItemUpdate
 
 
@@ -58,6 +58,8 @@ async def get_list(
             query = query.where(ClothingItem.subtype == filters.subtype)
         if filters.status:
             query = query.where(ClothingItem.status == filters.status)
+        if filters.tagging_status:
+            query = query.where(ClothingItem.tagging_status == filters.tagging_status)
         if filters.favorite is not None:
             query = query.where(ClothingItem.favorite == filters.favorite)
         if filters.colors:
@@ -194,6 +196,17 @@ async def create(
         await self.db.refresh(item, ["additional_images"])
         return item
 
+    # Tag attributes that also exist as first-class columns; `fit` is JSONB-only.
+    _TAG_COLUMN_ATTRS = (
+        "colors",
+        "primary_color",
+        "pattern",
+        "material",
+        "style",
+        "season",
+        "formality",
+    )
+
     async def update(self, item: ClothingItem, item_data: ItemUpdate) -> ClothingItem:
         update_data = item_data.model_dump(exclude_unset=True)
 
@@ -203,6 +216,9 @@ async def update(self, item: ClothingItem, item_data: ItemUpdate) -> ClothingIte
                 update_data["tags"] = {k: v for k, v in tags.items() if v is not None}
             else:
                 update_data["tags"] = tags.model_dump(exclude_none=True)
+            for attr in self._TAG_COLUMN_ATTRS:
+                if attr in update_data["tags"]:
+                    setattr(item, attr, update_data["tags"][attr])
 
         for field, value in update_data.items():
             setattr(item, field, value)
@@ -215,6 +231,16 @@ async def update(self, item: ClothingItem, item_data: ItemUpdate) -> ClothingIte
         result = await self.get_by_id(item.id, item.user_id)
         return result  # type: ignore[return-value]
 
+    async def mark_pending(self, item: ClothingItem, *, set_ready: bool = False) -> ClothingItem:
+        if set_ready:
+            item.status = ItemStatus.ready
+        item.tagging_status = TaggingStatus.pending
+        item.tagged_by = None
+        item.tagged_at = None
+        await self.db.flush()
+        result = await self.get_by_id(item.id, item.user_id)
+        return result  # type: ignore[return-value]
+
     async def delete(self, item: ClothingItem) -> None:
         await self.db.delete(item)
         await self.db.flush()

backend/app/workers/tagging.py

@@ -1,12 +1,13 @@
 import logging
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 from uuid import UUID
 
 from sqlalchemy import select
 
 from app.config import get_settings
-from app.models.item import ClothingItem, ItemStatus
+from app.models.item import ClothingItem, ItemStatus, TaggedBy, TaggingStatus
 from app.services.ai_service import AIService, ClothingTags
 from app.workers.db import get_db_session
 
@@ -47,6 +48,9 @@ def tags_to_item_fields(tags: ClothingTags, raw_response: str | None = None) ->
         "ai_confidence": tags.confidence,
         "ai_description": tags.description,  # Human-readable description
         "status": ItemStatus.ready,
+        "tagging_status": TaggingStatus.tagged,
+        "tagged_by": TaggedBy.auto,
+        "tagged_at": datetime.now(UTC),
     }
     if raw_response:
         fields["ai_raw_response"] = {"raw_text": raw_response}
@@ -60,6 +64,7 @@ async def mark_item_tagging_skipped(ctx: dict, item_id: str) -> None:
         item = result.scalar_one_or_none()
         if item and item.status == ItemStatus.processing:
             item.status = ItemStatus.ready
+            item.tagging_status = TaggingStatus.pending
             await db.commit()
     finally:
         await db.close()
@@ -165,6 +170,9 @@ async def tag_item_image(ctx: dict, item_id: str, image_path: str) -> dict[str,
                     "ai_raw_response",
                     "tags",
                     "ai_description",
+                    "tagging_status",
+                    "tagged_by",
+                    "tagged_at",
                 ):
                     setattr(item, field, value)
                 # Only update content fields if user hasn't set them (or they're default/unknown)