workflow.py

"""
Core orchestrator for the current session-level evolve_server pipeline.

Active flow:
1. Drain pending sessions from shared storage.
2. Summarize sessions and extract metadata.
3. Optionally backfill a session-level score with session_judge.
4. Aggregate sessions by referenced skill.
5. Evolve existing-skill groups or create new skills from no-skill groups.
6. Upload skills, persist registry state, and ack processed sessions.
"""

from __future__ import annotations

import asyncio
import hashlib
import logging
import time
from datetime import datetime, timezone
from typing import Any, Optional

from skillclaw.skill_bundle import bundle_tree_sha256
from skillclaw.validation_store import ValidationStore

from ..core.config import EvolveServerConfig
from ..core.constants import NO_SKILL_KEY, DecisionAction
from ..core.llm_client import AsyncLLMClient
from ..core.skill_registry import SkillIDRegistry
from ..core.utils import build_skill_md, parse_skill_content
from ..pipeline.aggregation import aggregate_sessions_by_skill
from ..pipeline.execution import (
    create_skill_from_sessions,
    evolve_skill_from_sessions,
    execute_merge,
    set_evolve_debug_dir,
)
from ..pipeline.session_judge import judge_sessions_parallel
from ..pipeline.skill_verifier import verify_skill_candidate
from ..pipeline.summarizer import set_summarizer_debug_dir, summarize_sessions_parallel
from ..storage.oss_helpers import (
    delete_session_keys,
    fetch_skill_content,
    list_session_keys,
    save_manifest,
    save_version_bundle,
)
from .common import EvolveEngineMixin

logger = logging.getLogger(__name__)


class EvolveServer(EvolveEngineMixin):
    """Session-level evolve server backed by shared object storage."""

    def __init__(
        self,
        config: EvolveServerConfig,
        *,
        mock: bool = False,
        mock_root: str | None = None,
    ) -> None:
        self.config = config
        self._mock = mock
        self._bucket = self._build_bucket(config, mock=mock, mock_root=mock_root)
        self._prefix = f"{config.group_id}/"
        self._llm = AsyncLLMClient(
            api_key=config.llm_api_key,
            base_url=config.llm_base_url,
            model=config.llm_model,
            max_tokens=config.llm_max_tokens,
            temperature=config.llm_temperature,
        )
        self._validation_store = ValidationStore(
            backend=self.config.storage_backend,
            endpoint=self.config.storage_endpoint,
            bucket=self.config.storage_bucket,
            access_key_id=self.config.storage_access_key_id,
            secret_access_key=self.config.storage_secret_access_key,
            region=self.config.storage_region,
            session_token=self.config.storage_session_token,
            local_root=self.config.local_root,
            group_id=self.config.group_id,
        )
        self._id_registry = SkillIDRegistry()
        self._nacos_skill_client = self._build_nacos_skill_client()
        self._running = False

        set_evolve_debug_dir(config.debug_dump_dir)
        set_summarizer_debug_dir(config.debug_dump_dir)
        if not self._uses_nacos_skill_registry():
            self._id_registry.load_from_oss(self._bucket, self._prefix)

    def _uses_nacos_skill_registry(self) -> bool:
        return str(getattr(self.config, "skill_storage_backend", "") or "").strip().lower() == "nacos"

    def _build_nacos_skill_client(self) -> Any | None:
        if not self._uses_nacos_skill_registry():
            return None
        from skillclaw.nacos_skill_hub import NacosSkillClient

        return NacosSkillClient(
            server=str(getattr(self.config, "nacos_server", "") or ""),
            namespace_id=str(getattr(self.config, "nacos_namespace_id", "") or "public"),
            access_token=str(getattr(self.config, "nacos_access_token", "") or ""),
            username=str(getattr(self.config, "nacos_username", "") or ""),
            password=str(getattr(self.config, "nacos_password", "") or ""),
        )

    def _load_remote_skills(self) -> dict[str, dict[str, Any]]:
        if self._nacos_skill_client is not None:
            skills: dict[str, dict[str, Any]] = {}
            for item in self._nacos_skill_client.list_skills():
                name = str(item.get("name") or "")
                if name:
                    skills[name] = item
            return skills
        return super()._load_remote_skills()

    def _load_remote_skill_record(self, name: str) -> Optional[dict[str, Any]]:
        rec = self._load_remote_skills().get(name)
        return rec if isinstance(rec, dict) else None

    @staticmethod
    def _overlay_manifest_metadata(
        skill: Optional[dict[str, Any]],
        manifest_record: Optional[dict[str, Any]],
    ) -> Optional[dict[str, Any]]:
        if not skill or not manifest_record:
            return skill
        category = str(manifest_record.get("category", "") or "").strip()
        if category and str(skill.get("category", "general") or "general").strip() == "general":
            skill["category"] = category
        if not str(skill.get("description", "") or "").strip():
            description = str(manifest_record.get("description", "") or "").strip()
            if description:
                skill["description"] = description
        return skill

    def _wait_nacos_publish(self, name: str, version: str, timeout: float = 30.0) -> bool:
        """Publish a Nacos skill version and wait for it to go online.

        Nacos v3 runs a publish pipeline on submit. The pipeline must approve
        before the publish call succeeds. This method polls for approval and
        retries publish until the version goes online with the latest label.
        """
        import time

        deadline = time.monotonic() + timeout
        published = False
        while time.monotonic() < deadline:
            try:
                detail = self._nacos_skill_client.get_skill(name)
                labels = detail.get("labels", {}) if detail else {}
                if labels.get("latest") == version:
                    return True
                if not published:
                    try:
                        self._nacos_skill_client.publish(name, version, update_latest_label=True)
                        logger.info("[EvolveServer] Nacos publish accepted for %s %s", name, version)
                        published = True
                    except Exception:
                        pass
            except Exception:
                pass
            time.sleep(2.0)
        logger.warning("[EvolveServer] Nacos publish not confirmed for %s %s within %.0fs", name, version, timeout)
        return False

    def _fetch_skill(self, name: str) -> Optional[str]:
        if self._nacos_skill_client is not None:
            try:
                from skillclaw.nacos_skill_hub import (
                    _nacos_published_version,
                    _nacos_working_version,
                    _nacos_zip_to_bundle,
                )

                record = self._load_remote_skill_record(name) or {}
                try:
                    detail = self._nacos_skill_client.get_skill(name) if record else {}
                except Exception:
                    detail = {}
                working = _nacos_working_version(record, detail)
                if working:
                    _status, version = working
                    zip_bytes = self._nacos_skill_client.download_skill_zip(name, version=version, admin=True)
                else:
                    label = str(getattr(self.config, "nacos_label", "") or "latest")
                    version = _nacos_published_version(record, detail, label=label)
                    if not version:
                        logger.info(
                            "[EvolveServer] Nacos skill %s has no published %s version",
                            name,
                            label,
                        )
                        return None
                    zip_bytes = self._nacos_skill_client.download_skill_zip(
                        name,
                        version=version,
                        label=label,
                    )
                bundle = _nacos_zip_to_bundle(zip_bytes)
                data = bundle.get("SKILL.md")
                return data.decode("utf-8") if data is not None else None
            except Exception as exc:
                logger.warning("[EvolveServer] failed to fetch Nacos skill %s: %s", name, exc)
                return None
        return fetch_skill_content(self._bucket, self._prefix, name)

    def _upload_skill(self, skill: dict, action: str) -> str:
        name = skill.get("name", "")
        if not name:
            return "skipped_missing_name"

        if self._nacos_skill_client is not None:
            from skillclaw.nacos_skill_hub import (
                _bundle_matches_remote,
                _bundle_to_nacos_zip,
                _nacos_working_version,
                _nacos_zip_to_bundle,
                _next_version,
            )

            md_content = build_skill_md(skill)
            md_bytes = md_content.encode("utf-8")
            bundle_files = {"SKILL.md": md_bytes}
            record = self._load_remote_skill_record(name) or {}
            try:
                detail = self._nacos_skill_client.get_skill(name) if record else {}
            except Exception:
                detail = {}
            working = _nacos_working_version(record, detail)
            if working:
                status, version = working
                try:
                    zip_bytes = self._nacos_skill_client.download_skill_zip(name, version=version, admin=True)
                    remote_bundle = _nacos_zip_to_bundle(zip_bytes)
                except Exception as exc:
                    logger.warning(
                        "[EvolveServer] skipping Nacos skill %s: failed to inspect %s version %s: %s",
                        name,
                        status,
                        version,
                        exc,
                    )
                    return f"skipped_existing_{status}"
                if _bundle_matches_remote(bundle_files, remote_bundle):
                    logger.info(
                        "[EvolveServer] skipped Nacos skill %s: %s version %s already matches",
                        name,
                        status,
                        version,
                    )
                    return f"skipped_existing_{status}"
                logger.info(
                    "[EvolveServer] skipped Nacos skill %s: %s version %s already exists",
                    name,
                    status,
                    version,
                )
                return f"skipped_existing_{status}"
            target_version = _next_version(record, detail)
            zip_bytes = _bundle_to_nacos_zip(name, bundle_files)
            self._nacos_skill_client.upload_skill_zip(
                zip_bytes=zip_bytes,
                filename=f"{name}-{target_version}.zip",
                overwrite=True,
                target_version=target_version,
            )
            publish_mode = str(getattr(self.config, "nacos_publish_mode", "") or "review").strip().lower()
            if publish_mode in {"review", "direct"}:
                self._nacos_skill_client.submit(name, target_version)
            publish_confirmed = False
            if publish_mode == "direct":
                publish_confirmed = self._wait_nacos_publish(name, target_version)
            logger.info(
                "[EvolveServer] uploaded skill %s to Nacos as %s via action=%s publish_mode=%s",
                name,
                target_version,
                action,
                publish_mode,
            )
            if publish_mode == "draft":
                return "uploaded_draft"
            if publish_mode == "review":
                return "uploaded_pending_review"
            if not publish_confirmed:
                return "uploaded_pending_publish"
            return "uploaded"

        skill_id = self._id_registry.get_or_create(name)
        md_content = build_skill_md(skill)
        object_key = f"{self._prefix}skills/{name}/SKILL.md"
        md_bytes = md_content.encode("utf-8")
        self._bucket.put_object(object_key, md_bytes)

        content_sha = hashlib.sha256(md_bytes).hexdigest()
        tree_sha = bundle_tree_sha256({"SKILL.md": md_bytes})
        bundle_record = {
            "format": "bundle_v1",
            "entrypoint": "SKILL.md",
            "tree_sha256": tree_sha,
            "files": [{"path": "SKILL.md", "sha256": content_sha, "size": len(md_bytes)}],
        }
        version = self._id_registry.record_update(
            name,
            content_sha,
            action=action,
            bundle_record=bundle_record,
        )
        save_version_bundle(self._bucket, self._prefix, name, version, {"SKILL.md": md_bytes})

        manifest = self._load_remote_skills()
        manifest[name] = {
            "name": name,
            "skill_id": skill_id,
            "version": version,
            "sha256": content_sha,
            "tree_sha256": tree_sha,
            "format": "bundle_v1",
            "entrypoint": "SKILL.md",
            "files": bundle_record["files"],
            "uploaded_by": "evolve_server",
            "uploaded_at": datetime.now(timezone.utc).isoformat(),
            "description": skill.get("description", ""),
            "category": skill.get("category", "general"),
        }
        save_manifest(self._bucket, self._prefix, manifest)
        logger.info(
            "[EvolveServer] uploaded skill %s (id=%s, v%d) to %s",
            name,
            skill_id,
            version,
            object_key,
        )
        return "uploaded"

    def _detect_conflict(self, name: str, incoming_skill: dict) -> bool:
        if self._nacos_skill_client is not None:
            existing_md = self._fetch_skill(name)
            if not existing_md:
                return False
            existing_sha = hashlib.sha256(existing_md.encode("utf-8")).hexdigest()
            incoming_md = build_skill_md(incoming_skill)
            incoming_sha = hashlib.sha256(incoming_md.encode("utf-8")).hexdigest()
            return existing_sha != incoming_sha
        existing_sha = self._id_registry.get_content_sha(name)
        if not existing_sha:
            return False
        incoming_md = build_skill_md(incoming_skill)
        incoming_sha = hashlib.sha256(incoming_md.encode("utf-8")).hexdigest()
        return existing_sha != incoming_sha

    async def _resolve_and_upload(self, skill: dict, action_type: str) -> tuple[str, bool]:
        name = skill.get("name", "")
        has_conflict = await self._call_storage(self._detect_conflict, name, skill)
        if not has_conflict:
            upload_status = await self._call_storage(self._upload_skill, skill, action_type)
            return self._upload_status_to_action(action_type, upload_status)

        logger.info("[EvolveServer] conflict detected for '%s' - merging", name)
        existing_md = await self._call_storage(self._fetch_skill, name)
        if not existing_md:
            upload_status = await self._call_storage(self._upload_skill, skill, action_type)
            return self._upload_status_to_action(action_type, upload_status)

        existing_skill = parse_skill_content(name, existing_md)
        existing_skill = self._overlay_manifest_metadata(
            existing_skill,
            self._load_remote_skill_record(name),
        )
        existing_skill["_version"] = self._id_registry.get_version(name)
        merged = await execute_merge(self._llm, existing_skill, skill)
        if merged and merged.get("name"):
            merged["name"] = name
            upload_status = await self._call_storage(self._upload_skill, merged, "merge")
            return self._upload_status_to_action("merge", upload_status)

        logger.warning("[EvolveServer] merge failed for '%s' - keeping incoming version", name)
        upload_status = await self._call_storage(self._upload_skill, skill, action_type)
        return self._upload_status_to_action(action_type, upload_status)

    @staticmethod
    def _upload_status_to_action(action_type: str, upload_status: str) -> tuple[str, bool]:
        if upload_status == "uploaded":
            return action_type, True
        if upload_status == "uploaded_pending_review":
            return f"{action_type}_pending_review", False
        if upload_status == "uploaded_pending_publish":
            return f"{action_type}_pending_publish", False
        if upload_status == "uploaded_draft":
            return f"{action_type}_draft", False
        return upload_status, False

    def _empty_judge_summary(self) -> dict[str, Any]:
        return {
            "enabled": bool(self.config.use_session_judge),
            "judged_sessions": 0,
            "scored_sessions": 0,
            "mean_score": None,
            "min_score": None,
            "max_score": None,
        }

    async def _run_session_judge(self, sessions: list[dict]) -> dict[str, Any]:
        summary = self._empty_judge_summary()
        if not self.config.use_session_judge or not sessions:
            return summary

        judged = await judge_sessions_parallel(self._llm, sessions)
        scores = [
            float(judge_scores["overall_score"])
            for session in sessions
            for judge_scores in [session.get("_judge_scores")]
            if isinstance(judge_scores, dict) and isinstance(judge_scores.get("overall_score"), (int, float))
        ]
        summary["judged_sessions"] = judged
        summary["scored_sessions"] = len(scores)
        if scores:
            summary["mean_score"] = round(sum(scores) / len(scores), 3)
            summary["min_score"] = round(min(scores), 3)
            summary["max_score"] = round(max(scores), 3)
        logger.info("[EvolveServer] judged %d sessions without benchmark scores", judged)
        return summary

    def _empty_skill_verifier_summary(self) -> dict[str, Any]:
        return {
            "enabled": bool(self.config.use_skill_verifier),
            "verified_skills": 0,
            "accepted": 0,
            "rejected": 0,
            "mean_score": None,
            "min_score": None,
            "max_score": None,
        }

    def _collect_skill_verifier_summary(self, records: list[dict[str, Any]]) -> dict[str, Any]:
        summary = self._empty_skill_verifier_summary()
        if not self.config.use_skill_verifier:
            return summary

        scores: list[float] = []
        for record in records:
            verification = record.get("verification")
            if not isinstance(verification, dict) or not verification.get("enabled"):
                continue
            summary["verified_skills"] += 1
            if verification.get("accepted"):
                summary["accepted"] += 1
            else:
                summary["rejected"] += 1
            score = verification.get("score")
            if isinstance(score, (int, float)) and not isinstance(score, bool):
                scores.append(float(score))

        if scores:
            summary["mean_score"] = round(sum(scores) / len(scores), 3)
            summary["min_score"] = round(min(scores), 3)
            summary["max_score"] = round(max(scores), 3)
        return summary

    def _empty_validation_publish_summary(self) -> dict[str, Any]:
        return {
            "publish_mode": self.config.publish_mode,
            "jobs_scanned": 0,
            "pending": 0,
            "published": 0,
            "rejected": 0,
            "skipped": 0,
        }

    def _build_validation_evidence(self, sessions: list[dict[str, Any]]) -> list[dict[str, Any]]:
        evidence: list[dict[str, Any]] = []
        for session in sessions[:8]:
            item: dict[str, Any] = {
                "session_id": str(session.get("session_id", "")),
                "summary": str(session.get("_summary", "")),
            }
            skills = session.get("_skills_referenced")
            if skills:
                item["skills_referenced"] = sorted(str(s or "") for s in skills if str(s or ""))
            judge_scores = session.get("_judge_scores")
            if isinstance(judge_scores, dict) and isinstance(judge_scores.get("overall_score"), (int, float)):
                item["judge_overall_score"] = float(judge_scores["overall_score"])
            if isinstance(session.get("_avg_prm"), (int, float)):
                item["avg_prm"] = float(session["_avg_prm"])
            evidence.append(item)
        return evidence

    def _build_replay_cases(self, sessions: list[dict[str, Any]]) -> list[dict[str, Any]]:
        preferred: list[dict[str, Any]] = []
        fallback: list[dict[str, Any]] = []

        for session in sessions[:6]:
            session_id = str(session.get("session_id", "") or "")
            turns = session.get("turns") or []
            if not isinstance(turns, list):
                continue
            for turn in turns:
                if not isinstance(turn, dict):
                    continue
                instruction = str(turn.get("prompt_text", "") or "").strip()
                reference_response = str(turn.get("response_text", "") or "").strip()
                if not instruction or not reference_response:
                    continue
                case = {
                    "session_id": session_id,
                    "turn_num": int(turn.get("turn_num", 0) or 0),
                    "instruction": instruction[:3000],
                    "reference_response": reference_response[:4000],
                    "had_tool_calls": bool(turn.get("tool_calls")),
                    "had_tool_results": bool(turn.get("tool_results") or turn.get("tool_observations")),
                }
                if not case["had_tool_calls"] and not case["had_tool_results"]:
                    preferred.append(case)
                else:
                    fallback.append(case)
                if len(preferred) >= 3:
                    return preferred[:3]

        if preferred:
            return preferred[:3]
        return fallback[:3]

    def _queue_validation_job(
        self,
        skill: dict[str, Any],
        action_type: str,
        sessions: list[dict[str, Any]],
        rationale: str,
        source: str,
        *,
        current_skill: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        name = str(skill.get("name", "") or "")
        skill_id = self._id_registry.get_or_create(name)
        job_id = self._validation_store.make_job_id(name)
        job = {
            "job_id": job_id,
            "status": "pending_validation",
            "created_at": datetime.now(timezone.utc).isoformat(),
            "candidate_skill_name": name,
            "candidate_skill_id": skill_id,
            "candidate_skill": skill,
            "current_skill": current_skill,
            "proposed_action": action_type,
            "source": source,
            "rationale": rationale,
            "session_ids": [session.get("session_id", "") for session in sessions],
            "session_evidence": self._build_validation_evidence(sessions),
            "replay_cases": self._build_replay_cases(sessions),
            "min_results": self.config.validation_required_results,
            "min_approvals": self.config.validation_required_approvals,
            "min_score": self.config.validation_min_mean_score,
            "max_rejections": self.config.validation_max_rejections,
        }
        self._validation_store.save_job(job)
        logger.info(
            "[EvolveServer] queued validation job %s for skill '%s' (publish_mode=validated)",
            job_id,
            name,
        )
        return {
            "action": "queued_for_validation",
            "proposed_action": action_type,
            "skill_name": name,
            "skill_id": skill_id,
            "version": None,
            "session_ids": job["session_ids"],
            "rationale": rationale,
            "source": source,
            "edit_summary": skill.get("edit_summary"),
            "uploaded": False,
            "validation_job_id": job_id,
        }

    async def _finalize_validation_jobs(self) -> tuple[list[dict[str, Any]], dict[str, Any]]:
        summary = self._empty_validation_publish_summary()
        if self.config.publish_mode != "validated":
            return [], summary

        records: list[dict[str, Any]] = []
        for job in self._validation_store.list_jobs():
            summary["jobs_scanned"] += 1
            job_id = str(job.get("job_id", "") or "")
            if not job_id:
                continue
            if self._validation_store.load_decision(job_id):
                continue

            results = self._validation_store.list_results(job_id)
            if not results:
                summary["pending"] += 1
                continue

            accepted = 0
            rejected = 0
            scores: list[float] = []
            for result in results:
                if result.get("accepted") is True:
                    accepted += 1
                else:
                    rejected += 1
                score = result.get("score")
                if isinstance(score, (int, float)) and not isinstance(score, bool):
                    scores.append(float(score))

            mean_score = round(sum(scores) / len(scores), 3) if scores else None
            publish_ready = (
                len(results) >= self.config.validation_required_results
                and accepted >= self.config.validation_required_approvals
                and mean_score is not None
                and mean_score >= self.config.validation_min_mean_score
            )
            reject_ready = rejected >= self.config.validation_max_rejections

            if publish_ready:
                candidate_skill = job.get("candidate_skill")
                if not isinstance(candidate_skill, dict) or not candidate_skill.get("name"):
                    self._validation_store.save_decision(
                        job_id,
                        {
                            "status": "rejected",
                            "reason": "candidate skill payload missing",
                            "result_count": len(results),
                            "accepted_count": accepted,
                            "rejected_count": rejected,
                            "mean_score": mean_score,
                        },
                    )
                    summary["rejected"] += 1
                    continue
                action_type = str(job.get("proposed_action", DecisionAction.CREATE) or DecisionAction.CREATE)
                actual_action, uploaded = await self._resolve_and_upload(candidate_skill, action_type)
                self._validation_store.save_decision(
                    job_id,
                    {
                        "status": "published" if uploaded else "skipped",
                        "published_action": actual_action,
                        "result_count": len(results),
                        "accepted_count": accepted,
                        "rejected_count": rejected,
                        "mean_score": mean_score,
                    },
                )
                if uploaded:
                    summary["published"] += 1
                else:
                    summary["skipped"] += 1
                records.append(
                    {
                        "action": "published_after_validation" if uploaded else actual_action,
                        "published_action": actual_action,
                        "skill_name": str(candidate_skill.get("name", "")),
                        "skill_id": self._id_registry.get_or_create(str(candidate_skill.get("name", ""))),
                        "version": self._id_registry.get_version(str(candidate_skill.get("name", ""))),
                        "session_ids": list(job.get("session_ids") or []),
                        "rationale": str(job.get("rationale", "") or ""),
                        "source": "validation_publish",
                        "uploaded": uploaded,
                        "validation_job_id": job_id,
                        "validation_results": {
                            "result_count": len(results),
                            "accepted_count": accepted,
                            "rejected_count": rejected,
                            "mean_score": mean_score,
                        },
                    }
                )
                continue

            if reject_ready:
                self._validation_store.save_decision(
                    job_id,
                    {
                        "status": "rejected",
                        "reason": "client validation rejected candidate",
                        "result_count": len(results),
                        "accepted_count": accepted,
                        "rejected_count": rejected,
                        "mean_score": mean_score,
                    },
                )
                summary["rejected"] += 1
                records.append(
                    {
                        "action": "validation_rejected",
                        "proposed_action": str(job.get("proposed_action", "")),
                        "skill_name": str(job.get("candidate_skill_name", "")),
                        "skill_id": str(job.get("candidate_skill_id", "")),
                        "version": None,
                        "session_ids": list(job.get("session_ids") or []),
                        "rationale": str(job.get("rationale", "") or ""),
                        "source": "validation_publish",
                        "uploaded": False,
                        "validation_job_id": job_id,
                        "validation_results": {
                            "result_count": len(results),
                            "accepted_count": accepted,
                            "rejected_count": rejected,
                            "mean_score": mean_score,
                        },
                    }
                )
                continue

            summary["pending"] += 1

        return records, summary

    async def _run_skill_verifier(
        self,
        skill: dict[str, Any],
        action_type: str,
        sessions: list[dict[str, Any]],
        *,
        current_skill: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        if not self.config.use_skill_verifier:
            return {"enabled": False}
        return await verify_skill_candidate(
            self._llm,
            skill,
            sessions,
            action_type,
            current_skill=current_skill,
            min_score=self.config.skill_verifier_min_score,
        )

    def _inherit_current_skill(
        self,
        evolved_skill: Optional[dict[str, Any]],
        current_skill: Optional[dict[str, Any]],
        *,
        overwrite_body: bool = False,
    ) -> None:
        if not evolved_skill or not current_skill:
            return
        if overwrite_body:
            evolved_skill["content"] = current_skill.get("content", "")
            evolved_skill["category"] = current_skill.get("category", "general")
        else:
            evolved_skill.setdefault("content", current_skill.get("content", ""))
            evolved_skill.setdefault("category", current_skill.get("category", "general"))
        evolved_skill.setdefault("extra_frontmatter", current_skill.get("extra_frontmatter") or {})

    async def _materialize_skill(
        self,
        evolved_skill: Optional[dict],
        action_type: str,
        sessions: list[dict[str, Any]],
        rationale: str,
        source: str,
        *,
        current_skill: Optional[dict[str, Any]] = None,
    ) -> Optional[dict]:
        if not evolved_skill or not evolved_skill.get("name"):
            return None

        if action_type == DecisionAction.IMPROVE and current_skill and current_skill.get("name"):
            name = current_skill["name"]
        else:
            name = self._sanitise_name(evolved_skill["name"])
        evolved_skill["name"] = name

        verification = await self._run_skill_verifier(
            evolved_skill,
            action_type,
            sessions,
            current_skill=current_skill,
        )
        session_ids = [session.get("session_id", "") for session in sessions]
        if verification.get("enabled") and not verification.get("accepted"):
            logger.info(
                "[EvolveServer] verifier rejected skill '%s': %s",
                name,
                verification.get("reason", "no reason provided"),
            )
            return {
                "action": "verification_rejected",
                "proposed_action": action_type,
                "skill_name": name,
                "skill_id": None,
                "version": None,
                "session_ids": session_ids,
                "rationale": rationale,
                "source": source,
                "edit_summary": evolved_skill.get("edit_summary"),
                "uploaded": False,
                "verification": verification,
            }

        skill_id = self._id_registry.get_or_create(name)
        evolved_skill["skill_id"] = skill_id
        if self.config.publish_mode == "validated":
            record = self._queue_validation_job(
                evolved_skill,
                action_type,
                sessions,
                rationale,
                source,
                current_skill=current_skill,
            )
            record["verification"] = verification
            return record

        actual_action, uploaded = await self._resolve_and_upload(evolved_skill, action_type)
        logger.info(
            "[EvolveServer] %s skill '%s' (id=%s, v%d)",
            actual_action,
            name,
            skill_id,
            self._id_registry.get_version(name),
        )
        return {
            "action": actual_action,
            "skill_name": name,
            "skill_id": skill_id,
            "version": self._id_registry.get_version(name),
            "session_ids": session_ids,
            "rationale": rationale,
            "source": source,
            "edit_summary": evolved_skill.get("edit_summary"),
            "uploaded": uploaded,
            "verification": verification,
        }

    async def _evolve_skill_group(
        self,
        skill_name: str,
        sessions: list[dict],
        existing_skill_names: list[str],
    ) -> Optional[dict]:
        current_md = await self._call_storage(self._fetch_skill, skill_name)
        current_skill = parse_skill_content(skill_name, current_md) if current_md else None
        current_skill = self._overlay_manifest_metadata(
            current_skill,
            await self._call_storage(self._load_remote_skill_record, skill_name),
        )

        result = await evolve_skill_from_sessions(
            self._llm,
            skill_name,
            sessions,
            current_skill,
            existing_skill_names,
        )
        if not result or result.get("action") == DecisionAction.SKIP:
            logger.info("[EvolveServer] skill '%s': LLM decided to skip", skill_name)
            return None

        action_type = result.get("action", DecisionAction.IMPROVE)
        evolved_skill = result.get("skill")
        if not evolved_skill:
            return None

        if action_type == DecisionAction.OPTIMIZE_DESC and current_skill:
            self._inherit_current_skill(evolved_skill, current_skill, overwrite_body=True)
        elif current_skill:
            self._inherit_current_skill(evolved_skill, current_skill)

        return await self._materialize_skill(
            evolved_skill,
            action_type,
            sessions,
            result.get("rationale", ""),
            "skill_group",
            current_skill=current_skill,
        )

    async def _handle_no_skill_sessions(
        self,
        sessions: list[dict],
        existing_skill_names: list[str],
    ) -> list[dict]:
        result = await create_skill_from_sessions(self._llm, sessions, existing_skill_names)
        if not result or result.get("action") == DecisionAction.SKIP:
            logger.info("[EvolveServer] no-skill sessions: LLM decided to skip")
            return []

        evolved_skill = result.get("skill")
        if not evolved_skill:
            return []

        record = await self._materialize_skill(
            evolved_skill,
            DecisionAction.CREATE,
            sessions,
            result.get("rationale", ""),
            "no_skill",
        )
        return [record] if record else []

    async def run_once(self) -> dict:
        logger.info("[EvolveServer] === starting evolution cycle ===")
        started_at = time.monotonic()

        sessions, session_keys = await self._drain_sessions()
        judge_summary = self._empty_judge_summary()
        skill_group_count = 0
        no_skill_sessions: list[dict] = []
        evolution_records: list[dict] = []
        had_processing_error = False

        if sessions:
            logger.info("[EvolveServer] summarizing %d sessions", len(sessions))
            await summarize_sessions_parallel(self._llm, sessions)
            judge_summary = await self._run_session_judge(sessions)

            grouped_sessions = aggregate_sessions_by_skill(sessions)
            no_skill_sessions = grouped_sessions.pop(NO_SKILL_KEY, [])
            skill_group_count = len(grouped_sessions)

            manifest = await self._call_storage(self._load_remote_skills)
            existing_skill_names = [item.get("name", "") for item in manifest.values()]

            if grouped_sessions:
                logger.info("[EvolveServer] evolving %d skill group(s)", skill_group_count)
            for skill_name, skill_sessions in grouped_sessions.items():
                try:
                    record = await self._evolve_skill_group(skill_name, skill_sessions, existing_skill_names)
                except Exception as exc:
                    logger.error("[EvolveServer] skill '%s' evolve failed: %s", skill_name, exc)
                    had_processing_error = True
                    continue
                if record:
                    evolution_records.append(record)

            if no_skill_sessions:
                logger.info("[EvolveServer] processing %d no-skill sessions", len(no_skill_sessions))
                try:
                    evolution_records.extend(
                        await self._handle_no_skill_sessions(no_skill_sessions, existing_skill_names)
                    )
                except Exception as exc:
                    logger.error("[EvolveServer] no-skill evolve failed: %s", exc)
                    had_processing_error = True
        else:
            logger.info("[EvolveServer] queue empty - checking pending validation publish jobs")

        published_records, validation_publish_summary = await self._finalize_validation_jobs()
        all_records = evolution_records + published_records

        if not self._uses_nacos_skill_registry():
            await self._call_storage(self._id_registry.save_to_oss, self._bucket, self._prefix)
        if session_keys and not had_processing_error:
            await self._call_storage(delete_session_keys, self._bucket, session_keys)
        elif session_keys and had_processing_error:
            logger.warning(
                "[EvolveServer] retaining %d session(s) in queue because this cycle had processing errors",
                len(session_keys),
            )

        elapsed = round(time.monotonic() - started_at, 1)
        uploaded_skills = sum(1 for record in all_records if record.get("uploaded"))
        queued_candidates = sum(1 for record in all_records if record.get("action") == "queued_for_validation")
        published_after_validation = sum(
            1 for record in all_records if record.get("action") == "published_after_validation"
        )
        skill_verifier_summary = self._collect_skill_verifier_summary(all_records)
        summary = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "elapsed_seconds": elapsed,
            "sessions": len(sessions),
            "skill_groups": skill_group_count,
            "no_skill_sessions": len(no_skill_sessions),
            "actions": len(all_records),
            "skills_evolved": uploaded_skills,
            "uploaded_skills": uploaded_skills,
            "candidates_queued": queued_candidates,
            "published_after_validation": published_after_validation,
            "evolutions": all_records,
            "session_judge": judge_summary,
            "skill_verifier": skill_verifier_summary,
            "validation_publish": validation_publish_summary,
            "had_processing_error": had_processing_error,
        }
        self._append_history(summary)
        logger.info(
            "[EvolveServer] === cycle done: %d sessions, %d skill groups, %d uploaded, %d queued in %.1fs ===",
            len(sessions),
            skill_group_count,
            uploaded_skills,
            queued_candidates,
            elapsed,
        )
        if uploaded_skills > 0:
            await self._notify_proxy_reload()
        return summary

    async def _notify_proxy_reload(self) -> None:
        mode = str(getattr(self.config, "skill_reload_mode", "") or "poll").strip().lower()
        url = str(getattr(self.config, "proxy_reload_url", "") or "").strip().rstrip("/")
        if mode != "callback" or not url:
            return
        headers: dict[str, str] = {}
        api_key = str(getattr(self.config, "proxy_reload_api_key", "") or "")
        if api_key:
            headers["Authorization"] = f"Bearer {api_key}"
        import httpx

        for attempt in range(3):
            try:
                async with httpx.AsyncClient(timeout=5.0) as client:
                    resp = await client.post(f"{url}/internal/reload-skills", headers=headers)
                    resp.raise_for_status()
                logger.info("[EvolveServer] notified proxy to reload skills: %s", url)
                return
            except Exception as exc:
                if attempt < 2:
                    await asyncio.sleep(1.0 * (attempt + 1))
                else:
                    logger.warning("[EvolveServer] proxy reload notify failed after 3 attempts: %s", exc)

    async def run_periodic(self) -> None:
        self._running = True
        logger.info("[EvolveServer] periodic mode: interval=%ds", self.config.interval_seconds)
        while self._running:
            try:
                await self.run_once()
            except Exception as exc:
                logger.error("[EvolveServer] cycle error: %s", exc, exc_info=True)
            await asyncio.sleep(self.config.interval_seconds)

    def stop(self) -> None:
        self._running = False

    def create_http_app(self):
        from fastapi import FastAPI
        from fastapi.responses import JSONResponse

        app = FastAPI(title="SkillClaw Evolve Server")

        @app.post("/trigger")
        async def trigger_evolve():
            return JSONResponse(content=await self.run_once())

        @app.get("/status")
        async def status():
            entries = (
                self._load_remote_skills() if self._uses_nacos_skill_registry() else self._id_registry.all_entries()
            )
            pending_keys = await self._call_storage(list_session_keys, self._bucket, self._prefix)
            return JSONResponse(
                content={
                    "running": self._running,
                    "pending_sessions": len(pending_keys),
                    "registered_skills": len(entries),
                    "skills": {
                        name: {
                            "skill_id": item.get("skill_id") or item.get("name") or name,
                            "version": item.get("version") or (item.get("labels") or {}).get("latest") or 0,
                        }
                        for name, item in entries.items()
                    },
                }
            )

        @app.get("/health")
        async def health():
            return {"status": "ok"}

        return app