feat(channels): answer channels — ask a human from the loop without stopping it

README.md

@@ -137,6 +137,29 @@ Specs are **constraint-oriented**: define what to do, what's off-limits, and mak
 > If you can write a shell command that verifies "done", it's a good owloop task.
 > If "done" requires a human to look and decide, it's not.
 
+## Answer Channels
+
+When the loop hits a `DECIDE` question overnight, it can ask you on your phone
+instead of stopping cold:
+
+```bash
+export OWLOOP_TELEGRAM_BOT_TOKEN=...   # from @BotFather
+export OWLOOP_TELEGRAM_CHAT_ID=...     # your chat with the bot
+owloop run --channels telegram --question-wait 30
+```
+
+- **Telegram** is bidirectional (long-polling — no public endpoint needed): the
+  question arrives with inline buttons; tap one, or reply `1` / `/skip` /
+  `/steer <guidance>`, and the loop applies your decision at the next iteration.
+- **ntfy** (`OWLOOP_NTFY_TOPIC`) is notify-only: pushes for questions, blockers,
+  and the end-of-run summary.
+- During quiet hours (22:00–08:00) questions are rendered as a safe binary —
+  approve the agent's recommendation or skip until morning. Both answers are
+  harmless: everything stays on the isolated branch, nothing merges overnight.
+- Only your configured chat is heard; replies from anyone else are silently
+  dropped. If nobody answers before `--question-wait` expires, the safe default
+  applies and the full question is preserved in `.owloop/questions/` for review.
+
 ## Compared To
 
 | | owloop | Claude Code `/goal` |

src/owloop/channels.py

@@ -0,0 +1,355 @@
+"""Answer channels — how the loop asks a human and hears back.
+
+Design (issue #53, Layer 6): the engine writes questions to disk first and
+treats channels as best-effort transport. Channel I/O must never interrupt the
+loop — ``post``/``notify``/``poll_replies`` swallow transport errors and the
+engine proceeds regardless.
+
+Night mode ("the drunk test"): inside quiet hours a question is always
+rendered as the same recommendation-first binary — approve the agent's plan or
+skip until morning — because both answers are safe by construction (worktree
+branch only, nothing merges overnight). The full multi-option question stays
+on disk for the sober morning. Outside quiet hours the full option list is
+sent.
+
+Transports are chosen for a dev machine with no public HTTP endpoint:
+Telegram's ``getUpdates`` long-poll is bidirectional with stdlib urllib only;
+ntfy.sh is a one-line notify. Only senders on the allowlist are heard —
+everyone else is silently dropped.
+"""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+from collections.abc import Mapping
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
+from owloop.questions import SKIP_OPTION_KEY, Question
+
+HTTP_TIMEOUT = 15  # seconds, on top of any long-poll timeout
+
+# Message body budget: keep under Telegram's 4096-char limit with headroom.
+MESSAGE_BUDGET = 3800
+
+STEER_VERB = "/steer"
+SKIP_VERB = "/skip"
+
+
+@dataclass
+class Reply:
+    """One inbound human message, normalized across channels."""
+
+    sender: str
+    text: str
+    question_id: str = ""  # explicit correlation (button/id-in-text), if any
+    option_key: str = ""  # set when the reply is a button press
+
+
+def is_night(hour: int, quiet_hours: tuple[int, int] = (22, 8)) -> bool:
+    """True when ``hour`` falls inside the quiet window (start may wrap midnight)."""
+    start, end = quiet_hours
+    if start == end:
+        return False
+    if start < end:
+        return start <= hour < end
+    return hour >= start or hour < end
+
+
+def _deadline_text(question: Question) -> str:
+    deadline = datetime.now() + timedelta(minutes=question.deadline_minutes)
+    return deadline.strftime("%H:%M")
+
+
+def render_question(question: Question, *, night: bool) -> tuple[str, list[tuple[str, str]]]:
+    """Render a question to plain text + a button list of (option key, label).
+
+    Night mode collapses to the safe binary; day mode lists every option.
+    Options and the default line are never truncated — context is (see
+    ``ensure_quality`` budgets).
+    """
+    header = f"🦉 owloop needs a decision  [{question.id}]"
+    spec_line = f"Spec: {question.spec} (iteration {question.iteration})" if question.spec else ""
+    deadline_line = (
+        f"⏰ No answer by {_deadline_text(question)} → the loop applies the safe default "
+        f"and this question tops the morning report."
+    )
+
+    if night:
+        # ensure_quality guarantees a recommendation; fall back defensively.
+        recommended = question.recommended_option() or question.options[0]
+        lines = [header]
+        if spec_line:
+            lines.append(spec_line)
+        lines.append("")
+        lines.append(f"I recommend: {recommended.label}")
+        if recommended.consequence:
+            lines.append(recommended.consequence)
+        lines.append("Everything stays on the branch for your morning review — nothing merges tonight.")
+        lines.append("")
+        lines.append(deadline_line)
+        lines.append(f"Reply: {recommended.key} to approve · {SKIP_VERB} to skip · {STEER_VERB} <text>")
+        buttons = [
+            (recommended.key, "👍 Go ahead"),
+            (SKIP_OPTION_KEY, "⏭️ Skip, ask me tomorrow"),
+        ]
+        return "\n".join(lines), buttons
+
+    lines = [header]
+    if spec_line:
+        lines.append(spec_line)
+    lines.append("")
+    lines.append(question.title)
+    if question.context:
+        lines.append("")
+        lines.append(question.context)
+    if question.tried:
+        lines.append("")
+        lines.append(f"Already tried: {question.tried}")
+    lines.append("")
+    for option in question.options:
+        marker = " ← recommended" if option.recommended else ""
+        lines.append(f"{option.key}) {option.label}{marker}")
+        if option.consequence:
+            lines.append(f"    {option.consequence}")
+    if question.branch:
+        lines.append("")
+        lines.append(f"Branch: {question.branch}")
+    lines.append("")
+    lines.append(deadline_line)
+    keys = " / ".join(option.key for option in question.options)
+    lines.append(f"Reply: {keys} · {STEER_VERB} <text>")
+
+    text = "\n".join(lines)
+    if len(text) > MESSAGE_BUDGET:
+        # Context is the only expendable part; options/default already fit.
+        overflow = len(text) - MESSAGE_BUDGET
+        trimmed = question.context[: max(0, len(question.context) - overflow - 1)] + "…"
+        text = text.replace(question.context, trimmed, 1)
+    buttons = [(option.key, option.label[:32]) for option in question.options]
+    return text, buttons
+
+
+class AnswerChannel:
+    """Base transport. Notify-only channels leave ``bidirectional`` False."""
+
+    name = "base"
+    bidirectional = False
+
+    def notify(self, text: str) -> bool:
+        raise NotImplementedError
+
+    def post_question(self, question: Question, text: str, buttons: list[tuple[str, str]]) -> bool:
+        return self.notify(text)
+
+    def poll_replies(self, timeout: float) -> list[Reply]:
+        return []
+
+    def ack(self, question: Question, text: str) -> None:
+        self.notify(text)
+
+
+class TelegramChannel(AnswerChannel):
+    """Telegram Bot API over stdlib urllib; long-poll, no public endpoint.
+
+    Only messages from the configured chat id are heard; everything else is
+    silently dropped (allowlist model, cf. Claude Code Channels).
+    """
+
+    name = "telegram"
+    bidirectional = True
+
+    def __init__(self, token: str, chat_id: str, api_base: str = "https://api.telegram.org"):
+        self.token = token
+        self.chat_id = str(chat_id)
+        self.api_base = api_base.rstrip("/")
+        self._offset = 0
+        self._question_messages: dict[str, int] = {}
+        self.last_error: str = ""
+
+    def _call(self, method: str, payload: dict, timeout: float = HTTP_TIMEOUT) -> dict | list | None:
+        url = f"{self.api_base}/bot{self.token}/{method}"
+        request = urllib.request.Request(
+            url,
+            data=json.dumps(payload).encode("utf-8"),
+            headers={"Content-Type": "application/json"},
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=timeout) as response:
+                data = json.loads(response.read().decode("utf-8"))
+        except (urllib.error.URLError, OSError, ValueError) as exc:
+            self.last_error = str(exc)
+            return None
+        if not isinstance(data, dict) or not data.get("ok"):
+            self.last_error = str(data)
+            return None
+        result = data.get("result")
+        return result if isinstance(result, (dict, list)) else {"value": result}
+
+    def notify(self, text: str) -> bool:
+        return self._call("sendMessage", {"chat_id": self.chat_id, "text": text}) is not None
+
+    def post_question(self, question: Question, text: str, buttons: list[tuple[str, str]]) -> bool:
+        keyboard = {
+            "inline_keyboard": [
+                [{"text": label, "callback_data": f"q:{question.id}:{key}"}] for key, label in buttons
+            ]
+        }
+        result = self._call(
+            "sendMessage",
+            {"chat_id": self.chat_id, "text": text, "reply_markup": keyboard},
+        )
+        if isinstance(result, dict) and "message_id" in result:
+            self._question_messages[question.id] = int(result["message_id"])
+            return True
+        return result is not None
+
+    def poll_replies(self, timeout: float) -> list[Reply]:
+        result = self._call(
+            "getUpdates",
+            {"offset": self._offset, "timeout": int(timeout)},
+            timeout=timeout + HTTP_TIMEOUT,
+        )
+        if not isinstance(result, list):
+            return []
+        replies: list[Reply] = []
+        for update in result:
+            if not isinstance(update, dict):
+                continue
+            self._offset = max(self._offset, int(update.get("update_id", 0)) + 1)
+            reply = self._parse_update(update)
+            if reply is not None:
+                replies.append(reply)
+        return replies
+
+    def _parse_update(self, update: dict) -> Reply | None:
+        callback = update.get("callback_query")
+        if isinstance(callback, dict):
+            chat = callback.get("message", {}).get("chat", {})
+            if str(chat.get("id", "")) != self.chat_id:
+                return None  # not on the allowlist: silently dropped
+            self._call("answerCallbackQuery", {"callback_query_id": callback.get("id", "")})
+            data = str(callback.get("data", ""))
+            parts = data.split(":", 2)
+            if len(parts) == 3 and parts[0] == "q":
+                sender = str(callback.get("from", {}).get("username", "") or chat.get("id", ""))
+                return Reply(sender=sender, text=data, question_id=parts[1], option_key=parts[2])
+            return None
+
+        message = update.get("message")
+        if not isinstance(message, dict):
+            return None
+        if str(message.get("chat", {}).get("id", "")) != self.chat_id:
+            return None  # not on the allowlist: silently dropped
+        text = str(message.get("text", "")).strip()
+        if not text:
+            return None
+        sender = str(message.get("from", {}).get("username", "") or self.chat_id)
+        return Reply(sender=sender, text=text)
+
+    def ack(self, question: Question, text: str) -> None:
+        message_id = self._question_messages.get(question.id)
+        if message_id is not None:
+            edited = self._call(
+                "editMessageText",
+                {"chat_id": self.chat_id, "message_id": message_id, "text": text},
+            )
+            if edited is not None:
+                return
+        self.notify(text)
+
+
+class NtfyChannel(AnswerChannel):
+    """ntfy.sh publisher — notify-only in v1 (a reply topic can come later)."""
+
+    name = "ntfy"
+    bidirectional = False
+
+    def __init__(self, topic: str, server: str = "https://ntfy.sh", token: str = ""):
+        self.topic = topic
+        self.server = server.rstrip("/")
+        self.token = token
+        self.last_error: str = ""
+
+    def notify(self, text: str) -> bool:
+        headers = {"Title": "owloop"}
+        if self.token:
+            headers["Authorization"] = f"Bearer {self.token}"
+        request = urllib.request.Request(
+            f"{self.server}/{urllib.parse.quote(self.topic)}",
+            data=text.encode("utf-8"),
+            headers=headers,
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=HTTP_TIMEOUT):
+                return True
+        except (urllib.error.URLError, OSError) as exc:
+            self.last_error = str(exc)
+            return False
+
+
+def match_reply(reply: Reply, pending: list[Question]) -> tuple[Question | None, str]:
+    """Correlate a reply to a pending question; returns (question, option key).
+
+    Ladder: button callback (explicit id) → id mentioned in text → bare option
+    key, accepted only when exactly one question is pending. Never guesses:
+    an ambiguous bare key matches nothing.
+    """
+    if reply.question_id:
+        for question in pending:
+            if question.id == reply.question_id:
+                return question, reply.option_key
+        return None, ""
+
+    text = reply.text.strip()
+    for question in pending:
+        if question.id in text:
+            remainder = text.replace(question.id, "").strip()
+            return question, remainder.split()[0] if remainder else ""
+
+    if len(pending) == 1:
+        token = text.split()[0] if text else ""
+        if token == SKIP_VERB:
+            return pending[0], SKIP_OPTION_KEY
+        if token == STEER_VERB or pending[0].find_option(token) is not None:
+            return pending[0], token
+    return None, ""
+
+
+def channels_from_env(names: list[str], env: Mapping[str, str]) -> list[AnswerChannel]:
+    """Build channels named in ``names`` from environment configuration.
+
+    Raises ValueError with a human-readable message when required variables
+    are missing, so the CLI can fail before the loop starts instead of at
+    2 a.m.
+    """
+    channels: list[AnswerChannel] = []
+    for name in names:
+        name = name.strip().lower()
+        if not name:
+            continue
+        if name == "telegram":
+            token = env.get("OWLOOP_TELEGRAM_BOT_TOKEN", "")
+            chat_id = env.get("OWLOOP_TELEGRAM_CHAT_ID", "")
+            if not token or not chat_id:
+                raise ValueError(
+                    "telegram channel needs OWLOOP_TELEGRAM_BOT_TOKEN and OWLOOP_TELEGRAM_CHAT_ID"
+                )
+            channels.append(TelegramChannel(token, chat_id))
+        elif name == "ntfy":
+            topic = env.get("OWLOOP_NTFY_TOPIC", "")
+            if not topic:
+                raise ValueError("ntfy channel needs OWLOOP_NTFY_TOPIC")
+            channels.append(
+                NtfyChannel(
+                    topic,
+                    server=env.get("OWLOOP_NTFY_SERVER", "https://ntfy.sh"),
+                    token=env.get("OWLOOP_NTFY_TOKEN", ""),
+                )
+            )
+        else:
+            raise ValueError(f"unknown channel: {name!r} (supported: telegram, ntfy)")
+    return channels