feat(llm): 表情标签统计口径修正、永久错误不重试、WebUI 观测空值注入

Author: TogetsuDo

pallas/product/llm/llm_daily_stats_store.py

@@ -85,6 +85,8 @@ def _metric_weight(key: str, value: object) -> int:
         return int(value.get("skip") or 0) + int(value.get("defer") or 0) + int(value.get("proceed") or 0)
     if key == "sticker_vision":
         return int(value.get("requests") or 0)
+    if key == "sticker_label":
+        return int(value.get("submitted") or 0)
     if key == "totals":
         return sum(int(v or 0) for v in value.values() if not isinstance(v, dict))
     if key == "by_task":
@@ -297,6 +299,7 @@ def _prefer_complete_metric(key: str, existing: Any, incoming: Any) -> Any:
     "totals",
     "gates",
     "sticker_vision",
+    "sticker_label",
 })
 
 
@@ -322,6 +325,7 @@ def merge_side_snapshot(existing: dict[str, Any] | None, snapshot: dict[str, Any
         "memory_rag",
         "gates",
         "sticker_vision",
+        "sticker_label",
     ):
         if key not in snapshot:
             continue

pallas/product/llm/model_admin.py

@@ -1159,15 +1159,34 @@ async def fetch_llm_task_stats(
         except Exception:
             pass
 
+    try:
+        from pallas.product.llm.sticker_label_observability import fetch_sticker_label_job_stats
+
+        sticker_label = await fetch_sticker_label_job_stats()
+    except Exception:
+        sticker_label = {}
+    if isinstance(sticker_label, dict):
+        ai_body = payload.get("ai") if isinstance(payload.get("ai"), dict) else {}
+        day = str(ai_body.get("day_key") or bot_snap.get("day_key") or today_key())
+        merged_ai = {
+            **ai_body,
+            "day_key": day,
+            "source": ai_body.get("source") or "bot",
+            "sticker_label": sticker_label,
+        }
+        payload["ai"] = _normalize_ai_task_stats_snapshot(merged_ai)
+        try:
+            write_llm_daily_stats_side(day, "ai", {**payload["ai"], "reachable": True})
+        except Exception:
+            pass
+
     try:
         from pallas.product.llm.sticker_vision import fetch_sticker_vision_stats
 
         sticker_vision = await fetch_sticker_vision_stats()
     except Exception:
         sticker_vision = {}
-    if isinstance(sticker_vision, dict) and (
-        int(sticker_vision.get("requests") or 0) > 0 or sticker_vision.get("recent")
-    ):
+    if isinstance(sticker_vision, dict):
         ai_body = payload.get("ai") if isinstance(payload.get("ai"), dict) else {}
         day = str(ai_body.get("day_key") or bot_snap.get("day_key") or today_key())
         merged_ai = {

pallas/product/llm/sticker_label_jobs.py

@@ -150,8 +150,12 @@ async def enqueue_sticker_label_candidate(*, cache_key: str, content: bytes, sou
         prompt_version=STICKER_LABEL_PROMPT_VERSION,
         min_confidence=STICKER_LABEL_MIN_CONFIDENCE,
     ):
+        from pallas.product.llm.task_metrics import record_bot_llm_task
+
+        record_bot_llm_task("sticker_label", "cache_hit")
         return False
     from pallas.core.shared.utils.media_cache import bind_image_content_hash
+    from pallas.product.llm.task_metrics import record_bot_llm_task
 
     await bind_image_content_hash(cache_key, content)
     job = WorkJob.create(
@@ -168,10 +172,8 @@ async def enqueue_sticker_label_candidate(*, cache_key: str, content: bytes, sou
             "observation": {"state": "queued"},
         },
     )
-    await build_work_job_store().requeue_terminal(job)
-    from pallas.product.llm.task_metrics import record_bot_llm_task
-
-    record_bot_llm_task("sticker_label", "submit_ok")
+    _reactivated_job, reactivated = await build_work_job_store().requeue_terminal(job)
+    record_bot_llm_task("sticker_label", "submit_ok" if reactivated else "background_coalesced")
     return True
 
 
@@ -303,6 +305,8 @@ async def handle_sticker_label_visual(payload: dict[str, Any]) -> None:
         })
         await save_sticker_label_observation(job_id, dict(payload), observation)
         logger.warning("sticker label failed: job_id={} err={}", job_id, type(exc).__name__)
+        if failure_state in {"parse_error", "no_vision"}:
+            return
         raise
     observation.update({
         "state": "labeled",

pallas/product/llm/sticker_label_observability.py

@@ -18,25 +18,51 @@
 
 
 def build_sticker_label_job_stats(records: list[dict[str, object]], *, recent_limit: int = 8) -> dict[str, object]:
-    pending = failed = 0
+    submitted = labeled = pending = failed = 0
+    timeout = parse_error = no_vision = circuit_open = cache_changed = 0
     recent_errors: list[dict[str, object]] = []
     for row in sorted(records, key=lambda item: float(item.get("created_at") or 0), reverse=True):
+        submitted += 1
         payload = row.get("payload") if isinstance(row.get("payload"), dict) else {}
         observation = payload.get("observation") if isinstance(payload.get("observation"), dict) else {}
         state = str(observation.get("state") or "queued")
         if state in {"queued", "running"}:
             pending += 1
+        elif state == "labeled":
+            labeled += 1
+        elif state == "timeout":
+            timeout += 1
+        elif state == "parse_error":
+            parse_error += 1
+        elif state == "no_vision":
+            no_vision += 1
+        elif state == "circuit_open":
+            circuit_open += 1
+        elif state == "cache_changed":
+            cache_changed += 1
+        elif state == "failed":
+            failed += 1
         error = str(observation.get("error") or row.get("last_error") or "").strip()
         if error:
-            failed += 1
             if len(recent_errors) < max(1, int(recent_limit)):
                 recent_errors.append({
                     "job_id": str(row.get("job_id") or "")[:64],
                     "created_at": float(row.get("created_at") or 0),
                     "state": state,
                     "error": error[:240],
                 })
-    return {"pending": pending, "failed": failed, "recent_errors": recent_errors}
+    return {
+        "submitted": submitted,
+        "labeled": labeled,
+        "pending": pending,
+        "failed": failed,
+        "timeout": timeout,
+        "parse_error": parse_error,
+        "no_vision": no_vision,
+        "circuit_open": circuit_open,
+        "cache_changed": cache_changed,
+        "recent_errors": recent_errors,
+    }
 
 
 async def fetch_sticker_label_job_stats(*, recent_limit: int = 8, aggregate_limit: int = 500) -> dict[str, object]:

pallas/product/llm/task_metrics.py

@@ -23,6 +23,7 @@
     "background_coalesced",
     "callback_ok",
     "callback_fail",
+    "cache_hit",
     "reply_gate_skip",
     "reply_gate_defer",
     "reply_gate_proceed",

tests/features/test_llm_daily_stats_closure.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
@@ -398,3 +398,45 @@ async def test_fetch_llm_task_stats_history_fallback_ignores_images_only_live(
     assert payload["ai"]["tokens"]["total_tokens"] == 200
     assert payload["ai"]["images"]["ok_count"] == 2
     assert payload["ai"]["images"]["image_count"] == 2
+
+
+@pytest.mark.asyncio
+async def test_fetch_llm_task_stats_exposes_zero_sticker_blocks_when_empty(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    bot_snapshot = {
+        "source": "bot",
+        "day_key": "2026-06-18",
+        "updated_at": 1.0,
+        "by_task": {},
+        "totals": {},
+    }
+    _patch_bot_snapshots(monkeypatch, bot_snapshot)
+    monkeypatch.setattr(
+        "pallas.product.llm.model_admin.today_key",
+        lambda: "2026-06-18",
+        raising=False,
+    )
+    _patch_bot_tokens(monkeypatch, {})
+    monkeypatch.setattr(
+        "pallas.product.llm.model_admin.load_llm_daily_stats_range",
+        lambda *, start_day, end_day: ([], start_day, end_day),
+    )
+    monkeypatch.setattr(
+        "pallas.product.llm.sticker_vision.fetch_sticker_vision_stats",
+        AsyncMock(return_value={"requests": 0, "recent": []}),
+    )
+    monkeypatch.setattr(
+        "pallas.product.llm.sticker_label_observability.fetch_sticker_label_job_stats",
+        AsyncMock(return_value={"submitted": 0, "recent_errors": []}),
+    )
+
+    payload = await fetch_llm_task_stats(start="2026-06-18", end="2026-06-18")
+
+    sticker_vision = payload["ai"].get("sticker_vision")
+    assert isinstance(sticker_vision, dict)
+    assert sticker_vision.get("requests") == 0
+    assert sticker_vision.get("recent") == []
+    sticker_label = payload["ai"].get("sticker_label")
+    assert isinstance(sticker_label, dict)
+    assert sticker_label.get("submitted") == 0

tests/product/llm/test_sticker_label_jobs.py

@@ -69,7 +69,7 @@ async def test_enqueue_candidate_uses_hash_locator_and_redacts_durable_payload(m
     from pallas.product.llm.sticker_labels import content_hash_for_bytes
 
     original = b"original-gif-bytes"
-    store = SimpleNamespace(requeue_terminal=AsyncMock(side_effect=lambda job: job))
+    store = SimpleNamespace(requeue_terminal=AsyncMock(side_effect=lambda job: (job, True)))
     repository = SimpleNamespace(get=AsyncMock(return_value=None))
     monkeypatch.setattr(sticker_label_jobs, "build_work_job_store", lambda: store)
     monkeypatch.setattr(sticker_label_jobs, "sticker_label_repository", lambda: repository)
@@ -140,7 +140,7 @@ async def test_low_confidence_or_old_prompt_label_requeues(monkeypatch: pytest.M
             )
         )
     )
-    store = SimpleNamespace(requeue_terminal=AsyncMock(side_effect=lambda job: job))
+    store = SimpleNamespace(requeue_terminal=AsyncMock(side_effect=lambda job: (job, True)))
     monkeypatch.setattr(sticker_label_jobs, "sticker_label_repository", lambda: repository)
     monkeypatch.setattr(sticker_label_jobs, "build_work_job_store", lambda: store)
     monkeypatch.setattr("pallas.core.shared.utils.media_cache.bind_image_content_hash", AsyncMock())
@@ -151,6 +151,71 @@ async def test_low_confidence_or_old_prompt_label_requeues(monkeypatch: pytest.M
     store.requeue_terminal.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_enqueue_candidate_counts_cache_hit_when_label_is_sufficient(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from pallas.product.llm import sticker_label_jobs
+    from pallas.product.llm.sticker_label_jobs import StickerLabelSource
+    from pallas.product.llm.sticker_labels import StickerSemanticLabel, content_hash_for_bytes
+    from pallas.product.llm.task_metrics import record_bot_llm_task
+
+    content = b"already-labeled"
+    repository = SimpleNamespace(
+        get=AsyncMock(
+            return_value=StickerSemanticLabel(
+                content_hash=content_hash_for_bytes(content),
+                is_sticker=True,
+                confidence=0.95,
+                prompt_version=1,
+            )
+        )
+    )
+    store = SimpleNamespace(requeue_terminal=AsyncMock())
+    metric = Mock()
+    monkeypatch.setattr(sticker_label_jobs, "sticker_label_repository", lambda: repository)
+    monkeypatch.setattr(sticker_label_jobs, "build_work_job_store", lambda: store)
+    monkeypatch.setattr("pallas.product.llm.task_metrics.record_bot_llm_task", metric)
+
+    assert not await sticker_label_jobs.enqueue_sticker_label_candidate(
+        cache_key="[CQ:image,file=already-labeled.image]",
+        content=content,
+        source=StickerLabelSource.FOLLOWUP_CANDIDATE,
+    )
+    store.requeue_terminal.assert_not_awaited()
+    metric.assert_any_call("sticker_label", "cache_hit")
+    assert not any(call.args[1] == "submit_ok" for call in metric.call_args_list)
+    assert record_bot_llm_task is not None
+
+
+@pytest.mark.asyncio
+async def test_enqueue_candidate_counts_coalesced_when_job_already_pending(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from pallas.product.llm import sticker_label_jobs
+    from pallas.product.llm.sticker_label_jobs import StickerLabelSource
+    from pallas.product.llm.sticker_labels import content_hash_for_bytes
+
+    content = b"dup-content"
+    repository = SimpleNamespace(get=AsyncMock(return_value=None))
+    store = SimpleNamespace(requeue_terminal=AsyncMock(side_effect=lambda job: (job, False)))
+    metric = Mock()
+    monkeypatch.setattr(sticker_label_jobs, "sticker_label_repository", lambda: repository)
+    monkeypatch.setattr(sticker_label_jobs, "build_work_job_store", lambda: store)
+    monkeypatch.setattr("pallas.core.shared.utils.media_cache.bind_image_content_hash", AsyncMock())
+    monkeypatch.setattr("pallas.product.llm.task_metrics.record_bot_llm_task", metric)
+
+    queued = await sticker_label_jobs.enqueue_sticker_label_candidate(
+        cache_key="[CQ:image,file=dup.image]",
+        content=content,
+        source=StickerLabelSource.FOLLOWUP_CANDIDATE,
+    )
+
+    assert queued is True
+    assert not any(call.args[1] == "submit_ok" for call in metric.call_args_list)
+    assert any(call.args == ("sticker_label", "background_coalesced") for call in metric.call_args_list)
+
+
 @pytest.mark.asyncio
 async def test_string_source_cannot_enqueue_a_label_job(monkeypatch: pytest.MonkeyPatch) -> None:
     from pallas.product.llm import sticker_label_jobs
@@ -413,3 +478,49 @@ async def test_cache_changed_is_completed_by_actual_worker(monkeypatch: pytest.M
 
     assert await worker.run_once()
     assert await store.claim(owner="other", lease_sec=1) is None
+
+
+@pytest.mark.asyncio
+async def test_permanent_label_errors_do_not_retry_worker(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from pallas.core.platform.work_jobs.models import WorkJob
+    from pallas.core.platform.work_jobs.store import MemoryWorkJobStore
+    from pallas.core.platform.work_jobs.worker import WorkJobWorker
+    from pallas.product.llm import sticker_label_jobs
+    from pallas.product.llm.sticker_labels import content_hash_for_bytes
+
+    sticker_label_jobs.reset_sticker_label_runtime_state_for_tests()
+    for case in ("parse_error", "no_vision"):
+        save_observation = AsyncMock()
+        monkeypatch.setattr(sticker_label_jobs, "save_sticker_label_observation", save_observation)
+        monkeypatch.setattr(
+            "pallas.core.shared.utils.media_cache.get_image_by_content_hash", AsyncMock(return_value=b"image")
+        )
+
+        async def fail_vision(_content: bytes) -> None:
+            if case == "parse_error":
+                raise ValueError("invalid sticker label JSON")
+            raise RuntimeError("no sticker vision endpoint")
+
+        monkeypatch.setattr(sticker_label_jobs, "label_sticker_with_vision", fail_vision)
+        store = MemoryWorkJobStore()
+        await store.enqueue(
+            WorkJob.create(
+                kind="sticker.label.visual",
+                payload={"content_hash": content_hash_for_bytes(b"image"), "observation": {"state": "queued"}},
+                idempotency_key=f"label:permanent:{case}",
+            )
+        )
+        worker = WorkJobWorker(
+            store=store, owner="worker", handlers={"sticker.label.visual": sticker_label_jobs.handle_sticker_label_visual}
+        )
+
+        assert await worker.run_once()
+        assert (await store.stats())["leased"] == 0
+        assert (await store.stats())["dead_lettered"] == 0
+        assert worker.metrics.snapshot()["retried_since_start"] == 0
+        assert (await store.stats())["pending"] == 0
+        assert await store.claim(owner="next", lease_sec=1) is None
+        assert save_observation.await_args.args[2]["state"] in {"parse_error", "no_vision"}
+    sticker_label_jobs.reset_sticker_label_runtime_state_for_tests()

tests/product/llm/test_sticker_label_observability.py

@@ -90,3 +90,44 @@ async def test_requeue_stale_labels_counts_active_jobs_as_skipped(beanie_fixture
     )
 
     assert result == {"requeued": 0, "queued": 0, "skipped": 1, "missing_cache": 0}
+
+
+def test_build_sticker_label_job_stats_counts_terminal_states_by_category() -> None:
+    from pallas.product.llm.sticker_label_observability import build_sticker_label_job_stats
+
+    records = [
+        {
+            "job_id": f"job-{state}",
+            "created_at": float(index),
+            "payload": {"observation": {"state": state, "error": f"err-{state}"}},
+            "last_error": None,
+        }
+        for index, state in enumerate(
+            ["labeled", "labeled", "timeout", "parse_error", "no_vision", "failed", "circuit_open", "cache_changed"]
+        )
+    ]
+    records.append({
+        "job_id": "job-pending",
+        "created_at": 100.0,
+        "payload": {"observation": {"state": "queued"}},
+        "last_error": None,
+    })
+    records.append({
+        "job_id": "job-running",
+        "created_at": 101.0,
+        "payload": {"observation": {"state": "running"}},
+        "last_error": None,
+    })
+
+    stats = build_sticker_label_job_stats(records, recent_limit=3)
+
+    assert stats["submitted"] == 10
+    assert stats["labeled"] == 2
+    assert stats["timeout"] == 1
+    assert stats["parse_error"] == 1
+    assert stats["no_vision"] == 1
+    assert stats["failed"] == 1
+    assert stats["circuit_open"] == 1
+    assert stats["cache_changed"] == 1
+    assert stats["pending"] == 2
+    assert len(stats["recent_errors"]) == 3