Function bodies 139 total

async def _run_bots(bot_names: list[str] | None = None) -> None:
    """Run one or more bots with long polling."""
    from telegram.ext import Application

    config = load_config()
    if not config or not config.get("bots"):
        console.print("[red]No bots configured. Run 'cclaw init' first.[/red]")
        return

    settings = config.get("settings", {})
    log_level = settings.get("log_level", "INFO")
    setup_logging(log_level)

    bots_to_run = config["bots"]
    if bot_names:
        bots_to_run = [b for b in bots_to_run if b["name"] in bot_names]
        if not bots_to_run:
            console.print("[red]No matching bots found.[/red]")
            return

    applications = []

    for bot_entry in bots_to_run:
        name = bot_entry["name"]
        try:
            bot_config = load_bot_config(name)
            if not bot_config:
                console.print(f"[yellow]Skipping {name}: bot.yaml not found.[/yellow]")
                continue

            token = bo

def start_bots(bot_name: str | None = None, daemon: bool = False) -> None:
    """Start bot(s), optionally as a daemon."""
    if daemon:
        _start_daemon()
        return

    bot_names = [bot_name] if bot_name else None

    try:
        asyncio.run(_run_bots(bot_names))
    except KeyboardInterrupt:
        pass

def _start_daemon() -> None:
    """Start cclaw as a launchd daemon."""
    plist_path = _plist_path()
    plist_path.parent.mkdir(parents=True, exist_ok=True)

    # Find cclaw in the same venv bin directory
    venv_bin = Path(sys.executable).parent
    cclaw_executable = venv_bin / "cclaw"
    if not cclaw_executable.exists():
        cclaw_executable = Path(sys.executable)
        cclaw_arguments = [str(cclaw_executable), "-m", "cclaw.cli", "start"]
    else:
        cclaw_arguments = [str(cclaw_executable), "start"]

    log_directory = cclaw_home() / "logs"
    log_directory.mkdir(parents=True, exist_ok=True)

    # Capture current PATH so the daemon can find claude CLI (npm global bin)
    current_path = os.environ.get("PATH", "/usr/bin:/bin:/usr/sbin:/sbin")

    newline = "\n"
    plist_content = f"""<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key

def stop_bots() -> None:
    """Stop the running daemon or foreground process."""
    plist_path = _plist_path()

    if plist_path.exists():
        result = subprocess.run(
            ["launchctl", "unload", str(plist_path)],
            capture_output=True,
            text=True,
        )
        plist_path.unlink(missing_ok=True)
        if result.returncode == 0:
            console.print("[green]Daemon stopped.[/green]")
        else:
            console.print(f"[yellow]launchctl unload: {result.stderr.strip()}[/yellow]")

    pid_file = _pid_file()
    if pid_file.exists():
        try:
            pid = int(pid_file.read_text().strip())
            os.kill(pid, signal.SIGTERM)
            console.print(f"[green]Sent SIGTERM to process {pid}.[/green]")
        except (ValueError, ProcessLookupError):
            pass
        pid_file.unlink(missing_ok=True)
    elif not plist_path.exists():
        console.print("[yellow]No running cclaw process found.[/yellow]")

def show_status() -> None:
    """Show the running status of cclaw."""
    from rich.table import Table

    config = load_config()

    pid_file = _pid_file()
    plist_path = _plist_path()

    if plist_path.exists():
        console.print("[green]Daemon: running (launchd)[/green]")
    elif pid_file.exists():
        try:
            pid = int(pid_file.read_text().strip())
            os.kill(pid, 0)
            console.print(f"[green]Process: running (PID {pid})[/green]")
        except (ValueError, ProcessLookupError):
            console.print("[yellow]Process: stale PID file[/yellow]")
            pid_file.unlink(missing_ok=True)
    else:
        console.print("[yellow]Status: not running[/yellow]")

    if not config or not config.get("bots"):
        console.print("[yellow]No bots configured.[/yellow]")
        return

    table = Table(title="Bot Status")
    table.add_column("Name", style="cyan")
    table.add_column("Telegram", style="green")
    table.add_column("Sessions

def list_builtin_skills() -> list[dict[str, Any]]:
    """List all available built-in skills.

    Scans subdirectories for those containing SKILL.md.
    Returns a list of dicts with name, description, and path.
    """
    directory = builtin_skills_directory()
    result = []

    for entry in sorted(directory.iterdir()):
        if entry.is_dir() and (entry / "SKILL.md").exists():
            description = ""
            skill_yaml_path = entry / "skill.yaml"
            if skill_yaml_path.exists():
                with open(skill_yaml_path) as file:
                    config = yaml.safe_load(file) or {}
                description = config.get("description", "")

            emoji = config.get("emoji", "") if skill_yaml_path.exists() else ""

            result.append(
                {
                    "name": entry.name,
                    "description": description,
                    "emoji": emoji,
                    "path": entry,
                }
            )

    

def get_builtin_skill_path(name: str) -> Path | None:
    """Return the path to a specific built-in skill template.

    Returns None if the skill does not exist.
    """
    path = builtin_skills_directory() / name
    if path.is_dir() and (path / "SKILL.md").exists():
        return path
    return None

def _write_session_settings(working_directory: str, allowed_tools: list[str]) -> None:
    """Write .claude/settings.json in the session directory with skill permissions."""
    if not allowed_tools:
        return

    claude_directory = Path(working_directory) / ".claude"
    claude_directory.mkdir(parents=True, exist_ok=True)

    settings_path = claude_directory / "settings.json"

    settings: dict[str, Any] = {}
    if settings_path.exists():
        with open(settings_path) as settings_file:
            settings = json.load(settings_file)

    permissions = settings.get("permissions", {})
    existing_allow = set(permissions.get("allow", []))
    existing_allow.update(allowed_tools)

    permissions["allow"] = sorted(existing_allow)
    settings["permissions"] = permissions

    with open(settings_path, "w") as settings_file:
        json.dump(settings, settings_file, indent=2)

def cancel_process(session_key: str) -> bool:
    """Cancel a running process for a session.

    Returns True if a process was found and killed, False otherwise.
    """
    process = _running_processes.get(session_key)
    if process and process.returncode is None:
        process.kill()
        logger.info("Cancelled Claude Code process for session %s", session_key)
        return True
    return False

def cancel_all_processes() -> int:
    """Kill all running Claude Code subprocesses.

    Called during shutdown to avoid waiting for long-running processes.
    Returns the number of processes killed.
    """
    killed = 0
    for session_key, process in list(_running_processes.items()):
        if process.returncode is None:
            process.kill()
            logger.info("Shutdown: killed process for session %s", session_key)
            killed += 1
    _running_processes.clear()
    return killed

async def run_claude(
    working_directory: str,
    message: str,
    extra_arguments: list[str] | None = None,
    timeout: int = DEFAULT_TIMEOUT,
    session_key: str | None = None,
    model: str | None = None,
    skill_names: list[str] | None = None,
    claude_session_id: str | None = None,
    resume_session: bool = False,
) -> str:
    """Run Claude Code CLI as a subprocess and return its output.

    Args:
        working_directory: Working directory for Claude Code.
        message: The prompt message to send.
        extra_arguments: Additional CLI arguments from bot config.
        timeout: Maximum execution time in seconds.
        session_key: Optional key for process tracking (enables /cancel).
        model: Claude model to use (sonnet, opus, haiku).
        skill_names: Optional list of skill names to inject MCP config and env vars.
        claude_session_id: Optional Claude Code session ID for continuity.
        resume_session: If True and claude_session_id is set,

def _prepare_skill_environment(
    working_directory: str,
    skill_names: list[str] | None,
) -> dict[str, str] | None:
    """Prepare MCP config and environment variables for skills."""
    if not skill_names:
        return None

    from cclaw.skill import collect_skill_environment_variables, merge_mcp_configs

    mcp_config = merge_mcp_configs(skill_names)
    if mcp_config:
        mcp_json_path = str(Path(working_directory) / ".mcp.json")
        with open(mcp_json_path, "w") as mcp_file:
            json.dump(mcp_config, mcp_file, indent=2)

    skill_environment_variables = collect_skill_environment_variables(skill_names)
    if skill_environment_variables:
        return {**os.environ, **skill_environment_variables}

    return None

def _extract_text_delta(data: dict[str, Any]) -> str | None:
    """Extract text from a stream-json event line.

    Handles two event types:
    - stream_event with content_block_delta (token-level, --verbose mode)
    - assistant message with text content blocks (turn-level)
    """
    event_type = data.get("type")

    if event_type == "stream_event":
        event = data.get("event", {})
        if event.get("type") == "content_block_delta":
            delta = event.get("delta", {})
            if delta.get("type") == "text_delta":
                return delta.get("text", "")

    return None

def _extract_assistant_text(data: dict[str, Any]) -> str | None:
    """Extract text from an assistant turn-level event (non-verbose mode)."""
    if data.get("type") == "assistant":
        message = data.get("message", {})
        content_blocks = message.get("content", [])
        texts = []
        for block in content_blocks:
            if block.get("type") == "text":
                texts.append(block.get("text", ""))
        if texts:
            return "".join(texts)
    return None

async def run_claude_streaming(
    working_directory: str,
    message: str,
    on_text_chunk: Callable[[str], Any] | None = None,
    extra_arguments: list[str] | None = None,
    timeout: int = DEFAULT_TIMEOUT,
    session_key: str | None = None,
    model: str | None = None,
    skill_names: list[str] | None = None,
    claude_session_id: str | None = None,
    resume_session: bool = False,
) -> str:
    """Run Claude Code CLI with streaming output.

    Uses --output-format stream-json --verbose --include-partial-messages
    to receive token-level text deltas. Calls on_text_chunk for each text
    chunk received. Returns the final complete text.

    If stream-json produces no result, falls back to accumulated text
    from streaming deltas or assistant turn events.

    Args:
        working_directory: Working directory for Claude Code.
        message: The prompt message to send.
        on_text_chunk: Async or sync callback for each text chunk.
        extra_arguments: Additi

def main(context: typer.Context) -> None:
    """cclaw - Telegram + Claude Code AI assistant."""
    if context.invoked_subcommand is None:
        from rich.console import Console

        from cclaw import __version__

        console = Console()
        console.print(f"[cyan]{ASCII_ART}[/cyan]")
        console.print(f"  [dim]v{__version__}[/dim]\n")
        console.print("Run [green]cclaw --help[/green] for available commands.\n")

def start(
    bot: str = typer.Option(None, help="Start specific bot only"),
    daemon: bool = typer.Option(False, help="Run as background daemon"),
) -> None:
    """Start bot(s)."""
    from cclaw.bot_manager import start_bots

    start_bots(bot_name=bot, daemon=daemon)

def bot_list() -> None:
    """List all bots."""
    from rich.console import Console
    from rich.table import Table

    from cclaw.config import load_config

    console = Console()
    config = load_config()

    if not config or not config.get("bots"):
        console.print("[yellow]No bots configured. Run 'cclaw init' or 'cclaw bot add'.[/yellow]")
        return

    table = Table(title="Registered Bots")
    table.add_column("Name", style="cyan")
    table.add_column("Model", style="magenta")
    table.add_column("Telegram", style="green")
    table.add_column("Path", style="dim")

    for bot_entry in config["bots"]:
        from cclaw.config import DEFAULT_MODEL, bot_directory, load_bot_config

        bot_config = load_bot_config(bot_entry["name"])
        telegram_username = bot_config.get("telegram_username", "N/A") if bot_config else "N/A"
        model = bot_config.get("model", DEFAULT_MODEL) if bot_config else DEFAULT_MODEL
        path = str(bot_directory(bot_entry["n

def bot_remove(name: str) -> None:
    """Remove a bot."""
    import shutil

    from rich.console import Console

    from cclaw.config import bot_directory as get_bot_directory
    from cclaw.config import load_config, save_config

    console = Console()
    config = load_config()

    if not config or not config.get("bots"):
        console.print("[red]No bots configured.[/red]")
        raise typer.Exit(1)

    bot_entry = next((b for b in config["bots"] if b["name"] == name), None)
    if not bot_entry:
        console.print(f"[red]Bot '{name}' not found.[/red]")
        raise typer.Exit(1)

    confirmed = typer.confirm(f"Remove bot '{name}'? This will delete all data.")
    if not confirmed:
        console.print("[yellow]Cancelled.[/yellow]")
        return

    target_directory = get_bot_directory(name)
    if target_directory.exists():
        shutil.rmtree(target_directory)

    config["bots"] = [b for b in config["bots"] if b["name"] != name]
    save_config(config)

    

def skills_callback(context: typer.Context) -> None:
    """Skill management."""
    if context.invoked_subcommand is None:
        from rich.console import Console
        from rich.table import Table

        from cclaw.builtin_skills import list_builtin_skills
        from cclaw.skill import bots_using_skill, list_skills

        console = Console()
        installed_skills = list_skills()
        installed_names = {skill["name"] for skill in installed_skills}

        builtin_skills = list_builtin_skills()
        not_installed_builtins = [
            skill for skill in builtin_skills if skill["name"] not in installed_names
        ]

        if not installed_skills and not not_installed_builtins:
            console.print("[yellow]No skills found. Run 'cclaw skills add' to create one.[/yellow]")
            return

        table = Table(title="All Skills")
        table.add_column("Name", style="cyan")
        table.add_column("Type", style="magenta")
        table.add_column("St

def logs_callback(
    context: typer.Context,
    lines: int = typer.Option(50, "--lines", "-n", help="Number of lines to show"),
    follow: bool = typer.Option(False, "--follow", "-f", help="Follow log output"),
) -> None:
    """Show today's log file."""
    if context.invoked_subcommand is not None:
        return

    import subprocess
    from datetime import datetime

    from rich.console import Console

    from cclaw.config import cclaw_home

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

    if not log_file.exists():
        console.print("[yellow]No log file for today.[/yellow]")
        raise typer.Exit()

    command = ["tail", f"-n{lines}"]
    if follow:
        command.append("-f")
    command.append(str(log_file))

    try:
        subprocess.run(command)
    except KeyboardInterrupt:
        pass

def logs_clean(
    days: int = typer.Option(7, "--days", "-d", help="Keep logs from the last N days"),
    dry_run: bool = typer.Option(False, "--dry-run", help="Show files to delete without deleting"),
) -> None:
    """Delete old log files, keeping the last N days (default: 7)."""
    from datetime import datetime, timedelta

    from rich.console import Console

    from cclaw.config import cclaw_home

    console = Console()
    log_directory = cclaw_home() / "logs"

    if not log_directory.exists():
        console.print("[yellow]No logs directory found.[/yellow]")
        return

    log_files = sorted(log_directory.glob("cclaw-*.log"))
    if not log_files:
        console.print("[yellow]No log files found.[/yellow]")
        return

    cutoff_date = datetime.now() - timedelta(days=days)
    cutoff_string = cutoff_date.strftime("%y%m%d")

    files_to_delete = []
    for log_file in log_files:
        # Extract YYMMDD from filename: cclaw-YYMMDD.log
        date_part = log_fi

def bot_model(
    name: str = typer.Argument(help="Bot name"),
    model: str = typer.Argument(None, help="Model to set (sonnet/opus/haiku)"),
) -> None:
    """Show or change the model for a bot."""
    from rich.console import Console

    from cclaw.config import (
        DEFAULT_MODEL,
        VALID_MODELS,
        is_valid_model,
        load_bot_config,
        save_bot_config,
    )

    console = Console()
    bot_config = load_bot_config(name)
    if not bot_config:
        console.print(f"[red]Bot '{name}' not found.[/red]")
        raise typer.Exit(1)

    if model is None:
        current = bot_config.get("model", DEFAULT_MODEL)
        console.print(f"[cyan]{name}[/cyan] model: [magenta]{current}[/magenta]")
        return

    if not is_valid_model(model):
        console.print(f"[red]Invalid model: {model}[/red]")
        console.print(f"Available: {', '.join(VALID_MODELS)}")
        raise typer.Exit(1)

    bot_config["model"] = model
    save_bot_config(name, bot_con

def bot_streaming(
    name: str = typer.Argument(help="Bot name"),
    value: str = typer.Argument(None, help="on or off"),
) -> None:
    """Show or toggle streaming mode for a bot."""
    from rich.console import Console

    from cclaw.config import DEFAULT_STREAMING, load_bot_config, save_bot_config

    console = Console()
    bot_config = load_bot_config(name)
    if not bot_config:
        console.print(f"[red]Bot '{name}' not found.[/red]")
        raise typer.Exit(1)

    if value is None:
        current = bot_config.get("streaming", DEFAULT_STREAMING)
        status_text = "on" if current else "off"
        console.print(f"[cyan]{name}[/cyan] streaming: [magenta]{status_text}[/magenta]")
        return

    if value.lower() == "on":
        bot_config["streaming"] = True
        save_bot_config(name, bot_config)
        console.print(f"[green]{name} streaming enabled[/green]")
    elif value.lower() == "off":
        bot_config["streaming"] = False
        save_bot_config(n

def bot_compact(
    name: str = typer.Argument(help="Bot name"),
    model: str = typer.Option("sonnet", help="Model for compaction"),
    yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation"),
) -> None:
    """Compact bot's MD files to save tokens."""
    import asyncio

    from rich.console import Console

    from cclaw.config import load_bot_config
    from cclaw.skill import regenerate_bot_claude_md, update_session_claude_md
    from cclaw.token_compact import (
        collect_compact_targets,
        format_compact_report,
        run_compact,
        save_compact_results,
    )

    console = Console()

    bot_config = load_bot_config(name)
    if not bot_config:
        console.print(f"[red]Bot '{name}' not found.[/red]")
        raise typer.Exit(1)

    targets = collect_compact_targets(name)
    if not targets:
        console.print("[yellow]No compactable files found.[/yellow]")
        return

    console.print(f"[cyan]Found {len(targets)} file(s) to

def bot_edit(name: str) -> None:
    """Edit bot configuration."""
    import subprocess

    from rich.console import Console

    from cclaw.config import bot_directory, load_bot_config

    console = Console()
    bot_config = load_bot_config(name)

    if not bot_config:
        console.print(f"[red]Bot '{name}' not found.[/red]")
        raise typer.Exit(1)

    bot_yaml_path = bot_directory(name) / "bot.yaml"
    editor = "vi"
    subprocess.run([editor, str(bot_yaml_path)])

def skill_builtins() -> None:
    """List available built-in skills."""
    from rich.console import Console
    from rich.table import Table

    from cclaw.builtin_skills import list_builtin_skills
    from cclaw.skill import is_skill

    console = Console()
    builtin_skills = list_builtin_skills()

    if not builtin_skills:
        console.print("[yellow]No built-in skills available.[/yellow]")
        return

    table = Table(title="Built-in Skills")
    table.add_column("Name", style="cyan")
    table.add_column("Description", style="dim")
    table.add_column("Installed", style="green")

    for skill in builtin_skills:
        installed = is_skill(skill["name"])
        installed_display = "[green]yes[/green]" if installed else "[dim]no[/dim]"
        table.add_row(skill["name"], skill["description"], installed_display)

    console.print(table)
    console.print("\nInstall with: [cyan]cclaw skills install <name>[/cyan]")

def skill_install(
    name: str = typer.Argument(None, help="Built-in skill name to install"),
) -> None:
    """Install a built-in skill (or list available ones)."""
    from rich.console import Console
    from rich.table import Table

    from cclaw.builtin_skills import list_builtin_skills
    from cclaw.skill import (
        activate_skill,
        check_skill_requirements,
        install_builtin_skill,
        is_skill,
    )

    console = Console()

    if name is None:
        builtin_skills = list_builtin_skills()
        if not builtin_skills:
            console.print("[yellow]No built-in skills available.[/yellow]")
            return

        table = Table(title="Built-in Skills")
        table.add_column("Name", style="cyan")
        table.add_column("Description", style="dim")
        table.add_column("Installed", style="green")

        for skill in builtin_skills:
            installed = is_skill(skill["name"])
            installed_display = "[green]yes[/green]" i

def skill_add() -> None:
    """Create a new skill interactively."""
    from rich.console import Console

    from cclaw.skill import (
        VALID_SKILL_TYPES,
        create_skill_directory,
        default_skill_yaml,
        generate_skill_markdown,
        is_skill,
        save_skill_config,
    )

    console = Console()

    from cclaw.utils import prompt_input

    name = prompt_input("Skill name:")
    if is_skill(name):
        console.print(f"[red]Skill '{name}' already exists.[/red]")
        raise typer.Exit(1)

    description = prompt_input("Description (optional):")

    use_tools = typer.confirm("Does this skill require tools (CLI, MCP, browser)?", default=False)

    selected_type = None
    required_commands: list[str] = []
    environment_variables: list[str] = []

    if use_tools:
        type_choices = ", ".join(VALID_SKILL_TYPES)
        selected_type = prompt_input(f"Skill type ({type_choices}):")
        if selected_type not in VALID_SKILL_TYPES:
         

def skill_remove(name: str) -> None:
    """Remove a skill."""
    from rich.console import Console

    from cclaw.skill import is_skill, remove_skill

    console = Console()

    if not is_skill(name):
        console.print(f"[red]Skill '{name}' not found.[/red]")
        raise typer.Exit(1)

    confirmed = typer.confirm(f"Remove skill '{name}'? This will detach it from all bots.")
    if not confirmed:
        console.print("[yellow]Cancelled.[/yellow]")
        return

    remove_skill(name)
    console.print(f"[green]Skill '{name}' removed.[/green]")

def skill_setup(name: str) -> None:
    """Set up a skill (check requirements and activate)."""
    from rich.console import Console

    from cclaw.skill import (
        activate_skill,
        check_skill_requirements,
        is_skill,
        load_skill_config,
        save_skill_config,
        skill_status,
    )

    console = Console()

    if not is_skill(name):
        console.print(f"[red]Skill '{name}' not found.[/red]")
        raise typer.Exit(1)

    already_active = skill_status(name) == "active"

    # Check if environment variables need configuration (even if already active)
    config = load_skill_config(name)
    has_unconfigured_environment_variables = False
    if config and config.get("environment_variables"):
        environment_variable_values = config.get("environment_variable_values", {})
        for variable in config["environment_variables"]:
            if not environment_variable_values.get(variable):
                has_unconfigured_environment_variable

def skill_test(name: str) -> None:
    """Test a skill's requirements."""
    from rich.console import Console

    from cclaw.skill import check_skill_requirements, is_skill

    console = Console()

    if not is_skill(name):
        console.print(f"[red]Skill '{name}' not found.[/red]")
        raise typer.Exit(1)

    errors = check_skill_requirements(name)
    if errors:
        console.print(f"[red]Requirements check failed for '{name}':[/red]")
        for error in errors:
            console.print(f"  [red]- {error}[/red]")
    else:
        console.print(f"[green]All requirements met for '{name}'.[/green]")

def skill_edit(name: str) -> None:
    """Edit a skill's SKILL.md in the default editor."""
    import os
    import subprocess

    from rich.console import Console

    from cclaw.skill import is_skill, skill_directory

    console = Console()

    if not is_skill(name):
        console.print(f"[red]Skill '{name}' not found.[/red]")
        raise typer.Exit(1)

    skill_md_path = skill_directory(name) / "SKILL.md"
    editor = os.environ.get("EDITOR", "vi")
    subprocess.run([editor, str(skill_md_path)])

def cron_list(bot: str = typer.Argument(help="Bot name")) -> None:
    """List cron jobs for a bot."""
    from rich.console import Console
    from rich.table import Table

    from cclaw.config import load_bot_config
    from cclaw.cron import list_cron_jobs, next_run_time

    console = Console()

    if not load_bot_config(bot):
        console.print(f"[red]Bot '{bot}' not found.[/red]")
        raise typer.Exit(1)

    jobs = list_cron_jobs(bot)
    if not jobs:
        console.print(f"[yellow]No cron jobs for '{bot}'. Run 'cclaw cron add {bot}'.[/yellow]")
        return

    table = Table(title=f"Cron Jobs - {bot}")
    table.add_column("Name", style="cyan")
    table.add_column("Schedule", style="magenta")
    table.add_column("Timezone", style="blue")
    table.add_column("Message", style="dim", max_width=40)
    table.add_column("Next Run", style="green")
    table.add_column("Status", style="yellow")

    for job in jobs:
        schedule_display = job.get("schedule") or f"a

def cron_add(bot: str = typer.Argument(help="Bot name")) -> None:
    """Add a cron job to a bot interactively."""
    from rich.console import Console

    from cclaw.config import load_bot_config
    from cclaw.cron import (
        add_cron_job,
        get_cron_job,
        parse_one_shot_time,
        validate_cron_schedule,
    )

    console = Console()

    if not load_bot_config(bot):
        console.print(f"[red]Bot '{bot}' not found.[/red]")
        raise typer.Exit(1)

    from cclaw.utils import prompt_input, prompt_multiline

    name = prompt_input("Job name:")
    if get_cron_job(bot, name):
        console.print(f"[red]Job '{name}' already exists.[/red]")
        raise typer.Exit(1)

    use_one_shot = typer.confirm("One-shot (run once at specific time)?", default=False)

    job: dict = {"name": name, "enabled": True}

    if use_one_shot:
        at_value = prompt_input("Run at (ISO datetime or duration like 30m/2h/1d):")
        parsed = parse_one_shot_time(at_value

def cron_remove(
    bot: str = typer.Argument(help="Bot name"),
    job: str = typer.Argument(help="Job name"),
) -> None:
    """Remove a cron job."""
    from rich.console import Console

    from cclaw.cron import remove_cron_job

    console = Console()

    if not remove_cron_job(bot, job):
        console.print(f"[red]Job '{job}' not found in bot '{bot}'.[/red]")
        raise typer.Exit(1)

    console.print(f"[green]Job '{job}' removed from '{bot}'.[/green]")

def cron_enable(
    bot: str = typer.Argument(help="Bot name"),
    job: str = typer.Argument(help="Job name"),
) -> None:
    """Enable a cron job."""
    from rich.console import Console

    from cclaw.cron import enable_cron_job

    console = Console()

    if not enable_cron_job(bot, job):
        console.print(f"[red]Job '{job}' not found in bot '{bot}'.[/red]")
        raise typer.Exit(1)

    console.print(f"[green]Job '{job}' enabled.[/green]")

def cron_disable(
    bot: str = typer.Argument(help="Bot name"),
    job: str = typer.Argument(help="Job name"),
) -> None:
    """Disable a cron job."""
    from rich.console import Console

    from cclaw.cron import disable_cron_job

    console = Console()

    if not disable_cron_job(bot, job):
        console.print(f"[red]Job '{job}' not found in bot '{bot}'.[/red]")
        raise typer.Exit(1)

    console.print(f"[green]Job '{job}' disabled.[/green]")

def cron_run(
    bot: str = typer.Argument(help="Bot name"),
    job: str = typer.Argument(help="Job name"),
) -> None:
    """Run a cron job immediately (for testing)."""
    import asyncio

    from rich.console import Console

    from cclaw.claude_runner import run_claude
    from cclaw.config import DEFAULT_MODEL, load_bot_config
    from cclaw.cron import cron_session_directory, get_cron_job

    console = Console()

    bot_config = load_bot_config(bot)
    if not bot_config:
        console.print(f"[red]Bot '{bot}' not found.[/red]")
        raise typer.Exit(1)

    cron_job = get_cron_job(bot, job)
    if not cron_job:
        console.print(f"[red]Job '{job}' not found in bot '{bot}'.[/red]")
        raise typer.Exit(1)

    message = cron_job.get("message", "")
    model = cron_job.get("model") or bot_config.get("model", DEFAULT_MODEL)
    job_skills = cron_job.get("skills") or bot_config.get("skills", [])
    command_timeout = bot_config.get("command_timeout", 300)
    work

def memory_show(bot: str = typer.Argument(help="Bot name")) -> None:
    """Show bot memory contents."""
    from rich.console import Console
    from rich.markdown import Markdown

    from cclaw.config import bot_directory, load_bot_config
    from cclaw.session import load_bot_memory

    console = Console()

    if not load_bot_config(bot):
        console.print(f"[red]Bot '{bot}' not found.[/red]")
        raise typer.Exit(1)

    content = load_bot_memory(bot_directory(bot))
    if not content:
        console.print(f"[yellow]No memories saved for '{bot}'.[/yellow]")
        return

    console.print(Markdown(content))

def memory_edit(bot: str = typer.Argument(help="Bot name")) -> None:
    """Edit bot memory in the default editor."""
    import os
    import subprocess

    from rich.console import Console

    from cclaw.config import bot_directory, load_bot_config
    from cclaw.session import memory_file_path

    console = Console()

    if not load_bot_config(bot):
        console.print(f"[red]Bot '{bot}' not found.[/red]")
        raise typer.Exit(1)

    path = memory_file_path(bot_directory(bot))
    if not path.exists():
        path.write_text("# Memory\n\n")

    editor = os.environ.get("EDITOR", "vi")
    subprocess.run([editor, str(path)])

def memory_clear(bot: str = typer.Argument(help="Bot name")) -> None:
    """Clear bot memory."""
    from rich.console import Console

    from cclaw.config import bot_directory, load_bot_config
    from cclaw.session import clear_bot_memory

    console = Console()

    if not load_bot_config(bot):
        console.print(f"[red]Bot '{bot}' not found.[/red]")
        raise typer.Exit(1)

    confirmed = typer.confirm(f"Clear all memory for '{bot}'?")
    if not confirmed:
        console.print("[yellow]Cancelled.[/yellow]")
        return

    clear_bot_memory(bot_directory(bot))
    console.print(f"[green]Memory cleared for '{bot}'.[/green]")

def _regenerate_all_bots_claude_md() -> None:
    """Regenerate CLAUDE.md for all bots and propagate to sessions."""
    from cclaw.config import bot_directory, load_config
    from cclaw.skill import regenerate_bot_claude_md, update_session_claude_md

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

    for bot_entry in config["bots"]:
        name = bot_entry["name"]
        regenerate_bot_claude_md(name)
        update_session_claude_md(bot_directory(name))

def global_memory_show() -> None:
    """Show global memory contents."""
    from rich.console import Console
    from rich.markdown import Markdown

    from cclaw.session import load_global_memory

    console = Console()

    content = load_global_memory()
    if not content:
        console.print("[yellow]No global memory saved yet.[/yellow]")
        return

    console.print(Markdown(content))

def global_memory_edit() -> None:
    """Edit global memory in the default editor."""
    import os
    import subprocess

    from rich.console import Console

    from cclaw.session import global_memory_file_path, save_global_memory

    console = Console()

    path = global_memory_file_path()
    if not path.exists():
        save_global_memory("# Global Memory\n\n")

    editor = os.environ.get("EDITOR", "vi")
    subprocess.run([editor, str(path)])

    # Regenerate all bots' CLAUDE.md to include updated global memory
    _regenerate_all_bots_claude_md()
    console.print("[green]Global memory updated. All bots' CLAUDE.md regenerated.[/green]")

def global_memory_clear() -> None:
    """Clear global memory."""
    from rich.console import Console

    from cclaw.session import clear_global_memory

    console = Console()

    confirmed = typer.confirm("Clear global memory? This affects all bots.")
    if not confirmed:
        console.print("[yellow]Cancelled.[/yellow]")
        return

    clear_global_memory()
    _regenerate_all_bots_claude_md()
    console.print("[green]Global memory cleared. All bots' CLAUDE.md regenerated.[/green]")

def heartbeat_status() -> None:
    """Show heartbeat status for all bots."""
    from rich.console import Console
    from rich.table import Table

    from cclaw.config import DEFAULT_MODEL, load_bot_config, load_config
    from cclaw.heartbeat import get_heartbeat_config

    console = Console()
    config = load_config()

    if not config or not config.get("bots"):
        console.print("[yellow]No bots configured.[/yellow]")
        return

    table = Table(title="Heartbeat Status")
    table.add_column("Bot", style="cyan")
    table.add_column("Enabled", style="green")
    table.add_column("Interval", style="magenta")
    table.add_column("Active Hours", style="dim")
    table.add_column("Model", style="yellow")

    for bot_entry in config["bots"]:
        name = bot_entry["name"]
        bot_config = load_bot_config(name)
        if not bot_config:
            continue

        heartbeat_config = get_heartbeat_config(name)
        enabled = heartbeat_config.get("enabled", Fal

def heartbeat_enable(bot: str = typer.Argument(help="Bot name")) -> None:
    """Enable heartbeat for a bot."""
    from rich.console import Console

    from cclaw.config import load_bot_config
    from cclaw.heartbeat import enable_heartbeat

    console = Console()

    if not load_bot_config(bot):
        console.print(f"[red]Bot '{bot}' not found.[/red]")
        raise typer.Exit(1)

    if enable_heartbeat(bot):
        console.print(f"[green]Heartbeat enabled for '{bot}'.[/green]")
    else:
        console.print(f"[red]Failed to enable heartbeat for '{bot}'.[/red]")
        raise typer.Exit(1)

def heartbeat_disable(bot: str = typer.Argument(help="Bot name")) -> None:
    """Disable heartbeat for a bot."""
    from rich.console import Console

    from cclaw.config import load_bot_config
    from cclaw.heartbeat import disable_heartbeat

    console = Console()

    if not load_bot_config(bot):
        console.print(f"[red]Bot '{bot}' not found.[/red]")
        raise typer.Exit(1)

    if disable_heartbeat(bot):
        console.print(f"[green]Heartbeat disabled for '{bot}'.[/green]")
    else:
        console.print(f"[red]Failed to disable heartbeat for '{bot}'.[/red]")
        raise typer.Exit(1)

def heartbeat_run(bot: str = typer.Argument(help="Bot name")) -> None:
    """Run heartbeat immediately (for testing)."""
    import asyncio

    from rich.console import Console

    from cclaw.claude_runner import run_claude
    from cclaw.config import DEFAULT_MODEL, load_bot_config
    from cclaw.heartbeat import (
        HEARTBEAT_OK_MARKER,
        heartbeat_session_directory,
        load_heartbeat_markdown,
    )

    console = Console()

    bot_config = load_bot_config(bot)
    if not bot_config:
        console.print(f"[red]Bot '{bot}' not found.[/red]")
        raise typer.Exit(1)

    heartbeat_content = load_heartbeat_markdown(bot)
    if not heartbeat_content:
        console.print(
            f"[yellow]No HEARTBEAT.md found. Run 'cclaw heartbeat enable {bot}' first.[/yellow]"
        )
        raise typer.Exit(1)

    model = bot_config.get("model", DEFAULT_MODEL)
    attached_skills = bot_config.get("skills", [])
    command_timeout = bot_config.get("command_timeout"