Function bodies 139 total

def heartbeat_edit(bot: str = typer.Argument(help="Bot name")) -> None:
    """Edit HEARTBEAT.md for a bot."""
    import os
    import subprocess

    from rich.console import Console

    from cclaw.config import load_bot_config
    from cclaw.heartbeat import (
        default_heartbeat_content,
        heartbeat_session_directory,
    )

    console = Console()

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

    session_directory = heartbeat_session_directory(bot)
    heartbeat_md_path = session_directory / "HEARTBEAT.md"

    if not heartbeat_md_path.exists():
        heartbeat_md_path.write_text(default_heartbeat_content())

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

def load_config() -> dict[str, Any] | None:
    """Load the global config.yaml. Returns None if it doesn't exist."""
    path = config_path()
    if not path.exists():
        return None
    with open(path) as file:
        return yaml.safe_load(file)

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

def save_bot_config(name: str, bot_config: dict[str, Any]) -> None:
    """Save a bot's bot.yaml and generate CLAUDE.md."""
    directory = bot_directory(name)
    directory.mkdir(parents=True, exist_ok=True)

    sessions_directory = directory / "sessions"
    sessions_directory.mkdir(exist_ok=True)

    with open(directory / "bot.yaml", "w") as file:
        yaml.dump(bot_config, file, default_flow_style=False, allow_unicode=True)

    # Lazy import to avoid circular dependency with skill module
    from cclaw.skill import compose_claude_md

    claude_md_content = compose_claude_md(
        bot_name=name,
        personality=bot_config.get("personality", ""),
        description=bot_config.get("description", ""),
        skill_names=bot_config.get("skills", []),
        bot_path=directory,
    )
    with open(directory / "CLAUDE.md", "w") as file:
        file.write(claude_md_content)

def generate_claude_md(name: str, personality: str, description: str) -> str:
    """Generate CLAUDE.md content for a bot."""
    return f"""# {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.).
"""

def default_config() -> dict[str, Any]:
    """Return a default config structure."""
    return {
        "bots": [],
        "settings": {
            "log_level": "INFO",
            "command_timeout": 300,
        },
    }

def add_bot_to_config(name: str) -> None:
    """Add a bot entry to the global config."""
    config = load_config() or default_config()
    if "bots" not in config:
        config["bots"] = []

    existing = [b for b in config["bots"] if b["name"] == name]
    if not existing:
        config["bots"].append({"name": name, "path": str(bot_directory(name))})

    save_config(config)

def remove_bot_from_config(name: str) -> None:
    """Remove a bot entry from the global config."""
    config = load_config()
    if not config or "bots" not in config:
        return
    config["bots"] = [b for b in config["bots"] if b["name"] != name]
    save_config(config)

def load_cron_config(name: str) -> dict[str, Any]:
    """Load a bot's cron.yaml. Returns empty config if not found."""
    path = bot_directory(name) / "cron.yaml"
    if not path.exists():
        return {"jobs": []}
    with open(path) as file:
        data = yaml.safe_load(file)
    if not data or "jobs" not in data:
        return {"jobs": []}
    return data

def save_cron_config(name: str, config: dict[str, Any]) -> None:
    """Save a bot's cron.yaml."""
    path = bot_directory(name) / "cron.yaml"
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w") as file:
        yaml.dump(config, file, default_flow_style=False, allow_unicode=True)

def bot_exists(name: str) -> bool:
    """Check if a bot with this name already exists."""
    config = load_config()
    if not config or "bots" not in config:
        return False
    return any(b["name"] == name for b in config["bots"])

def resolve_job_timezone(job: dict[str, Any]) -> ZoneInfo | timezone:
    """Resolve the timezone for a cron job.

    Returns ZoneInfo for named timezones (e.g., 'Asia/Seoul'),
    or timezone.utc as default.
    """
    timezone_name = job.get("timezone")
    if timezone_name:
        try:
            return ZoneInfo(timezone_name)
        except (KeyError, ValueError):
            logger.warning("Invalid timezone '%s', falling back to UTC", timezone_name)
    return timezone.utc

def load_cron_config(bot_name: str) -> dict[str, Any]:
    """Load a bot's cron.yaml. Returns empty config if it doesn't exist."""
    path = cron_config_path(bot_name)
    if not path.exists():
        return {"jobs": []}
    with open(path) as file:
        data = yaml.safe_load(file)
    if not data or "jobs" not in data:
        return {"jobs": []}
    return data

def save_cron_config(bot_name: str, config: dict[str, Any]) -> None:
    """Save a bot's cron.yaml."""
    path = cron_config_path(bot_name)
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w") as file:
        yaml.dump(config, file, default_flow_style=False, allow_unicode=True)

def get_cron_job(bot_name: str, job_name: str) -> dict[str, Any] | None:
    """Get a specific cron job by name."""
    for job in list_cron_jobs(bot_name):
        if job.get("name") == job_name:
            return job
    return None

def add_cron_job(bot_name: str, job: dict[str, Any]) -> None:
    """Add a cron job to a bot's configuration.

    If the job has an 'at' field with a relative duration (e.g., '10m', '2h'),
    it is converted to an absolute ISO datetime at add time so that the
    scheduler can correctly detect when the time has passed.
    """
    config = load_cron_config(bot_name)
    existing_names = {j["name"] for j in config["jobs"]}
    if job["name"] in existing_names:
        raise ValueError(f"Job '{job['name']}' already exists")

    # Convert relative duration to absolute ISO datetime
    if "at" in job:
        at_value = job["at"]
        duration_match = re.match(r"^(\d+)([mhd])$", str(at_value).strip())
        if duration_match:
            at_time = parse_one_shot_time(at_value)
            if at_time:
                job["at"] = at_time.isoformat()
                logger.info(
                    "Converted relative 'at' value '%s' to absolute '%s'",
                    at_value,
  

def remove_cron_job(bot_name: str, job_name: str) -> bool:
    """Remove a cron job by name. Returns True if found and removed."""
    config = load_cron_config(bot_name)
    original_count = len(config["jobs"])
    config["jobs"] = [j for j in config["jobs"] if j.get("name") != job_name]
    if len(config["jobs"]) == original_count:
        return False
    save_cron_config(bot_name, config)
    return True

def enable_cron_job(bot_name: str, job_name: str) -> bool:
    """Enable a cron job. Returns True if found."""
    config = load_cron_config(bot_name)
    for job in config["jobs"]:
        if job.get("name") == job_name:
            job["enabled"] = True
            save_cron_config(bot_name, config)
            return True
    return False

def disable_cron_job(bot_name: str, job_name: str) -> bool:
    """Disable a cron job. Returns True if found."""
    config = load_cron_config(bot_name)
    for job in config["jobs"]:
        if job.get("name") == job_name:
            job["enabled"] = False
            save_cron_config(bot_name, config)
            return True
    return False

def parse_one_shot_time(at_value: str) -> datetime | None:
    """Parse a one-shot time value.

    Supports:
    - ISO 8601 datetime: "2026-02-20T15:00:00"
    - Duration shorthand: "30m", "2h", "1d"

    Returns a timezone-aware datetime or None if parsing fails.
    """
    # Try duration shorthand first
    duration_match = re.match(r"^(\d+)([mhd])$", at_value.strip())
    if duration_match:
        amount = int(duration_match.group(1))
        unit = duration_match.group(2)
        now = datetime.now(timezone.utc)
        if unit == "m":
            return now + timedelta(minutes=amount)
        elif unit == "h":
            return now + timedelta(hours=amount)
        elif unit == "d":
            return now + timedelta(days=amount)

    # Try ISO 8601 datetime
    try:
        parsed = datetime.fromisoformat(at_value)
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        return parsed
    except ValueError:
        return None

def next_run_time(job: dict[str, Any]) -> datetime | None:
    """Calculate the next run time for a job.

    Returns None if the job has no valid schedule.
    The returned datetime is timezone-aware in the job's configured timezone.
    """
    if "at" in job:
        at_time = parse_one_shot_time(job["at"])
        return at_time

    schedule = job.get("schedule")
    if not schedule or not validate_cron_schedule(schedule):
        return None

    job_timezone = resolve_job_timezone(job)
    now = datetime.now(job_timezone)
    cron = croniter(schedule, now)
    return cron.get_next(datetime).replace(tzinfo=job_timezone)

def cron_session_directory(bot_name: str, job_name: str) -> Path:
    """Return the cron session directory for a job, ensuring it exists."""
    directory = bot_directory(bot_name) / "cron_sessions" / job_name
    directory.mkdir(parents=True, exist_ok=True)

    # Copy bot's CLAUDE.md if not present
    bot_claude_md = bot_directory(bot_name) / "CLAUDE.md"
    session_claude_md = directory / "CLAUDE.md"
    if not session_claude_md.exists() and bot_claude_md.exists():
        import shutil

        shutil.copy2(bot_claude_md, session_claude_md)

    return directory

async def execute_cron_job(
    bot_name: str,
    job: dict[str, Any],
    bot_config: dict[str, Any],
    send_message_callback: Any,
) -> None:
    """Execute a single cron job and send results via callback.

    Args:
        bot_name: Name of the bot.
        job: The cron job configuration dict.
        bot_config: The bot's configuration.
        send_message_callback: Async callable(user_id, text) to send messages.
    """
    from cclaw.claude_runner import run_claude
    from cclaw.config import DEFAULT_MODEL
    from cclaw.utils import markdown_to_telegram_html, split_message

    job_name = job["name"]
    raw_message = job.get("message", "")
    model = job.get("model") or bot_config.get("model", DEFAULT_MODEL)
    job_skills = job.get("skills") or bot_config.get("skills", [])
    command_timeout = bot_config.get("command_timeout", 300)

    working_directory = str(cron_session_directory(bot_name, job_name))

    from cclaw.session import load_bot_memory, load_global_memor

async def run_cron_scheduler(
    bot_name: str,
    bot_config: dict[str, Any],
    application: Any,
    stop_event: asyncio.Event,
) -> None:
    """Run the cron scheduler loop for a bot.

    Checks every 30 seconds for jobs that need to run.
    Uses application.bot.send_message as the message callback.

    Args:
        bot_name: Name of the bot.
        bot_config: The bot's configuration.
        application: The telegram Application instance.
        stop_event: Event that signals shutdown.
    """
    last_run_times: dict[str, datetime] = {}

    logger.info("Cron scheduler started for bot '%s'", bot_name)

    while not stop_event.is_set():
        try:
            jobs = list_cron_jobs(bot_name)

            for job in jobs:
                job_name = job.get("name")
                if not job_name:
                    continue

                if not job.get("enabled", True):
                    continue

                should_run = False

                if "at" in job:

def get_heartbeat_config(bot_name: str) -> dict[str, Any]:
    """Read the heartbeat section from bot.yaml."""
    bot_config = load_bot_config(bot_name)
    if not bot_config:
        return dict(DEFAULT_HEARTBEAT_CONFIG)
    return bot_config.get("heartbeat", dict(DEFAULT_HEARTBEAT_CONFIG))

def save_heartbeat_config(bot_name: str, heartbeat_config: dict[str, Any]) -> None:
    """Save the heartbeat section to bot.yaml."""
    bot_config = load_bot_config(bot_name)
    if not bot_config:
        return
    bot_config["heartbeat"] = heartbeat_config
    save_bot_config(bot_name, bot_config)

def enable_heartbeat(bot_name: str) -> bool:
    """Enable heartbeat for a bot.

    Creates default HEARTBEAT.md if missing. Returns True if successful.
    """
    bot_config = load_bot_config(bot_name)
    if not bot_config:
        return False
    heartbeat_config = bot_config.get("heartbeat", dict(DEFAULT_HEARTBEAT_CONFIG))
    heartbeat_config["enabled"] = True
    bot_config["heartbeat"] = heartbeat_config
    save_bot_config(bot_name, bot_config)

    # Create default HEARTBEAT.md if it doesn't exist
    session_directory = heartbeat_session_directory(bot_name)
    heartbeat_md_path = session_directory / "HEARTBEAT.md"
    if not heartbeat_md_path.exists():
        heartbeat_md_path.write_text(default_heartbeat_content())

    return True

def disable_heartbeat(bot_name: str) -> bool:
    """Disable heartbeat for a bot. Returns True if successful."""
    bot_config = load_bot_config(bot_name)
    if not bot_config:
        return False
    heartbeat_config = bot_config.get("heartbeat", dict(DEFAULT_HEARTBEAT_CONFIG))
    heartbeat_config["enabled"] = False
    bot_config["heartbeat"] = heartbeat_config
    save_bot_config(bot_name, bot_config)
    return True

def is_within_active_hours(active_hours: dict[str, str], now: datetime | None = None) -> bool:
    """Check if the current time is within the active hours range.

    Uses local time. Supports overnight ranges (e.g. start=22:00, end=06:00).

    Args:
        active_hours: Dict with 'start' and 'end' keys in HH:MM format.
        now: Optional datetime for testing. Uses local time if None.
    """
    if now is None:
        now = datetime.now()

    start_str = active_hours.get("start", "00:00")
    end_str = active_hours.get("end", "23:59")

    start_hour, start_minute = map(int, start_str.split(":"))
    end_hour, end_minute = map(int, end_str.split(":"))

    current_minutes = now.hour * 60 + now.minute
    start_minutes = start_hour * 60 + start_minute
    end_minutes = end_hour * 60 + end_minute

    if start_minutes <= end_minutes:
        # Normal range (e.g. 07:00 - 23:00)
        return start_minutes <= current_minutes <= end_minutes
    else:
        # Overnight range (e.g.

def heartbeat_session_directory(bot_name: str) -> Path:
    """Return the heartbeat session directory, ensuring it exists.

    Creates the directory and copies bot's CLAUDE.md if not present.
    """
    directory = bot_directory(bot_name) / "heartbeat_sessions"
    directory.mkdir(parents=True, exist_ok=True)

    # Create workspace subdirectory
    workspace = directory / "workspace"
    workspace.mkdir(exist_ok=True)

    # Copy bot's CLAUDE.md if not present
    bot_claude_md = bot_directory(bot_name) / "CLAUDE.md"
    session_claude_md = directory / "CLAUDE.md"
    if not session_claude_md.exists() and bot_claude_md.exists():
        shutil.copy2(bot_claude_md, session_claude_md)

    return directory

def default_heartbeat_content() -> str:
    """Return the default HEARTBEAT.md template."""
    return """# Heartbeat Checklist

실행할 항목이 없으면 HEARTBEAT_OK만 출력하세요.

## 확인 항목

- workspace/에 미완성 작업이 있는지 확인
- 마지막 대화 이후 8시간 이상 지났으면 가벼운 안부

## 규칙

- 긴급하지 않으면 메시지 보내지 말 것
- HEARTBEAT_OK는 반드시 응답 마지막에 포함할 것
"""

def load_heartbeat_markdown(bot_name: str) -> str:
    """Load the HEARTBEAT.md content for a bot."""
    directory = heartbeat_session_directory(bot_name)
    heartbeat_md_path = directory / "HEARTBEAT.md"
    if not heartbeat_md_path.exists():
        return ""
    return heartbeat_md_path.read_text()

async def execute_heartbeat(
    bot_name: str,
    bot_config: dict[str, Any],
    send_message_callback: Callable,
) -> None:
    """Execute a heartbeat check and notify users if needed.

    1. Prepare heartbeat session directory
    2. Read HEARTBEAT.md and compose prompt
    3. Run Claude with the prompt
    4. If response contains HEARTBEAT_OK → log only, no message
    5. If response does NOT contain HEARTBEAT_OK → send to allowed_users

    Args:
        bot_name: Name of the bot.
        bot_config: The bot's configuration.
        send_message_callback: Async callable(chat_id, text, ...) to send messages.
    """
    from cclaw.claude_runner import run_claude
    from cclaw.config import DEFAULT_MODEL
    from cclaw.utils import markdown_to_telegram_html, split_message

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

    working_directory = str(heartbeat_session

async def run_heartbeat_scheduler(
    bot_name: str,
    bot_config: dict[str, Any],
    application: Any,
    stop_event: asyncio.Event,
) -> None:
    """Run the heartbeat scheduler loop for a bot.

    Repeats at interval_minutes intervals:
    1. Check stop_event → exit if set
    2. Check is_within_active_hours() → skip if outside range
    3. Call execute_heartbeat()
    4. Sleep for interval_minutes

    Args:
        bot_name: Name of the bot.
        bot_config: The bot's configuration.
        application: The telegram Application instance.
        stop_event: Event that signals shutdown.
    """
    heartbeat_config = bot_config.get("heartbeat", {})
    interval_minutes = heartbeat_config.get("interval_minutes", 30)
    active_hours = heartbeat_config.get("active_hours", {"start": "07:00", "end": "23:00"})
    interval_seconds = interval_minutes * 60

    logger.info(
        "Heartbeat scheduler started for bot '%s' (interval: %dm, active: %s-%s)",
        bot_name,
      

def check_claude_code() -> EnvironmentCheckResult:
    """Check if Claude Code CLI is installed."""
    path = shutil.which("claude")
    if not path:
        return EnvironmentCheckResult(
            name="Claude Code",
            available=False,
            version="",
            message="Claude Code is not installed.\n\n"
            "  Install:\n"
            "    npm install -g @anthropic-ai/claude-code\n\n"
            "  Then run again: cclaw init",
        )
    try:
        result = subprocess.run(["claude", "--version"], capture_output=True, text=True, timeout=10)
        version = result.stdout.strip() or result.stderr.strip()
    except (subprocess.TimeoutExpired, OSError):
        version = "unknown"

    return EnvironmentCheckResult(name="Claude Code", available=True, version=version, message="")

def check_node() -> EnvironmentCheckResult:
    """Check if Node.js is installed."""
    path = shutil.which("node")
    if not path:
        return EnvironmentCheckResult(
            name="Node.js",
            available=False,
            version="",
            message="Node.js is not installed. Required for Claude Code.",
        )
    try:
        result = subprocess.run(["node", "--version"], capture_output=True, text=True, timeout=10)
        version = result.stdout.strip()
    except (subprocess.TimeoutExpired, OSError):
        version = "unknown"

    return EnvironmentCheckResult(name="Node.js", available=True, version=version, message="")

def check_python() -> EnvironmentCheckResult:
    """Check Python version."""
    import sys

    version = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
    return EnvironmentCheckResult(name="Python", available=True, version=version, message="")

def display_environment_checks(checks: list[EnvironmentCheckResult]) -> bool:
    """Display environment check results. Returns True if all passed."""
    console.print("\nChecking environment...")
    all_passed = True
    for check in checks:
        if check.available:
            console.print(f"  [green]OK[/green] {check.name} {check.version}")
        else:
            console.print(f"  [red]FAIL[/red] {check.name}")
            if check.message:
                console.print(f"\n  {check.message}")
            all_passed = False
    return all_passed

async def validate_telegram_token(token: str) -> dict | None:
    """Validate a Telegram bot token. Returns bot info dict or None."""
    from telegram import Bot

    try:
        bot = Bot(token=token)
        bot_info = await bot.get_me()
        return {
            "username": f"@{bot_info.username}",
            "botname": bot_info.first_name,
        }
    except Exception:
        return None

def prompt_telegram_token() -> tuple[str, dict]:
    """Prompt user for Telegram bot token with retry. Returns (token, bot_info)."""
    from cclaw.utils import prompt_input

    console.print("\n[bold]Connecting Telegram bot.[/bold]")
    console.print()
    console.print("  1. Send a DM to @BotFather on Telegram.")
    console.print("  2. Create a bot with the /newbot command.")
    console.print("  3. Enter the issued token below.")
    console.print()

    for attempt in range(MAXIMUM_TOKEN_RETRY):
        token = prompt_input("Bot Token:")
        console.print("Verifying token...")

        bot_info = asyncio.run(validate_telegram_token(token))
        if bot_info:
            console.print(
                f"[green]OK[/green] Bot verified: {bot_info['username']} ({bot_info['botname']})"
            )
            return token, bot_info

        remaining = MAXIMUM_TOKEN_RETRY - attempt - 1
        if remaining > 0:
            console.print(f"[red]Invalid token. {remaining} attem

def prompt_bot_profile() -> dict:
    """Prompt user for bot profile information. Returns profile dict."""
    from cclaw.utils import prompt_input, prompt_multiline

    console.print("\n[bold]Setting up bot profile.[/bold]\n")

    while True:
        name = prompt_input("Bot name (English, used as directory name):")
        name = name.strip().lower().replace(" ", "-")
        if not name.isascii() or not name.replace("-", "").isalnum():
            console.print("[red]Use only English letters, numbers, and hyphens.[/red]")
            continue
        if bot_exists(name):
            console.print(f"[red]Bot '{name}' already exists.[/red]")
            continue
        break

    personality = prompt_multiline("Bot personality:")
    description = prompt_multiline("Bot role/tasks:")

    return {
        "name": name,
        "personality": personality,
        "description": description,
    }

def _restart_daemon() -> None:
    """Restart the cclaw daemon to pick up new bot."""
    from cclaw.bot_manager import start_bots, stop_bots

    console.print("\n[yellow]Restarting daemon to register new bot...[/yellow]")
    try:
        stop_bots()
        start_bots(daemon=True)
    except Exception as error:
        console.print(f"[red]Failed to restart daemon: {error}[/red]")
        console.print("  Restart manually: [bold]cclaw stop && cclaw start --daemon[/bold]")

def create_bot(token: str, bot_info: dict, profile: dict) -> None:
    """Create bot configuration files."""
    bot_config = {
        "telegram_token": token,
        "telegram_username": bot_info["username"],
        "telegram_botname": bot_info["botname"],
        "description": profile["description"],
        "personality": profile["personality"],
        "allowed_users": [],
        "claude_args": [],
        "streaming": False,
        "heartbeat": {
            "enabled": False,
            "interval_minutes": 30,
            "active_hours": {
                "start": "07:00",
                "end": "23:00",
            },
        },
    }

    save_bot_config(profile["name"], bot_config)
    add_bot_to_config(profile["name"])

    home = cclaw_home()
    console.print()
    console.print(
        Panel(
            f"[green]OK[/green] {profile['name']} created!\n\n"
            f"  Name:      {profile['name']}\n"
            f"  Personality: {profile['personality']}\n"
       

def run_onboarding() -> None:
    """Run the full onboarding flow."""
    console.print("[bold]Starting cclaw initial setup.[/bold]")

    checks = run_environment_checks()
    if not display_environment_checks(checks):
        raise typer.Exit(1)

    console.print("\n[green]Environment check passed![/green]")

    token, bot_info = prompt_telegram_token()
    profile = prompt_bot_profile()
    create_bot(token, bot_info, profile)

def run_doctor() -> None:
    """Run environment and configuration diagnostics."""
    console.print("[bold]cclaw doctor[/bold]\n")

    checks = run_environment_checks()
    display_environment_checks(checks)

    console.print()

    config = load_config()
    if config is None:
        console.print("[yellow]No config.yaml found. Run 'cclaw init' first.[/yellow]")
        return

    console.print("[green]OK[/green] config.yaml found")
    console.print(f"  Log level: {config.get('settings', {}).get('log_level', 'N/A')}")

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

    console.print(f"\n[bold]Bots ({len(bots)}):[/bold]")

    for bot_entry in bots:
        name = bot_entry["name"]
        bot_config = load_bot_config(name)
        if not bot_config:
            console.print(f"  [red]FAIL[/red] {name}: bot.yaml missing")
            continue

        token = bot_config.get("telegram_token", "")
   

def collect_session_chat_ids(bot_path: Path) -> list[int]:
    """Collect chat IDs from existing session directories.

    Scans ``sessions/chat_<id>/`` directories and returns the list of chat IDs.
    Used as a fallback when ``allowed_users`` is empty but the bot needs to
    send proactive messages (cron results, heartbeat notifications).
    """
    sessions_directory = bot_path / "sessions"
    if not sessions_directory.exists():
        return []
    chat_ids: list[int] = []
    for child in sorted(sessions_directory.iterdir()):
        if child.is_dir() and child.name.startswith("chat_"):
            try:
                chat_id = int(child.name.removeprefix("chat_"))
                chat_ids.append(chat_id)
            except ValueError:
                continue
    return chat_ids

def ensure_session(bot_path: Path, chat_id: int) -> Path:
    """Ensure a session directory exists with required files.

    Creates the session directory, copies the bot's CLAUDE.md into it,
    and creates the workspace subdirectory.

    Returns the session directory path.
    """
    directory = session_directory(bot_path, chat_id)
    directory.mkdir(parents=True, exist_ok=True)

    workspace = directory / "workspace"
    workspace.mkdir(exist_ok=True)

    session_claude_md = directory / "CLAUDE.md"
    bot_claude_md = bot_path / "CLAUDE.md"

    if not session_claude_md.exists() and bot_claude_md.exists():
        shutil.copy2(bot_claude_md, session_claude_md)

    return directory

def _list_all_conversation_files(session_directory: Path) -> list[Path]:
    """List all conversation files (dated + legacy) in the session directory."""
    if not session_directory.exists():
        return []
    files = list(session_directory.glob(CONVERSATION_GLOB_PATTERN))
    legacy_file = session_directory / "conversation.md"
    if legacy_file.exists():
        files.append(legacy_file)
    return files

def conversation_status_summary(session_directory: Path) -> str:
    """Return a human-readable summary of conversation files in the session."""
    if not session_directory.exists():
        return "No conversation yet"

    conversation_files = list(session_directory.glob(CONVERSATION_GLOB_PATTERN))
    legacy_file = session_directory / "conversation.md"

    total_size = 0
    file_count = 0

    for file in conversation_files:
        total_size += file.stat().st_size
        file_count += 1

    if legacy_file.exists():
        total_size += legacy_file.stat().st_size
        file_count += 1

    if file_count == 0:
        return "No conversation yet"

    return f"{total_size:,} bytes ({file_count} files)"

def reset_session(bot_path: Path, chat_id: int) -> None:
    """Reset a session by deleting all conversation files (keep workspace)."""
    directory = session_directory(bot_path, chat_id)
    for conversation_file in _list_all_conversation_files(directory):
        conversation_file.unlink()
    clear_claude_session_id(directory)