obs(ml): tag_and_embed logs file + phase + timing; failures name them

The task logged nothing and SoftTimeLimitExceeded stringifies to empty, so a
timeout surfaced as a bare 'SoftTimeLimitExceeded()' with no clue which file or
why (operator-flagged 2026-06-08).

- Log start (id/path/mime/bytes/video?), per-phase timing (load_models, video
  probe/sample/infer, tag, embed, persist), and a success summary.
- Track a  + file ; on SoftTimeLimitExceeded log it and re-raise
  SoftTimeLimitExceeded WITH that context (keeps the 'timeout' task_run status
  but gives the activity a real error_message: which file, which phase, elapsed).
- On other exceptions, log context then re-raise the ORIGINAL (preserves
  autoretry for OSError/DBAPIError/OperationalError).

Now a stuck run names the culprit — most likely a slow video (frame sampling is
up to 10x60s ffmpeg) or a huge image; the phase log will say which.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/tasks/ml.py

@@ -6,8 +6,10 @@ apply_allowlist_tags sweeps which are 'maintenance' lane. Sync sessions
 (Celery workers are sync processes), same pattern as FC-2a tasks.
 """
 
+import logging
 from pathlib import Path
 
+from celery.exceptions import SoftTimeLimitExceeded
 from sqlalchemy import select
 from sqlalchemy.exc import DBAPIError, OperationalError
 
@@ -15,6 +17,8 @@ from ..celery_app import celery
 from ..models import ImageRecord, MLSettings
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
+log = logging.getLogger(__name__)
+
 IMAGES_ROOT = Path("/images")
 VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v", ".wmv", ".flv"}
 
@@ -50,67 +54,136 @@ def tag_and_embed(self, image_id: int) -> dict:
     SigLIP embeddings. On no-frames returns status='no_frames' (not an error).
     """
     import os
+    import time
 
     from ..services.ml.embedder import get_embedder
     from ..services.ml.tagger import get_tagger
 
-    SessionLocal = _sync_session_factory()
-    with SessionLocal() as session:
-        record = session.get(ImageRecord, image_id)
-        if record is None:
-            return {"status": "missing", "image_id": image_id}
-        settings = session.execute(
-            select(MLSettings).where(MLSettings.id == 1)
-        ).scalar_one()
+    # Phase + file context, so a timeout/crash names WHICH file and WHERE it
+    # died instead of a bare SoftTimeLimitExceeded() (operator-flagged 2026-06-08:
+    # the activity told them nothing about the file or why). `ctx` is enriched
+    # once the record is loaded; both feed the worker log AND the re-raised
+    # exception message (which becomes the activity's error_message).
+    started = time.monotonic()
+    phase = "open_session"
+    ctx = f"image_id={image_id}"
 
-        src = Path(record.path)
-        if not src.is_file():
-            return {"status": "file_missing", "image_id": image_id}
+    def _elapsed() -> float:
+        return time.monotonic() - started
 
-        tagger = get_tagger()
-        embedder = get_embedder()
+    try:
+        SessionLocal = _sync_session_factory()
+        with SessionLocal() as session:
+            record = session.get(ImageRecord, image_id)
+            if record is None:
+                return {"status": "missing", "image_id": image_id}
+            settings = session.execute(
+                select(MLSettings).where(MLSettings.id == 1)
+            ).scalar_one()
 
-        if _is_video(src):
-            # Layer-3 isolation: ffprobe (a separate process) validates
-            # the container before we burn ~20 GPU ops sampling frames
-            # from it. A corrupt video that would crash the frame
-            # decoder is rejected cleanly here instead of taking down
-            # the ml-worker. Operator-flagged 2026-05-28.
-            from ..utils import safe_probe
-            vprobe = safe_probe.probe_video(src)
-            if not vprobe.ok:
-                return {
-                    "status": "bad_video", "image_id": image_id,
-                    "reason": vprobe.reason,
-                }
-            frames = _sample_video_frames(
-                src, int(os.environ.get("VIDEO_ML_FRAMES", "10"))
+            src = Path(record.path)
+            is_vid = _is_video(src)
+            ctx = (
+                f"image_id={image_id} path={record.path} mime={record.mime} "
+                f"bytes={record.size_bytes} video={is_vid}"
             )
-            if not frames:
-                return {"status": "no_frames", "image_id": image_id}
-            preds = _maxpool_predictions([tagger.infer(f) for f in frames])
-            import numpy as np
+            log.info("tag_and_embed start: %s", ctx)
+            if not src.is_file():
+                log.warning("tag_and_embed file missing on disk: %s", ctx)
+                return {"status": "file_missing", "image_id": image_id}
 
-            embedding = np.mean(
-                [embedder.infer(f) for f in frames], axis=0
-            ).astype("float32")
-            for f in frames:
-                f.unlink(missing_ok=True)
-        else:
-            raw = tagger.infer(src)
-            preds = {
-                name: {"category": p.category, "confidence": p.confidence}
-                for name, p in raw.items()
-            }
-            embedding = embedder.infer(src)
+            phase = "load_models"
+            tagger = get_tagger()
+            embedder = get_embedder()
 
-        record.tagger_predictions = preds
-        record.tagger_model_version = settings.tagger_model_version
-        record.siglip_embedding = embedding.tolist()
-        record.siglip_model_version = settings.embedder_model_version
-        session.add(record)
-        session.commit()
+            if is_vid:
+                # Layer-3 isolation: ffprobe (a separate process) validates
+                # the container before we burn ~20 GPU ops sampling frames
+                # from it. A corrupt video that would crash the frame
+                # decoder is rejected cleanly here instead of taking down
+                # the ml-worker. Operator-flagged 2026-05-28.
+                phase = "video_probe"
+                from ..utils import safe_probe
+                vprobe = safe_probe.probe_video(src)
+                if not vprobe.ok:
+                    log.warning(
+                        "tag_and_embed bad video (%s): %s", vprobe.reason, ctx
+                    )
+                    return {
+                        "status": "bad_video", "image_id": image_id,
+                        "reason": vprobe.reason,
+                    }
+                phase = "video_sample_frames"
+                t0 = time.monotonic()
+                frames = _sample_video_frames(
+                    src, int(os.environ.get("VIDEO_ML_FRAMES", "10"))
+                )
+                log.info(
+                    "tag_and_embed sampled %d frame(s) in %.1fs: %s",
+                    len(frames), time.monotonic() - t0, ctx,
+                )
+                if not frames:
+                    return {"status": "no_frames", "image_id": image_id}
+                phase = "video_infer"
+                import numpy as np
 
+                preds = _maxpool_predictions([tagger.infer(f) for f in frames])
+                embedding = np.mean(
+                    [embedder.infer(f) for f in frames], axis=0
+                ).astype("float32")
+                for f in frames:
+                    f.unlink(missing_ok=True)
+            else:
+                phase = "tag"
+                t0 = time.monotonic()
+                raw = tagger.infer(src)
+                log.info(
+                    "tag_and_embed tagged in %.1fs (%d tags): %s",
+                    time.monotonic() - t0, len(raw), ctx,
+                )
+                preds = {
+                    name: {"category": p.category, "confidence": p.confidence}
+                    for name, p in raw.items()
+                }
+                phase = "embed"
+                t0 = time.monotonic()
+                embedding = embedder.infer(src)
+                log.info(
+                    "tag_and_embed embedded in %.1fs: %s",
+                    time.monotonic() - t0, ctx,
+                )
+
+            phase = "persist"
+            record.tagger_predictions = preds
+            record.tagger_model_version = settings.tagger_model_version
+            record.siglip_embedding = embedding.tolist()
+            record.siglip_model_version = settings.embedder_model_version
+            session.add(record)
+            session.commit()
+    except SoftTimeLimitExceeded:
+        log.error(
+            "tag_and_embed TIMED OUT after %.0fs in phase=%s: %s",
+            _elapsed(), phase, ctx,
+        )
+        # Re-raise as SoftTimeLimitExceeded (preserves the 'timeout' status in
+        # the task_run signal) but WITH context, so the activity error_message
+        # names the file + phase instead of being empty.
+        raise SoftTimeLimitExceeded(
+            f"timed out in phase={phase} after {_elapsed():.0f}s ({ctx})"
+        ) from None
+    except Exception:
+        # OSError/DBAPIError/OperationalError are autoretried — re-raise the
+        # ORIGINAL so the type is preserved; just make sure it's logged with
+        # context first.
+        log.exception(
+            "tag_and_embed FAILED in phase=%s after %.0fs: %s",
+            phase, _elapsed(), ctx,
+        )
+        raise
+
+    log.info(
+        "tag_and_embed ok in %.1fs (%d tags): %s", _elapsed(), len(preds), ctx
+    )
     apply_allowlist_tags.delay(image_id=image_id)
     return {"status": "ok", "image_id": image_id, "tags": len(preds)}