diff --git a/coworker/connectors/adapters.py b/coworker/connectors/adapters.py index 3d64d9528..d38ced173 100644 --- a/coworker/connectors/adapters.py +++ b/coworker/connectors/adapters.py @@ -110,9 +110,11 @@ async def _on_update(update, _context): if event is not None: await self.handle_message(event) - self._app.add_handler( - MessageHandler(filters.TEXT & ~filters.COMMAND, _on_update) - ) + # NOT `& ~filters.COMMAND`: that discards every message beginning with "/", + # which is the only syntax people use to talk to a bot. With no CommandHandler + # registered either, /help and /status were dropped before anything saw them — + # no reply, no record, indistinguishable from a dead bot. + self._app.add_handler(MessageHandler(filters.TEXT, _on_update)) await self._app.initialize() await self._app.start() await self._app.updater.start_polling(drop_pending_updates=True) diff --git a/coworker/connectors/commands.py b/coworker/connectors/commands.py new file mode 100644 index 000000000..6cd1edb74 --- /dev/null +++ b/coworker/connectors/commands.py @@ -0,0 +1,123 @@ +"""Slash commands for a bound chat — so the bot is somewhere you can ask things, +not just somewhere prompts arrive. + +The rule this module exists to enforce: **an inbound message is never met with +silence.** A prompt mirrored to a phone is only half a conversation if the person +holding the phone can't ask what is waiting, can't tell whether their reply landed, +and gets nothing back when they type something the parser doesn't understand. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable, Optional + +HELP = """Commands: +/pending - what is waiting on you, and how to answer it +/status - workers, routines, and anything stuck +/runs - how the last few scheduled runs went +/help - this message + +To answer a prompt, reply with its tag: +approve [ow:...] - just this once +always [ow:...] - and stop asking for that routine +deny [ow:...] - refuse it""" + + +@dataclass +class PendingItem: + id: str + title: str + detail: str = "" + worker: str = "" + routine: str = "" + is_approval: bool = True + + +@dataclass +class RunLine: + routine: str + when: str + status: str + + +@dataclass +class ChatContext: + """What the commands can see. A narrow view, passed in, so this module stays + testable without a server and can't reach anything it wasn't handed.""" + + pending: Callable[[], list[PendingItem]] + workers: Callable[[], list[str]] + routines: Callable[[], list[str]] + runs: Callable[[], list[RunLine]] + + +def _pending_block(items: list[PendingItem]) -> str: + if not items: + return "Nothing is waiting on you." + lines = [f"{len(items)} waiting:" if len(items) > 1 else "1 waiting:"] + for it in items: + who = it.worker or "a worker" + where = f" · {it.routine}" if it.routine else "" + lines.append(f"\n{who}{where}\n{it.title}") + if it.detail: + lines.append(it.detail) + # The exact strings to send — never make someone assemble a reply. + if it.is_approval: + lines.append(f"approve [ow:{it.id}] · always [ow:{it.id}] · deny [ow:{it.id}]") + else: + lines.append(f"Reply with your answer plus [ow:{it.id}]") + return "\n".join(lines) + + +def handle_command(text: str, ctx: ChatContext) -> Optional[str]: + """The reply for a slash command, or None if this isn't one. + + Tolerant on purpose: a leading slash is optional and a @botname suffix is + stripped, because that is what people actually type. + """ + raw = (text or "").strip() + if not raw: + return None + word = raw.split()[0].lower().lstrip("/").split("@")[0] + + if word in {"help", "commands", "start"}: + return HELP + + if word in {"pending", "waiting", "inbox", "queue"}: + return _pending_block(ctx.pending()) + + if word == "status": + workers, routines = ctx.workers(), ctx.routines() + pending = ctx.pending() + lines = [ + f"{len(workers)} workers: {', '.join(workers)}" if workers else "No workers yet.", + ] + lines += routines if routines else ["No routines."] + lines.append("") + lines.append(_pending_block(pending)) + return "\n".join(lines) + + if word in {"runs", "history"}: + runs = ctx.runs() + if not runs: + return "No scheduled runs recorded yet." + return "\n".join(f"{r.routine} · {r.when} · {r.status}" for r in runs) + + return None + + +def unmatched_reply(text: str, ctx: ChatContext) -> str: + """What to say when a message matched no command and no waiting prompt. + + Silence was the old behaviour: the message was filed as unrouted and the sender + told nothing, which is indistinguishable from the bot being dead. + """ + items = ctx.pending() + if items: + return ( + "I couldn't match that to anything. To answer what's waiting, send one of " + "these lines (the tag is how I know which prompt you mean):\n\n" + + _pending_block(items) + ) + return f"I couldn't match that to anything, and nothing is waiting on you.\n\n{HELP}" diff --git a/coworker/server/manager.py b/coworker/server/manager.py index cf9b740fa..2385bd78d 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -43,6 +43,7 @@ from ..roots import RootDir from ..workspace_trust import WorkspaceTrustStore from ..automation import Schedule, ScheduledTask, Scheduler, TaskRun, TaskStore +from ..automation.models import rule_parts from ..connectors import ( Gateway, MessageSource, @@ -171,6 +172,16 @@ def _grant_offered(outcome, request) -> bool: return True +def _when(epoch: float) -> str: + """A timestamp for a chat: the day and time, in the box's own zone.""" + import datetime + + try: + return datetime.datetime.fromtimestamp(epoch).strftime("%a %H:%M") + except Exception: + return "?" + + def _approval_body(request) -> str: """Approval card body: the tool's reason (if any) plus a compact preview of its args, so a mirrored 'Run `write_file`?' shows the path/content rather than just the tool name. @@ -1084,7 +1095,13 @@ async def approve(request): self.persist_session(session_id) await self.mirror_inbox_item(item) resolution = await self.inbox.wait(item.id) - return self.approval_outcome(resolution, request, session_id) + outcome = self.approval_outcome(resolution, request, session_id) + # Say what happened, in the chat that asked. Answering a prompt and hearing + # nothing back is indistinguishable from the message not arriving — and for + # "always" the outcome is genuinely not knowable from the reply alone, since + # a grant can be refused for a tool that cannot carry one. + await self._ack_resolution(item, resolution, None, request) + return outcome return approve @@ -4439,10 +4456,39 @@ async def approver(request): self.persist_session(session_id) await self.mirror_inbox_item(item) resolution = await self.inbox.wait(item.id) - return self.approval_outcome(resolution, request, session_id) + outcome = self.approval_outcome(resolution, request, session_id) + await self._ack_resolution(item, resolution, task, request) + return outcome return approver + async def _ack_resolution(self, item, resolution: str, task, request) -> None: + """Tell the bound chat what an answer actually did.""" + binding = self.inbox_routing.binding_for(item.inbox) + if not (binding.channel and self.gateway is not None): + return + tool = getattr(request, "tool_name", "that") + who = task.title if task is not None else "the worker" + if resolution == "deny": + text = f"Declined — {who} did not run `{tool}`." + elif resolution == "always_task" and task is not None: + fresh = self.task_store.get(task.id) + granted = fresh is not None and tool in { + t for t, _ in map(rule_parts, fresh.always_allowed_tools) + } + text = ( + f"Approved, and `{tool}` is now standing for {task.title} — it won't ask again." + if granted + else f"Approved once. `{tool}` can't hold a standing grant, so it will ask again." + ) + else: + text = f"Approved — {who} is continuing." + target = f"{binding.channel}:{binding.target}" + try: + await self.gateway.deliver(target, text) + except Exception: + logger.debug("resolution ack failed", exc_info=True) + def _seed_task_permissions(self, engine: TurnEngine, task) -> None: """Apply a task's standing allowances to an engine: target-bound rules feed the permission engine's matcher (connector tools included — the target binding is the @@ -4567,6 +4613,87 @@ async def _on_interaction(self, event) -> None: except Exception: pass + # -- chat commands ---------------------------------------------------------- + def _chat_context(self): + """The narrow view of state that slash commands may read.""" + from ..connectors.commands import ChatContext, PendingItem, RunLine + from ..inbox import KIND_APPROVAL + + def pending() -> list: + out = [] + for item in self.inbox.pending(): + data = getattr(item, "data", None) or {} + out.append( + PendingItem( + id=item.id, + title=item.title, + detail=(item.body or "").strip().splitlines()[0] if item.body else "", + worker=self._session_agent_name(item.session_id), + routine=str(data.get("task_title") or ""), + is_approval=item.kind == KIND_APPROVAL, + ) + ) + return out + + def workers() -> list[str]: + return [a.get("name") or a.get("id", "") for a in self.list_agents() if a.get("enabled", True)] + + def routines() -> list[str]: + lines = [] + for task in self.task_store.list(): + state = "paused" if not task.enabled else (task.last_status or "not yet run") + lines.append(f"{task.title} · {task.schedule.human()} · {state}") + return lines + + def runs() -> list: + out = [] + for task in self.task_store.list(): + for run in self.task_store.runs(task.id)[:3]: + out.append( + RunLine( + routine=task.title, + when=_when(run.started_at), + status=run.status or "?", + ) + ) + return out[:6] + + return ChatContext(pending=pending, workers=workers, routines=routines, runs=runs) + + def _session_agent_name(self, session_id: str) -> str: + record = self.session_store.load(session_id) if session_id else None + agent = getattr(record, "agent", "") if record else "" + for row in self.list_agents(): + if row.get("id") == agent: + return row.get("name") or agent + return agent or "" + + async def handle_chat_command(self, event, text: str) -> bool: + """Answer a slash command in the chat it came from. True if handled.""" + from ..connectors.commands import handle_command + + reply = handle_command(text, self._chat_context()) + if reply is None: + return False + await self._reply_in_chat(event, reply) + return True + + async def _reply_in_chat(self, event, text: str) -> None: + source = getattr(event, "source", None) + if self.gateway is None or source is None: + return + from ..connectors.base import format_target + + target = format_target( + getattr(source, "platform", ""), + str(getattr(source, "chat_id", "") or ""), + getattr(source, "thread_id", None), + ) + try: + await self.gateway.deliver(target, text) + except Exception: + logger.debug("chat reply failed", exc_info=True) + # -- inbox replies over messaging connectors -------------------------------- def _resolve_inbox_reply(self, event) -> bool: """Try to handle an inbound Slack/Telegram message as an Inbox reply. Returns True if the @@ -4770,6 +4897,11 @@ async def _dispatch_inbound(self, event) -> None: pass return return # channel with no subscribers — nobody is listening + # A slash command is answered here, in the chat it came from — before any + # routing, because "what is waiting on me?" is a question about this bot, not + # something to hand to an agent. + if await self.handle_chat_command(event, text): + return # DM (or any non-channel): route to the designated session, else park it for visibility. dm = self.dm_session() if dm and self._inbound_connector_allowed(dm, src.platform): @@ -4779,10 +4911,17 @@ async def _dispatch_inbound(self, event) -> None: self.unrouted.record( src.target, who, text, reason="connector muted for DM session" ) + await self._reply_in_chat(event, "That worker isn't listening to this chat right now.") else: + # Parked for the app, AND answered here. Silence was the old behaviour: a + # message vanished with no reply, which is indistinguishable from a dead bot + # — and it happened to every "approve" that arrived without its tag. self.unrouted.record( src.target, who, text, reason="no DM session designated" ) + from ..connectors.commands import unmatched_reply + + await self._reply_in_chat(event, unmatched_reply(text, self._chat_context())) # -- mention router (§31) ---------------------------------------------------- async def _route_mention(self, event, ms: MessageSource, subs) -> None: diff --git a/tests/test_chat_commands.py b/tests/test_chat_commands.py new file mode 100644 index 000000000..776cfa53d --- /dev/null +++ b/tests/test_chat_commands.py @@ -0,0 +1,90 @@ +"""The bot answers. Every time. + +An inbound message used to have three fates: it matched a prompt tag, it was handed +to a designated session, or it was filed as unrouted and the sender told nothing. +That last one is most of them — and it is what a person meets when they type +"approve" without a tag, or "help", or ask what is waiting. +""" + +from __future__ import annotations + +from coworker.connectors.commands import ( + ChatContext, + HELP, + PendingItem, + RunLine, + handle_command, + unmatched_reply, +) + + +def _ctx(pending=(), workers=("The Watcher",), routines=("Daily watch · 7am · ok",), runs=()): + return ChatContext( + pending=lambda: list(pending), + workers=lambda: list(workers), + routines=lambda: list(routines), + runs=lambda: list(runs), + ) + + +def _item(**kw): + base = dict(id="abc123", title="Run `web_search`?", detail="query: anything", + worker="The Watcher", routine="Daily watch") + base.update(kw) + return PendingItem(**base) + + +def test_help_lists_the_commands_and_the_answer_lines(): + reply = handle_command("/help", _ctx()) + assert "/pending" in reply and "/status" in reply + assert "approve [ow:...]" in reply and "always [ow:...]" in reply + + +def test_slash_is_optional_and_botname_suffix_is_ignored(): + assert handle_command("help", _ctx()) == HELP + assert handle_command("/help@crew_bot", _ctx()) == HELP + + +def test_pending_names_the_worker_and_gives_the_exact_reply_lines(): + reply = handle_command("/pending", _ctx(pending=[_item()])) + assert "The Watcher" in reply and "Daily watch" in reply + # The three answers, ready to copy — nobody should have to assemble one. + assert "approve [ow:abc123]" in reply + assert "always [ow:abc123]" in reply + assert "deny [ow:abc123]" in reply + + +def test_pending_says_so_when_nothing_waits(): + assert "Nothing is waiting" in handle_command("/pending", _ctx()) + + +def test_a_question_is_not_offered_approve_or_deny(): + reply = handle_command("/pending", _ctx(pending=[_item(is_approval=False, title="Which one?")])) + assert "approve [ow:" not in reply + assert "[ow:abc123]" in reply + + +def test_status_covers_workers_routines_and_what_is_waiting(): + reply = handle_command("/status", _ctx(pending=[_item()])) + assert "The Watcher" in reply and "Daily watch" in reply and "1 waiting" in reply + + +def test_runs_reports_recent_outcomes(): + reply = handle_command("/runs", _ctx(runs=[RunLine("Daily watch", "Mon 07:00", "ok")])) + assert "Daily watch" in reply and "ok" in reply + + +def test_ordinary_text_is_not_a_command(): + assert handle_command("what is happening", _ctx()) is None + + +def test_unmatched_text_offers_the_waiting_prompt(): + reply = unmatched_reply("approve", _ctx(pending=[_item()])) + assert "couldn't match" in reply + assert "approve [ow:abc123]" in reply + + +def test_unmatched_text_with_an_empty_queue_falls_back_to_help(): + reply = unmatched_reply("approve", _ctx()) + assert "nothing is waiting" in reply.lower() + assert "/pending" in reply diff --git a/tests/test_telegram_command_filter.py b/tests/test_telegram_command_filter.py new file mode 100644 index 000000000..c4238478a --- /dev/null +++ b/tests/test_telegram_command_filter.py @@ -0,0 +1,56 @@ +"""Slash commands must reach the handler. + +The adapter registered `filters.TEXT & ~filters.COMMAND`, which discards every +message beginning with "/" — and no CommandHandler was registered either, so a bot +with a documented command surface silently dropped every command sent to it. Plain +text arrived normally, which is what made it look like the commands were broken +rather than never delivered. + +The distinction is Telegram's `bot_command` entity, not the leading slash: a message +typed as a command carries that entity, and that is what `filters.COMMAND` matches. +""" + +from __future__ import annotations + +import datetime + +import pytest + +telegram = pytest.importorskip("telegram") +from telegram import Chat, Message, MessageEntity, Update, User # noqa: E402 +from telegram.ext import filters # noqa: E402 + +from coworker.connectors.adapters import telegram_message_to_event # noqa: E402 + + +def _update(text: str, *, as_command: bool) -> Update: + entities = ( + [MessageEntity(type=MessageEntity.BOT_COMMAND, offset=0, length=len(text.split()[0]))] + if as_command + else [] + ) + message = Message( + message_id=1, + date=datetime.datetime.now(datetime.timezone.utc), + chat=Chat(id=42, type=Chat.PRIVATE), + from_user=User(id=7, first_name="D", is_bot=False), + text=text, + entities=entities, + ) + return Update(update_id=1, message=message) + + +@pytest.mark.parametrize("text", ["/help", "/pending", "/status@some_bot"]) +def test_commands_reach_the_handler(text): + assert filters.TEXT.check_update(_update(text, as_command=True)) + # What the adapter used to register, and why the commands vanished. + assert not (filters.TEXT & ~filters.COMMAND).check_update(_update(text, as_command=True)) + + +def test_plain_text_still_reaches_the_handler(): + assert filters.TEXT.check_update(_update("approve [ow:abc123]", as_command=False)) + + +def test_a_command_becomes_an_event_like_any_other_message(): + event = telegram_message_to_event(_update("/pending", as_command=True).effective_message) + assert event is not None and event.text == "/pending"