service/runner.py

from __future__ import annotations

import hashlib
import random
import re
import threading
import time
from collections import Counter, deque
from datetime import datetime, timezone
import urllib.parse
from typing import Any, Dict

import requests
from loguru import logger
import asyncio
import os
import sys

sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from orchestrator.agent_utils import get_llm, get_browser_with_auth
from browser_use import Agent

from .config import ServiceConfig
from .stability_controller import StabilityController, CaptchaSolverHook, CaptchaChallenge, CaptchaSolution
from .storage import LocalJsonStorage
from .tasks import CallbackState, CallbackStatus, TaskStatus
from engines import MediaCrawlerEngine, SpiderXHSEngine

class AgenticCaptchaSolver(CaptchaSolverHook):
    def __init__(self):
        super().__init__()

    async def solve(self, challenge: CaptchaChallenge) -> CaptchaSolution:
        logger.info(f"AgenticCaptchaSolver invoked for task {challenge.task_id}, engine: {challenge.engine_name}")
        
        # Determine the storage state path based on payload
        payload = challenge.payload
        storage_state_path = payload.get("storage_state_path") or payload.get("storage_state")
        
        llm = get_llm()
        browser = get_browser_with_auth(storage_state_path)
        
        prompt = """
        页面出现验证码，请观察当前页面的截图。
        如果是滑块验证码，请拖动滑块到缺口位置。
        如果是点选验证码，请按顺序点击图中要求的文字。
        如果成功通过验证码，请返回 success，否则返回 failed。
        """
        
        agent = Agent(task=prompt, llm=llm, browser=browser)
        
        try:
            result = await agent.run()
            final_text = result.final_result() if hasattr(result, 'final_result') else str(result)
            ok = "success" in final_text.lower()
            return CaptchaSolution(ok=ok, meta={"agent_result": final_text})
        except Exception as e:
            logger.error(f"AgenticCaptchaSolver failed: {e}")
            return CaptchaSolution(ok=False, meta={"error": str(e)})
        finally:
            await browser.close()


class _RunnerCounters:
    def __init__(self, *, recent_window_seconds: int = 300, recent_maxlen: int = 2000):
        self._lock = threading.Lock()
        self._recent_window_seconds = int(recent_window_seconds)
        self._recent: deque[tuple[float, bool]] = deque(maxlen=int(recent_maxlen))
        self._started_total = 0
        self._succeeded_total = 0
        self._failed_total = 0
        self._engine_usage_total: Counter[str] = Counter()
        self._engine_succeeded_total: Counter[str] = Counter()
        self._engine_failed_total: Counter[str] = Counter()

    def record_started(self) -> None:
        with self._lock:
            self._started_total += 1

    def _prune_recent(self, now_ts: float) -> None:
        window = float(self._recent_window_seconds)
        while self._recent:
            ts, _ = self._recent[0]
            if now_ts - float(ts) <= window:
                return
            self._recent.popleft()

    def record_finished(self, *, engine: str, ok: bool) -> None:
        name = str(engine or "unknown")
        now_ts = time.time()
        with self._lock:
            self._engine_usage_total[name] += 1
            if bool(ok):
                self._succeeded_total += 1
                self._engine_succeeded_total[name] += 1
            else:
                self._failed_total += 1
                self._engine_failed_total[name] += 1
            self._recent.append((now_ts, bool(ok)))
            self._prune_recent(now_ts)

    def snapshot(self) -> Dict[str, Any]:
        now_ts = time.time()
        with self._lock:
            self._prune_recent(now_ts)
            recent_total = len(self._recent)
            recent_failed = sum(1 for _, ok in self._recent if not ok)
            recent_rate = float(recent_failed) / float(recent_total) if recent_total > 0 else 0.0
            return {
                "started_total": int(self._started_total),
                "succeeded_total": int(self._succeeded_total),
                "failed_total": int(self._failed_total),
                "engine_usage_total": dict(self._engine_usage_total),
                "engine_succeeded_total": dict(self._engine_succeeded_total),
                "engine_failed_total": dict(self._engine_failed_total),
                "recent_window_seconds": int(self._recent_window_seconds),
                "recent_total": int(recent_total),
                "recent_failed": int(recent_failed),
                "recent_failure_rate": float(recent_rate),
            }


class TaskRunner:
    def __init__(self, *, storage: LocalJsonStorage, config: ServiceConfig):
        self.storage = storage
        self.config = config
        self._engine_api = SpiderXHSEngine(proxy=config.proxy)
        self._engine_browser = MediaCrawlerEngine(
            proxy=config.proxy,
            storage_state_paths=config.mediacrawler_storage_state_paths,
        )
        self._ops_counters = _RunnerCounters()
        self._stability = StabilityController(
            engine_fallback_threshold=int(getattr(config, "engine_fallback_threshold", 3) or 3),
        )
        self._stability.set_captcha_solver(AgenticCaptchaSolver())

    def get_ops_stats(self) -> Dict[str, Any]:
        return self._ops_counters.snapshot()

    @staticmethod
    def _normalize_engine_choice(choice: str) -> str:
        value = (choice or "").strip().lower()
        if value in ("api", "spider_xhs", "xhs_api"):
            return "api"
        if value in ("browser", "mediacrawler", "xhs_browser"):
            return "browser"
        if value in ("auto", ""):
            return "auto"
        return value

    @staticmethod
    def _first_str(*values: Any) -> str | None:
        for value in values:
            if value is None:
                continue
            if isinstance(value, str):
                v = value.strip()
                if v != "":
                    return v
            try:
                v = str(value).strip()
                if v != "":
                    return v
            except Exception:
                continue
        return None

    @staticmethod
    def _extract_scope(task_type: str, payload: Dict[str, Any]) -> str:
        url_candidate = TaskRunner._first_str(
            payload.get("note_url"),
            payload.get("url"),
            payload.get("user_url"),
        )
        if url_candidate:
            try:
                parsed = urllib.parse.urlparse(url_candidate)
                if parsed.hostname:
                    return parsed.hostname.lower()
            except Exception:
                pass
        return (task_type or "unknown").strip().lower() or "unknown"

    @staticmethod
    def _safe_url_for_log(url: str | None) -> str:
        if url is None:
            return ""
        u = str(url).strip()
        if u == "":
            return ""
        try:
            parsed = urllib.parse.urlparse(u)
            host = parsed.hostname or ""
            if parsed.port:
                host = f"{host}:{parsed.port}"
            if host == "":
                return ""
            scheme = parsed.scheme or "http"
            path = parsed.path or ""
            return f"{scheme}://{host}{path}"
        except Exception:
            return ""

    @staticmethod
    def _safe_error_text(text: Any) -> str:
        if text is None:
            return ""
        raw = str(text)
        if raw.strip() == "":
            return ""
        raw = re.sub(
            r"https?://[^\s]+",
            lambda m: TaskRunner._safe_url_for_log(m.group(0)) or "",
            raw,
        )
        raw = re.sub(r"(?i)(xsec_token=)[^&\s]+", r"\1***", raw)
        raw = re.sub(r"(?i)(cookies?=)[^&\s]+", r"\1***", raw)
        raw = re.sub(r"(?i)(authorization:\s*)(bearer\s+)?[^\s]+", r"\1***", raw)
        raw = raw.strip()
        if len(raw) > 200:
            return raw[:200] + "..."
        return raw

    def _safe_run_engine(self, engine: Any, task: Any) -> Dict[str, Any]:
        try:
            output = engine.run(task)
            return output.to_dict()
        except Exception as e:
            payload = getattr(task, "payload", {}) or {}
            source_type = getattr(task, "task_type", "") or ""
            source_ref = str(
                payload.get("note_url")
                or payload.get("url")
                or payload.get("query")
                or payload.get("keyword")
                or payload.get("user_id")
                or ""
            )
            dedup_payload = f"{source_type}:{source_ref}".encode("utf-8", errors="ignore")
            return {
                "raw": None,
                "normalized": None,
                "meta": {
                    "task_id": getattr(task, "id", None) or "",
                    "source_engine": "api" if getattr(engine, "name", "") == "spider_xhs" else "browser",
                    "engine_name": getattr(engine, "name", None) or "",
                    "source_type": source_type,
                    "source_ref": source_ref,
                    "operator": "system",
                    "ingested_at": datetime.now(timezone.utc).isoformat(),
                    "dedup_key": hashlib.sha1(dedup_payload).hexdigest(),
                    "ok": False,
                    "error_kind": "parse",
                    "error_message": str(e),
                },
            }

    def _persist_raw_html_snapshot(self, task_id: str, engine: Any, result: Dict[str, Any]) -> str | None:
        engine_name = getattr(engine, "name", None)
        if engine_name != getattr(self._engine_browser, "name", "mediacrawler"):
            return None
        raw = result.get("raw")
        if not isinstance(raw, dict):
            return None
        html = raw.get("html")
        if not isinstance(html, str) or html.strip() == "":
            return None
        path = self.storage.write_raw_html_snapshot(task_id, html)
        try:
            rel = path.relative_to(self.storage.base_dir)
            return str(rel)
        except Exception:
            return str(path)

    @staticmethod
    def _to_dict(model: Any) -> Dict[str, Any]:
        if hasattr(model, "model_dump"):
            return model.model_dump()
        if hasattr(model, "dict"):
            return model.dict()
        return dict(model)

    @staticmethod
    def _callback_idempotency_key(task_id: str, result: Dict[str, Any]) -> str:
        meta = result.get("meta") if isinstance(result, dict) else None
        dedup_key = meta.get("dedup_key") if isinstance(meta, dict) else None
        payload = f"{task_id}:{dedup_key or ''}".encode("utf-8", errors="ignore")
        return hashlib.sha256(payload).hexdigest()

    @staticmethod
    def _sleep_backoff(attempt: int, *, base_s: float = 0.6, max_s: float = 10.0) -> None:
        exp = min(max_s, base_s * (2**attempt))
        jitter = random.uniform(0.0, exp * 0.25)
        time.sleep(exp + jitter)

    def _push_callback(self, task_id: str, result: Dict[str, Any]) -> CallbackState | None:
        callback_url = getattr(self.config, "callback_url", None)
        if callback_url is None or str(callback_url).strip() == "":
            return None

        existing = self.storage.get_task(task_id)
        existing_state = getattr(existing, "callback", None) if existing is not None else None
        if existing_state is not None and getattr(existing_state, "status", None) == CallbackStatus.succeeded:
            return existing_state

        idempotency_key = self._callback_idempotency_key(task_id, result)
        state = CallbackState(
            status=CallbackStatus.pending,
            callback_url=callback_url,
            idempotency_key=idempotency_key,
            attempts=int(getattr(existing_state, "attempts", 0) or 0),
            last_attempt_at=getattr(existing_state, "last_attempt_at", None),
            last_http_status=getattr(existing_state, "last_http_status", None),
            last_error=getattr(existing_state, "last_error", None),
            next_retry_at=getattr(existing_state, "next_retry_at", None),
        )
        self.storage.update_task(task_id, callback=state)

        body = {"task_id": task_id, "result": result}
        headers = {"Idempotency-Key": idempotency_key}

        max_attempts = 5
        safe_callback_url = self._safe_url_for_log(callback_url)
        logger.bind(task_id=task_id, status="callback").info(
            "callback_started callback_url={} attempts={}",
            safe_callback_url,
            int(state.attempts),
        )
        for attempt in range(max_attempts):
            state.attempts = int(state.attempts) + 1
            state.last_attempt_at = time.time()
            state.next_retry_at = None
            self.storage.update_task(task_id, callback=state)
            try:
                resp = requests.post(callback_url, json=body, headers=headers, timeout=(5.0, 20.0))
                state.last_http_status = int(resp.status_code)
                if 200 <= int(resp.status_code) < 300:
                    state.status = CallbackStatus.succeeded
                    state.last_error = None
                    state.next_retry_at = None
                    self.storage.update_task(task_id, callback=state)
                    self.storage.delete_callback_failure(task_id)
                    logger.bind(task_id=task_id, status="callback").info(
                        "callback_succeeded callback_url={} http_status={} attempts={}",
                        safe_callback_url,
                        int(resp.status_code),
                        int(state.attempts),
                    )
                    return state
                state.last_error = f"http_status={resp.status_code}"
                if int(resp.status_code) not in (429, 500, 502, 503, 504) and attempt >= 0:
                    break
            except Exception as e:
                state.last_http_status = None
                state.last_error = str(e)

            if attempt < max_attempts - 1:
                self._sleep_backoff(attempt)

        state.status = CallbackStatus.failed
        state.next_retry_at = time.time() + 60.0
        self.storage.update_task(task_id, callback=state)
        self.storage.write_callback_failure(
            task_id,
            {
                "task_id": task_id,
                "callback_url": callback_url,
                "idempotency_key": idempotency_key,
                "body": body,
                "callback": self._to_dict(state),
                "saved_at": time.time(),
            },
        )
        logger.bind(task_id=task_id, status="callback").warning(
            "callback_failed callback_url={} attempts={} last_http_status={} last_error={}",
            safe_callback_url,
            int(state.attempts),
            str(state.last_http_status) if state.last_http_status is not None else "",
            self._safe_error_text(state.last_error),
        )
        return state

    def import_extension_result(self, task_id: str, result: Dict[str, Any]) -> TaskRecord:
        task = self.storage.get_task(task_id)
        if task is None:
            raise KeyError("task not found")

        self.storage.write_result(task_id, result)
        updated = self.storage.update_task(task_id, status=TaskStatus.rpa_imported, engine="extension_rpa", error=None)
        try:
            self._push_callback(task_id, result)
        except Exception:
            pass

        return updated or self.storage.get_task(task_id) or task

    def retry_callback(self, task_id: str) -> CallbackState:
        callback_url = getattr(self.config, "callback_url", None)
        if callback_url is None or str(callback_url).strip() == "":
            raise ValueError("callback_url not configured")

        task = self.storage.get_task(task_id)
        if task is None:
            raise KeyError("task not found")

        result = self.storage.read_result(task_id)
        if result is None:
            raise RuntimeError("result not ready")

        failure = self.storage.read_callback_failure(task_id) or {}
        url = str(failure.get("callback_url") or callback_url)
        body = failure.get("body")
        if not isinstance(body, dict):
            body = {"task_id": task_id, "result": result}
        idempotency_key = str(
            failure.get("idempotency_key") or self._callback_idempotency_key(task_id, result)
        )

        body_task_id = body.get("task_id")
        if body_task_id != task_id:
            body["task_id"] = task_id
        if "result" not in body:
            body["result"] = result

        state = CallbackState(
            status=CallbackStatus.pending,
            callback_url=url,
            idempotency_key=idempotency_key,
            attempts=int(getattr(getattr(task, "callback", None), "attempts", 0) or 0),
        )
        self.storage.update_task(task_id, callback=state)

        headers = {"Idempotency-Key": idempotency_key}
        max_attempts = 5
        safe_url = self._safe_url_for_log(url)
        logger.bind(task_id=task_id, status="callback").info(
            "callback_retry_started callback_url={} attempts={}",
            safe_url,
            int(state.attempts),
        )
        for attempt in range(max_attempts):
            state.attempts = int(state.attempts) + 1
            state.last_attempt_at = time.time()
            state.next_retry_at = None
            self.storage.update_task(task_id, callback=state)
            try:
                resp = requests.post(url, json=body, headers=headers, timeout=(5.0, 20.0))
                state.last_http_status = int(resp.status_code)
                if 200 <= int(resp.status_code) < 300:
                    state.status = CallbackStatus.succeeded
                    state.last_error = None
                    state.next_retry_at = None
                    self.storage.update_task(task_id, callback=state)
                    self.storage.delete_callback_failure(task_id)
                    logger.bind(task_id=task_id, status="callback").info(
                        "callback_retry_succeeded callback_url={} http_status={} attempts={}",
                        safe_url,
                        int(resp.status_code),
                        int(state.attempts),
                    )
                    return state
                state.last_error = f"http_status={resp.status_code}"
                if int(resp.status_code) not in (429, 500, 502, 503, 504) and attempt >= 0:
                    break
            except Exception as e:
                state.last_http_status = None
                state.last_error = str(e)

            if attempt < max_attempts - 1:
                self._sleep_backoff(attempt)

        state.status = CallbackStatus.failed
        state.next_retry_at = time.time() + 60.0
        self.storage.update_task(task_id, callback=state)
        self.storage.write_callback_failure(
            task_id,
            {
                "task_id": task_id,
                "callback_url": url,
                "idempotency_key": idempotency_key,
                "body": body,
                "callback": self._to_dict(state),
                "saved_at": time.time(),
            },
        )
        logger.bind(task_id=task_id, status="callback").warning(
            "callback_retry_failed callback_url={} attempts={} last_http_status={} last_error={}",
            safe_url,
            int(state.attempts),
            str(state.last_http_status) if state.last_http_status is not None else "",
            self._safe_error_text(state.last_error),
        )
        return state

    def run_task(self, task_id: str) -> None:
        task = self.storage.get_task(task_id)
        if task is None:
            return

        payload = getattr(task, "payload", {}) or {}
        scope = self._extract_scope(getattr(task, "task_type", "") or "", payload)
        task_type = str(getattr(task, "task_type", "") or "")
        engine_req = str(getattr(task, "engine", None) or "")
        logger.bind(task_id=task_id, task_type=task_type, engine=engine_req, status="running").info(
            "task_started scope={}",
            scope,
        )
        self._ops_counters.record_started()
        self.storage.update_task(task_id, status=TaskStatus.running)

        try:
            if (getattr(task, "task_type", "") or "").strip().lower() == "selftest":
                operator = str(payload.get("operator") or "system")
                dedup_payload = f"selftest:{task_id}".encode("utf-8", errors="ignore")
                meta: Dict[str, Any] = {
                    "task_id": task_id,
                    "source_engine": "selftest",
                    "engine_name": "selftest",
                    "source_type": "selftest",
                    "source_ref": "",
                    "operator": operator,
                    "ingested_at": datetime.now(timezone.utc).isoformat(),
                    "dedup_key": hashlib.sha1(dedup_payload).hexdigest(),
                    "ok": True,
                    "primary_engine": "selftest",
                    "final_engine": "selftest",
                }
                result: Dict[str, Any] = {
                    "raw": {"message": "ok"},
                    "normalized": {"kind": "selftest"},
                    "meta": meta,
                }
                self.storage.update_task(task_id, engine="selftest")
                self.storage.write_result(task_id, result)
                self.storage.update_task(task_id, status=TaskStatus.succeeded, error=None)
                self._ops_counters.record_finished(engine="selftest", ok=True)
                logger.bind(task_id=task_id, task_type=task_type, engine="selftest", status="succeeded").info(
                    "task_succeeded"
                )
                try:
                    self._push_callback(task_id, result)
                except Exception:
                    pass
                return

            engine_choice = self._normalize_engine_choice(task.engine or self.config.engine_strategy or "auto")
            exec_state = self._stability.init_execution(
                task_engine_choice=engine_choice,
                scope=scope,
                engine_api=self._engine_api,
                engine_browser=self._engine_browser,
            )

            primary_name = str(getattr(exec_state.primary_engine, "name", exec_state.primary_key))
            retry_count = int(getattr(task, "retry_count", 0) or 0)

            final_engine = exec_state.primary_engine
            final_name = primary_name
            final_result: Dict[str, Any] | None = None
            paused_status: TaskStatus | None = None

            while True:
                payload_now = getattr(task, "payload", {}) or {}
                plan = self._stability.build_attempt_plan(state=exec_state, payload=payload_now)

                if plan.status in (TaskStatus.retrying, TaskStatus.fallback_running):
                    retry_count += 1

                self.storage.update_task(
                    task_id,
                    status=plan.status,
                    engine=plan.engine_name,
                    retry_count=retry_count,
                    payload=plan.payload,
                )
                task.payload = plan.payload
                task.engine = plan.engine_name

                result = self._safe_run_engine(plan.engine, task)
                meta = result.get("meta") if isinstance(result.get("meta"), dict) else {}
                ok = bool(meta.get("ok"))
                error_kind = str(meta.get("error_kind") or "").strip().lower() or None

                self._stability.observe_result(
                    state=exec_state,
                    engine_name=plan.engine_name,
                    ok=ok,
                    error_kind=error_kind,
                )

                if ok:
                    self._stability.report_success(resources=plan.resources)
                    final_engine = plan.engine
                    final_name = plan.engine_name
                    final_result = result
                    break

                self._stability.report_failure(resources=plan.resources, error_kind=error_kind)
                decision = self._stability.decide_after_failure(
                    state=exec_state,
                    engine_name=plan.engine_name,
                    error_kind=error_kind,
                )

                if decision.action == "pause" and error_kind in ("captcha", "risk"):
                    logger.info(f"Triggering AgenticCaptchaSolver for error_kind={error_kind}")
                    challenge = self._stability.build_captcha_challenge(
                        engine_name=plan.engine_name,
                        scope=scope,
                        task_id=task_id,
                        payload=plan.payload,
                    )
                    try:
                        solution = asyncio.run(self._stability.try_solve_captcha(challenge=challenge))
                        if solution and solution.ok:
                            logger.info("AgenticCaptchaSolver succeeded! Retrying task...")
                            TaskRunner._sleep_backoff(int(exec_state.attempts_by_key.get(exec_state.current_key, 1) or 1))
                            continue
                        else:
                            logger.warning(f"AgenticCaptchaSolver failed. Solution: {solution}")
                    except Exception as e:
                        logger.error(f"Failed to run AgenticCaptchaSolver: {e}")

                if decision.action == "retry":
                    TaskRunner._sleep_backoff(int(exec_state.attempts_by_key.get(exec_state.current_key, 1) or 1))
                    continue

                if decision.action == "fallback" and decision.next_key is not None:
                    self._stability.switch_to(state=exec_state, key=decision.next_key, fallback_reason=decision.fallback_reason)
                    continue

                final_engine = plan.engine
                final_name = plan.engine_name
                final_result = result
                if decision.action == "pause":
                    paused_status = decision.status or TaskStatus.waiting_rpa
                break

            if final_result is None:
                raise RuntimeError("runner_no_result")

            meta = final_result.get("meta") if isinstance(final_result.get("meta"), dict) else {}
            meta = dict(meta or {})
            meta["primary_engine"] = primary_name
            meta["final_engine"] = str(final_name)
            if exec_state.fallback_reason is not None:
                meta["fallback_reason"] = exec_state.fallback_reason
            raw_html_path = self._persist_raw_html_snapshot(task.id, final_engine, final_result)
            if raw_html_path is not None:
                meta["raw_html_path"] = raw_html_path
            final_result["meta"] = meta

            self.storage.write_result(task.id, final_result)

            if paused_status is not None:
                err_msg = str(meta.get("error_message") or "engine_failed")
                err_kind = str(meta.get("error_kind") or "parse")
                self.storage.update_task(
                    task_id,
                    status=paused_status,
                    error={"kind": err_kind, "message": err_msg, "engine": str(final_name)},
                )
                return

            if bool(meta.get("ok")):
                self.storage.update_task(task_id, status=TaskStatus.succeeded, error=None)
                self._ops_counters.record_finished(engine=str(final_name), ok=True)
                logger.bind(task_id=task_id, task_type=task_type, engine=str(final_name), status="succeeded").info(
                    "task_succeeded"
                )
            else:
                err_msg = str(meta.get("error_message") or "engine_failed")
                err_kind = str(meta.get("error_kind") or "parse")
                self.storage.update_task(
                    task_id,
                    status=TaskStatus.failed,
                    error={"kind": err_kind, "message": err_msg, "engine": str(final_name)},
                )
                self._ops_counters.record_finished(engine=str(final_name), ok=False)
                logger.bind(task_id=task_id, task_type=task_type, engine=str(final_name), status="failed").error(
                    "task_failed error_kind={} error_message={}",
                    str(meta.get("error_kind") or ""),
                    self._safe_error_text(err_msg),
                )
            try:
                existing = self.storage.get_task(task_id)
                if paused_status is None and existing is not None and existing.status != TaskStatus.waiting_rpa:
                    self._push_callback(task_id, final_result)
            except Exception:
                pass
        except Exception as e:
            source_type = getattr(task, "task_type", "") if task is not None else ""
            source_ref = ""
            engine_choice = "auto"
            try:
                engine_choice = self._normalize_engine_choice(
                    getattr(task, "engine", None) or self.config.engine_strategy or "auto"
                )
            except Exception:
                engine_choice = "auto"
            primary_engine = self._engine_api
            if engine_choice == "browser":
                primary_engine = self._engine_browser
            try:
                payload = getattr(task, "payload", {}) or {}
                source_ref = str(
                    payload.get("note_url")
                    or payload.get("url")
                    or payload.get("query")
                    or payload.get("keyword")
                    or payload.get("user_id")
                    or ""
                )
            except Exception:
                source_ref = ""
            dedup_payload = f"{source_type}:{source_ref}".encode("utf-8", errors="ignore")
            primary_name = getattr(primary_engine, "name", "spider_xhs")
            result = {
                "raw": None,
                "normalized": None,
                "meta": {
                    "task_id": task_id,
                    "source_engine": "api" if primary_name == "spider_xhs" else "browser",
                    "engine_name": primary_name,
                    "source_type": source_type,
                    "source_ref": source_ref,
                    "operator": "system",
                    "ingested_at": datetime.now(timezone.utc).isoformat(),
                    "dedup_key": hashlib.sha1(dedup_payload).hexdigest(),
                    "ok": False,
                    "primary_engine": primary_name,
                    "final_engine": primary_name,
                    "fallback_reason": "runner_exception",
                    "error_kind": "parse",
                    "error_message": str(e),
                },
            }
            self.storage.write_result(task_id, result)
            self.storage.update_task(task_id, status=TaskStatus.failed, error={"kind": "parse", "message": str(e)})
            self._ops_counters.record_finished(engine=str(primary_name), ok=False)
            logger.bind(task_id=task_id, task_type=task_type, engine=str(primary_name), status="failed").error(
                "task_failed error_kind=parse error_message={}",
                self._safe_error_text(str(e)),
            )
            try:
                self._push_callback(task_id, result)
            except Exception:
                pass