"""Self-evolution executor.

Runs an isolated review agent over an idle conversation's transcript and, if a
clear signal is found, lets it edit memory / skills via a restricted toolset.
Conservative by design: most runs return ``[SILENT]`` and change nothing.

Flow:
    1. Build a transcript from the session's new (since last pass) messages.
    2. Snapshot MEMORY.md + daily file + editable skills (for undo) -> backup_id.
    3. Run an isolated agent (same model, restricted tools, evolution prompt).
    4. If output is [SILENT], or no workspace file actually changed -> done.
    5. Otherwise -> record to the evolution log, inject an [EVOLUTION] note into
       the user session (so the main agent can honor "undo"), and push the
       summary to the user's channel.

Reuses existing infrastructure (AgentBridge.create_agent, ToolManager,
remember_scheduled_output, channel_factory) rather than introducing a fork.
"""

from __future__ import annotations

import re
import threading
from datetime import datetime
from pathlib import Path
from typing import List, Optional

from common.log import logger

from agent.evolution.backup import create_backup
from agent.evolution.config import get_evolution_config
from agent.evolution.prompts import (
    EVOLUTION_MARKER,
    EVOLUTION_SYSTEM_PROMPT,
    SILENT_TOKEN,
    build_review_user_message,
)
from agent.evolution.record import append_session_evolution

# Tools the isolated evolution agent is allowed to use. Everything else is
# withheld so a review pass can only read context, run workspace scripts, and
# edit memory/skill files. bash is needed by skill-creator's init script and is
# confined to the workspace by _BashWorkspaceGuard.
_ALLOWED_TOOLS = {"read", "write", "edit", "ls", "bash", "memory_search", "memory_get"}

# Cap concurrent evolution passes so a burst of idle sessions can't spawn many
# background model runs at once. Extra sessions simply wait for the next scan.
_MAX_CONCURRENT = 2
_running_lock = threading.Lock()
_running_count = 0


def _builtin_skill_names() -> set:
    """Names of skills shipped with the product (project-root ``skills/``).

    These are protected: the evolution agent must never edit them, even though
    a same-named copy exists in the workspace at runtime. The project dir is the
    authoritative list of what counts as built-in.
    """
    try:
        # executor.py -> agent/evolution -> agent -> project root
        project_root = Path(__file__).resolve().parents[2]
        builtin_dir = project_root / "skills"
        if not builtin_dir.is_dir():
            return set()
        names = set()
        for entry in builtin_dir.iterdir():
            if entry.is_dir() and not entry.name.startswith("."):
                names.add(entry.name)
        return names
    except Exception:
        return set()


def _build_transcript(messages: List[dict], max_chars: int = 12000) -> str:
    """Render the session messages into a compact text transcript."""
    lines: List[str] = []
    for msg in messages:
        role = msg.get("role", "")
        if role not in ("user", "assistant"):
            continue
        content = msg.get("content", "")
        text = _extract_text(content)
        if not text.strip():
            continue
        speaker = "User" if role == "user" else "Assistant"
        lines.append(f"{speaker}: {text.strip()}")
    transcript = "\n".join(lines)
    # Keep the most RECENT context if oversized (tail is most relevant).
    if len(transcript) > max_chars:
        transcript = "...(earlier omitted)...\n" + transcript[-max_chars:]
    return transcript


def _extract_text(content) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for block in content:
            if isinstance(block, dict) and block.get("type") == "text":
                parts.append(block.get("text", ""))
            elif isinstance(block, str):
                parts.append(block)
        return "\n".join(parts)
    return ""


def _select_tools(all_tools: list) -> list:
    return [t for t in all_tools if getattr(t, "name", None) in _ALLOWED_TOOLS]


# Tools whose writes must be confined to the workspace during evolution.
_WRITE_TOOLS = {"write", "edit"}


class _WorkspaceWriteGuard:
    """Wraps a write/edit tool so it can ONLY write inside the workspace.

    Hard engineering guard (not prompt-based): any write resolving outside the
    workspace — e.g. the project's bundled ``skills/`` dir — is rejected. This
    protects built-in skills regardless of what the model attempts.
    """

    def __init__(self, inner, workspace_dir: str):
        self._inner = inner
        self._ws = Path(workspace_dir).resolve()
        # Mirror the attributes the agent runtime reads off a tool.
        self.name = inner.name
        self.description = inner.description
        self.params = inner.params

    def __getattr__(self, item):
        return getattr(self._inner, item)

    def execute_tool(self, params):
        # The agent runtime calls execute_tool (not execute); route it through
        # our guarded execute so the path checks always run.
        try:
            return self.execute(params)
        except Exception as e:
            logger.error(f"[Evolution] guarded tool error: {e}")
            from agent.tools.base_tool import ToolResult
            return ToolResult.fail(f"Error: {e}")

    def execute(self, args):
        path = (args.get("path") or "").strip()
        if path:
            try:
                resolved = Path(self._inner._resolve_path(path)).resolve()
                from agent.tools.base_tool import ToolResult
                # Confine writes to the workspace. This protects the product's
                # bundled skills (which live outside the workspace) from ever
                # being modified, no matter what path the model attempts.
                if self._ws not in resolved.parents and resolved != self._ws:
                    return ToolResult.fail(
                        "Error: evolution may only write inside the workspace; "
                        f"path '{path}' is outside and was blocked."
                    )
            except Exception:
                pass
        return self._inner.execute(args)


class _BashWorkspaceGuard:
    """Wraps the bash tool so evolution can only run commands inside the
    workspace.

    Evolution needs bash for skill-creator's init script, but it runs
    unattended in the background, so a raw shell is too broad. This guard:
      - forces the command to execute with cwd = workspace,
      - rejects commands that reference an absolute path or ``..`` segment
        pointing OUTSIDE the workspace (the common ways to escape it).
    It is a coarse textual check, not a sandbox — paired with the model's
    instruction to only run skill-creator scripts, it keeps writes local.
    """

    def __init__(self, inner, workspace_dir: str):
        self._inner = inner
        self._ws = Path(workspace_dir).resolve()
        # Pin the shell's working directory to the workspace.
        try:
            self._inner.cwd = str(self._ws)
        except Exception:
            pass
        self.name = inner.name
        self.description = inner.description
        self.params = inner.params

    def __getattr__(self, item):
        return getattr(self._inner, item)

    def execute_tool(self, params):
        try:
            return self.execute(params)
        except Exception as e:
            logger.error(f"[Evolution] guarded bash error: {e}")
            from agent.tools.base_tool import ToolResult
            return ToolResult.fail(f"Error: {e}")

    def _escapes_workspace(self, command: str) -> bool:
        # Absolute paths that are not under the workspace.
        for tok in re.findall(r'(?:^|\s)(/[^\s\'";|&]+)', command):
            try:
                resolved = Path(tok).resolve()
            except Exception:
                continue
            if self._ws != resolved and self._ws not in resolved.parents:
                return True
        # Parent-dir traversal that climbs above the workspace.
        for tok in re.findall(r'[^\s\'";|&]*\.\.[^\s\'";|&]*', command):
            try:
                resolved = (self._ws / tok).resolve()
            except Exception:
                continue
            if self._ws != resolved and self._ws not in resolved.parents:
                return True
        return False

    def execute(self, args):
        from agent.tools.base_tool import ToolResult
        command = (args.get("command") or "").strip()
        if command and self._escapes_workspace(command):
            return ToolResult.fail(
                "Error: evolution may only run commands inside the workspace; "
                "this command references a path outside it and was blocked."
            )
        return self._inner.execute(args)


def _guard_tools(tools: list, workspace_dir: str) -> list:
    """Wrap write/edit/bash tools with workspace guards; leave others as-is."""
    guarded = []
    for t in tools:
        name = getattr(t, "name", None)
        if name in _WRITE_TOOLS:
            guarded.append(_WorkspaceWriteGuard(t, workspace_dir))
        elif name == "bash":
            guarded.append(_BashWorkspaceGuard(t, workspace_dir))
        else:
            guarded.append(t)
    return guarded


# Workspace subtrees worth watching for evolution-induced changes. AGENT.md is
# watched too: evolution may rarely refine the assistant's persona/style there.
_WATCH_SUBDIRS = ("MEMORY.md", "AGENT.md", "skills", "knowledge", "output")
# Subpaths under memory/ to ignore: evolution's own bookkeeping + the nightly
# dream diary, none of which count as a user-facing change signal.
_MEMORY_IGNORE = (".evolution_backups", "dreams", "evolution")
# Files the skill subsystem maintains automatically (the enable/disable index).
# Not an evolution result, so a rewrite must not count as a change signal.
_WATCH_IGNORE_NAMES = ("skills_config.json",)


def _workspace_snapshot(workspace_dir) -> dict:
    """Map relative path -> (mtime, size) for watched files. Cheap, no reads."""
    ws = Path(workspace_dir)
    snap: dict = {}
    for name in _WATCH_SUBDIRS:
        root = ws / name
        if root.is_file():
            try:
                st = root.stat()
                snap[name] = (st.st_mtime, st.st_size)
            except OSError:
                pass
            continue
        if not root.is_dir():
            continue
        for p in root.rglob("*"):
            if not p.is_file():
                continue
            if p.name in _WATCH_IGNORE_NAMES:
                continue
            try:
                st = p.stat()
                snap[str(p.relative_to(ws))] = (st.st_mtime, st.st_size)
            except OSError:
                pass

    # Watch the daily memory files (memory/*.md and per-user dailies) since
    # evolution now records learnings there. Skip backups/dreams bookkeeping.
    mem_dir = ws / "memory"
    if mem_dir.is_dir():
        for p in mem_dir.rglob("*.md"):
            rel_parts = p.relative_to(mem_dir).parts
            if rel_parts and rel_parts[0] in _MEMORY_IGNORE:
                continue
            try:
                st = p.stat()
                snap[str(p.relative_to(ws))] = (st.st_mtime, st.st_size)
            except OSError:
                pass
    return snap


def _workspace_changed(workspace_dir, pre: dict) -> bool:
    """True if any watched file was added, removed, or modified since ``pre``."""
    return _workspace_snapshot(workspace_dir) != pre


def run_evolution_for_session(
    agent_bridge,
    session_id: str,
    channel_type: str = "",
    receiver: str = "",
    user_id: Optional[str] = None,
    idle_minutes: float = 0.0,
) -> bool:
    """Run one evolution pass for a session. Returns True if it changed anything.

    Safe to call from a background thread. All failures are swallowed and
    logged — evolution must never disrupt the main pipeline.
    """
    cfg = get_evolution_config()
    if not cfg.enabled:
        return False

    # Concurrency gate: bound how many evolution passes run at once.
    global _running_count
    with _running_lock:
        if _running_count >= _MAX_CONCURRENT:
            logger.info(
                f"[Evolution] busy ({_running_count}/{_MAX_CONCURRENT} running); "
                f"skipping session={session_id} this scan"
            )
            return False
        _running_count += 1

    try:
        agent = agent_bridge.agents.get(session_id) or agent_bridge.default_agent
        if not agent:
            return False

        with agent.messages_lock:
            all_messages = list(agent.messages)
        total_msgs = len(all_messages)
        # In-memory evolution cursor: only review messages added since the last
        # pass so a long session doesn't re-judge (and re-write) old content.
        # Stored on the agent instance; lost on restart (acceptable — at worst
        # one redundant pass right after a restart, gated by the file-change
        # check downstream so it won't double-write identical memory).
        done = int(getattr(agent, "_evo_done_msg_count", 0))
        if done > total_msgs:
            done = 0  # history was trimmed/reset; start fresh
        new_messages = all_messages[done:]
        transcript = _build_transcript(new_messages)
        if not transcript.strip():
            # Routine no-op: the per-minute scan hits every idle session. Advance
            # the cursor so we don't re-scan the same tail; no log (pure noise).
            agent._evo_done_msg_count = total_msgs
            return False

        logger.info(
            f"[Evolution] ▶ Reviewing session={session_id} "
            f"(idle {idle_minutes:.1f}min, {len(new_messages)} new/{total_msgs} msgs, "
            f"~{len(transcript)} chars)"
        )

        # Resolve workspace + files to snapshot for undo.
        from agent.memory.config import get_default_memory_config
        mem_cfg = get_default_memory_config()
        workspace_dir = mem_cfg.get_workspace()
        if user_id:
            memory_file = Path(workspace_dir) / "memory" / "users" / user_id / "MEMORY.md"
        else:
            memory_file = Path(workspace_dir) / "MEMORY.md"
        skills_dir = mem_cfg.get_skills_dir()

        # Snapshot MEMORY.md + every NON-protected skill's SKILL.md. Protected
        # built-in skills are excluded from backup because they must never be
        # edited in the first place.
        protected_names = _builtin_skill_names()
        # Back up both MEMORY.md and today's daily file: evolution now writes to
        # the daily file, but MEMORY.md is cheap to snapshot and keeps undo safe
        # if the model ever edits it.
        today_daily = Path(workspace_dir) / "memory" / (
            datetime.now().strftime("%Y-%m-%d") + ".md"
        )
        if user_id:
            today_daily = Path(workspace_dir) / "memory" / "users" / user_id / (
                datetime.now().strftime("%Y-%m-%d") + ".md"
            )
        # AGENT.md (persona) is backed up too so a rare persona edit is undoable.
        # Persona is workspace-global (not per-user): it always lives at the
        # workspace root, regardless of user_id.
        agent_file = Path(workspace_dir) / "AGENT.md"
        backup_files = [Path(memory_file), today_daily, agent_file]
        if skills_dir.exists():
            for skill_md in skills_dir.rglob("SKILL.md"):
                # The skill dir is the SKILL.md's parent (or an ancestor for
                # collections); guard by checking the immediate top-level dir.
                try:
                    top = skill_md.relative_to(skills_dir).parts[0]
                except (ValueError, IndexError):
                    continue
                if top in protected_names:
                    continue
                backup_files.append(skill_md)
        backup_id = create_backup(workspace_dir, backup_files)
        _backup_n = sum(1 for f in backup_files if Path(f).exists())

        # Snapshot the whole workspace (path -> mtime/size) so we can reliably
        # detect ANY file change — including new output files written when
        # finishing an unfinished task, which are not in backup_files.
        pre_snapshot = _workspace_snapshot(workspace_dir)

        # Build the isolated review agent: same model, restricted tools, with a
        # hard guard that confines all writes to the workspace (protects the
        # project's bundled skills from ever being modified).
        review_tools = _guard_tools(
            _select_tools(list(getattr(agent, "tools", []) or [])),
            str(workspace_dir),
        )
        review_agent = agent_bridge.create_agent(
            system_prompt="",
            tools=review_tools,
            description="Self-evolution review agent",
            max_steps=cfg.max_steps,
            workspace_dir=str(workspace_dir),
            skill_manager=getattr(agent, "skill_manager", None),
            memory_manager=getattr(agent, "memory_manager", None),
            enable_skills=True,
            runtime_info=getattr(agent, "runtime_info", None),
        )
        # Mark this as a restricted review agent so runtime MCP reconciliation
        # (ToolManager.sync_mcp_into_agent) will NOT silently re-inject MCP tools
        # that _select_tools()/_guard_tools() intentionally withheld. Without this
        # flag the review boundary would be re-opened on the first LLM turn.
        review_agent._evolution_restricted = True
        # Reuse the live model so it follows the user's configured model.
        review_agent.model = agent.model
        # Inject the evolution task brief AFTER the full system prompt: the agent
        # gets the full context (tools, workspace, user preferences, memory, time)
        # AND its evolution-specific instructions on top, instead of one
        # overwriting the other.
        review_agent.extra_system_suffix = EVOLUTION_SYSTEM_PROMPT

        logger.info(
            f"[Evolution] backup {backup_id} ({_backup_n} files) → running review agent"
        )
        user_msg = build_review_user_message(transcript, protected_skills=list(protected_names))
        result = review_agent.run_stream(user_msg, clear_history=True)
        result = (result or "").strip()

        # These messages are now reviewed; advance the cursor so the next pass
        # only looks at messages added after this point (silent or not).
        agent._evo_done_msg_count = total_msgs

        # Respect an explicit silent verdict: empty, exactly [SILENT], or text
        # that STARTS with [SILENT] means the model chose to stay quiet.
        if not result or result.startswith(SILENT_TOKEN):
            logger.info(f"[Evolution] ✗ No change for session={session_id} ([SILENT])")
            return False

        # Anti-nag backstop: if the model wrote a summary but actually changed no
        # watched file, stay silent — never notify about work that didn't happen.
        if not _workspace_changed(workspace_dir, pre_snapshot):
            logger.info(
                f"[Evolution] ✗ session={session_id}: text produced but no file "
                f"changed — staying silent"
            )
            return False

        # The model produced a real summary. Strip any stray [SILENT] tokens it
        # left mid-text, then notify.
        result = result.replace(SILENT_TOKEN, "").strip()
        if not result:
            logger.info(f"[Evolution] ✗ No change for session={session_id} ([SILENT])")
            return False

        logger.info(f"[Evolution] ✓ session={session_id} evolved:\n{result}")
        append_session_evolution(workspace_dir, result, backup_id=backup_id, user_id=user_id)
        # Inject an [EVOLUTION] note so the main agent can honor "undo".
        _inject_evolution_record(agent_bridge, session_id, channel_type, result, backup_id)
        # The injection appended its own messages ([SCHEDULED]/[EVOLUTION]).
        # Advance the cursor past them so the next scan does not treat
        # evolution's own bookkeeping as new user content and re-trigger.
        try:
            with agent.messages_lock:
                agent._evo_done_msg_count = len(agent.messages)
        except Exception:
            pass

        # Push the summary to the user's channel. The "did a file actually
        # change" gate above is the only throttle we need: real evolutions are
        # rare, so no extra opt-in switch or daily-count limit is required.
        if channel_type and receiver:
            _notify_user(channel_type, receiver, result)

        return True

    except Exception as e:
        logger.warning(f"[Evolution] Run failed for session={session_id}: {e}")
        return False
    finally:
        with _running_lock:
            _running_count -= 1


def _inject_evolution_record(
    agent_bridge, session_id: str, channel_type: str, summary: str, backup_id: Optional[str]
) -> None:
    """Add an [EVOLUTION] note to the user session so the main agent can undo."""
    try:
        note = f"{EVOLUTION_MARKER} {summary}"
        if backup_id:
            note += f"\n(backup_id: {backup_id}; to undo, restore this backup)"
        # Reuse the scheduler-output injection path: isolated execution, only a
        # compact record lands in the user session.
        agent_bridge.remember_scheduled_output(
            session_id=session_id,
            content=note,
            channel_type=channel_type,
            task_description="self-evolution",
        )
    except Exception as e:
        logger.debug(f"[Evolution] Failed to inject evolution record: {e}")


def _notify_user(channel_type: str, receiver: str, summary: str) -> None:
    """Push the evolution summary to the user's channel as a new message."""
    try:
        from bridge.context import Context, ContextType
        from bridge.reply import Reply, ReplyType
        from channel.channel_factory import create_channel

        context = Context(ContextType.TEXT, summary)
        context["receiver"] = receiver
        context["isgroup"] = False
        context["session_id"] = receiver
        # Channels that reply to an original message need msg=None for a fresh push.
        if channel_type in ("feishu", "dingtalk", "wecom_bot", "qq"):
            context["msg"] = None
        if channel_type == "feishu":
            context["receive_id_type"] = "open_id"

        channel = create_channel(channel_type)
        if not channel:
            return

        # Web is request-response: a background push needs a synthetic request_id
        # plus a request->session mapping so the channel can route the message to
        # the user's polling queue (same approach the scheduler uses).
        if channel_type == "web":
            import uuid
            request_id = f"evolution_{uuid.uuid4().hex[:8]}"
            context["request_id"] = request_id
            if hasattr(channel, "request_to_session"):
                channel.request_to_session[request_id] = receiver

        channel.send(Reply(ReplyType.TEXT, summary), context)
        logger.info(f"[Evolution] Notified user via {channel_type}")
    except Exception as e:
        logger.warning(f"[Evolution] Failed to notify user: {e}")
