Function bodies 139 total

def get_claude_session_id(session_directory: Path) -> str | None:
    """Read stored Claude Code session ID."""
    path = session_directory / CLAUDE_SESSION_ID_FILE
    if path.exists():
        return path.read_text().strip()
    return None

def load_conversation_history(
    session_directory: Path, max_turns: int = MAX_CONVERSATION_HISTORY_TURNS
) -> str | None:
    """Read last N turns from conversation files.

    Searches conversation-YYMMDD.md files from newest to oldest,
    collecting turns until max_turns is reached.
    Falls back to legacy conversation.md if no dated files exist.

    Returns None if no conversation files exist or all are empty.
    """
    # Dated files in reverse chronological order (newest first)
    conversation_files = sorted(
        session_directory.glob(CONVERSATION_GLOB_PATTERN),
        reverse=True,
    )

    # Legacy fallback
    if not conversation_files:
        legacy_file = session_directory / "conversation.md"
        if legacy_file.exists():
            conversation_files = [legacy_file]

    if not conversation_files:
        return None

    all_sections: list[str] = []

    for conversation_file in conversation_files:
        content = conversation_file.read_text()
       

def log_conversation(session_directory: Path, role: str, content: str) -> None:
    """Append a conversation entry to today's conversation file.

    Writes to conversation-YYMMDD.md (UTC date).

    Args:
        session_directory: Path to the session directory.
        role: Either 'user' or 'assistant'.
        content: The message content.
    """
    conversation_file = _conversation_file_for_today(session_directory)
    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")

    entry = f"\n## {role} ({timestamp})\n\n{content}\n"

    with open(conversation_file, "a") as file:
        file.write(entry)

def list_workspace_files(session_directory: Path) -> list[str]:
    """List all files in the session's workspace directory.

    Returns a list of relative file paths within the workspace.
    """
    workspace = session_directory / "workspace"
    if not workspace.exists():
        return []

    files = []
    for file_path in sorted(workspace.rglob("*")):
        if file_path.is_file():
            files.append(str(file_path.relative_to(workspace)))
    return files

def load_bot_memory(bot_path: Path) -> str | None:
    """Load the bot's MEMORY.md content.

    Returns None if MEMORY.md doesn't exist or is empty.
    """
    path = memory_file_path(bot_path)
    if not path.exists():
        return None
    content = path.read_text()
    if not content.strip():
        return None
    return content

def load_global_memory() -> str | None:
    """Load the global GLOBAL_MEMORY.md content.

    Returns None if GLOBAL_MEMORY.md doesn't exist or is empty.
    """
    path = global_memory_file_path()
    if not path.exists():
        return None
    content = path.read_text()
    if not content.strip():
        return None
    return content

def save_global_memory(content: str) -> None:
    """Save content to the global GLOBAL_MEMORY.md file."""
    from cclaw.config import ensure_home

    ensure_home()
    global_memory_file_path().write_text(content)

def list_skills() -> list[dict[str, Any]]:
    """List all recognized skills (directories containing SKILL.md)."""
    directory = skills_directory()
    if not directory.exists():
        return []

    from cclaw.builtin_skills import get_builtin_skill_path

    result = []
    for entry in sorted(directory.iterdir()):
        if entry.is_dir() and (entry / "SKILL.md").exists():
            config = load_skill_config(entry.name)
            emoji = config.get("emoji", "") if config else ""

            # Fallback: read emoji from builtin template if not in installed config
            if not emoji:
                builtin_path = get_builtin_skill_path(entry.name)
                if builtin_path:
                    builtin_yaml_path = builtin_path / "skill.yaml"
                    if builtin_yaml_path.exists():
                        with open(builtin_yaml_path) as builtin_file:
                            builtin_config = yaml.safe_load(builtin_file) or {}
                        

def load_skill_config(name: str) -> dict[str, Any] | None:
    """Load a skill's skill.yaml. Returns None if it doesn't exist."""
    path = skill_directory(name) / "skill.yaml"
    if not path.exists():
        return None
    with open(path) as file:
        return yaml.safe_load(file) or {}

def load_skill_markdown(name: str) -> str | None:
    """Load a skill's SKILL.md content. Returns None if it doesn't exist."""
    path = skill_directory(name) / "SKILL.md"
    if not path.exists():
        return None
    return path.read_text()

def skill_status(name: str) -> str:
    """Return the status of a skill: active, inactive, or not_found."""
    if not is_skill(name):
        return "not_found"

    config = load_skill_config(name)
    if config is None:
        # Markdown-only skill (no skill.yaml) is always active
        return "active"

    return config.get("status", "inactive")

def skill_type(name: str) -> str | None:
    """Return the skill type. None means markdown-only (no tools)."""
    config = load_skill_config(name)
    if config is None:
        return None
    return config.get("type")

def generate_skill_markdown(name: str, description: str = "") -> str:
    """Generate default SKILL.md content."""
    return f"""# {name}

{description}

## Instructions

(Describe what this skill does and how the bot should use it.)
"""

def default_skill_yaml(
    name: str,
    description: str = "",
    skill_type: str | None = None,
    required_commands: list[str] | None = None,
    environment_variables: list[str] | None = None,
) -> dict[str, Any]:
    """Return a default skill.yaml structure."""
    config: dict[str, Any] = {
        "name": name,
        "description": description,
        "status": "inactive",
    }
    if skill_type:
        config["type"] = skill_type
    if required_commands:
        config["required_commands"] = required_commands
    if environment_variables:
        config["environment_variables"] = environment_variables
    return config

def remove_skill(name: str) -> bool:
    """Remove a skill directory entirely.

    Also detaches it from all bots that use it.
    Returns True if the skill was found and removed.
    """
    directory = skill_directory(name)
    if not directory.exists():
        return False

    # Detach from all bots using this skill
    for bot_name in bots_using_skill(name):
        detach_skill_from_bot(bot_name, name)

    shutil.rmtree(directory)
    return True

def check_skill_requirements(name: str) -> list[str]:
    """Check if skill requirements are met.

    Returns a list of error messages. Empty list means all OK.
    """
    errors: list[str] = []
    config = load_skill_config(name)
    if config is None:
        return errors  # Markdown-only, no requirements

    install_hints = config.get("install_hints", {})
    for command in config.get("required_commands", []):
        if not shutil.which(command):
            hint = install_hints.get(command)
            if hint:
                errors.append(f"Required command not found: {command}\n    Install: {hint}")
            else:
                errors.append(f"Required command not found: {command}")

    for variable in config.get("environment_variables", []):
        # Only check if the variable key is defined, value can be set during setup
        pass

    return errors

def activate_skill(name: str) -> None:
    """Set a skill's status to active."""
    config = load_skill_config(name)
    if config is None:
        return
    config["status"] = "active"
    save_skill_config(name, config)

def get_bot_skills(bot_name: str) -> list[str]:
    """Get the list of skill names attached to a bot."""
    bot_config = load_bot_config(bot_name)
    if not bot_config:
        return []
    return bot_config.get("skills", [])

def attach_skill_to_bot(bot_name: str, skill_name: str) -> None:
    """Attach a skill to a bot and regenerate CLAUDE.md."""
    bot_config = load_bot_config(bot_name)
    if not bot_config:
        return

    skills = bot_config.get("skills", [])
    if skill_name not in skills:
        skills.append(skill_name)
        bot_config["skills"] = skills
        save_bot_config(bot_name, bot_config)

def detach_skill_from_bot(bot_name: str, skill_name: str) -> None:
    """Detach a skill from a bot and regenerate CLAUDE.md."""
    bot_config = load_bot_config(bot_name)
    if not bot_config:
        return

    skills = bot_config.get("skills", [])
    if skill_name in skills:
        skills.remove(skill_name)
        bot_config["skills"] = skills
        save_bot_config(bot_name, bot_config)

def bots_using_skill(skill_name: str) -> list[str]:
    """Return a list of bot names that have this skill attached."""
    from cclaw.config import load_config

    config = load_config()
    if not config or not config.get("bots"):
        return []

    result = []
    for bot_entry in config["bots"]:
        bot_name = bot_entry["name"]
        bot_config = load_bot_config(bot_name)
        if bot_config and skill_name in bot_config.get("skills", []):
            result.append(bot_name)
    return result

def compose_claude_md(
    bot_name: str,
    personality: str,
    description: str,
    skill_names: list[str] | None = None,
    bot_path: Path | None = None,
) -> str:
    """Compose a full CLAUDE.md combining bot profile and skill content.

    When skill_names is empty or None, output is identical to generate_claude_md().
    When bot_path is provided, a Memory section is appended with instructions
    for the bot to save and retrieve long-term memories.
    """
    sections = [
        f"# {bot_name}",
        "",
        "## Personality",
        personality,
        "",
        "## Role",
        description,
        "",
        "## Rules",
        "- Respond in Korean.",
        "- Save generated files to the workspace/ directory.",
        "- Always ask for confirmation before executing dangerous commands "
        "(delete, restart, etc.).",
        "- **절대로 Markdown 표(table)를 사용하지 마라.** "
        "Telegram에서 표는 깨진다. "
        "대신 이모지 + 한 줄씩 나열하라. "
        "예시:\n"
        

def regenerate_bot_claude_md(bot_name: str) -> None:
    """Regenerate a bot's CLAUDE.md based on current bot.yaml (including skills)."""
    bot_config = load_bot_config(bot_name)
    if not bot_config:
        return

    directory = bot_directory(bot_name)
    content = compose_claude_md(
        bot_name=bot_name,
        personality=bot_config.get("personality", ""),
        description=bot_config.get("description", ""),
        skill_names=bot_config.get("skills", []),
        bot_path=directory,
    )
    claude_md_path = directory / "CLAUDE.md"
    claude_md_path.write_text(content)

def update_session_claude_md(bot_path: Path) -> None:
    """Propagate the bot's CLAUDE.md to all existing sessions."""
    bot_claude_md = bot_path / "CLAUDE.md"
    if not bot_claude_md.exists():
        return

    content = bot_claude_md.read_text()
    sessions_directory = bot_path / "sessions"
    if not sessions_directory.exists():
        return

    for session in sessions_directory.iterdir():
        if session.is_dir():
            session_claude_md = session / "CLAUDE.md"
            session_claude_md.write_text(content)

def load_skill_mcp_config(name: str) -> dict[str, Any] | None:
    """Load MCP configuration from a skill's mcp.json."""
    path = skill_directory(name) / "mcp.json"
    if not path.exists():
        return None
    with open(path) as file:
        return json.load(file)

def merge_mcp_configs(skill_names: list[str]) -> dict[str, Any] | None:
    """Merge MCP configurations from multiple skills.

    Returns a combined mcpServers dict, or None if no skills have MCP config.
    """
    merged_servers: dict[str, Any] = {}

    for skill_name in skill_names:
        mcp_config = load_skill_mcp_config(skill_name)
        if mcp_config and "mcpServers" in mcp_config:
            merged_servers.update(mcp_config["mcpServers"])

    if not merged_servers:
        return None

    return {"mcpServers": merged_servers}

def install_builtin_skill(name: str) -> Path:
    """Install a built-in skill template to the user's skills directory.

    Copies all template files (SKILL.md, skill.yaml, etc.) from the
    built-in skill package to ~/.cclaw/skills/<name>/.

    Raises:
        ValueError: If the name is not a recognized built-in skill.
        FileExistsError: If the skill is already installed.

    Returns:
        The path to the installed skill directory.
    """
    from cclaw.builtin_skills import get_builtin_skill_path

    template_path = get_builtin_skill_path(name)
    if template_path is None:
        raise ValueError(f"Unknown built-in skill: {name}")

    target = skill_directory(name)
    if target.exists():
        raise FileExistsError(f"Skill '{name}' is already installed at {target}")

    target.mkdir(parents=True)
    for source_file in template_path.iterdir():
        if source_file.is_file():
            shutil.copy2(source_file, target / source_file.name)

    logger.info("Inst

def collect_skill_allowed_tools(skill_names: list[str]) -> list[str]:
    """Collect allowed_tools from all specified skills.

    Returns a merged list of tool patterns for --allowedTools flag.
    """
    result: list[str] = []

    for skill_name in skill_names:
        config = load_skill_config(skill_name)
        if not config:
            continue

        tools = config.get("allowed_tools", [])
        if tools:
            result.extend(tools)

    return result

def collect_skill_environment_variables(skill_names: list[str]) -> dict[str, str]:
    """Collect environment variables from all specified skills.

    Returns a merged dict of environment variable name → value.
    """
    result: dict[str, str] = {}

    for skill_name in skill_names:
        config = load_skill_config(skill_name)
        if not config:
            continue

        env_values = config.get("environment_variable_values", {})
        if env_values:
            result.update(env_values)

    return result

def collect_compact_targets(bot_name: str) -> list[CompactTarget]:
    """Collect all files eligible for compaction.

    Targets: MEMORY.md, user-created SKILL.md (not builtins), HEARTBEAT.md.
    """
    from cclaw.builtin_skills import is_builtin_skill
    from cclaw.skill import skill_directory

    targets: list[CompactTarget] = []

    bot_path = bot_directory(bot_name)
    bot_config = load_bot_config(bot_name)
    if not bot_config:
        return targets

    # 1. MEMORY.md
    memory_path = bot_path / "MEMORY.md"
    if memory_path.exists():
        content = memory_path.read_text()
        if content.strip():
            targets.append(
                CompactTarget(
                    label="MEMORY.md",
                    file_path=memory_path,
                    content=content,
                    line_count=len(content.splitlines()),
                    token_count=estimate_token_count(content),
                    document_type=DOCUMENT_TYPE_MEMORY,
                )

async def compact_content(
    content: str,
    document_type: str,
    working_directory: str,
    model: str = "sonnet",
    timeout: int = 120,
) -> str:
    """Compress a single document using Claude Code.

    Returns the compacted content string.
    """
    from cclaw.claude_runner import run_claude

    prompt = COMPACT_PROMPT.format(document_type=document_type, content=content)

    result = await run_claude(
        working_directory=working_directory,
        message=prompt,
        model=model,
        timeout=timeout,
    )

    return result.strip()

async def run_compact(bot_name: str, model: str = "sonnet") -> list[CompactResult]:
    """Run compaction on all eligible targets for a bot.

    Processes targets sequentially. Individual failures do not stop remaining targets.
    """
    targets = collect_compact_targets(bot_name)
    results: list[CompactResult] = []

    for target in targets:
        try:
            with tempfile.TemporaryDirectory() as temporary_directory:
                compacted = await compact_content(
                    content=target.content,
                    document_type=target.document_type,
                    working_directory=temporary_directory,
                    model=model,
                )
            results.append(
                CompactResult(
                    target=target,
                    compacted_content=compacted,
                    compacted_lines=len(compacted.splitlines()),
                    compacted_tokens=estimate_token_count(compacted),
                )
        

def format_compact_report(bot_name: str, results: list[CompactResult]) -> str:
    """Generate a human-readable compaction report."""
    lines = [f"\U0001f4e6 Token Compact: {bot_name}", ""]

    total_before = 0
    total_after = 0

    for result in results:
        if result.error:
            lines.append(f"\u274c {result.target.label}")
            lines.append(f"  Error: {result.error}")
            lines.append("")
            continue

        before_tokens = result.target.token_count
        after_tokens = result.compacted_tokens
        saved_tokens = before_tokens - after_tokens
        percentage = result.savings_percentage

        total_before += before_tokens
        total_after += after_tokens

        if "MEMORY" in result.target.label:
            icon = "\U0001f4dd"
        elif "Skill" in result.target.label:
            icon = "\U0001f9e9"
        else:
            icon = "\U0001f493"

        lines.append(f"{icon} {result.target.label}")
        lines.append(f"  

def save_compact_results(results: list[CompactResult]) -> None:
    """Save successfully compacted content back to original files.

    Only writes results that have no error.
    """
    for result in results:
        if result.error:
            continue
        result.target.file_path.write_text(result.compacted_content)
        logger.info("Saved compacted %s (%s)", result.target.label, result.target.file_path)

def prompt_input(label: str, default: str | None = None) -> str:
    """Prompt for single-line input using builtin input() for IME compatibility."""
    from rich.console import Console

    if default is not None:
        Console().print(f"{label} [dim](default: {default})[/dim] ", end="")
        value = input().strip()
        return value if value else default
    else:
        Console().print(f"{label} ", end="")
        return input().strip()

def prompt_multiline(label: str) -> str:
    """Prompt for multi-line input. Empty line finishes input."""
    from rich.console import Console

    Console().print(f"{label} [dim](empty line to finish)[/dim]")
    lines = []
    while True:
        line = input()
        if line == "":
            break
        lines.append(line)
    return "\n".join(lines).strip()

def split_message(text: str, limit: int = TELEGRAM_MESSAGE_LIMIT) -> list[str]:
    """Split a long message into chunks that fit Telegram's message limit.

    Tries to split at newline boundaries when possible.
    """
    if len(text) <= limit:
        return [text]

    chunks = []
    while text:
        if len(text) <= limit:
            chunks.append(text)
            break

        split_index = text.rfind("\n", 0, limit)
        if split_index == -1 or split_index < limit // 2:
            split_index = limit

        chunks.append(text[:split_index])
        text = text[split_index:].lstrip("\n")

    return chunks

def markdown_to_telegram_html(text: str) -> str:
    """Convert Markdown formatting to Telegram-compatible HTML.

    Handles: **bold**, *italic*, `code`, ```code blocks```, [links](url)
    """
    # Extract links before escaping so URLs stay intact
    link_placeholder = {}
    link_counter = 0

    def _replace_link(match: re.Match) -> str:
        nonlocal link_counter
        placeholder = f"\x00LINK{link_counter}\x00"
        link_placeholder[placeholder] = (match.group(1), match.group(2))
        link_counter += 1
        return placeholder

    text = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", _replace_link, text)

    text = html.escape(text)

    text = re.sub(r"```(\w*)\n(.*?)```", r"<pre>\2</pre>", text, flags=re.DOTALL)
    text = re.sub(r"```(.*?)```", r"<pre>\1</pre>", text, flags=re.DOTALL)

    text = re.sub(r"`([^`]+)`", r"<code>\1</code>", text)

    # Headings → bold
    text = re.sub(r"^#{1,6}\s+(.+)$", r"<b>\1</b>", text, flags=re.MULTILINE)

    text = re.sub(r"\*\*(.+?)

def setup_logging(log_level: str = "INFO") -> None:
    """Configure logging with daily rotation to ~/.cclaw/logs/."""
    log_directory = cclaw_home() / "logs"
    log_directory.mkdir(parents=True, exist_ok=True)

    today = datetime.now().strftime("%y%m%d")
    log_file = log_directory / f"cclaw-{today}.log"

    logging.basicConfig(
        level=getattr(logging, log_level.upper(), logging.INFO),
        format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
        handlers=[
            logging.FileHandler(log_file),
            logging.StreamHandler(),
        ],
    )