Codecast

Codecast is a distributed system for controlling AI CLI tools on remote machines through chat bots. It supports Discord, Telegram, and Lark (Feishu), and works with Claude CLI, Codex (OpenAI), Gemini CLI, and OpenCode.

The system lets you start, manage, and interact with AI sessions on GPU servers, cloud VMs, or any SSH-accessible machine directly from your phone or desktop chat client.

Why Codecast?

When working with remote development servers -- GPU nodes behind firewalls, lab machines accessible only via jump hosts, cloud instances without a GUI -- you often need to run AI CLI tools in those environments. Codecast bridges the gap by letting you manage those sessions through familiar chat interfaces without opening a terminal.

Key Features

SSH Tunnel Management -- Automatic SSH connections with ProxyJump support, port forwarding, and connection health monitoring. The daemon only binds to 127.0.0.1, so all communication is secured through SSH tunnels.

Auto-Deployment -- The daemon is a single static Rust binary. It is automatically deployed to remote machines via SCP on first connection. No Node.js, npm, or manual setup is needed on the remote side.

Multi-CLI Support -- Start sessions using Claude CLI, Codex (OpenAI), Gemini CLI, or OpenCode. Choose the CLI per session with the --cli flag on /start.

Session Routing -- A SQLite-backed session registry maps chat channels to active AI sessions across multiple machines. Sessions can be detached, resumed, and destroyed independently.

Message Queuing -- When the AI is busy processing a request, additional messages are queued and processed in order. If the SSH connection drops mid-stream, responses are buffered and replayed on reconnect.

Streaming Responses -- Responses stream back in real-time via Server-Sent Events (SSE), with partial text updates rendered progressively in chat. Long responses are automatically split to fit platform message limits.

Tool Display Modes -- Three modes control how tool calls are displayed during a response: timer (shows elapsed time, sends all results at the end), append (shows each tool call progressively), and batch (accumulates tool calls into a summary at the end).

Interactive Questions -- When the AI uses AskUserQuestion, each platform presents the question with interactive controls: buttons on Discord, an inline keyboard on Telegram, and interactive cards on Lark.

File Forwarding -- When AI responses reference file paths, Codecast can automatically download matching files from the remote machine and send them to your chat.

Permission Modes -- Four modes control AI autonomy: auto (bypass all permissions), code (auto-accept edits, confirm bash), plan (read-only analysis), and ask (confirm everything). Switch modes at any time during a session.

Skills Sync -- Share CLAUDE.md and .claude/skills/ files across projects on remote machines. Skills are synced from a local directory to remote project paths on session creation, without overwriting existing files.

Web UI and TUI -- A browser-based Web UI and an interactive terminal UI (TUI) are available in addition to the chat bot interface.

Model Switching -- Switch the AI model mid-session with /model without restarting the session.

Supported Platforms

Project Structure

codecast/
├── src/
│   ├── head/                        # Head Node (Python)
│   │   ├── cli.py                   # CLI entry point
│   │   ├── main.py                  # Head node entry
│   │   ├── config.py                # Config loader
│   │   ├── engine.py                # Core command engine
│   │   ├── ssh_manager.py           # SSH connections & tunnels
│   │   ├── session_router.py        # SQLite session registry
│   │   ├── daemon_client.py         # JSON-RPC + SSE client
│   │   ├── message_formatter.py     # Message formatting
│   │   ├── file_forward.py          # File forwarding
│   │   ├── platform/                # Bot adapters
│   │   │   ├── protocol.py          # Platform adapter interface
│   │   │   ├── discord_adapter.py
│   │   │   ├── telegram_adapter.py
│   │   │   └── lark_adapter.py
│   │   ├── tui/                     # Terminal UI (Textual)
│   │   └── webui/                   # Web UI (aiohttp)
│   └── daemon/                      # Daemon (Rust)
│       ├── main.rs                  # Axum HTTP server
│       ├── server.rs                # JSON-RPC router, SSE streaming
│       ├── session_pool.rs          # CLI process management
│       ├── message_queue.rs         # Message buffering
│       ├── cli_adapter/             # Multi-CLI adapters
│       │   ├── claude.rs
│       │   ├── codex.rs
│       │   ├── gemini.rs
│       │   └── opencode.rs
│       └── types.rs                 # Type definitions
├── docs/                            # This documentation
└── tests/                           # Python tests (855+ tests)

Quick Links

• Getting Started -- Install and run Codecast
• Configuration Guide -- All config.yaml options
• Bot Command Reference -- Every chat command explained
• Architecture Overview -- Understand the two-tier design

Getting Started

This guide walks you through setting up Codecast from scratch. By the end you will have a bot running on Discord, Telegram, or Lark that lets you interact with AI CLI tools on a remote machine.

Prerequisites

Local Machine (Head Node)

• Python 3.10 or later
• pip (Python package manager)
• SSH keys configured for your remote machines (or password access)
• A bot token for at least one of: Discord, Telegram, or Lark

Remote Machines

• SSH access from your local machine
• At least one AI CLI installed and authenticated:
  • Claude CLI (claude in PATH)
  • Codex (codex in PATH)
  • Gemini CLI (gemini in PATH)
  • OpenCode (opencode in PATH)

No Node.js or npm is required on remote machines. The Codecast daemon is a single static Rust binary that is deployed automatically via SCP.

Bot Setup

Discord

1. Go to the Discord Developer Portal.
2. Create a new application, then under "Bot" create a bot and copy the token.
3. Enable the "Message Content Intent" under "Privileged Gateway Intents".
4. Under "OAuth2 > URL Generator", select scopes bot and applications.commands, with permissions: Send Messages, Manage Messages, Read Message History.
5. Use the generated URL to invite the bot to your server.

Telegram

1. Message @BotFather on Telegram.
2. Send /newbot and follow the prompts.
3. Copy the token BotFather provides.

Lark (Feishu)

1. Go to the Lark Open Platform and create an application.
2. Under "Permissions & Scopes", add: im:message, im:message:send_as_bot.
3. Under "Event Subscriptions", add the im.message.receive_v1 event.
4. Copy the App ID and App Secret.
5. Set up a webhook endpoint or use Lark's built-in bot messaging.

Installation

Option 1: Install from PyPI (recommended)

pip install codecast

This installs all dependencies and provides the codecast command.

Option 2: Install from source

git clone https://github.com/Chivier/codecast.git
cd codecast
pip install -e .

The pip install -e . command installs the package in editable mode and provides the codecast command.

Building the Daemon Manually (optional)

If daemon.auto_deploy is enabled (the default), the daemon binary is deployed to remote machines automatically on first connection. If you need to build it manually:

cargo build --release

The output binary is target/release/codecast-daemon. Copy it to ~/.codecast/daemon/codecast-daemon on your local machine and it will be deployed from there.

Installing the AI CLI on Remote Machines

Each remote machine needs at least one AI CLI installed. For Claude CLI:

# On the remote machine
npm install -g @anthropic-ai/claude-code
claude auth login

Verify the CLI works:

claude -p "Hello" --output-format stream-json

Configuration

1. Create the config file

The primary config location is ~/.codecast/config.yaml. Create the directory and copy the example:

mkdir -p ~/.codecast
cp /path/to/codecast/config.example.yaml ~/.codecast/config.yaml

Alternatively, you can place config.yaml in the current working directory as a development fallback.

2. Set environment variables

Export your bot tokens:

export DISCORD_TOKEN="your-discord-bot-token"
export TELEGRAM_TOKEN="your-telegram-bot-token"
export LARK_APP_ID="your-lark-app-id"
export LARK_APP_SECRET="your-lark-app-secret"

Config values support ${ENV_VAR} syntax, so tokens are never hardcoded in the file.

3. Configure your remote machines

Edit ~/.codecast/config.yaml to add your machines under peers::

peers:
  gpu-1:
    host: gpu1.example.com
    user: your-user
    daemon_port: 9100
    default_paths:
      - /home/your-user/project-a
      - /home/your-user/project-b

bot:
  discord:
    token: ${DISCORD_TOKEN}
    allowed_channels:
      - 1234567890123456789

default_mode: auto

See the Configuration Guide for all available options, including Telegram, Lark, ProxyJump, and file forwarding.

First-Time Setup with the TUI Wizard

If this is your first time using Codecast, the TUI (terminal UI) provides an interactive setup wizard:

codecast tui

The TUI lets you configure machines, test SSH connections, and start sessions without editing YAML directly.

Running Codecast

Start the head node:

codecast

With a specific config file:

codecast /path/to/config.yaml

You should see output like:

INFO: Discord bot configured
INFO: Telegram bot configured
INFO: Codecast started with 2 bot(s)
INFO: Peers: gpu-1
INFO: Default mode: auto

Starting Your First Session

1. Open Discord, Telegram, or Lark.

2. In an allowed channel or chat, use the /start command:

   /start gpu-1 /home/your-user/project-a
   

3. Codecast will:

   • Establish an SSH tunnel to gpu-1
   • Deploy the daemon binary if not already present (auto-deploy)
   • Start the daemon process on the remote machine
   • Sync skills files if configured
   • Create an AI session in the project directory

4. Send a message to interact:

   What files are in this project?
   

5. The response streams back in real-time.

To start a session with a specific CLI other than Claude:

/start gpu-1 /home/your-user/project --cli codex

Supported CLI types: claude, codex, gemini, opencode.

Stopping Codecast

Press Ctrl+C or send SIGTERM to the process. The head node will:

1. Stop all bots gracefully
2. Close the daemon client HTTP session
3. Close all SSH tunnels
4. Cancel pending tasks

Sessions on remote daemons are not destroyed on head node shutdown. Resume them later with /resume.

Troubleshooting

"Could not connect to machine" -- Check that the SSH host, user, and key are correct. Test with ssh user@host from your terminal. If the machine is behind a jump host, see the proxy_jump option in the Configuration Guide.

"Daemon not found after deploy" -- Check that ~/.codecast/daemon/ exists on the remote machine and the binary is executable. Check ~/.codecast/daemon.log on the remote machine for errors.

"claude: command not found" -- Claude CLI is not installed or not in PATH on the remote machine. The daemon inherits the SSH session's PATH; ensure the CLI is accessible via a normal SSH login.

Bot does not respond -- Confirm the bot token is valid and the channel or user ID is in the allowed list. For Discord, check that the Message Content Intent is enabled.

Session appears stuck -- Use /interrupt or /stop in the chat to interrupt the current AI operation. Use /status to check queue state.

Configuration Guide

Codecast reads its configuration from a YAML file. The file is searched in this order:

1. A path provided as a CLI argument: codecast /path/to/config.yaml
2. ~/.codecast/config.yaml (recommended location)
3. ./config.yaml in the current working directory (development fallback)

To get started, copy the example config:

mkdir -p ~/.codecast
cp /path/to/codecast/config.example.yaml ~/.codecast/config.yaml

Environment Variable Expansion

All string values support ${ENV_VAR} syntax. If the variable is not set, the expression is left as-is.

bot:
  discord:
    token: ${DISCORD_TOKEN}

Path values also support ~ expansion (e.g. ~/.ssh/id_rsa expands to /home/user/.ssh/id_rsa).

Passwords can be read from a file using the file: prefix:

password: file:~/.secrets/my-password

The file should contain only the password text (whitespace is trimmed).

peers

Defines the remote machines (peers) that Codecast can connect to. Each key is a machine ID used in commands like /start gpu-1 /path.

peers:
  gpu-1:
    host: gpu1.example.com
    user: your-ssh-user
    ssh_key: ~/.ssh/id_rsa
    port: 22
    proxy_jump: gateway
    password: file:~/.secrets/gpu1-password
    daemon_port: 9100
    default_paths:
      - /home/your-user/project-a
      - /home/your-user/project-b

Peer transport types

For ssh transport (the default), use these fields:

For http transport, use these fields:

ProxyJump example

For machines behind a bastion host:

peers:
  gateway:
    host: bastion.example.com
    user: admin

  gpu-2:
    host: gpu2.lab.internal
    user: researcher
    proxy_jump: gateway
    daemon_port: 9100
    default_paths:
      - /data/experiments

The gateway peer is used only as a jump host and does not need default_paths.

bot

Configures the bot connections. At least one bot must be configured for Codecast to start.

bot.discord

bot:
  discord:
    token: ${DISCORD_TOKEN}
    allowed_channels:
      - 123456789012345678
    admin_users:
      - 987654321098765432
    command_prefix: "/"

bot.telegram

bot:
  telegram:
    token: ${TELEGRAM_TOKEN}
    allowed_users:
      - 123456789
    allowed_chats:
      - -1001234567890
    admin_users:
      - 123456789

bot.lark

bot:
  lark:
    app_id: ${LARK_APP_ID}
    app_secret: ${LARK_APP_SECRET}
    allowed_chats:
      - "oc_abcdef1234567890"
    admin_users:
      - "ou_abcdef1234567890"
    use_cards: true

bot.webui

Enables the browser-based Web UI alongside the chat bots.

bot:
  webui:
    enabled: true
    port: 8080
    host: 127.0.0.1

default_mode

default_mode: auto

The default permission mode for new sessions. Can be changed per-session with /mode.

tool_batch_size

tool_batch_size: 15

The number of consecutive tool_use events (file reads, shell commands, etc.) that are compressed into a single summary message. Reduces chat noise during large operations. Default is 15.

skills

skills:
  shared_dir: ./skills
  sync_on_start: true

daemon

Controls how the Codecast daemon (a single static Rust binary) is deployed to and managed on remote machines.

daemon:
  install_dir: ~/.codecast/daemon
  auto_deploy: true
  log_file: ~/.codecast/daemon.log

No Node.js or npm is required on remote machines. The daemon is a self-contained binary with no external runtime dependencies.

file_pool

Controls how files uploaded to bot chats are staged for use in AI sessions.

file_pool:
  max_size: 1073741824
  pool_dir: ~/.codecast/file-pool
  remote_dir: /tmp/codecast/files
  allowed_types:
    - text/plain
    - text/markdown
    - application/pdf
    - image/*
    - video/*
    - audio/*

file_forward

Controls automatic forwarding of files from remote machines to chat when the AI's response contains matching file paths.

file_forward:
  enabled: true
  download_dir: ~/.codecast/downloads
  default_max_size: 5242880
  default_auto: false
  rules:
    - pattern: "*.png"
      max_size: 10485760
      auto: true
    - pattern: "*.log"
      max_size: 1048576
      auto: false

Each rule in rules has:

When auto is false for a matched file, the bot sends a prompt asking whether to forward it.

Complete Example

peers:
  gateway:
    host: bastion.university.edu
    user: admin

  gpu-1:
    host: gpu-node-1.internal
    user: researcher
    proxy_jump: gateway
    daemon_port: 9100
    default_paths:
      - /data/ml-project
      - /data/nlp-experiments

  cloud-dev:
    host: 203.0.113.42
    user: ubuntu
    ssh_key: ~/.ssh/cloud-key.pem
    daemon_port: 9100
    default_paths:
      - /home/ubuntu/webapp

bot:
  discord:
    token: ${DISCORD_TOKEN}
    allowed_channels:
      - 1234567890123456789
    admin_users:
      - 9876543210987654321

  telegram:
    token: ${TELEGRAM_TOKEN}
    allowed_users:
      - 987654321
    admin_users:
      - 987654321

  lark:
    app_id: ${LARK_APP_ID}
    app_secret: ${LARK_APP_SECRET}
    allowed_chats:
      - "oc_abcdef1234567890"

  webui:
    enabled: false
    port: 8080
    host: 127.0.0.1

default_mode: auto

tool_batch_size: 15

skills:
  shared_dir: ./skills
  sync_on_start: true

daemon:
  install_dir: ~/.codecast/daemon
  auto_deploy: true
  log_file: ~/.codecast/daemon.log

file_pool:
  max_size: 1073741824
  pool_dir: ~/.codecast/file-pool
  remote_dir: /tmp/codecast/files

file_forward:
  enabled: false
  download_dir: ~/.codecast/downloads
  default_max_size: 5242880
  default_auto: false
  rules: []

Bot Command Reference

This page documents all commands available across the Discord, Telegram, and Lark bots.

Command Summary

/start

Start a new AI session on a remote machine.

Usage:

/start <machine_id> <path> [--cli <type>]

Arguments:

Shorthand flags: --codex, --gemini, --opencode can be used instead of --cli <type>.

Examples:

/start gpu-1 /home/user/my-project
/start gpu-1 /home/user/my-project --cli codex
/start gpu-1 /home/user/my-project --gemini

What happens:

1. An SSH tunnel is established to the machine (if not already active).
2. The daemon is deployed and started if not already running.
3. Skills files are synced to the project directory if configured.
4. A new AI session is created on the daemon.
5. The session is registered in the local database.
6. A confirmation message shows the session name and current mode.

Session names are auto-assigned in adjective-noun format, for example bright-falcon or smooth-dove. You can rename a session with /rename.

Discord: Slash command with autocomplete for machine (from configured machines) and path (from default_paths in config).

/resume

Resume a previously detached session.

Usage:

/resume <session_name_or_id>

Arguments:

Examples:

/resume bright-falcon
/resume a1b2c3d4-e5f6-7890-abcd-ef1234567890

What happens:

1. The session is looked up in the local database by name or ID.
2. An SSH tunnel is established to the session's machine.
3. The daemon is notified to resume the session.
4. The session is re-registered as active on the current channel.
5. Future messages continue the conversation context.

/new

Start a new AI session in the same directory as the current session, automatically detaching the current one.

Usage:

/new

Equivalent to /exit followed by /start with the same machine, path, and CLI type. Useful for getting a clean context without re-entering connection details.

/clear

Destroy the current session and immediately start a fresh one in the same directory.

Usage:

/clear

Unlike /new, the old session is fully destroyed rather than detached.

/exit

Detach from the current session without destroying it.

Usage:

/exit

The AI process on the remote machine keeps running. Use /resume with the session name to reconnect later.

Example output:

Detached from session on gpu-1:/home/user/project
Use /resume bright-falcon to reconnect.

/stop and /interrupt

Interrupt the AI's current operation.

Usage:

/stop
/interrupt

Both commands are equivalent. They:

1. Send an interrupt signal to the running AI process.
2. Clear the message queue.
3. Leave the session active for future messages.

Output:

• If the AI was processing: "Interrupted current operation."
• If the AI was idle: "No active operation to interrupt."

/ls

List machines or sessions.

Usage:

/ls machine
/ls session [machine_id]

Examples:

/ls machine
/ls session
/ls session gpu-1

Machine listing output:

Machines:
  gpu-1 (gpu1.example.com) [online, daemon running]
    Paths: /home/user/project-a, /home/user/project-b
  gpu-2 (gpu2.lab.internal) [offline]

Session listing output:

Sessions:
  bright-falcon  gpu-1:/home/user/project  [bypass] active
  smooth-dove    gpu-1:/home/user/other    [code]   detached

/rm-session

Destroy a specific session by name or ID.

Usage:

/rm-session <name_or_id>

Examples:

/rm-session bright-falcon
/rm-session a1b2c3d4-e5f6-7890-abcd-ef1234567890

This kills the AI process for that session and marks it as destroyed in the database.

/rm

Destroy all sessions matching a machine and path.

Usage:

/rm <machine_id> <path>

Example:

/rm gpu-1 /home/user/project

All active and detached sessions on the given machine/path combination are destroyed.

/mode

Switch the permission mode for the current session.

Usage:

/mode <auto|code|plan|ask>

Example:

/mode plan

Discord: Dropdown choice with descriptions for each mode.

/model

Switch the AI model for the current session.

Usage:

/model <model_name>

Examples:

/model claude-sonnet-4-20250514
/model claude-opus-4-20250514

The model change takes effect for the next message sent to the session. Use /status to confirm the active model.

/tool-display

Switch how tool calls (file reads, shell commands, etc.) are displayed while the AI is working.

Usage:

/tool-display <timer|append|batch|buffer>

Example:

/tool-display timer

/rename

Rename the current session.

Usage:

/rename <new_name>

Arguments:

Example:

/rename fast-hawk

The new name is stored in the session registry and can be used with /resume.

/status

Show the current session's status and queue statistics.

Usage:

/status

Example output:

Session: bright-falcon
Machine: gpu-1
Path: /home/user/project
Mode: bypass
Status: active
CLI: claude
Model: claude-sonnet-4-20250514
Queue: 0 pending messages
Buffered: 0 responses

/health

Check daemon health on a remote machine.

Usage:

/health [machine_id]

If no machine is specified, checks the machine of the current session, or checks all connected machines.

Example output:

Daemon Health - gpu-1
Status: OK
Uptime: 2h 15m 30s
Sessions: 3 (idle: 2, busy: 1)

/monitor

Monitor session details and queue state on a remote machine.

Usage:

/monitor [machine_id]

Example output:

Monitor - gpu-1 (uptime: 2h 15m 30s, 2 session(s))

  bright-falcon  idle [bypass | claude-sonnet-4-20250514]
    Path: /home/user/project
    Client: connected | Queue: 0 pending, 0 buffered

  smooth-dove  busy [code | claude-sonnet-4-20250514]
    Path: /home/user/other
    Client: disconnected | Queue: 1 pending, 5 buffered

/add-machine

Add a new remote machine to the configuration.

Usage:

/add-machine <name> [host] [user]
/add-machine --from-ssh

Examples:

/add-machine gpu-3 10.0.1.52 alice
/add-machine gpu-3 --from-ssh

The --from-ssh option reads ~/.ssh/config and presents an interactive selection of hosts to import. The machine is persisted to config.yaml immediately. The daemon is deployed on first /start.

/remove-machine

Remove a machine from the configuration.

Usage:

/remove-machine <machine_id>

If active or detached sessions exist on the machine, you are asked to confirm. The machine entry is deleted from config.yaml.

/update

Pull the latest code and restart the Head Node. Admin only.

Usage:

/update

Runs git pull in the project directory, then replaces the running process. Requires your user ID in admin_users in the config.

/restart

Restart the Head Node without pulling new code. Admin only.

Usage:

/restart

Useful for picking up config changes or recovering from a degraded state. Requires your user ID in admin_users in the config.

/help

Show the list of available commands.

Usage:

/help

Sending Messages

After starting or resuming a session, any message sent in the channel that is not a recognized command is forwarded to the AI. If you type something that starts with / but is not a known bot command, it is also forwarded to the AI directly -- useful for passing slash commands to the AI CLI itself.

Responses stream back in real-time. While the AI is processing, a cursor indicator or timer shows progress. On Discord, a "bot is typing..." indicator and periodic status updates keep you informed during long operations.

If you send a message while the AI is still processing the previous one, the new message is queued and processed automatically in order.

Interactive Questions (AskUserQuestion)

When the AI uses the AskUserQuestion tool, Codecast presents the question with interactive controls rather than as plain text:

• Discord -- Buttons below the message. Click to select.
• Telegram -- An inline keyboard. Tap to select.
• Lark -- An interactive card. Tap to select.

For multiple-choice questions, each option appears as a separate button or key. Your selection is sent back to the AI as the response.

File Forwarding

When the AI response contains a file path that matches a configured forwarding rule, Codecast can automatically download the file from the remote machine and send it to your chat. This happens without any manual command.

File forwarding is configured in config.yaml under file_forward. See the Configuration Guide for setup details.

Platform Differences

Architecture Overview

Codecast uses a two-tier architecture consisting of a Head Node (local orchestrator) and one or more Daemons (remote agents). The Head Node handles user interaction and SSH connections. Each remote machine runs a Daemon that spawns CLI processes and streams results back.

System Diagram

┌─────────────────────────────────────────────────────────────────┐
│                        User Devices                             │
│                                                                 │
│   ┌──────────┐    ┌──────────┐    ┌──────────┐                  │
│   │ Discord  │    │ Telegram │    │   Lark   │                  │
│   │ Client   │    │ Client   │    │  Client  │                  │
│   └────┬─────┘    └────┬─────┘    └────┬─────┘                  │
└────────┼──────────────┼───────────────┼──────────────────────────┘
         │              │               │
         │  Platform APIs               │
         │              │               │
┌────────▼──────────────▼───────────────▼──────────────────────────┐
│                  HEAD NODE  (Python, asyncio)                    │
│                                                                  │
│   ┌──────────────────────────────────────────────────────────┐   │
│   │  PlatformAdapter protocol                                │   │
│   │  discord_adapter.py  telegram_adapter.py  lark_adapter.py│   │
│   └────────────────────────┬─────────────────────────────────┘   │
│                            │  set_input_handler / send_message   │
│                            ▼                                     │
│   ┌─────────────────────────────────────┐  ┌──────────────────┐  │
│   │  BotEngine  (engine.py)             │──│  SessionRouter   │  │
│   │  cmd_* handlers, _forward_message   │  │  (SQLite)        │  │
│   └──────────────┬──────────────────────┘  └──────────────────┘  │
│                  │                                               │
│   ┌──────────────▼──────────────┐  ┌──────────────────────────┐  │
│   │  DaemonClient               │  │  SSHManager              │  │
│   │  JSON-RPC + SSE client      │  │  asyncssh tunnels        │  │
│   └──────────────┬──────────────┘  └────────────┬─────────────┘  │
└──────────────────┼──────────────────────────────┼────────────────┘
                   │                              │
                   │  JSON-RPC over SSH tunnel    │  SSH port forwarding
                   │                              │
┌──────────────────▼──────────────────────────────▼────────────────┐
│                  DAEMON  (Rust, tokio)           REMOTE MACHINE  │
│                                                                  │
│   ┌─────────────────────────────────────┐                        │
│   │  Axum RPC Server  (server.rs)       │ ◄── 127.0.0.1:9100     │
│   │  POST /rpc  (JSON + SSE)            │                        │
│   └──────────────┬──────────────────────┘                        │
│                  │                                               │
│   ┌──────────────▼──────────────┐  ┌──────────────────────────┐  │
│   │  SessionPool                │──│  MessageQueue            │  │
│   │  (session_pool.rs)          │  │  (message_queue.rs)      │  │
│   └──────────────┬──────────────┘  └──────────────────────────┘  │
│                  │  spawn per message                            │
│                  ▼                                               │
│   ┌─────────────────────────────────────┐                        │
│   │  CliAdapter trait  (cli_adapter/)   │                        │
│   │  claude.rs / codex.rs / gemini.rs   │                        │
│   │  opencode.rs                        │                        │
│   └──────────────┬──────────────────────┘                        │
│                  │  spawn subprocess                             │
│                  ▼                                               │
│   ┌─────────────────────────────────────┐                        │
│   │  Claude/Codex/Gemini/OpenCode      │                        │
│   │  non-interactive stream-json CLI   │                        │
│   └─────────────────────────────────────┘                        │
└──────────────────────────────────────────────────────────────────┘

Data Flow

A typical user interaction follows this path:

1. User sends a message or command via Discord, Telegram, or Lark.
2. The PlatformAdapter receives the event and calls the registered InputHandler callback on the BotEngine.
3. BotEngine.handle_input() routes the input: commands go to cmd_* handlers; regular messages go to _forward_message().
4. For message forwarding, the SessionRouter resolves the active session for this channel.
5. SSHManager.ensure_tunnel() establishes (or reuses) an SSH port-forwarding tunnel to the remote machine.
6. DaemonClient.send_message() sends a session.send JSON-RPC request over the tunnel and returns an async SSE event iterator.
7. The Daemon receives the request, selects a CliAdapter for the session's CLI type, and spawns a subprocess (for Claude, e.g. claude -p <message> --output-format stream-json --resume <sdkSessionId>).
8. The CLI process writes JSON-lines to stdout. The daemon parses each line via CliAdapter.parse_output_line() and converts it to a StreamEvent.
9. Each StreamEvent is serialized and sent back to the Head Node as an SSE data: frame.
10. BotEngine._forward_message() handles each event: accumulating partial deltas for streaming display, forwarding tool_use notifications, and capturing the SDK session ID from the result event.
11. When Claude finishes (emits a result event), the SDK session ID is stored in the SessionRouter for future --resume calls.

Key Design Decisions

Per-Message Spawn

The daemon spawns a fresh CLI process for each user message rather than keeping a long-running process with stdin open. For Claude, the command pattern is:

claude -p "user message" --output-format stream-json --verbose \
       [--resume <sdkSessionId>] [--dangerously-skip-permissions]

The --resume flag passes the SDK session ID from the previous result event, maintaining conversation continuity across process boundaries.

Benefits:

• Clean process state and memory for every message
• No zombie process management
• Natural recovery from crashes: just spawn a new process on the next send
• The pattern generalizes across all CLI backends (Claude, Codex, Gemini, OpenCode)

SSH Tunnels for Security

The daemon binds exclusively to 127.0.0.1. It is never network-reachable. All Head Node access passes through SSH port forwarding:

localhost:1xxxx  ──SSH tunnel──▶  remote:127.0.0.1:9100

This means:

• No firewall changes are needed on remote machines
• SSH handles authentication and encryption
• ProxyJump chains are supported for machines behind bastion hosts
• Localhost machines skip SSH entirely (auto-detected)

BotEngine + PlatformAdapter Composition

The Head Node uses composition rather than inheritance. BotEngine holds a PlatformAdapter instance and contains all command and streaming logic. Each platform (Discord, Telegram, Lark) implements the PlatformAdapter protocol independently, with no shared base class.

This means:

• Platform adapters are independently testable
• New platforms can be added without touching BotEngine
• The engine can be driven by a test adapter for integration tests

CliAdapter Trait for Multi-CLI Support

The daemon uses a CliAdapter trait to abstract over different CLI backends. A fresh adapter instance is created per run_cli_process() call via create_adapter(). Each adapter implements:

• build_command() and build_resume_command() for constructing the subprocess invocation
• parse_output_line() for parsing JSON-lines output into StreamEvent values
• instructions_file() and skills_dir() for skill sync

Currently supported CLI types: claude, codex, gemini, opencode.

SQLite for Session State

The Head Node uses SQLite (sessions.db) to persist session mappings between chat channels and remote daemon sessions. This ensures:

• Sessions survive Head Node restarts
• Multiple platform adapters (Discord, Telegram, Lark) share the same registry
• Session history enables /resume after detach
• The session_log table records detached sessions with their SDK session IDs

SSE for Streaming Responses

The session.send RPC method responds with an SSE stream (Content-Type: text/event-stream) instead of a single JSON body. This enables:

• Real-time streaming of Claude's output as it is generated
• Progressive rendering in chat (partial text updates with a ▌ cursor indicator)
• Keepalive pings every 30 seconds to prevent idle tunnel timeouts
• Graceful buffering of events when the client disconnects mid-stream

Component Responsibilities

Head Node Overview

The Head Node is the local orchestrator component of Codecast. It runs on your local machine (or a control server) and manages all user-facing interactions, SSH connections, session state, and communication with remote daemons.

Technology Stack

• Language: Python 3.10+
• SSH: asyncssh for async SSH connections and tunnels
• HTTP Client: aiohttp for JSON-RPC and SSE streaming
• Discord: discord.py v2 with slash commands
• Telegram: python-telegram-bot v20+ with async handlers
• Lark/Feishu: lark-oapi SDK
• Database: SQLite via Python's built-in sqlite3 module
• Config: YAML via ruamel.yaml (comment-preserving round-trip editing)

Module Map

src/head/
├── cli.py                   # CLI entry point: argparse, subcommand dispatch
├── main.py                  # Head node entry: loads config, starts adapters, shutdown
├── config.py                # Config dataclasses, YAML loader, env var expansion
├── engine.py                # BotEngine: all command logic and message forwarding
├── ssh_manager.py           # SSH connections, tunnels, daemon deployment
├── session_router.py        # SQLite-backed session registry
├── daemon_client.py         # JSON-RPC + SSE client for daemon communication
├── message_formatter.py     # Output formatting, message splitting, tool batching
├── file_forward.py          # File forwarding: detect and forward file paths to users
├── platform/
│   ├── protocol.py          # PlatformAdapter protocol, MessageHandle, FileAttachment
│   ├── discord_adapter.py   # Discord PlatformAdapter (discord.py v2)
│   ├── telegram_adapter.py  # Telegram PlatformAdapter (python-telegram-bot v20+)
│   ├── lark_adapter.py      # Lark/Feishu PlatformAdapter
│   ├── format_utils.py      # markdown_to_telegram_html() and other format helpers
│   └── __init__.py
├── tui/                     # Interactive TUI (Textual)
└── webui/                   # Web UI (aiohttp)

Module Dependencies

main.py
  ├── config.py              (load_config, Config)
  ├── ssh_manager.py         (SSHManager)
  ├── session_router.py      (SessionRouter)
  ├── daemon_client.py       (DaemonClient)
  ├── platform/discord_adapter.py   (DiscordAdapter)
  ├── platform/telegram_adapter.py  (TelegramAdapter)
  └── platform/lark_adapter.py      (LarkAdapter)

engine.py  (BotEngine)
  ├── platform/protocol.py   (PlatformAdapter, MessageHandle, FileAttachment)
  ├── ssh_manager.py         (SSHManager)
  ├── session_router.py      (SessionRouter)
  ├── daemon_client.py       (DaemonClient)
  ├── message_formatter.py   (formatting functions)
  └── file_forward.py        (FileForwardMatcher)

platform/discord_adapter.py  (DiscordAdapter)
  ├── platform/protocol.py   (PlatformAdapter, MessageHandle, FileAttachment)
  └── message_formatter.py   (split_message, format_error, display_mode)

platform/telegram_adapter.py  (TelegramAdapter)
  ├── platform/protocol.py   (PlatformAdapter, MessageHandle, FileAttachment)
  ├── message_formatter.py   (split_message)
  └── platform/format_utils.py (markdown_to_telegram_html)

ssh_manager.py
  └── config.py              (Config, PeerConfig)

session_router.py
  └── (standalone, uses sqlite3)

daemon_client.py
  └── (standalone, uses aiohttp)

message_formatter.py
  └── (standalone, no dependencies)

file_forward.py
  └── (standalone)

Architecture: BotEngine + PlatformAdapter

The Head Node uses a composition pattern rather than inheritance. BotEngine (in engine.py) contains all command and streaming logic and holds a PlatformAdapter instance for platform-specific I/O.

Each platform adapter (Discord, Telegram, Lark) implements the PlatformAdapter protocol defined in platform/protocol.py. The protocol includes methods for:

• Sending and editing messages (send_message, edit_message, delete_message)
• File operations (download_file, send_file)
• Interaction state (start_typing, stop_typing)
• Interactive questions (send_question) for AskUserQuestion handling
• Lifecycle (start, stop)

When a user message arrives, the platform adapter calls the registered InputHandler callback on the BotEngine, passing the channel ID, text, optional user ID, and any file attachments.

Lifecycle

1. Startup (main.py): Load config, create shared infrastructure instances (SSHManager, SessionRouter, DaemonClient), create one BotEngine per platform adapter, start adapters.
2. Command handling: User sends /start gpu-1 /path via Discord/Telegram/Lark. The adapter calls engine.handle_input(), which routes to cmd_start(). This calls SSHManager to set up the tunnel and DaemonClient to create a session.
3. Message forwarding: User sends a regular message. BotEngine resolves the active session via SessionRouter, forwards to DaemonClient, streams the SSE response back to chat in real time.
4. Shutdown: SIGTERM/SIGINT triggers graceful cleanup — stop adapters, close HTTP sessions, close SSH tunnels.

Shared Infrastructure

The core infrastructure components are created once in main.py and shared across all platform adapters:

• SSHManager: One instance manages all SSH connections and tunnels. Thread-safe through asyncio's single-threaded event loop.
• SessionRouter: One SQLite database (sessions.db) tracks sessions across all platforms (Discord, Telegram, Lark).
• DaemonClient: One aiohttp session handles all RPC calls to remote daemons.

Each platform gets its own BotEngine instance, but all engines share the same SSHManager, SessionRouter, and DaemonClient.

Entry Point (main.py)

File: head/main.py

The entry point for the Codecast Head Node. This module bootstraps the entire system by loading configuration, initializing shared components, starting bots, and handling graceful shutdown.

Purpose

• Load and validate config.yaml
• Create shared infrastructure (SSHManager, SessionRouter, DaemonClient)
• Initialize and start Discord and/or Telegram bots
• Handle graceful shutdown on SIGTERM/SIGINT

Main Function

async def main(config_path: str = "config.yaml") -> None

The main() coroutine is the primary entry point. It performs the following steps:

1. Configuration Loading

config = load_config(config_path)

Loads config.yaml (or a custom path passed as a command-line argument). If the file is missing or invalid, the process exits with an error message.

2. Shared Component Initialization

ssh_manager = SSHManager(config)
session_router = SessionRouter(db_path=str(Path(__file__).parent / "sessions.db"))
daemon_client = DaemonClient()

These three components are created once and shared across all bots:

• SSHManager: Manages SSH connections and tunnels to all configured machines. Takes the full config to access machine definitions and daemon deployment settings.
• SessionRouter: SQLite-backed session registry. The database is stored as head/sessions.db (next to the Python source files).
• DaemonClient: Stateless JSON-RPC client with a shared aiohttp session.

3. Bot Initialization

discord_bot = DiscordBot(ssh_manager, session_router, daemon_client, config)
telegram_bot = TelegramBot(ssh_manager, session_router, daemon_client, config)

Each bot is created only if its token is configured. If neither bot has a valid token, the process exits with an error.

4. Bot Startup

task = asyncio.create_task(discord_bot.start(), name="discord")
task = asyncio.create_task(telegram_bot.start(), name="telegram")

Bots run as concurrent asyncio tasks. The main coroutine then waits for either:

• A shutdown signal (SIGTERM/SIGINT)
• A bot task to crash (first completed)

5. Graceful Shutdown

def handle_shutdown(sig: signal.Signals) -> None:
    shutdown_event.set()

Signal handlers for SIGTERM and SIGINT set a shutdown event. When triggered:

1. All bots are stopped via bot.stop()
2. The DaemonClient's HTTP session is closed
3. All SSH tunnels are closed via ssh_manager.close_all()
4. Remaining asyncio tasks are cancelled

Command-Line Usage

# Default config
python -m head.main

# Custom config path
python -m head.main /path/to/config.yaml

The config path is read from sys.argv[1] if provided, defaulting to "config.yaml".

Logging

The module configures Python's logging at the INFO level with the format:

2026-03-14 10:00:00 [codecast] INFO: message

All modules under head/ use logging.getLogger(__name__) and inherit this configuration.

Error Handling

• Missing config file: logs error and exits with code 1
• Empty machines dict: logs error and exits with code 1
• No bots configured (no tokens): logs error and exits with code 1
• Bot crash during runtime: logs the exception, triggers shutdown
• Cleanup errors: logged as warnings, do not prevent other cleanup steps

Config Loader (config.py)

File: head/config.py

Handles loading, parsing, and validating the config.yaml configuration file. Defines all configuration dataclasses and provides environment variable expansion.

Purpose

• Define typed configuration structure using Python dataclasses
• Load and parse YAML configuration files
• Expand ${ENV_VAR} references in string values
• Expand ~ in file paths

Dataclasses

MachineConfig

Represents a single remote machine.

@dataclass
class MachineConfig:
    id: str                              # Machine identifier (key from YAML)
    host: str                            # Hostname or IP
    user: str                            # SSH username
    ssh_key: Optional[str] = None        # Path to SSH private key
    port: int = 22                       # SSH port
    proxy_jump: Optional[str] = None     # Jump host machine ID
    proxy_command: Optional[str] = None  # SSH ProxyCommand string
    password: Optional[str] = None       # Password or "file:/path"
    daemon_port: int = 9100              # Remote daemon port
    node_path: Optional[str] = None      # Path to Node.js on remote
    default_paths: list[str] = []        # Common project paths

DiscordConfig

@dataclass
class DiscordConfig:
    token: str                           # Bot token
    allowed_channels: list[int] = []     # Channel ID whitelist (empty = all)
    command_prefix: str = "/"            # Command prefix

TelegramConfig

@dataclass
class TelegramConfig:
    token: str                           # Bot token
    allowed_users: list[int] = []        # User ID whitelist (empty = all)

BotConfig

@dataclass
class BotConfig:
    discord: Optional[DiscordConfig] = None
    telegram: Optional[TelegramConfig] = None

SkillsConfig

@dataclass
class SkillsConfig:
    shared_dir: str = "./skills"         # Local skills directory
    sync_on_start: bool = True           # Sync on session creation

DaemonDeployConfig

@dataclass
class DaemonDeployConfig:
    install_dir: str = "~/.codecast/daemon"   # Remote install path
    auto_deploy: bool = True                        # Auto-deploy daemon
    log_file: str = "~/.codecast/daemon.log"  # Remote log file

Config

Top-level configuration container:

@dataclass
class Config:
    machines: dict[str, MachineConfig] = {}
    bot: BotConfig = BotConfig()
    default_mode: str = "auto"
    skills: SkillsConfig = SkillsConfig()
    daemon: DaemonDeployConfig = DaemonDeployConfig()

Key Functions

load_config(config_path: str) -> Config

Main entry point for configuration loading.

1. Reads the YAML file
2. Recursively expands ${ENV_VAR} references through _process_value()
3. Parses machines section into MachineConfig objects (using the YAML key as the machine id)
4. Parses bot.discord and bot.telegram sections
5. Parses default_mode, skills, and daemon sections

Raises:

• FileNotFoundError if the config file does not exist
• ValueError if the config file is empty

expand_env_vars(value: str) -> str

Replaces ${VARIABLE_NAME} patterns with the corresponding environment variable value. If the variable is not set, the original ${...} expression is left unchanged.

# Example:
expand_env_vars("token: ${DISCORD_TOKEN}")
# → "token: my-actual-token"  (if DISCORD_TOKEN is set)
# → "token: ${DISCORD_TOKEN}" (if DISCORD_TOKEN is not set)

expand_path(path: str) -> str

Combines environment variable expansion with ~ (home directory) expansion. Used for file paths like ssh_key.

expand_path("~/.ssh/id_rsa")
# → "/home/user/.ssh/id_rsa"

_process_value(value: Any) -> Any

Recursively processes all values in the config dictionary, expanding environment variables in strings and recursing into dicts and lists. Non-string, non-container values are returned unchanged.

Connection to Other Modules

• main.py calls load_config() at startup
• SSHManager receives the full Config object and reads MachineConfig instances for SSH connections and DaemonDeployConfig for deployment settings
• Bot classes receive Config to access bot tokens and settings

BotEngine (engine.py)

File: src/head/engine.py

The central command engine for Codecast. BotEngine contains all command routing, session management, and message forwarding logic. It uses composition: a PlatformAdapter instance handles platform-specific I/O while the engine handles all shared behavior.

This replaces the old BotBase ABC inheritance pattern. Instead of subclassing, each platform creates an adapter and passes it to a BotEngine instance.

Purpose

• Route user input to command handlers (cmd_* methods) or forward to the active Claude session
• Manage session lifecycle: create, resume, detach, destroy
• Stream responses from the daemon back to chat with configurable display modes
• Handle AskUserQuestion interactive flows
• Manage file forwarding via FileForwardMatcher
• Control concurrency: prevent simultaneous streaming to the same channel

Class: BotEngine

class BotEngine:
    adapter: PlatformAdapter        # Platform-specific I/O
    ssh: SSHManager                 # SSH tunnel management
    router: SessionRouter           # SQLite session registry
    daemon: DaemonClient            # JSON-RPC + SSE client
    config: Config                  # Loaded configuration
    file_pool: Any                  # Optional file pool for file forwarding
    file_forward: FileForwardMatcher | None  # File forwarding rules
    _streaming: set[str]            # Channels currently streaming
    _stop_requested: set[str]       # Channels with a pending stop request
    _init_shown: set[str]           # Sessions that have shown the init message

Command Dispatcher

handle_input(channel_id, text, user_id=None, attachments=None)

Main entry point for all user input from a platform adapter. Logic:

1. If there is a pending interactive flow (SSH import wizard, remove confirmation), route there first.
2. If the text starts with /, call _handle_command().
3. Otherwise, call _forward_message() to send to the active Claude session.

_handle_command(channel_id, text, user_id=None)

Parses the command name and dispatches to the appropriate handler. Uses maxsplit=2 to preserve path arguments that may contain spaces, except for variadic commands like /add-machine.

Full command table:

Unknown commands (not starting with a recognized prefix) are forwarded to the active Claude session as regular messages, allowing users to send slash-prefixed prompts directly to Claude.

All command handlers are wrapped in error handling that catches DaemonConnectionError, DaemonError, and generic exceptions, formatting them as error messages back to the user.

Command Implementations

cmd_start(channel_id, args, silent_init=False)

Creates a new session: /start <machine_id> <path> [cli_type]

1. Validates arguments (machine ID and path required)
2. Resolves the path: git URLs are expanded to {project_path}/{repo_name}, bare names are expanded to {project_path}/{name}
3. Calls ssh.ensure_tunnel() to establish the SSH port-forwarding tunnel
4. Calls ssh.sync_skills() to copy shared skills to the remote machine
5. Optionally clones a git repo if a URL was provided
6. Calls daemon.create_session() to register the session on the daemon
7. Registers the session in the router with router.register()
8. Sends a confirmation message with session name, mode, and model

The silent_init parameter suppresses the "Starting session..." message (used by Discord slash commands which send their own initial response).

cmd_resume(channel_id, args)

Resumes a detached or previously active session: /resume <session_id_or_name>

1. Looks up the session by name or daemon ID in the router
2. Calls ssh.ensure_tunnel() for the session's machine
3. Calls daemon.resume_session() with the SDK session ID if available
4. Re-registers the session as active via router.register()

cmd_ls(channel_id, args)

Lists machines or sessions: /ls machine or /ls session [machine]

cmd_exit(channel_id)

Detaches from the current session. The daemon-side session (and Claude process) is not destroyed. The user can resume with /resume.

cmd_rm(channel_id, args)

Destroys all sessions matching a machine/path combination.

cmd_mode(channel_id, args)

Changes the permission mode: /mode <auto|code|plan|ask>

Calls daemon.set_mode() on the daemon, which takes effect on the next spawned process. Also updates the local session state in the router.

cmd_model(channel_id, args)

Sets the model for the current session: /model <model_name>

Calls daemon.set_model() on the daemon, which takes effect on the next spawned process.

cmd_tool_display(channel_id, args)

Switches the tool display mode for the current session: /tool-display <timer|append|batch>

Stored in the session router and applied by _forward_message() on the next stream.

cmd_interrupt(channel_id)

Interrupts the current operation: /interrupt or /stop

1. Adds the channel to _stop_requested to signal the active stream loop to exit
2. Calls daemon.interrupt_session() to send SIGTERM to the running CLI process

cmd_health(channel_id, args)

Checks daemon health for a specific machine, the current session's machine, or all connected machines.

cmd_monitor(channel_id, args)

Shows detailed per-session monitoring data (status, queue depth, connected state) from the daemon.

Message Forwarding

_forward_message(channel_id, text, attachments=None)

Forwards a user message to the active Claude session and streams the response back to chat.

Concurrency control: The _streaming set tracks channels with an active stream. A second message to a streaming channel is rejected with a "Claude is still processing" notice. The _stop_requested set signals the loop to exit early when /interrupt is called.

Streaming display flow:

1. Resolve the session from the router; error if none active
2. Call ssh.ensure_tunnel() to confirm the tunnel is up
3. Call daemon.send_message(), which returns an async SSE event iterator
4. Handle events according to the session's tool_display mode

Tool display mode logic:

For timer mode (default):

• Show a timer message while tools are running
• Collect all tool events; send them together at the end

For append mode:

• Send or edit an activity message immediately for each tool_use event
• Show accumulated tool lines plus a partial-text snippet

For batch mode:

• Accumulate tool events silently
• Compress them into a single summary at the end via compress_tool_messages()

Per-event handling (all modes):

• partial: Accumulate text in a buffer. Every STREAM_UPDATE_INTERVAL seconds (1.5s), send or edit a message with current buffer plus ▌ cursor. If the buffer exceeds STREAM_BUFFER_FLUSH_SIZE (1800 chars), finalize the current message and start a new one.
• text: Complete text block. Edit the streaming message to its final form, or send as new message(s) if no streaming message exists. Split at platform limits if needed.
• tool_use with tool == "AskUserQuestion": Parse the question input using format_ask_user_question(), then call adapter.send_question() for each question to display platform-native interactive buttons/keyboard.
• tool_use (other tools): Route to the active tool display mode handler.
• result: Capture the SDK session ID and update the router with router.update_sdk_session().
• system (subtype init): On the first system event for a session, display the connected model and current mode.
• queued: Notify the user that their message is queued, with its position number.
• error: Display the error message.
• interrupted: Display an interruption notice.
• ping: Ignored (daemon keepalive).

5. After the stream ends, flush any remaining buffer content as a final message.

AskUserQuestion Handling

When Claude invokes the AskUserQuestion tool, the stream emits a tool_use event with tool == "AskUserQuestion" and structured input containing a list of question dicts.

The engine calls format_ask_user_question() to parse the input, then calls adapter.send_question() for each question. Platform adapters that support inline buttons (Discord, Telegram) render them as clickable buttons or an inline keyboard. The user's button click is forwarded back to the engine as a regular input event, which gets sent to Claude as the response.

File Forwarding

If config.file_forward.enabled is true, the engine initializes a FileForwardMatcher. After each streamed response, the engine scans the output for file paths matching the forwarding rules. Matched files are downloaded from the remote machine via SSH and sent to the chat channel via adapter.send_file().

Constants

SSH Manager (ssh_manager.py)

File: head/ssh_manager.py

Manages SSH connections, port-forwarding tunnels, remote daemon deployment, and skills synchronization. This is the bridge between the local Head Node and remote machines.

Purpose

• Maintain a pool of SSH connections and tunnels to remote machines
• Create local port-forwarding tunnels to access remote daemons
• Deploy daemon code to remote machines via SCP
• Start and health-check daemons on remote machines
• Sync skills files to remote project directories
• List machines with their online/daemon status

Classes

SSHTunnel

Represents an active SSH tunnel to a remote machine.

class SSHTunnel:
    machine_id: str          # Machine this tunnel connects to
    local_port: int          # Local port (e.g., 19100)
    conn: SSHClientConnection  # asyncssh connection
    listener: SSHListener    # Port forwarding listener

Properties:

• alive -- Returns True if the underlying SSH connection is still open.

Methods:

• close() -- Closes the port forwarding listener and SSH connection.

SSHManager

Main class managing all SSH operations.

class SSHManager:
    config: Config
    machines: dict[str, MachineConfig]
    tunnels: dict[str, SSHTunnel]      # machine_id -> active tunnel

Key Methods

ensure_tunnel(machine_id: str) -> int

Ensures an SSH tunnel exists to the specified machine. Returns the local port number for accessing the daemon.

Flow:

1. Check if a tunnel already exists and is alive -- return existing local port
2. If the tunnel is dead, close and remove it
3. Allocate a new local port (starting from 19100, incrementing)
4. Establish SSH connection via _connect_ssh()
5. Create local port forwarding: 127.0.0.1:<local_port> -> 127.0.0.1:<daemon_port>
6. Ensure the daemon is running on the remote machine via _ensure_daemon()
7. Store the tunnel and return the local port

_connect_ssh(machine: MachineConfig) -> SSHClientConnection

Establishes an SSH connection to a machine. Handles:

• SSH key authentication: Uses client_keys if ssh_key is configured
• Password authentication: Supports direct passwords and file:/path syntax
• ProxyJump: Connects through a jump host by first establishing a connection to the jump machine, then using it as a tunnel for the final connection
• Known hosts: Disabled (known_hosts=None) for simplicity in trusted environments

_ensure_daemon(machine_id: str, conn: SSHClientConnection) -> None

Ensures the daemon process is running on the remote machine.

Flow:

1. Check if a node.*dist/server.js process is already running via pgrep
2. If running, return immediately
3. Check if daemon code exists at install_dir (both dist/server.js and node_modules/)
4. If missing and auto_deploy is enabled, call _deploy_daemon()
5. Start the daemon with nohup, setting:
   • DAEMON_PORT environment variable
   • PATH including the Node.js binary directory and ~/.local/bin (for Claude CLI)
6. Poll the health endpoint (health.check RPC) every 2 seconds for up to 30 seconds
7. Raise RuntimeError if the daemon does not respond within the timeout

_deploy_daemon(machine_id: str, conn: SSHClientConnection) -> None

Deploys daemon code to a remote machine via SCP.

Flow:

1. Build the daemon locally if daemon/dist/ does not exist (npm run build)
2. Create the remote install directory
3. SCP package.json and package-lock.json to the remote
4. SCP the entire dist/ directory recursively
5. Run npm install --production on the remote machine
6. If npm is in a non-standard location, derive its path from node_path

sync_skills(machine_id: str, remote_path: str) -> None

Syncs skills files from the local skills.shared_dir to a remote project path.

Behavior:

• Skips entirely if skills.sync_on_start is false
• Copies CLAUDE.md to the remote project root, but only if it does not already exist there
• Copies the .claude/skills/ directory recursively to the remote project
• Uses existing SSH tunnel connection if available, otherwise creates a new connection
• Errors are logged as warnings and do not fail the session creation

list_machines() -> list[dict]

Lists all configured machines with their online and daemon status.

Behavior:

• Skips machines that are only used as jump hosts (referenced by proxy_jump and having no default_paths)
• For each machine, attempts an SSH connection with a 15-second timeout
• If reachable, checks if the daemon process is running via pgrep
• Returns a list of dicts with: id, host, user, status (online/offline), daemon (running/stopped/unknown), default_paths

get_local_port(machine_id: str) -> Optional[int]

Returns the local tunnel port for a machine if a live tunnel exists, otherwise None.

close_all() -> None

Closes all SSH tunnels and connections. Called during graceful shutdown.

Port Allocation

Local ports for SSH tunnels are allocated sequentially starting from 19100:

gpu-1 -> localhost:19100
gpu-2 -> localhost:19101
gpu-3 -> localhost:19102
...

This simple allocation works because the Head Node manages all tunnels in a single process.

Connection to Other Modules

• main.py creates the SSHManager with the full config and calls close_all() on shutdown
• BotBase calls ensure_tunnel() before every daemon RPC call and sync_skills() on /start
• BotBase calls list_machines() for the /ls machine command
• BotBase calls get_local_port() for the /health command when checking all connected machines

Session Router (session_router.py)

File: head/session_router.py

Manages session state in a local SQLite database. Maps bot channels (Discord or Telegram) to active Claude sessions on remote machines.

Purpose

• Maintain a persistent registry of sessions across Head Node restarts
• Map chat channels to remote Claude sessions
• Track session lifecycle: active -> detached -> destroyed
• Log session history for resume capabilities
• Provide query methods for session lookup by channel, daemon ID, or machine/path

Database Schema

sessions table

Stores the current state of each session. Primary key is channel_id (one active session per channel).

session_log table

Append-only log of detached sessions. Used for session resume lookups.

Indexed on machine_id and daemon_session_id for fast lookups.

Session Dataclass

@dataclass
class Session:
    channel_id: str           # e.g., "discord:123456"
    machine_id: str           # e.g., "gpu-1"
    path: str                 # e.g., "/home/user/project"
    daemon_session_id: str    # UUID from daemon
    sdk_session_id: Optional[str]  # Claude SDK session ID
    status: str               # "active" | "detached" | "destroyed"
    mode: str                 # "auto" | "code" | "plan" | "ask"
    created_at: str           # ISO 8601
    updated_at: str           # ISO 8601

Key Methods

resolve(channel_id: str) -> Optional[Session]

Find the active session for a channel. Returns None if no active session exists. This is the primary lookup used when forwarding user messages to Claude.

register(channel_id, machine_id, path, daemon_session_id, mode) -> None

Register a new active session for a channel. If an active session already exists on this channel, it is automatically detached first (moved to the session log). The new session is inserted with status active.

update_sdk_session(channel_id: str, sdk_session_id: str) -> None

Update the SDK session ID for an active session. Called when a result event is received from Claude, which contains the session ID needed for future --resume calls.

update_mode(channel_id: str, mode: str) -> None

Update the permission mode for the active session on a channel. Called when the user changes mode with /mode.

detach(channel_id: str) -> Optional[Session]

Detach the active session on a channel without destroying it. The session is:

1. Copied to session_log with the current timestamp as detached_at
2. Status is updated to detached in the sessions table

Returns the detached session, or None if no active session was found. Detached sessions can be resumed later with /resume.

destroy(channel_id: str) -> Optional[Session]

Mark a session as destroyed. Unlike detach, this does not log the session. Returns the destroyed session or None.

list_sessions(machine_id: Optional[str]) -> list[Session]

List all sessions, optionally filtered by machine ID. Returns sessions ordered by updated_at descending (most recent first). Includes sessions in all statuses.

list_active_sessions() -> list[Session]

List only sessions with status active.

find_session_by_daemon_id(daemon_session_id: str) -> Optional[Session]

Find a session by its daemon-assigned UUID. Searches both the active sessions table and the session_log table. Used by /resume to locate previously detached sessions.

find_sessions_by_machine_path(machine_id: str, path: str) -> list[Session]

Find all sessions on a specific machine and path. Used by /rm to destroy sessions matching a machine/path combination.

Connection to Other Modules

• main.py creates the SessionRouter with the database path
• BotBase calls resolve() before every message forward, register() on /start, detach() on /exit, destroy() via /rm, and query methods for /ls, /resume, /status
• BotBase calls update_sdk_session() when a result event provides the Claude SDK session ID
• BotBase calls update_mode() when the user changes the permission mode

Daemon Client (daemon_client.py)

File: head/daemon_client.py

JSON-RPC client for communicating with the Remote Agent Daemon over SSH tunnels. Handles both regular JSON responses and SSE (Server-Sent Events) streaming responses.

Purpose

• Send JSON-RPC requests to the daemon's HTTP endpoint
• Parse SSE streams for session.send (streaming Claude responses)
• Provide typed methods for each RPC operation
• Handle connection errors and daemon-reported errors

Class: DaemonClient

class DaemonClient:
    timeout: int = 300  # Default timeout in seconds

Internal Methods

_url(local_port: int) -> str

Builds the RPC endpoint URL: http://127.0.0.1:{local_port}/rpc

_rpc_call(local_port, method, params) -> dict

Makes a JSON-RPC call with the given method and parameters. Uses a 30-second timeout for non-streaming calls. Raises DaemonError if the response contains an error, or DaemonConnectionError if the HTTP request fails.

Session Management Methods

create_session(local_port, path, mode) -> str

Creates a new Claude session on the remote machine.

• Params: path (project directory), mode (permission mode)
• Returns: sessionId (UUID string)

send_message(local_port, session_id, message, idle_timeout) -> AsyncIterator[dict]

Sends a message to a Claude session and streams back events via SSE.

This is the core method for interacting with Claude. It:

1. Sends a session.send JSON-RPC request
2. Reads the response as an SSE stream (text/event-stream)
3. Parses each data: {...} line as JSON
4. Yields parsed event dicts to the caller
5. Returns when it receives data: [DONE]

Timeout behavior:

• Total timeout: 15 minutes (900 seconds)
• Idle timeout (per-read): configurable, defaults to 300 seconds (5 minutes)
• If no events are received within the idle timeout, yields an error event

Error handling:

• asyncio.TimeoutError -> yields an error event about stream idle timeout
• aiohttp.ClientError -> yields a connection error event

resume_session(local_port, session_id, sdk_session_id) -> dict

Resumes a previously detached session. If sdk_session_id is provided, it is passed to the daemon to set up --resume for future Claude invocations.

Returns a dict with ok (bool) and fallback (bool indicating if a fresh session was created with history injection).

destroy_session(local_port, session_id) -> bool

Destroys a session and kills any running Claude process. Returns True on success.

list_sessions(local_port) -> list[dict]

Lists all sessions on a remote daemon. Returns a list of session info dicts.

set_mode(local_port, session_id, mode) -> bool

Sets the permission mode for a session. Returns True on success.

interrupt_session(local_port, session_id) -> dict

Interrupts the current Claude operation for a session by sending SIGTERM to the Claude CLI process. Returns a dict with:

• ok (bool): Always True if the session exists
• interrupted (bool): True if there was an active operation to interrupt

health_check(local_port) -> dict

Checks daemon health. Returns session counts, uptime, memory usage, daemon version, and PID.

monitor_sessions(local_port) -> dict

Gets detailed monitoring information for all sessions, including queue stats.

reconnect_session(local_port, session_id) -> list[dict]

Reconnects to a session and retrieves any buffered events that were generated while the client was disconnected.

get_queue_stats(local_port, session_id) -> dict

Gets message queue statistics for a session: pending user messages, pending responses, and client connection state.

Cleanup

close() -> None

Closes the underlying aiohttp session. Called during Head Node shutdown.

Exception Classes

DaemonError

Raised when the daemon returns an error response in the JSON-RPC result.

class DaemonError(Exception):
    code: int  # Error code from daemon

DaemonConnectionError

Raised when the HTTP connection to the daemon fails (network error, connection refused, etc.).

Connection to Other Modules

• main.py creates the DaemonClient and calls close() on shutdown
• BotBase calls all session management methods in response to user commands and message forwarding
• SSHManager provides the local_port that maps to the remote daemon via SSH tunnel

Discord Adapter (discord_adapter.py)

File: src/head/platform/discord_adapter.py

Discord platform adapter implementing the PlatformAdapter protocol. Uses discord.py v2 with slash commands, autocomplete, typing indicators, heartbeat status updates, and interactive button views for AskUserQuestion.

Purpose

• Implement the Discord-specific platform layer for Codecast
• Register slash commands with Discord's application command system and provide autocomplete
• Show typing indicators and heartbeat messages during long Claude operations
• Present AskUserQuestion prompts as Discord UI buttons or select menus
• Handle Discord's 2000-character message limit with smart splitting

Class: DiscordAdapter

Implements PlatformAdapter.

class DiscordAdapter:
    platform_name: str = "discord"
    max_message_length: int = 2000
    _config: DiscordConfig
    _bot: commands.Bot
    _on_input: InputHandler | None
    _channels: dict[str, Messageable]
    _typing_tasks: dict[str, Task]
    _heartbeat_msgs: dict[str, Message]
    _deferred_interactions: dict[str, Interaction]
    _engine_ref: EngineType | None    # back-reference for heartbeat

Channel ID Format

Discord channels use the prefix discord: in the unified channel ID (e.g., discord:123456789012345678). The BotEngine and SessionRouter always use these prefixed IDs.

Slash Commands

All commands are registered as Discord application commands (slash commands) with full autocomplete support. Commands are synced to all guilds on on_ready.

/start <machine> <path> [cli_type]

• machine autocomplete: Lists all configured machines, excluding jump hosts. Filters by the current input text.
• path autocomplete: Lists default_paths from the selected machine's config. Falls back to paths from all machines if no machine is selected yet.
• cli_type: Optional choice from claude, codex, gemini, opencode.
• Sends the initial "Starting..." response via interaction.response.send_message(), then processes asynchronously.

/resume <session_id>

Takes a session name or UUID string. No autocomplete.

/ls <target> [machine]

• target: Choice dropdown between "machine" and "session".
• machine: Autocomplete with configured machine IDs (optional, for session filtering).
• Uses deferred response since listing may take time.

/mode <mode>

Choice dropdown with descriptions:

• "bypass - Full auto (skip all permissions)" → auto
• "code - Auto accept edits, confirm bash" → code
• "plan - Read-only analysis" → plan
• "ask - Confirm everything" → ask

/exit, /status, /help, /interrupt

Simple commands with no parameters. All use deferred responses.

/rm <machine> <path>

Machine autocomplete, manual path input. Deferred response.

/health [machine], /monitor [machine]

Optional machine parameter with autocomplete. Deferred response.

/tool-display <mode>

Choice dropdown for timer, append, batch.

/model <model_name>

Text input for the model identifier.

Deferred Interactions

Discord requires a response within 3 seconds. For operations that may take longer, the adapter calls interaction.response.defer() to acknowledge the command immediately. The next send_message() call for that channel uses interaction.followup.send() instead of channel.send().

The _defer_and_register() method stores the pending interaction. It is consumed on first use and then cleared.

Typing Indicator

While Claude processes a message, the adapter shows "Bot is typing..." in the channel:

async def _start_typing(channel_id: str) -> None:
    # Sends typing() context manager every 8s (Discord indicator lasts ~10s)

The typing loop runs as a background asyncio task. It is cancelled by stop_typing() when the stream completes.

Heartbeat Status Updates

For long-running operations, the adapter sends periodic status messages every HEARTBEAT_INTERVAL (30 seconds) to keep users informed. These prevent the perception of a frozen bot.

async def _heartbeat_loop(channel_id: str, start_time: float, event_tracker: dict) -> None:
    # Every 30 seconds, sends/updates a message like:
    # "[1m30s] Claude is working... Using tool: Write"

The heartbeat message reflects the current processing state:

• Tool name set: "Using tool: {tool_name}"
• Partial events with content: "Writing response..."
• Tool use/result events: "Processing tool results..."
• Default: "Thinking..."

The heartbeat message is deleted when the operation completes.

The _forward_message_with_heartbeat() method wraps the BotEngine's streaming loop with typing and heartbeat management. The Discord adapter sets up this wrapper by storing a back-reference to the engine.

AskUserQuestion: Interactive View

When BotEngine calls adapter.send_question(), the Discord adapter presents options as interactive UI elements using _AskUserQuestionView:

class _AskUserQuestionView(discord.ui.View):
    timeout = 300  # 5 minutes

• For 5 or fewer options: renders as discord.ui.Button instances (one per option, secondary style)
• For more than 5 options: renders as a discord.SelectMenu with up to 25 options

When the user clicks a button or selects from the menu, the view calls the on_input callback with the selected option text. This feeds back into engine.handle_input() as a normal message.

Platform Methods

send_message(channel_id, text) -> MessageHandle

1. If a pending deferred interaction exists for this channel, uses interaction.followup.send() first
2. Splits the text with split_message(max_len=2000)
3. Sends each chunk; if a chunk fails (e.g., invalid Discord markdown), retries with formatting stripped
4. Returns a MessageHandle wrapping the last sent discord.Message object

edit_message(handle, text) -> None

Edits an existing Discord message using the handle's raw field (the original discord.Message). Truncates to 2000 characters. Falls back to sending a new message if the edit fails (e.g., message deleted).

delete_message(handle) -> None

Deletes the message referenced by the handle.

download_file(attachment, dest) -> Path

Downloads the file from the Discord CDN URL to the local path using aiohttp.

send_file(channel_id, path, caption="") -> MessageHandle

Uploads a file to the Discord channel as a discord.File attachment.

start_typing(channel_id) -> None / stop_typing(channel_id) -> None

Starts or cancels the background typing loop for the channel.

send_question(channel_id, header, options, multi_select=False) -> MessageHandle

Sends the question header text and an _AskUserQuestionView with the given options. The view expires after 300 seconds.

Event Handling

on_ready

Logs the bot username and ID, then calls bot.tree.sync() to sync slash commands to all guilds.

on_message

Handles regular (non-slash-command) messages:

1. Ignores messages from the bot itself and other bots
2. Ignores messages starting with / (handled by slash commands)
3. Checks the allowed_channels whitelist (if configured)
4. Calls _on_input(channel_id, text, user_id, attachments) which routes to BotEngine

Constants

Connection to Other Modules

• Implements PlatformAdapter from platform/protocol.py
• main.py creates and starts the DiscordAdapter, then passes it to BotEngine
• Uses message_formatter for split_message(), format_error(), and display_mode()
• BotEngine drives all command and streaming logic via the registered InputHandler

Telegram Adapter (telegram_adapter.py)

File: src/head/platform/telegram_adapter.py

Telegram platform adapter implementing the PlatformAdapter protocol. Uses python-telegram-bot v20+ with async handlers, HTML formatting (with Markdown-to-HTML conversion), inline keyboard buttons for AskUserQuestion, and rate limit handling.

Purpose

• Implement the Telegram-specific platform layer for Codecast
• Register command and message handlers with the python-telegram-bot Application
• Handle Telegram's 4096-character message limit with smart splitting
• Present AskUserQuestion prompts as Telegram inline keyboards
• Handle rate limiting (RetryAfter exceptions) gracefully
• Convert Discord-style Markdown output to Telegram HTML format

Class: TelegramAdapter

Implements PlatformAdapter.

class TelegramAdapter:
    platform_name: str = "telegram"
    max_message_length: int = 4096
    _config: TelegramConfig
    _app: Application | None
    _bot: Bot | None
    _on_input: InputHandler | None
    _last_messages: dict[str, int]     # channel_id -> last message_id
    _typing_tasks: dict[str, Task]     # active typing indicators

Channel ID Format

Telegram channels use the prefix telegram: in the unified channel ID (e.g., telegram:123456789). Chat IDs are extracted from this prefix when making Bot API calls.

Access Control

Access is controlled by two config lists:

• allowed_users: If non-empty, only these user IDs may interact with the bot
• admin_users: User IDs allowed to run admin commands (/restart, /update)

Handlers

The adapter registers two handler types with the python-telegram-bot Application:

Command Handler

Registered via CommandHandler for each recognized command name:

command_names = [
    "start", "resume", "ls", "list", "exit", "rm", "remove",
    "destroy", "mode", "model", "status", "health", "monitor",
    "interrupt", "stop", "tool_display", "help", "clear", "new",
    "add_machine", "remove_machine", "rename", "restart", "update",
]

The _handle_telegram_command() method:

1. Validates that the message and user exist
2. Checks allowed_users permission
3. Re-prefixes the command text with / if Telegram stripped it
4. Calls _on_input(channel_id, text, user_id) which routes to BotEngine

Message Handler

A MessageHandler with filter filters.TEXT & ~filters.COMMAND handles non-command messages. The _handle_telegram_message() method also checks allowed_users before forwarding.

Callback Query Handler

A CallbackQueryHandler handles inline keyboard button presses from AskUserQuestion views. When the user presses a button:

1. The handler acknowledges the callback query
2. Edits the original message to show the selected choice
3. Calls _on_input(channel_id, selected_option, user_id) which feeds back to BotEngine

Platform Methods

send_message(channel_id, text) -> MessageHandle

1. Splits the text with split_message(max_len=4096)
2. Converts Markdown to Telegram HTML via markdown_to_telegram_html()
3. Sends with parse_mode=ParseMode.HTML
4. On BadRequest (parse error), retries with plain text
5. On RetryAfter (rate limit), waits the specified number of seconds and retries
6. Returns a MessageHandle wrapping the last sent message; caches the message ID in _last_messages

edit_message(handle, text) -> None

1. Extracts the message ID from the handle
2. Converts Markdown to HTML
3. Calls bot.edit_message_text() with parse_mode=ParseMode.HTML
4. On BadRequest, retries with plain text
5. On RetryAfter, waits and retries

delete_message(handle) -> None

Calls bot.delete_message() using the message ID from the handle.

download_file(attachment, dest) -> Path

For Telegram file attachments (which provide a file ID rather than a URL), calls bot.get_file() then file.download_to_drive() to write to the destination path.

send_file(channel_id, path, caption="") -> MessageHandle

Checks the file size against TELEGRAM_FILE_SIZE_LIMIT (20 MB). If within limit, sends as a document with bot.send_document(). If oversized, sends an error message instead.

start_typing(channel_id) -> None / stop_typing(channel_id) -> None

Starts or cancels a background asyncio task that calls bot.send_chat_action(action="typing") every 5 seconds (Telegram's typing indicator lasts about 5 seconds).

send_question(channel_id, header, options, multi_select=False) -> MessageHandle

Builds a Telegram InlineKeyboardMarkup with one button per option. Button callback_data is set to the option text (truncated to 64 bytes, Telegram's limit for callback data).

For multi_select=True, options are prefixed with a checkbox indicator and the header notes that multiple selections are accepted. (Multi-select is handled by collecting multiple button presses until the user sends a confirmation message.)

HTML Formatting

Telegram uses HTML formatting rather than Markdown for safe message delivery. The markdown_to_telegram_html() function (in platform/format_utils.py) converts:

• **bold** → <b>bold</b>
• *italic* or _italic_ → <i>italic</i>
• `code` → <code>code</code>
• ```lang\n...\n``` → <pre><code class="language-lang">...</code></pre>
• Escaped HTML characters in non-code sections

Lifecycle

start() -> None

1. Builds the Application using the configured token
2. Registers all command handlers
3. Registers the message handler and callback query handler
4. Sets bot commands in the Telegram UI via bot.set_my_commands()
5. Initializes and starts the Application
6. Starts polling for updates

stop() -> None

1. Stops the updater (polling)
2. Stops and shuts down the Application

Differences from Discord Adapter

Connection to Other Modules

• Implements PlatformAdapter from platform/protocol.py
• main.py creates and starts the TelegramAdapter, then passes it to BotEngine
• Uses message_formatter for split_message()
• Uses platform/format_utils.py for markdown_to_telegram_html()
• BotEngine drives all command and streaming logic via the registered InputHandler

Message Formatter (message_formatter.py)

File: src/head/message_formatter.py

Handles message splitting for platform character limits and formatting of various output types for display in Discord, Telegram, and Lark.

Purpose

• Split long messages into chunks that respect platform limits (Discord: 2000, Telegram: 4096)
• Smart splitting that avoids breaking code blocks and prefers natural boundaries
• Format tool use events in multiple styles (full, line, activity, compressed batch)
• Handle AskUserQuestion parsing and text rendering
• Format machine lists, session lists, status reports, health checks, and monitoring data
• Map internal mode names to user-facing display names

Mode Display Names

Internal mode names are mapped to user-facing names:

The auto mode is displayed as bypass to make it clear that all permission prompts are skipped.

def display_mode(mode: str) -> str

Message Splitting

split_message(text: str, max_len: int = 2000) -> list[str]

Splits a long message into chunks that fit within the platform's character limit.

Splitting priority (highest to lowest):

1. Code block awareness: If a split would land inside a code block (odd number of ``` markers before the split point), the split is moved to before the opening ```. This prevents sending a message with an unclosed code block.
2. Paragraph boundary (\n\n): Preferred split point; must be at least 30% into the text.
3. Line boundary (\n): Next best option; also requires 30% minimum position.
4. Sentence boundary (. , ! , ? , ; ): Requires 50% minimum position.
5. Word boundary (space): Requires 50% minimum position.
6. Forced split: At exactly max_len if no natural boundary is found.

Empty chunks are filtered out of the result.

Tool Formatting Functions

format_tool_use(event: dict) -> str

Formats a single tool_use event for full display. Used in append display mode and for single-tool responses.

With a status message:

**[Tool: Bash]** Running command...

With structured input (truncated to 500 chars):

**[Tool: Write]**

{"file_path": "/path/to/file", "content": "..."}

With no message or input:

**[Tool: Glob]**

format_tool_line(event: dict) -> str

Formats a single tool_use event as a compact one-liner for activity messages. Used when building the accumulated tool call list in timer and append display modes.

  `WebFetch` — https://api.github.com/repos/...
  `Write` — {"file_path": "/home/user/..."}
  `Bash`

Input/message text is truncated to 120 characters.

compress_tool_messages(events: list[dict]) -> str

Compresses multiple tool_use events into a single summary message. Used in batch display mode when there is more than one tool call in a response.

For a single event, delegates to format_tool_use(). For multiple events:

**[Tools: 3 calls]**
  `Read` — /home/user/project/main.py
  `Bash` — {"command": "pytest tests/ -v"}
  `Write` — {"file_path": "/home/user/project/..."}

Each line is truncated to 120 characters.

format_activity_message(tool_lines: list[str], thinking: str = "", cursor: bool = True) -> str

Builds a live activity message showing accumulated tool calls and an optional thinking snippet. Used in timer and append display modes to provide a continuously updated status message.

**[Tools: 2 calls]**
  `Read` — /home/user/project/main.py
  `Bash` — {"command": "pytest"}
> *...running test suite...*
▌

Parameters:

• tool_lines: One line per tool call from format_tool_line()
• thinking: Current partial-text snippet (last 200 chars shown)
• cursor: Whether to append the ▌ cursor indicator

AskUserQuestion Functions

format_ask_user_question(questions: list[dict]) -> list[tuple[str, list[str], bool]]

Parses the structured input from a Claude AskUserQuestion tool invocation into a list of (header, options, multi_select) tuples.

Input format (from Claude's tool input JSON):

[
    {
        "header": "Which framework should I use?",
        "options": [
            {"description": "FastAPI (async, modern)"},
            {"description": "Flask (simple, synchronous)"}
        ],
        "multiSelect": false
    }
]

Output:

[("Which framework should I use?", ["FastAPI (async, modern)", "Flask (simple, synchronous)"], False)]

The adapter then calls adapter.send_question() for each tuple in the list.

format_question_text(header: str, options: list[str], multi_select: bool = False) -> str

Formats a question with numbered options as plain text. Used as a fallback for platforms that do not support inline buttons, and for logging.

**Which framework should I use?**
  1. FastAPI (async, modern)
  2. Flask (simple, synchronous)

For multi-select:

**Which components need updating?**
_(Select one or more — reply with numbers separated by commas)_
  1. Authentication module
  2. Database layer
  3. API endpoints

List and Status Formatting

format_machine_list(machines: list[dict]) -> str

Formats the machine list for /ls machine:

**Machines:**
🟢 **gpu-1** (gpu1.example.com) ⚡
  Paths: `/home/user/project-a`, `/home/user/project-b`
🔴 **gpu-2** (gpu2.lab.internal) 💤
  Paths: `/home/user/experiments`

Icons: online (🟢) / offline (🔴), daemon running (⚡) / stopped (💤). Localhost machines are tagged with [localhost].

format_session_list(sessions: list) -> str

Formats the session list for /ls session. Delegates to format_session_info() for each session.

format_session_info(session) -> str

Formats a single session. Handles both Session objects from the SessionRouter and dict objects from the daemon API.

For router sessions:

● **smooth-dove** `a1b2c3d4...` **gpu-1**:`/home/user/project` [bypass] (active)

For daemon API dicts:

◉ `e5f6g7h8...` **/home/user/other** [code | claude-sonnet-4-20250514] (busy)

Status icons: ● active/idle, ◉ busy, ○ detached, ✕ destroyed/error, ? unknown.

format_error(error: str) -> str

**Error:** message text

format_status(session, queue_stats=None) -> str

Formats the /status output, including session name, machine, path, CLI type, mode, tool display mode, status, session IDs, and queue statistics.

format_health(machine_id, health) -> str

Formats the /health output with uptime, session counts by status, memory usage, and process info.

format_monitor(machine_id, monitor) -> str

Formats the /monitor output with detailed per-session information including queue depth and client connection state.

Connection to Other Modules

• BotEngine (engine.py) imports all formatting functions
• discord_adapter.py imports split_message, format_error, display_mode
• telegram_adapter.py imports split_message

Daemon Overview

The Daemon is the remote agent component of Codecast. It runs on each remote machine (GPU server, cloud VM, etc.) and provides a JSON-RPC + SSE interface for managing CLI sessions.

Technology Stack

• Language: Rust (2021 edition)
• Async runtime: tokio
• HTTP framework: Axum
• Serialization: serde / serde_json
• Process management: tokio::process
• UUID generation: uuid crate (v4)
• Time: chrono (UTC timestamps)
• Logging: tracing + tracing-subscriber

Module Map

src/daemon/
├── main.rs              # Axum server entry: port binding, TLS, graceful shutdown
├── server.rs            # JSON-RPC router (POST /rpc), SSE streaming, AppState
├── session_pool.rs      # Session registry, per-message spawn, CliAdapter dispatch
├── message_queue.rs     # Per-session user message + response buffering
├── skill_manager.rs     # Skills sync from ~/.codecast/skills to project dirs
├── types.rs             # StreamEvent, SessionStatus, PermissionMode, RPC types
├── cli_adapter/
│   ├── mod.rs           # CliAdapter trait, create_adapter() factory, CLI_TYPES
│   ├── claude.rs        # Claude CLI adapter
│   ├── codex.rs         # OpenAI Codex adapter
│   ├── gemini.rs        # Google Gemini CLI adapter
│   └── opencode.rs      # OpenCode adapter
├── auth.rs              # Token-based auth middleware, TokenStore
├── config.rs            # DaemonConfig: load from ~/.codecast/daemon.yaml
└── tls.rs               # TLS certificate generation/loading for HTTPS mode

Module Dependencies

main.rs
  ├── server.rs       (AppState, build_router)
  ├── session_pool.rs (SessionPool)
  ├── skill_manager.rs (SkillManager)
  ├── auth.rs         (TokenStore)
  ├── config.rs       (DaemonConfig)
  └── types.rs        (PermissionMode)

server.rs
  ├── session_pool.rs (SessionPool)
  ├── skill_manager.rs (SkillManager — for session.create skills sync)
  ├── auth.rs         (auth_middleware)
  └── types.rs        (RpcRequest, RpcResponse, PermissionMode)

session_pool.rs
  ├── cli_adapter/mod.rs (create_adapter, CliAdapter)
  ├── message_queue.rs   (MessageQueue)
  └── types.rs           (SessionStatus, PermissionMode, StreamEvent, SessionInfo, QueueStats)

message_queue.rs
  └── types.rs           (StreamEvent, QueueStats)

cli_adapter/
  └── types.rs           (PermissionMode, StreamEvent)

types.rs
  └── (standalone, type definitions only)

Architecture: Per-Message Spawn

The daemon uses a per-message spawn architecture:

1. session.create registers session metadata (path, mode, CLI type) but does not spawn a process.
2. session.send selects the appropriate CliAdapter, builds the subprocess command, and spawns it for the duration of one message exchange.
3. The process exits after producing its full output. The SDK/thread session ID is captured from the result event.
4. The next session.send spawns a new process with --resume <sdkSessionId> (or the equivalent for non-Claude CLIs) to continue the conversation.

This design provides:

• Clean process isolation per message
• Automatic memory cleanup between messages
• No zombie process management
• Natural recovery from CLI crashes

Security

The daemon binds exclusively to 127.0.0.1 by default. It is not network-reachable without SSH port forwarding.

#![allow(unused)]
fn main() {
let addr: SocketAddr = format!("{}:{}", host, actual_port).parse().unwrap();
let listener = TcpListener::bind(addr).await?;
}

In HTTP mode (the default for SSH-tunnel access), the daemon requires no credentials because access is controlled by SSH authentication on the tunnel.

In HTTPS/auth mode (when config.requires_auth() is true), the daemon uses Bearer token authentication via auth_middleware and serves over TLS. This mode is used when the daemon is accessed directly over a network rather than through SSH.

Port Discovery

On startup, the daemon tries to bind to the configured port (default: 9100). If the port is in use, it increments and retries up to port + 100. The actual port is:

1. Printed to stdout as DAEMON_PORT=<port> for the Head Node to capture during startup
2. Written to ~/.codecast/daemon.port for subsequent discovery

Lifecycle

1. Start: The Head Node's SSHManager deploys the daemon binary via SCP and launches it over SSH. The DAEMON_PORT environment variable (or ~/.codecast/daemon.yaml) controls the listen port.
2. Running: The daemon accepts JSON-RPC requests on POST /rpc. Each request is dispatched to the appropriate handler in server.rs.
3. Shutdown: SIGTERM or SIGINT triggers graceful shutdown. session_pool.destroy_all() is called to kill all running CLI processes and clear queues. The port file at ~/.codecast/daemon.port is removed.

Environment / Config

The daemon loads configuration from ~/.codecast/daemon.yaml. Environment variables override config file values:

RPC Server (server.rs)

File: src/daemon/server.rs

Axum-based JSON-RPC server that provides the HTTP endpoint for all daemon operations. Handles method routing, SSE streaming for session.send, keepalive pings, auth middleware, and graceful shutdown.

Purpose

• Provide a single POST /rpc endpoint for all JSON-RPC methods
• Route requests to method-specific handlers based on the method field
• Stream responses for session.send via SSE (Server-Sent Events)
• Send keepalive pings to prevent idle SSH tunnel timeouts
• Apply Bearer token auth middleware when the daemon is in auth mode
• Handle graceful shutdown on SIGTERM/SIGINT

AppState

Shared application state injected into every request handler via Axum's State extractor:

#![allow(unused)]
fn main() {
pub struct AppState {
    pub session_pool: SessionPool,
    pub skill_manager: SkillManager,
    pub start_time: Instant,
    pub shutdown: Arc<Notify>,
    pub config: DaemonConfig,
    pub token_store: TokenStore,
}
}

Router

The router is built by build_router(state):

#![allow(unused)]
fn main() {
pub fn build_router(state: Arc<AppState>) -> Router {
    Router::new()
        .route("/rpc", post(handle_rpc))
        .layer(axum::middleware::from_fn_with_state(
            state.clone(),
            crate::auth::auth_middleware,
        ))
        .with_state(state)
}
}

The auth middleware runs on every request. In plain HTTP mode (default for SSH tunnel access), the middleware is a no-op pass-through. In auth mode (HTTPS), it validates the Authorization: Bearer <token> header against the token store.

Method Routing

All requests arrive at POST /rpc. The method field in the JSON body determines the handler:

Missing method field returns error -32600 (Invalid request). Unknown method returns -32601 (Method not found).

SSE Streaming (session.send)

handle_send_message is the only handler that returns an SSE stream rather than a JSON body. It uses Axum's Sse response type:

#![allow(unused)]
fn main() {
// SSE response headers (set by axum::response::sse::Sse):
// Content-Type: text/event-stream
// Cache-Control: no-cache
// Connection: keep-alive
// X-Accel-Buffering: no   (disables nginx response buffering)
}

The handler spawns a session_pool.send() task. Events from the task are forwarded onto an mpsc channel. A ReceiverStream wraps the channel receiver and drives the SSE response.

Keepalive Pings

A keepalive task runs alongside the stream, sending a Ping {} event every 30 seconds:

#![allow(unused)]
fn main() {
// Every 30 seconds:
StreamEvent::Ping {}
// Serializes to: data: {"type":"ping"}\n\n
}

These prevent idle SSH tunnel timeouts and proxy connection closures.

Client Disconnect Handling

Axum detects client disconnection when the SSE stream's Sink reports an error. The keepalive task is cancelled. If a disconnect is detected while the CLI is still running, the session pool's client_disconnect() method is called to start buffering subsequent events. These buffered events can be retrieved later via session.reconnect.

Stream Termination

The stream ends with data: [DONE]\n\n after the terminal event (result, error, or interrupted) has been sent.

Method Handlers

handle_create_session

1. Validates the required path param
2. Optionally reads mode, model, and cli_type params
3. Calls skill_manager.sync_to_project(path, cli_type) to copy shared skills
4. Calls session_pool.create(path, mode, model, cli_type)
5. Returns { "sessionId": "uuid" }

handle_send_message

See SSE Streaming section above. Params: sessionId (required), message (required).

handle_resume_session

Delegates to session_pool.resume(session_id, sdk_session_id). Returns { "ok": true, "fallback": false }.

handle_destroy_session

Delegates to session_pool.destroy(session_id). Returns { "ok": true }.

handle_list_sessions

Returns { "sessions": [...] } with SessionInfo for all sessions.

handle_set_mode

Delegates to session_pool.set_mode(session_id, mode). Returns { "ok": true }.

handle_set_model

Delegates to session_pool.set_model(session_id, model). Returns { "ok": true }.

handle_interrupt_session

Delegates to session_pool.interrupt(session_id). Returns { "ok": true, "interrupted": bool }.

handle_queue_stats

Returns QueueStats for a session: { "userPending": N, "responsePending": N, "clientConnected": bool }.

handle_reconnect

Calls session_pool.client_reconnect(session_id) and returns any buffered events. Returns { "bufferedEvents": [...] }.

handle_health_check

Returns daemon health info:

{
    "ok": true,
    "sessions": 3,
    "sessionsByStatus": { "idle": 2, "busy": 1 },
    "uptime": 3600,
    "memory": { "rss": 45, "heapUsed": 20, "heapTotal": 30 },
    "rustVersion": "1.78.0",
    "pid": 12345
}

Memory values are in megabytes. rustVersion reports the Rust toolchain version.

handle_monitor_sessions

Returns detailed session info including per-session queue stats:

{
    "sessions": [
        {
            "sessionId": "...",
            "path": "/home/user/project",
            "status": "busy",
            "mode": "auto",
            "cliType": "claude",
            "model": "claude-sonnet-4-20250514",
            "sdkSessionId": "sdk-uuid",
            "createdAt": "2026-03-29T10:00:00Z",
            "lastActivityAt": "2026-03-29T10:05:00Z",
            "queue": {
                "userPending": 1,
                "responsePending": 0,
                "clientConnected": true
            }
        }
    ],
    "totalSessions": 1,
    "uptime": 3600
}

JSON-RPC Helpers

RpcResponse is defined in types.rs with two constructors:

#![allow(unused)]
fn main() {
RpcResponse::success(result: Value, id: Option<String>) -> RpcResponse
RpcResponse::error(code: i32, message: impl Into<String>, id: Option<String>) -> RpcResponse
}

Standard error codes:

• -32600: Invalid request (missing method)
• -32601: Method not found
• -32602: Invalid params (missing required params)
• -32000: Internal/application error (session not found, path invalid, etc.)

Connection to Other Modules

• Uses SessionPool for all session lifecycle operations
• Uses SkillManager for skills sync on session.create
• Uses TokenStore via auth_middleware for request authentication
• Imports types from types.rs for request/response typing

Session Pool (session_pool.rs)

File: src/daemon/session_pool.rs

Manages CLI sessions using a per-message spawn architecture. Each user message spawns a fresh CLI subprocess via the appropriate CliAdapter, maintaining conversation continuity via session resume flags (--resume for Claude, equivalent for other CLIs).

Purpose

• Maintain a registry of session metadata (path, mode, CLI type, status, SDK session ID)
• Spawn CLI subprocesses for individual messages via the CliAdapter trait
• Convert CLI stdout JSON-lines to StreamEvent values
• Handle message queuing when the CLI is busy
• Manage process lifecycle (spawn, monitor, interrupt, kill)
• Track client connection state for response buffering

Architecture: Per-Message Spawn

Rather than keeping a long-running CLI process with stdin open, the SessionPool spawns a fresh process for each message:

# First message (no session ID yet):
claude -p "user message" --output-format stream-json --verbose \
       [--dangerously-skip-permissions]

# Subsequent messages (using --resume to continue conversation):
claude -p "user message" --output-format stream-json --verbose \
       --resume <sdkSessionId> \
       [--dangerously-skip-permissions]

The CliAdapter trait abstracts this pattern across all supported CLIs. Each adapter implements build_command() for the first message and build_resume_command() for subsequent messages.

A fresh adapter instance is created for each run_cli_process() call via create_adapter(). This ensures any per-run state (such as cumulative text tracking) is reset cleanly between message turns.

Internal Session State

#![allow(unused)]
fn main() {
struct InternalSession {
    session_id: String,
    path: String,
    mode: PermissionMode,
    cli_type: String,
    status: SessionStatus,
    sdk_session_id: Option<String>,
    created_at: DateTime<Utc>,
    last_activity_at: DateTime<Utc>,
    process: Option<Child>,       // Running CLI process (only during processing)
    queue: MessageQueue,          // Per-session message + response queue
    processing: bool,             // Whether a message is currently being processed
    model: Option<String>,        // Model name reported by CLI init event
}
}

The sessions map is wrapped in Arc<Mutex<HashMap<String, InternalSession>>> for async-safe access.

Key Methods

create(path, mode, model, cli_type) -> Result<String, String>

Creates a new session entry. Lightweight — no CLI process is spawned.

1. Resolves the path: expands ~, and expands bare project names to ~/Projects/<name>
2. Validates that the resolved path exists on the filesystem
3. Generates a UUID for the session ID
4. Inserts an InternalSession with status: Idle, no process, and a fresh MessageQueue
5. Returns the session ID

send(session_id, message) -> mpsc::Receiver<StreamEvent>

Sends a message to a session. Returns a channel receiver that yields stream events.

If the session is busy (another message in flight):

• Enqueues the message via queue.enqueue_user()
• Sends a single Queued { position } event on the receiver
• Returns immediately

If the session is idle:

• Sets status: Busy and processing: true
• Spawns a tokio task that calls run_cli_process()
• The task forwards events onto the mpsc channel

run_cli_process(session_id, message) -> impl Stream<Item = StreamEvent>

Internal method that spawns the CLI subprocess and yields events.

Adapter selection and command building:

#![allow(unused)]
fn main() {
let adapter = create_adapter(&session.cli_type);
let command = if let Some(sdk_id) = &session.sdk_session_id {
    adapter.build_resume_command(message, mode, cwd, sdk_id, model)
} else {
    adapter.build_command(message, mode, cwd, model)
};
}

Process spawn:

#![allow(unused)]
fn main() {
let mut child = command
    .current_dir(&session.path)
    .stdin(Stdio::null())
    .stdout(Stdio::piped())
    .stderr(Stdio::piped())
    .spawn()?;
}

Stdin is set to null because the prompt is passed via CLI arguments in non-interactive mode.

Output processing:

stdout is read line-by-line using tokio::io::BufReader and AsyncBufReadExt::lines(). Each line is passed to adapter.parse_output_line(), which returns zero or more StreamEvent values. Events are forwarded onto the mpsc channel.

stderr is logged at the level specified by adapter.stderr_log_level() (typically debug for Claude, warn for others).

Session ID extraction:

On the first line of output, adapter.extract_session_id() is called. If an ID is found, it is stored as session.sdk_session_id for future build_resume_command() calls.

Model name capture:

System { subtype: Some("init"), model, .. } events are used to update session.model.

Terminal events:

The generator stops forwarding events after receiving a Result, Error, or Interrupted event. The subprocess is awaited and then cleaned up.

Cleanup after process exit:

1. session.process is set to None
2. session.processing is set to false
3. session.status is set to Idle
4. If the process exited with a non-zero code, an Error event is emitted
5. If there are queued user messages, process_queued_message() is called to auto-process the next one

resume(session_id, sdk_session_id?) -> Result<ResumeResult, String>

In per-message spawn mode, this simply updates sdk_session_id so the next send() uses the resume command. Also calls queue.on_client_reconnect() to mark the client as reconnected.

Returns { ok: true, fallback: false }.

destroy(session_id) -> bool

1. Sends SIGTERM to any running CLI process
2. Waits up to 5 seconds for the process to exit; sends SIGKILL if it does not
3. Sets status: Destroyed
4. Clears the message queue
5. Removes the session from the pool

set_mode(session_id, mode) -> bool

Updates session.mode. Takes effect on the next send() call since the mode is passed to adapter.build_command().

set_model(session_id, model) -> bool

Updates session.model. Takes effect on the next send() call.

interrupt(session_id) -> Result<bool, String>

1. Sends SIGTERM to the running CLI process
2. Clears the message queue (cancels any pending messages)
3. Returns true if there was an active process to interrupt, false if the session was idle

list_sessions() -> Vec<SessionInfo>

Returns SessionInfo for all non-destroyed sessions. SessionInfo is a serializable snapshot (dates as ISO 8601 strings, no runtime-only fields).

client_disconnect(session_id) / client_reconnect(session_id) -> Vec<StreamEvent>

Proxy methods for MessageQueue client state management. Called by server.rs when the SSE connection drops or is re-established.

get_queue_stats(session_id) -> QueueStats

Returns QueueStats { user_pending, response_pending, client_connected } for a session.

destroy_all()

Destroys all sessions. Called during daemon shutdown.

Connection to Other Modules

• server.rs creates a single SessionPool instance (inside AppState) and calls its methods for all session-related RPC handlers
• Uses cli_adapter via create_adapter() to build and parse CLI subprocesses
• Uses MessageQueue for per-session message buffering and client state tracking
• Imports types from types.rs (SessionStatus, PermissionMode, StreamEvent, SessionInfo, QueueStats)

Message Queue (message_queue.rs)

File: src/daemon/message_queue.rs

Per-session message queue with three responsibilities: buffering user messages when the CLI is busy, buffering response events when the SSH connection drops, and tracking client connection state.

Purpose

• User message buffering: When Claude is processing a message, additional user messages are queued and processed in order after the current message completes.
• Response buffering: When the SSH connection (and thus the SSE stream) drops mid-response, events are buffered and can be replayed when the client reconnects via session.reconnect.
• Client connection tracking: Tracks whether the Head Node client is currently connected so the system knows whether to buffer response events.

Struct: MessageQueue

#![allow(unused)]
fn main() {
pub struct MessageQueue {
    user_pending: VecDeque<QueuedUserMessage>,
    response_pending: VecDeque<QueuedResponse>,
    client_connected: bool,  // Starts as true
}
}

Each session has its own MessageQueue instance, created when the session is created in SessionPool::create().

Data Types

QueuedUserMessage

#![allow(unused)]
fn main() {
pub struct QueuedUserMessage {
    pub message: String,
    pub timestamp: u64,  // Unix milliseconds
}
}

QueuedResponse (private)

#![allow(unused)]
fn main() {
struct QueuedResponse {
    event: StreamEvent,
    timestamp: u64,  // Unix milliseconds
}
}

User Message Buffering

enqueue_user(message: String) -> usize

Adds a user message to the back of the queue. Returns the new queue length (used as the queue position in the Queued event sent back to the user). Called by SessionPool::send() when the session is busy.

dequeue_user() -> Option<QueuedUserMessage>

Removes and returns the next user message from the front of the queue. Returns None if empty. Called by SessionPool after completing a message to auto-process the next one.

has_user_pending() -> bool

Returns true if there are queued user messages. Used by SessionPool to decide whether to call process_queued_message() after a message completes.

Response Buffering

buffer_response(event: StreamEvent, force: bool)

Buffers a response event. When force is false (normal path), events are only buffered when client_connected is false. When force is true, the event is always buffered — used by server.rs when it detects an SSE client disconnect before the session pool has been notified.

on_client_reconnect() -> Vec<StreamEvent>

Marks the client as reconnected and returns all buffered response events (draining the queue). Called by the session.reconnect RPC handler. The caller sends the returned events back to the newly reconnected client.

Client Connection State

is_client_connected() -> bool

Returns the current connection state.

on_client_disconnect()

Sets client_connected = false. After this call, response events will be buffered via buffer_response() rather than discarded.

Cleanup

clear()

Clears both the user message queue and the response buffer. Called when a session is destroyed or interrupted.

stats() -> QueueStats

Returns a QueueStats snapshot for monitoring:

#![allow(unused)]
fn main() {
pub struct QueueStats {
    pub user_pending: usize,
    pub response_pending: usize,
    pub client_connected: bool,
}
}

Used by the /status and /monitor commands to report queue depth.

Flow Example

User sends msg1  →  SessionPool: start processing msg1

User sends msg2  →  queue.enqueue_user("msg2") → returns position 1
                    SessionPool: yields Queued { position: 1 } to client
                    (msg2 sits in queue while msg1 processes)

User sends msg3  →  queue.enqueue_user("msg3") → returns position 2
                    SessionPool: yields Queued { position: 2 } to client

SSH drops        →  server.rs detects SSE close event
                    queue.on_client_disconnect()
                    (subsequent response events buffered via buffer_response(event, false))

msg1 completes   →  SessionPool: dequeue_user() → returns msg2
                    SessionPool: start processing msg2 (events buffered since disconnected)

msg2 completes   →  SessionPool: dequeue_user() → returns msg3
                    SessionPool: start processing msg3

SSH reconnects   →  client sends session.reconnect RPC
                    queue.on_client_reconnect() → returns buffered events for msg2
                    server.rs sends buffered events to client
                    client is now back in sync; msg3 is still processing

Connection to Other Modules

• session_pool.rs creates one MessageQueue per session and calls its methods for message queuing and client state management
• server.rs calls queue.on_client_disconnect() when the SSE connection closes, and uses the session.reconnect handler to call queue.on_client_reconnect()
• Imports StreamEvent and QueueStats from types.rs

Skill Manager

Skills sync is a two-stage process that spans both the Head Node and the Daemon. Understanding which component does what prevents confusion when debugging skill sync issues.

Overview

Skills are shared instruction files (CLAUDE.md, AGENTS.md, GEMINI.md) and skill documents (.claude/skills/, etc.) that provide Claude and other CLIs with reusable context for your projects.

The sync pipeline works as follows:

Local machine                     Remote machine
─────────────────────             ─────────────────────────────────────
~/.codecast/skills/          SCP  ~/.codecast/skills/
  CLAUDE.md              ──────▶    CLAUDE.md
  .claude/skills/                   .claude/skills/
    coding-standards.md               coding-standards.md

                                     ▼ (on session.create)

                                  /home/user/project/
                                    CLAUDE.md        (if not already present)
                                    .claude/skills/
                                      coding-standards.md

Stage 1: Head Node to Remote Machine (SSHManager)

File: src/head/ssh_manager.py Method: sync_skills(machine_id)

When cmd_start() or cmd_resume() is called, the BotEngine calls ssh.sync_skills() to populate the remote machine's shared skills directory.

The SSHManager copies from the local skills.shared_dir (configured in config.yaml, defaults to ~/.codecast/skills) to ~/.codecast/skills on the remote machine via SCP (asyncssh's scp support).

This stage happens once per session creation (or on every connection, depending on configuration). It ensures the daemon's source directory is populated before any session starts.

Stage 2: Remote Skills Dir to Project (SkillManager in Rust)

File: src/daemon/skill_manager.rs Struct: SkillManager

When session.create is called, server.rs calls skill_manager.sync_to_project(project_path, cli_type). This copies from ~/.codecast/skills to the specific project directory.

#![allow(unused)]
fn main() {
pub struct SkillManager {
    skills_source_dir: PathBuf,  // Default: ~/.codecast/skills
}
}

sync_to_project(project_path, cli_type) -> SyncResult

Uses the CliAdapter to determine the correct file names for the target CLI:

#![allow(unused)]
fn main() {
let adapter = create_adapter(cli_type);
let instructions_file = adapter.instructions_file();  // "CLAUDE.md", "AGENTS.md", "GEMINI.md"
let skills_dir = adapter.skills_dir();                // Some(".claude/skills/"), or None
}

Then:

1. Copies {source}/{instructions_file} to {project}/{instructions_file} — only if the target does not already exist
2. If skills_dir is Some, recursively copies {source}/{skills_dir} to {project}/{skills_dir} — skipping any file that already exists in the target

Returns SyncResult { synced: Vec<String>, skipped: Vec<String> }.

No-Overwrite Policy

The SkillManager never overwrites existing files. This is a deliberate design choice:

• Projects may have their own CLAUDE.md with project-specific instructions that should take precedence
• Shared skills provide a baseline; project-specific files override them
• After initial sync, the project's customized files are preserved across new sessions

Supported CLI Adapters

Example

Given this source structure:

~/.codecast/skills/
├── CLAUDE.md
└── .claude/
    └── skills/
        ├── coding-standards.md
        └── review-checklist.md

And this project state:

/home/user/project/
├── CLAUDE.md                    # Already exists — not overwritten
└── .claude/
    └── skills/
        └── coding-standards.md  # Already exists — not overwritten

Result:

• synced: [".claude/skills/review-checklist.md"]
• skipped: ["CLAUDE.md (already exists)", ".claude/skills/coding-standards.md (already exists)"]

Debugging Skill Sync

If skills are not appearing in a project:

1. Confirm the local ~/.codecast/skills/ directory exists and contains the expected files
2. After running /start, SSH into the remote machine and check ~/.codecast/skills/ — if empty, Stage 1 (SCP) failed
3. If Stage 1 succeeded but the project directory is missing files, check the daemon logs for sync_to_project output — the files may already exist in the project (no-overwrite policy)
4. To force-resync a specific file, delete it from the project directory on the remote machine and run /start again

Connection to Other Modules

• ssh_manager.py (Head Node) is responsible for populating ~/.codecast/skills on the remote machine
• server.rs calls skill_manager.sync_to_project() during session.create
• cli_adapter/mod.rs provides instructions_file() and skills_dir() per CLI type

Type Definitions (types.rs)

File: src/daemon/types.rs

Central type definitions for the daemon's RPC protocol, session management, and stream events. All types use serde for JSON serialization/deserialization.

RPC Protocol Types

RpcRequest

#![allow(unused)]
fn main() {
#[derive(Debug, Deserialize)]
pub struct RpcRequest {
    pub method: Option<String>,
    pub params: Option<Value>,  // serde_json::Value
    pub id: Option<String>,
}
}

RpcResponse

#![allow(unused)]
fn main() {
#[derive(Debug, Serialize)]
pub struct RpcResponse {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub result: Option<Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<RpcError>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub id: Option<String>,
}

#[derive(Debug, Serialize)]
pub struct RpcError {
    pub code: i32,
    pub message: String,
}
}

Constructors:

#![allow(unused)]
fn main() {
RpcResponse::success(result: Value, id: Option<String>) -> RpcResponse
RpcResponse::error(code: i32, message: impl Into<String>, id: Option<String>) -> RpcResponse
}

Session Types

SessionStatus

#![allow(unused)]
fn main() {
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum SessionStatus {
    Idle,
    Busy,
    Error,
    Destroyed,
}
}

PermissionMode

#![allow(unused)]
fn main() {
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum PermissionMode {
    #[default]
    Auto,
    Code,
    Plan,
    Ask,
}
}

PermissionMode implements Default as Auto.

to_claude_flags(self) -> Vec<&'static str>

Maps a PermissionMode to Claude CLI flags. Currently only Auto has a corresponding flag:

#![allow(unused)]
fn main() {
impl PermissionMode {
    pub fn to_claude_flags(self) -> Vec<&'static str> {
        match self {
            PermissionMode::Auto => vec!["--dangerously-skip-permissions"],
            _ => vec![],
        }
    }
}
}

Each CliAdapter implementation decides how to use the mode value — the ClaudeAdapter calls to_claude_flags(), while other adapters may map modes differently.

Stream Event Types

StreamEvent is a tagged enum serialized with #[serde(tag = "type", rename_all = "lowercase")]. The type field in JSON identifies the variant.

#![allow(unused)]
fn main() {
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "lowercase")]
pub enum StreamEvent {
    Text {
        content: Option<String>,
        raw: Option<Value>,
    },
    #[serde(rename = "tool_use")]
    ToolUse {
        tool: Option<String>,
        input: Option<Value>,
        message: Option<String>,
        raw: Option<Value>,
    },
    Result {
        session_id: Option<String>,
        raw: Option<Value>,
    },
    Queued {
        position: usize,
    },
    Error {
        message: String,
    },
    System {
        subtype: Option<String>,
        session_id: Option<String>,
        model: Option<String>,
        raw: Option<Value>,
    },
    Partial {
        content: Option<String>,
        raw: Option<Value>,
    },
    Ping {},
    Interrupted {},
}
}

All Option fields are skipped in JSON serialization when None (#[serde(skip_serializing_if = "Option::is_none")]).

Variant Reference

is_terminal() -> bool

#![allow(unused)]
fn main() {
pub fn is_terminal(&self) -> bool {
    matches!(
        self,
        StreamEvent::Result { .. } | StreamEvent::Error { .. } | StreamEvent::Interrupted {}
    )
}
}

session_id() -> Option<&str>

Extracts the session_id field from System or Result variants. Returns None for all other variants.

Session Info Types

SessionInfo

Serializable session snapshot used in session.list and monitor.sessions responses. All DateTime fields are serialized as ISO 8601 strings. Uses camelCase field names in JSON.

#![allow(unused)]
fn main() {
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionInfo {
    pub session_id: String,
    pub path: String,
    pub status: SessionStatus,
    pub mode: PermissionMode,
    pub cli_type: String,          // "claude", "codex", "gemini", "opencode"
    pub sdk_session_id: Option<String>,
    pub model: Option<String>,
    pub created_at: String,        // ISO 8601
    pub last_activity_at: String,  // ISO 8601
}
}

QueueStats

#![allow(unused)]
fn main() {
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct QueueStats {
    pub user_pending: usize,
    pub response_pending: usize,
    pub client_connected: bool,
}
}

Connection to Other Modules

• All daemon modules import types from this file
• server.rs uses RpcRequest, RpcResponse, PermissionMode
• session_pool.rs uses SessionStatus, PermissionMode, StreamEvent, SessionInfo, QueueStats
• message_queue.rs uses StreamEvent, QueueStats
• cli_adapter/mod.rs and each adapter module use PermissionMode, StreamEvent

JSON-RPC Protocol

The daemon exposes a single HTTP endpoint at POST /rpc that accepts JSON-RPC requests. All communication between the Head Node and the daemon uses this protocol.

Endpoint

POST http://127.0.0.1:{port}/rpc
Content-Type: application/json

The daemon only binds to 127.0.0.1 by default. Access is through SSH port forwarding managed by the Head Node's SSHManager. In auth mode (HTTPS), access may be direct with a Bearer token.

Request Format

{
    "method": "session.create",
    "params": { "path": "/home/user/project", "mode": "auto" },
    "id": "optional-request-id"
}

Response Format

Success:

{
    "result": { "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890" },
    "id": "optional-request-id"
}

Error:

{
    "error": { "code": -32602, "message": "Missing required param: path" },
    "id": "optional-request-id"
}

Error Codes

Methods

session.create

Create a new Claude session. Lightweight — no CLI process is spawned until a message is sent.

Request:

{
    "method": "session.create",
    "params": {
        "path": "/home/user/project",
        "mode": "auto",
        "model": "claude-sonnet-4-20250514",
        "cli_type": "claude"
    }
}

Response:

{
    "result": {
        "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
    }
}

Side effects:

• Skills are synced to the project directory via skill_manager.sync_to_project()
• The path is validated to exist on the filesystem

session.send

Send a message to a Claude session. Unlike other methods, this returns an SSE stream instead of a JSON response.

Request:

{
    "method": "session.send",
    "params": {
        "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        "message": "What files are in this project?"
    }
}

Response: SSE stream (Content-Type: text/event-stream)

data: {"type":"system","subtype":"init","model":"claude-sonnet-4-20250514","session_id":"sdk-123"}

data: {"type":"partial","content":"Let me "}

data: {"type":"partial","content":"look at the files..."}

data: {"type":"tool_use","tool":"Bash","input":{"command":"ls -la"}}

data: {"type":"text","content":"Here are the files in this project:\n\n- src/\n- Cargo.toml"}

data: {"type":"result","session_id":"sdk-session-uuid-here"}

data: [DONE]

If the session is busy processing another message:

data: {"type":"queued","position":1}

data: [DONE]

See SSE Stream Events for full event type documentation.

Side effects:

• Spawns a CLI subprocess for the duration of the message
• Captures the SDK session ID from the result event for future --resume
• After completion, auto-processes the next queued message if any

session.resume

Resume a previously detached session. Updates the SDK session ID so the next send() uses --resume (or equivalent).

Request:

{
    "method": "session.resume",
    "params": {
        "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        "sdkSessionId": "sdk-session-uuid-here"
    }
}

Response:

{
    "result": {
        "ok": true,
        "fallback": false
    }
}

session.destroy

Destroy a session and kill any running CLI process.

Request:

{
    "method": "session.destroy",
    "params": {
        "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
    }
}

Response:

{
    "result": { "ok": true }
}

Side effects:

• Sends SIGTERM to any running CLI process (SIGKILL after 5 seconds if not exited)
• Clears message queues
• Removes the session from the pool

session.list

List all sessions on the daemon.

Request:

{
    "method": "session.list",
    "params": {}
}

No parameters required.

Response:

{
    "result": {
        "sessions": [
            {
                "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
                "path": "/home/user/project",
                "status": "idle",
                "mode": "auto",
                "cliType": "claude",
                "sdkSessionId": "sdk-uuid",
                "model": "claude-sonnet-4-20250514",
                "createdAt": "2026-03-29T10:00:00Z",
                "lastActivityAt": "2026-03-29T10:05:00Z"
            }
        ]
    }
}

session.set_mode

Change the permission mode for a session. Takes effect on the next message (next process spawn).

Request:

{
    "method": "session.set_mode",
    "params": {
        "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        "mode": "code"
    }
}

Response:

{
    "result": { "ok": true }
}

session.set_model

Override the model for a session. Takes effect on the next message.

Request:

{
    "method": "session.set_model",
    "params": {
        "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        "model": "claude-opus-4-20250115"
    }
}

Response:

{
    "result": { "ok": true }
}

session.interrupt

Interrupt Claude's current operation. Sends SIGTERM to the running CLI process.

Request:

{
    "method": "session.interrupt",
    "params": {
        "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
    }
}

Response:

{
    "result": {
        "ok": true,
        "interrupted": true
    }
}

Side effects:

• Sends SIGTERM to the CLI process
• Clears the message queue

session.queue_stats

Get message queue statistics for a session.

Request:

{
    "method": "session.queue_stats",
    "params": {
        "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
    }
}

Response:

{
    "result": {
        "userPending": 2,
        "responsePending": 0,
        "clientConnected": true
    }
}

session.reconnect

Reconnect to a session and retrieve any buffered response events that arrived while the client was disconnected.

Request:

{
    "method": "session.reconnect",
    "params": {
        "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
    }
}

Response:

{
    "result": {
        "bufferedEvents": [
            {"type": "partial", "content": "Here is "},
            {"type": "text", "content": "Here is the answer."},
            {"type": "result", "session_id": "sdk-uuid"}
        ]
    }
}

Side effects:

• Marks the client as reconnected
• Clears the response buffer after replay

health.check

Check daemon health and system information.

Request:

{
    "method": "health.check",
    "params": {}
}

Response:

{
    "result": {
        "ok": true,
        "sessions": 3,
        "sessionsByStatus": {
            "idle": 2,
            "busy": 1
        },
        "uptime": 3600,
        "memory": {
            "rss": 45,
            "heapUsed": 20,
            "heapTotal": 30
        },
        "rustVersion": "1.78.0",
        "pid": 12345
    }
}

monitor.sessions

Get detailed monitoring information for all sessions, including per-session queue stats.

Request:

{
    "method": "monitor.sessions",
    "params": {}
}

Response:

{
    "result": {
        "sessions": [
            {
                "sessionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
                "path": "/home/user/project",
                "status": "busy",
                "mode": "auto",
                "cliType": "claude",
                "model": "claude-sonnet-4-20250514",
                "sdkSessionId": "sdk-uuid",
                "createdAt": "2026-03-29T10:00:00Z",
                "lastActivityAt": "2026-03-29T10:05:00Z",
                "queue": {
                    "userPending": 1,
                    "responsePending": 0,
                    "clientConnected": true
                }
            }
        ],
        "totalSessions": 1,
        "uptime": 3600
    }
}

SSE Stream Events

When the Head Node sends a session.send RPC call, the daemon responds with a Server-Sent Events (SSE) stream. This document describes all event types that can appear in the stream.

SSE Format

Events are sent as data: lines with JSON payloads, separated by double newlines:

data: {"type":"partial","content":"Hello"}

data: {"type":"text","content":"Hello, world!"}

data: [DONE]

The stream ends with data: [DONE] (not a JSON payload). The JSON payloads are serialized StreamEvent values from types.rs, using the type field as a tag.

Terminal Events

Three event types signal the end of a message exchange. The daemon stops sending events after any of these:

Event Types

system

System events provide metadata about the session. The most common is the init subtype, sent at the start of each message exchange when Claude starts up.

{
    "type": "system",
    "subtype": "init",
    "session_id": "sdk-session-uuid",
    "model": "claude-sonnet-4-20250514"
}

The Head Node uses the init event to display a "Connected to model | Mode: mode" message on the first interaction with a session.

partial

Streaming text deltas. These arrive as the CLI generates text, providing real-time output that can be rendered progressively.

{
    "type": "partial",
    "content": "Let me "
}

The Head Node accumulates partial deltas in a buffer and periodically updates the chat message with the current buffer plus a ▌ cursor indicator. When a complete text event arrives, it replaces the accumulated partials.

partial events can also carry partial_json content during tool use streaming (JSON being assembled incrementally). The Head Node renders these the same way as text partials.

text

A complete text block from the CLI. Represents a finished content block in the response.

{
    "type": "text",
    "content": "Here is the complete analysis of your project...",
    "raw": { ... }
}

If partial events were being accumulated, the text event's content replaces the partial buffer. If no partials were sent (e.g., a short response), the text is sent as a new message.

tool_use

Indicates the CLI is invoking a tool (file write, bash command, web fetch, etc.).

{
    "type": "tool_use",
    "tool": "Write",
    "input": {
        "file_path": "/home/user/project/README.md",
        "content": "# My Project\n..."
    },
    "raw": { ... }
}

With a status message (from tool progress events):

{
    "type": "tool_use",
    "tool": "Bash",
    "message": "Running command...",
    "raw": { ... }
}

Special case: AskUserQuestion

When tool is "AskUserQuestion", the input field contains a structured question list:

{
    "type": "tool_use",
    "tool": "AskUserQuestion",
    "input": [
        {
            "header": "Which framework should I use?",
            "options": [
                {"description": "FastAPI (async, modern)"},
                {"description": "Flask (simple, synchronous)"}
            ],
            "multiSelect": false
        }
    ]
}

The Head Node passes this to format_ask_user_question() and then calls adapter.send_question() to display platform-native interactive buttons.

result

Indicates the CLI has finished processing the message. Contains the SDK session ID needed for conversation continuity.

{
    "type": "result",
    "session_id": "sdk-session-uuid-here",
    "raw": { ... }
}

The Head Node captures session_id and stores it via router.update_sdk_session() for future --resume calls.

This is a terminal event.

queued

Sent immediately when the session is busy and the new message has been queued.

{
    "type": "queued",
    "position": 2
}

The Head Node displays: "Message queued (position: 2). Claude is busy with a previous request."

When the queued message is eventually processed, its events will flow through a new SSE stream from the implicit next session.send call (initiated by the daemon after the previous message completes). If the client is disconnected at that point, the events are buffered for session.reconnect.

error

An error occurred during processing.

{
    "type": "error",
    "message": "Claude process exited abnormally (code=1)"
}

Common error sources:

• CLI process exiting with non-zero code
• CLI process spawn failure (binary not found, permission denied)
• Stream idle timeout (no events for an extended period)
• SSH connection loss detected by the daemon

This is a terminal event.

ping

Keepalive event sent every 30 seconds to prevent idle SSH tunnel timeouts.

{
    "type": "ping"
}

The Head Node ignores these events. They exist solely to keep the HTTP connection alive through SSH tunnels and proxies that close idle connections.

interrupted

Sent when the operation was interrupted, either via session.interrupt or by an external SIGTERM to the CLI process.

{
    "type": "interrupted"
}

This is a terminal event.

Event Flow Examples

Simple Text Response

data: {"type":"system","subtype":"init","model":"claude-sonnet-4-20250514","session_id":"sdk-123"}
data: {"type":"partial","content":"The "}
data: {"type":"partial","content":"answer "}
data: {"type":"partial","content":"is 42."}
data: {"type":"text","content":"The answer is 42."}
data: {"type":"result","session_id":"sdk-123"}
data: [DONE]

Tool Use with Text

data: {"type":"system","subtype":"init","model":"claude-sonnet-4-20250514"}
data: {"type":"partial","content":"Let me check..."}
data: {"type":"tool_use","tool":"Bash","input":{"command":"ls -la"}}
data: {"type":"tool_use","tool":"Bash","message":"Running command..."}
data: {"type":"partial","content":"Here are the files:\n"}
data: {"type":"partial","content":"- src/\n- Cargo.toml"}
data: {"type":"text","content":"Here are the files:\n- src/\n- Cargo.toml"}
data: {"type":"result","session_id":"sdk-456"}
data: [DONE]

AskUserQuestion

data: {"type":"system","subtype":"init","model":"claude-sonnet-4-20250514"}
data: {"type":"partial","content":"I need to clarify a few things."}
data: {"type":"tool_use","tool":"AskUserQuestion","input":[{"header":"Which approach?","options":[{"description":"Option A"},{"description":"Option B"}],"multiSelect":false}]}
data: {"type":"result","session_id":"sdk-789"}
data: [DONE]

Queued Message

data: {"type":"queued","position":1}
data: [DONE]

Error During Processing

data: {"type":"system","subtype":"init","model":"claude-sonnet-4-20250514"}
data: {"type":"partial","content":"Let me "}
data: {"type":"error","message":"Claude process exited abnormally (code=1)"}
data: [DONE]

Keepalive During Long Operation

data: {"type":"system","subtype":"init","model":"claude-sonnet-4-20250514"}
data: {"type":"partial","content":"Analyzing..."}
data: {"type":"ping"}
data: {"type":"partial","content":" the codebase structure"}
data: {"type":"ping"}
data: {"type":"tool_use","tool":"Glob","input":{"pattern":"**/*.rs"}}
data: {"type":"text","content":"I found 15 Rust files..."}
data: {"type":"result","session_id":"sdk-789"}
data: [DONE]

Interrupted Operation

data: {"type":"system","subtype":"init","model":"claude-sonnet-4-20250514"}
data: {"type":"partial","content":"Let me analyze this large codebase..."}
data: {"type":"tool_use","tool":"Glob","input":{"pattern":"**/*"}}
data: {"type":"interrupted"}
data: [DONE]