1. Scopo e il requisito status-visibility

Un Channel è l'adapter tra un protocollo di messaging esterno e il gateway. Fa due mestieri: ingest (traduce input esterno in ingest(channel, sender, message)) e render (traduce la risposta di myclaw nelle convenzioni del protocollo). Il requisito non opzionale di fase 1 è status visibility: nessun utente aspetta muto.

Cosa copre questo doc

• Il Protocol Channel.
• Le due prime implementazioni: CLI e Telegram.
• I pattern di status visibility per canale.
• La formattazione dei messaggi (plain, markdown, inline buttons).

Cosa non copre

• L'ingestion pipeline del gateway (→ gateway).
• Il flusso di approvazione (→ approval_ux).
• Il pairing (→ pairing.html).
• Canali futuri (voce, Signal, MQTT) — stesso Protocol, impl separate.

2. Il Protocol Channel

Un Channel è un oggetto asincrono con due responsabilità: pump input verso il gateway, push output verso il mondo. Espone capability flag che il gateway usa per dimensionare l'esperienza (es. "supporta bottoni inline?" → approvazioni con UI ricca vs testo puro).

from typing import Protocol, Literal
from dataclasses import dataclass
from uuid import UUID

@dataclass
class ChannelCapabilities:
    supports_inline_buttons: bool     # Telegram, Signal: sì. CLI: no.
    supports_streaming: bool          # CLI WS: sì. Telegram: edit message.
    supports_typing_indicator: bool   # Telegram: sì. CLI: simulato con spinner.
    supports_markdown: Literal["none", "basic", "full"]
    supports_voice: bool              # futuro
    max_message_length: int           # Telegram: 4096
    max_messages_per_minute: int      # rate-limit outbound

class Channel(Protocol):
    name: str                         # "cli" | "telegram" | ...
    capabilities: ChannelCapabilities

    async def start(self, gateway: Gateway) -> None:
        """Avvia pump (long-poll, listener, etc)."""
        ...

    async def shutdown(self) -> None: ...

    async def send(
        self,
        sender: str,
        content: OutboundMessage,
    ) -> str:
        """
        Invia un messaggio al sender. Ritorna un message_id opaco che
        permette edit successivi (per status updates e approval buttons).
        """
        ...

    async def edit(
        self,
        sender: str,
        message_id: str,
        content: OutboundMessage,
    ) -> None: ...

    async def show_typing(self, sender: str, seconds: float = 3.0) -> None:
        """Mostra indicatore di attività. No-op se capability lo vieta."""
        ...

    async def present_approval(
        self,
        sender: str,
        request: ApprovalRequest,
    ) -> str:
        """Renderizza una ApprovalRequest secondo le capability del canale."""
        ...

    def default_approval_timeout(self) -> timedelta:
        """
        Timeout di default per una richiesta di approvazione su questo canale.
        CLI: 30s · Telegram DM: 2m · canale di casa: 5m · voce: 15s.
        Fonte canonica della tabella: approval_ux.html §8.
        Il Gateway legge da qui, non hardcoda valori.
        """
        ...

Message model: OutboundMessage

@dataclass
class OutboundMessage:
    text: str
    format: Literal["plain", "markdown"] = "plain"
    buttons: list[Button] | None = None      # None se no inline
    replace_previous: bool = False           # vedi nota sotto
    replace_target: str | None = None        # message_id da rimpiazzare
    attachment: Attachment | None = None     # file / image, futuro

@dataclass
class Button:
    label: str
    callback_data: str
    style: Literal["primary", "secondary", "danger"] = "secondary"
    disabled_until_ms: int = 0               # pausa lettura approval_ux §5

3. CLI: Unix socket + WebSocket streaming

Il canale CLI è quello che Roberto userà dal terminale del PC. Massima interattività, supporto streaming, latenza sub-5ms.

Architettura

1. Binario myclaw chat (in src/myclaw/cli/chat.py).
2. Alla partenza, connette a /tmp/myclaw.sock (Unix domain socket, chmod 600, uid check).
3. Parla HTTP sul socket (richiesta iniziale) + upgrade a WebSocket su /ws/internal.
4. Il WebSocket riceve eventi tipizzati del loop ReAct: thought_start, thought_partial_text, tool_call_start, tool_call_result, final_response.
5. Li renderizza con spinner + print progressivi.

Rendering del loop in CLI

$ myclaw chat
myclaw · supervised · session s_4a7b...

> riassumi gli errori di stamattina

◐ sto cercando il log di systemd...
  tool: shell_run(journalctl --since "today" --priority err)
✓ letto (42 righe, 1.3KB)

◐ sto riassumendo...

Stamattina ci sono stati 3 errori:
- alle 08:14, wpa_supplicant ha perso l'associazione (ri-collegato alle 08:14:30)
- alle 09:22, docker ha fallito a pullare un'immagine (network timeout)
- alle 11:04, smartd ha rilevato un contatore SMART anomalo su /dev/sda

>

Lo spinner (◐◓◑◒) gira durante ogni passo. Il nome del tool è sempre esposto (§5). I tool_result sono riassunti in una riga.

Streaming del final_response

Il testo della risposta finale arriva in streaming token-per-token (SSE da agent_runtime, inoltrato via WS). Il CLI stampa on-the-fly per l'effetto "l'agente sta parlando". Latenza prima-lettera < 1s su frontier model.

4. Telegram: long-poll + inline buttons

Bot setup

1. Creazione bot via BotFather, ottenimento token.
2. Token in config/secrets.env come MYCLAW_TELEGRAM_TOKEN.
3. Bot privacy mode ON (il bot riceve solo messaggi che lo menzionano esplicitamente in gruppo).
4. Comandi registrati: /start, /help, /undo, /bye.

Ingestion loop

# src/myclaw/channels/telegram.py (schematico)
async def pump(self, gateway):
    offset = 0
    while not self._stopped:
        updates = await self._api("getUpdates", offset=offset, timeout=25)
        for upd in updates:
            offset = upd["update_id"] + 1
            if "message" in upd:
                await self._handle_message(gateway, upd["message"])
            elif "callback_query" in upd:
                await self._handle_callback(gateway, upd["callback_query"])

Presentazione di una ApprovalRequest in Telegram

Il canale riceve present_approval(sender, request) dal gateway e compone un messaggio con inline buttons secondo le specifiche di approval_ux §4:

# Messaggio:
Vuoi che scarichi?
📄 rapporto_marzo.pdf
da drive.google.com
→ ~/downloads/ (~2.4 MB)
*reversibile · classe: fs_write:~/downloads/**

# Inline keyboard:
[✅ attendi 3s...]  (disabled, timer sostituisce progressivamente)
[❌ No]
[✓ approva classe per 10 min]

Il "disabled per 3s" è simulato editando il messaggio dopo 3s con il pulsante abilitato (Telegram non ha veri "disabled buttons", si fa con l'edit). Contatore visibile all'utente: "attendi 2s..." → "attendi 1s..." → "Approva ✅".

Outbound rate limit

Telegram accetta max ~30 messaggi/secondo. Il canale fa il throttling interno con una token-bucket queue. Per un singolo utente in casa questo non sarà mai un problema, ma il codice è ready per aggiungere famiglia.

5. Come si mostra "sto pensando": per canale

Il gateway emette eventi del loop in tempo reale. Ogni canale ha la responsabilità di tradurli in feedback visibile all'utente. La tabella seguente mostra, per tipo di evento, cosa fa ciascun canale. Questa è la reificazione concreta della critica #4.

6. Formattazione dei messaggi

Il runtime emette testo in un formato canonico interno (chiamato MyclawMD: sottoinsieme markdown + marker speciali per status). Il canale traduce.

7. Contratto Python

Oltre ai Channel e tipi già definiti sopra:

class ChannelRegistry(Protocol):
    """Registra le implementazioni Channel attive al boot del gateway."""
    def register(self, channel: Channel) -> None: ...
    def get(self, name: str) -> Channel: ...
    def list_all(self) -> list[Channel]: ...

class ChannelEvent(Protocol):
    """Evento che il canale riceve dal gateway per aggiornare l'utente."""
    event_type: str
    trace_id: UUID
    sender: str
    payload: dict

# Tipi di evento emessi dal runtime verso i canali:
# - thought_start
# - thought_stream(partial_text)
# - tool_call_start(name, args_summary)
# - tool_call_result(outcome_summary, error=None)
# - approval_required(request_id, request)
# - approval_resolved(request_id, granted, via)
# - final_response_stream(partial_text)
# - final_response_complete(text)
# - policy_denied(reason)
# - budget_hit(current, hard_cap)
# - trace_closed(outcome, cost, wall_ms)

Eccezioni

8. Alternative considerate

9. Test di conformità

10. Riferimenti

Continua a leggere

myclaw — channel microprogettazione v1.0 — 2026-04-21
Secondo dei 4 classici. Prossimo: tool.html.