diff --git a/README.md b/README.md index 9e94a60..9929c26 100644 --- a/README.md +++ b/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 `, 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` | diff --git a/src/owloop/channels.py b/src/owloop/channels.py new file mode 100644 index 0000000..5f71e38 --- /dev/null +++ b/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} ") + 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 = "\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 diff --git a/src/owloop/cli.py b/src/owloop/cli.py index 551e27d..5c03031 100644 --- a/src/owloop/cli.py +++ b/src/owloop/cli.py @@ -19,6 +19,7 @@ from owloop import _brand from owloop.adapters import get_adapter from owloop.backpressure import discover_and_save +from owloop.channels import channels_from_env from owloop.engine import EngineConfig, OwloopEngine, RunSummary from owloop.paths import resolve_specs_dir from owloop.report import ReportGenerator @@ -709,7 +710,16 @@ def _run_engine( session_id: str | None = None, resume: bool = False, no_tui: bool = False, dry_run: bool = False, max_tokens_per_iteration: int = 0, + channels: str = "", question_wait: int = 30, ) -> None: + channel_list = [] + if channels: + try: + channel_list = channels_from_env(channels.split(","), os.environ) + except ValueError as exc: + Console(no_color=no_color).print(f"[red]Error:[/] {exc}") + raise SystemExit(1) from None + config = EngineConfig( project_dir=Path.cwd(), max_iterations=max_iterations, @@ -722,6 +732,8 @@ def _run_engine( session_id=session_id, resume=resume, dry_run=dry_run, + channels=channel_list, + question_wait_minutes=question_wait, ) adapter = get_adapter( agent, @@ -870,8 +882,21 @@ def _print_dry_run_report(console: Console, summary: RunSummary) -> None: help="Kill a single iteration early if it exceeds N tokens (0 = unlimited; " "supports k/w/m shorthand).", show_default=True, ) +@click.option( + "--channels", default="", + help="Comma-separated answer channels (telegram, ntfy). Telegram needs " + "OWLOOP_TELEGRAM_BOT_TOKEN + OWLOOP_TELEGRAM_CHAT_ID; ntfy needs " + "OWLOOP_NTFY_TOPIC. A DECIDE question is pushed to the channel and, on " + "telegram, can be answered from your phone without stopping the loop.", +) +@click.option( + "--question-wait", type=int, default=30, show_default=True, + help="Minutes to wait for a human answer to a DECIDE question before " + "applying the safe default (only with a bidirectional channel).", +) @_common_run_options def run(max_iterations: int, resume: bool, dry_run: bool, no_tui: bool, max_tokens_per_iteration: int, + channels: str, question_wait: int, worktree: bool, model: str, agent: str, verifier_model: str | None, subagents: bool, idle_timeout: float, max_duration: int, max_tokens: int) -> None: """Start the autonomous coding loop.""" @@ -893,6 +918,8 @@ def run(max_iterations: int, resume: bool, dry_run: bool, no_tui: bool, max_toke no_tui=no_tui, dry_run=dry_run, max_tokens_per_iteration=max_tokens_per_iteration, + channels=channels, + question_wait=question_wait, ) diff --git a/src/owloop/engine.py b/src/owloop/engine.py index 3a679ac..c1387d4 100644 --- a/src/owloop/engine.py +++ b/src/owloop/engine.py @@ -21,13 +21,22 @@ import time import uuid from collections.abc import Callable -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any +from owloop import questions as questions_mod from owloop import spec_queue from owloop.adapters import AgentAdapter, AgentResult +from owloop.channels import ( + STEER_VERB, + AnswerChannel, + Reply, + is_night, + match_reply, + render_question, +) from owloop.learnings import ( append_learning, extract_learnings, @@ -59,7 +68,14 @@ If you hit an external blocker (missing API key, unavailable dependency), output `BLOCKED:reason` and the loop will stop. If you need a human decision before continuing, output `DECIDE:question` and the -loop will stop to ask for guidance. +loop will ask the human through its notification channels. When you emit DECIDE, +first emit a structured question block so the human can answer from a phone: + +{"title": "one-line question", "context": "what you were doing and why you are stuck (max 600 chars)", "tried": "what you attempted and the key error", "options": [{"key": "1", "label": "...", "consequence": "one line", "recommended": true}, {"key": "2", "label": "...", "consequence": "one line"}]} + +Options must be mutually exclusive, each with a one-line consequence, and +exactly one marked recommended. A previously answered question appears in your +prompt as steering — follow it instead of asking again. If you discover a useful operational fact during this iteration (e.g., "tests require a running database", "use poetry not pip"), wrap it in @@ -130,6 +146,15 @@ class EngineConfig: # When True, run exactly one iteration, skip push, revert any commit the # iteration made, and produce a DryRunReport instead of looping. dry_run: bool = False + # Answer channels (issue #53, Layer 6). Best-effort transport: a dead + # channel never interrupts the loop. When a bidirectional channel is + # configured, a DECIDE question waits up to ``question_wait_minutes`` for + # a human answer before falling back to stopping the run. + channels: list[AnswerChannel] = field(default_factory=list) + question_wait_minutes: int = 30 + # Quiet hours (start, end) during which questions are rendered as the + # safe recommendation-first binary ("the drunk test"). + quiet_hours: tuple[int, int] = (22, 8) @dataclass @@ -675,6 +700,17 @@ def _build_prompt_with_context(self, prompt_text: str) -> str: f"{steering}" ) + next_spec = spec_queue.get_next_ready_spec(self.specs_dir) + if next_spec is not None: + spec_steering = questions_mod.load_spec_steering(self.cwd, next_spec.name) + if spec_steering: + self._emit("spec_steering_loaded", spec=next_spec.name) + sections.append( + f"A human answered questions about `{next_spec.name}` through a " + "notification channel. Follow these decisions instead of asking again.\n\n" + f"{spec_steering}" + ) + notes = self._read_run_notes() if notes is not None: self._emit("run_notes_loaded", path=str(self.cwd / "run-notes.md")) @@ -741,6 +777,146 @@ def _run_acceptance_criteria(self, spec_name: str | None) -> tuple[int, int]: failed += 1 return passed, failed + # ---- answer channels (issue #53, Layer 6) ------------------------------- + + def _bidirectional_channels(self) -> list[AnswerChannel]: + return [channel for channel in self.config.channels if channel.bidirectional] + + def _notify_channels(self, text: str) -> None: + """Best-effort broadcast; a dead channel costs delivery, never the loop.""" + for channel in self.config.channels: + try: + delivered = channel.notify(text) + except Exception as exc: # transport bugs must not kill the run + delivered = False + self._emit("channel_error", channel=channel.name, error=str(exc)) + if not delivered: + self._emit("notify_failed", channel=channel.name) + + def _raise_question(self, kind: str, result: IterationResult) -> questions_mod.Question: + """Persist a Question (source of truth) and post it to every channel.""" + status = self._spec_status() + question = questions_mod.build_question( + kind, + result.promise_payload, + result.stdout, + spec=status["first_incomplete"] or "", + iteration=result.iteration, + branch=self._current_branch(), + deadline_minutes=self.config.question_wait_minutes, + ) + questions_mod.save_question(self.cwd, question) + self._emit( + "question_created", + question_id=question.id, + title=question.title, + low_fidelity=question.low_fidelity, + ) + + night = is_night(datetime.now().hour, self.config.quiet_hours) + text, buttons = render_question(question, night=night) + posted = False + for channel in self.config.channels: + try: + if channel.post_question(question, text, buttons): + posted = True + self._emit("question_posted", question_id=question.id, channel=channel.name) + else: + self._emit("question_post_failed", question_id=question.id, channel=channel.name) + except Exception as exc: + self._emit("channel_error", channel=channel.name, error=str(exc)) + if posted: + question.status = "posted" + questions_mod.save_question(self.cwd, question) + return question + + def _route_reply(self, reply: Reply, pending: list[questions_mod.Question]) -> questions_mod.Question | None: + """Apply one inbound reply; returns the question it answered, if any.""" + question, key = match_reply(reply, pending) + if question is None: + return None + if key == STEER_VERB: + text = reply.text.split(None, 1)[1] if len(reply.text.split(None, 1)) > 1 else "" + questions_mod.record_answer(self.cwd, question, key="", text=text, sender=reply.sender) + elif key == questions_mod.SKIP_OPTION_KEY or question.find_option(key) is not None: + # Skip is always a valid answer, even when it isn't a listed + # option — the reply syntax advertises /skip on every question. + questions_mod.record_answer(self.cwd, question, key=key, text=reply.text, sender=reply.sender) + else: + return None + questions_mod.apply_answer_as_steering(self.cwd, question) + ack_text = ( + f"✅ {question.id} answered by {reply.sender} — applies at the next " + f"iteration on {question.spec or 'this run'}." + ) + for channel in self._bidirectional_channels(): + try: + channel.ack(question, ack_text) + except Exception as exc: + self._emit("channel_error", channel=channel.name, error=str(exc)) + self._emit("question_answered", question_id=question.id, key=question.answer_key, by=question.answered_by) + return question + + def _await_answer(self, question: questions_mod.Question) -> str: + """Wait (bounded) for a human answer; returns 'answered' | 'skipped' | 'expired'. + + Only called for DECIDE. Without a bidirectional channel there is + nothing to wait for and the question expires immediately. + """ + channels = self._bidirectional_channels() + if not channels or self.config.question_wait_minutes <= 0: + questions_mod.mark_expired(self.cwd, question) + return "expired" + + deadline = time.monotonic() + self.config.question_wait_minutes * 60 + self._emit("question_waiting", question_id=question.id, minutes=self.config.question_wait_minutes) + while time.monotonic() < deadline: + remaining = deadline - time.monotonic() + cycle_start = time.monotonic() + for channel in channels: + try: + replies = channel.poll_replies(timeout=min(20.0, max(1.0, remaining))) + except Exception as exc: + self._emit("channel_error", channel=channel.name, error=str(exc)) + replies = [] + for reply in replies: + answered = self._route_reply(reply, [question]) + if answered is not None: + return "skipped" if answered.answer_key == questions_mod.SKIP_OPTION_KEY else "answered" + # Real transports long-poll (block) inside poll_replies; if a + # cycle returned instantly with nothing (e.g. transport error), + # pace the loop instead of busy-spinning until the deadline. + if time.monotonic() - cycle_start < 1.0: + time.sleep(min(5.0, max(0.0, deadline - time.monotonic()))) + questions_mod.mark_expired(self.cwd, question) + self._emit("question_expired", question_id=question.id) + for channel in channels: + try: + channel.ack(question, f"⏰ {question.id} expired — safe default applied (loop stopped for review).") + except Exception as exc: + self._emit("channel_error", channel=channel.name, error=str(exc)) + return "expired" + + def _idle_poll(self, delay: float) -> None: + """Between iterations: the long-poll *is* the sleep. + + Replies arriving here are routed to any pending questions on disk; + they apply at the next iteration boundary, never mid-iteration. + """ + channels = self._bidirectional_channels() + if not channels or delay <= 0: + time.sleep(delay) + return + pending = questions_mod.pending_questions(self.cwd) + for channel in channels: + try: + replies = channel.poll_replies(timeout=delay / len(channels)) + except Exception as exc: + self._emit("channel_error", channel=channel.name, error=str(exc)) + continue + for reply in replies: + self._route_reply(reply, pending) + def run_iteration(self, iteration: int) -> IterationResult: owloop_dir = resolve_owloop_dir(self.cwd) prompt_file = owloop_dir / "PROMPT_build.md" @@ -1016,12 +1192,23 @@ def run(self) -> RunSummary: stopped_reason = "blocked" blocker = result.promise_payload self._emit("blocked", payload=result.promise_payload) + self._notify_channels( + f"🦉 owloop is blocked: {result.promise_payload}\n" + f"The loop stopped on branch {branch}; nothing merges until you look." + ) break elif result.promise_state == "DECIDE": - stopped_reason = "decide" - decision_question = result.promise_payload - self._emit("decide", payload=result.promise_payload) - break + question = self._raise_question("decide", result) + resolution = self._await_answer(question) + if resolution == "answered": + # Not a failure: the next iteration retries this spec + # with the human's decision injected as steering. + consecutive_failures = 0 + else: + stopped_reason = "decide" + decision_question = result.promise_payload + self._emit("decide", payload=result.promise_payload) + break else: consecutive_failures += 1 if consecutive_failures >= self.config.max_consecutive_failures: @@ -1050,7 +1237,7 @@ def run(self) -> RunSummary: self.config.base_retry_delay * (2**backoff_level), self.config.max_retry_delay, ) - time.sleep(delay) + self._idle_poll(delay) except KeyboardInterrupt: stopped_reason = "interrupted" self._emit("interrupted", iteration=iteration) @@ -1078,6 +1265,14 @@ def run(self) -> RunSummary: dry_run_report=dry_run_report, ) self._write_summary(summary) + if stopped_reason not in ("blocked",): # blocked already sent its own message + spec_state = self._spec_status() + done = spec_state["spec_count"] - spec_state["incomplete_count"] + self._notify_channels( + f"🌅 owloop run finished: {stopped_reason} after {iteration} iteration(s).\n" + f"Specs: {done}/{spec_state['spec_count']} complete · branch {branch}\n" + f"Tokens: {self.tokens_used:,}" + ) return summary def _write_summary(self, summary: RunSummary) -> None: diff --git a/src/owloop/questions.py b/src/owloop/questions.py new file mode 100644 index 0000000..aa1d7b0 --- /dev/null +++ b/src/owloop/questions.py @@ -0,0 +1,296 @@ +"""Structured question protocol for human-in-the-loop decisions. + +When an iteration ends with ``DECIDE:...`` (or hits a +blocker), the agent may precede the promise with a structured block: + + {"title": "...", "context": "...", "options": [...]} + +The engine enriches that block with mechanical facts it owns (spec name, +iteration, branch) and enforces a quality contract before anything reaches a +notification channel: at least two options, exactly one recommendation, and a +mandatory no-answer default. Bare ``DECIDE:text`` payloads without a block are +synthesized into a low-fidelity question so old-format agents keep working. + +Questions are persisted as JSON under ``.owloop/questions/`` — the file is the +source of truth; channel delivery is best-effort transport on top of it. +Answers are applied as per-spec steering under ``.owloop/steering/`` which the +next iteration loads into its prompt. +""" + +from __future__ import annotations + +import json +import re +import uuid +from dataclasses import asdict, dataclass, field +from datetime import datetime +from pathlib import Path + +from owloop.paths import resolve_owloop_dir + +QUESTION_BLOCK_RE = re.compile(r"(.*?)", re.DOTALL) + +QUESTIONS_DIR_NAME = "questions" +STEERING_DIR_NAME = "steering" + +# Rendering budgets (chars). Context and evidence are truncated to these; +# options and the default line are never truncated — they are the decision. +CONTEXT_BUDGET = 600 +TRIED_BUDGET = 400 + +SKIP_OPTION_KEY = "skip" + + +@dataclass +class QuestionOption: + key: str + label: str + consequence: str = "" + recommended: bool = False + + +@dataclass +class Question: + """One decision the loop needs a human for.""" + + id: str + kind: str # "decide" | "blocked" + title: str + spec: str = "" + iteration: int = 0 + context: str = "" + tried: str = "" + options: list[QuestionOption] = field(default_factory=list) + default_action: str = "stop_run" # what happens if nobody answers + deadline_minutes: int = 30 + branch: str = "" + low_fidelity: bool = False # synthesized from a bare DECIDE:text payload + status: str = "pending" # pending → posted → answered → applied | expired + answer_key: str = "" + answer_text: str = "" + answered_by: str = "" + answered_at: str = "" + created_at: str = "" + + def recommended_option(self) -> QuestionOption | None: + for option in self.options: + if option.recommended: + return option + return None + + def find_option(self, key: str) -> QuestionOption | None: + for option in self.options: + if option.key == key: + return option + return None + + def as_dict(self) -> dict: + return asdict(self) + + +def questions_dir(project_dir: Path) -> Path: + return resolve_owloop_dir(project_dir) / QUESTIONS_DIR_NAME + + +def steering_dir(project_dir: Path) -> Path: + return resolve_owloop_dir(project_dir) / STEERING_DIR_NAME + + +def parse_question_block(stdout: str) -> dict | None: + """Extract the agent's ```` JSON block, or None if absent/invalid.""" + match = QUESTION_BLOCK_RE.search(stdout) + if match is None: + return None + try: + data = json.loads(match.group(1)) + except json.JSONDecodeError: + return None + return data if isinstance(data, dict) else None + + +def _new_question_id(spec: str) -> str: + stem = Path(spec).stem if spec else "run" + # Keep ids short: Telegram callback_data caps at 64 bytes and carries + # "q::