diff --git a/docs/architecture.md b/docs/architecture.md index 319b4f4..aae66b6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -36,7 +36,7 @@ The contracts live in `contracts/`. | Package | Responsibility | |---|---| | `agent/agent.py`, `prompt.py`, `think.py`, `plan.py` | The brain. When a reflex wakes it, the model (today Claude Haiku 5.5, at high effort for its rounds and medium when it answers the owner) reads a short briefing and calls tools until every open invoice has its decision: pay now, pay on a date, hold, or ask the owner, each with a reason, plus alarms and notes for later and a summary for the owner. It pays during the session and reads what happened (paid with its transaction, refused by the contract, or Circle's own error), then acts on it: looks up why, holds, or tells the owner what only they can fix. Capped per session (turns, tokens, 10 payments) and per day (USD 1 of model use by default, estimated from the tokens and the public prices; past it the agent is not woken until the next day, and the owner is told once). A payment it sent stands even if the session is cut off; its other choices count once the session finishes. The rules still decide what may be paid; the agent decides what is paid and when, and pays it. On by default, and nothing is paid without it; `EMBCO_AUTONOMY=observe` decides and explains without paying | -| `services/telegram` | The agent on Telegram, a process of its own (`embco telegram`). A chat is linked with a one-time code from the dashboard (hashed, 10 minutes, once); any other chat can do nothing. `/stop` turns the whole agent off and pauses the contract from the agent's wallet, `/pause` pauses payments while the agent keeps watching and answering, `/status` tells its state: none of them go through the model, and only the dashboard turns it on again. Any other text is a message to the agent, whose answers come back to the chat. The bot token never reaches a log or an error | +| `services/telegram` | The agent on Telegram, a process of its own (`embco telegram`). A chat is linked with a one-time code from the dashboard (hashed, 10 minutes, once); any other chat can do nothing. `/stop` turns the whole agent off and pauses the contract from the agent's wallet, `/pause` pauses payments while the agent keeps watching and answering, `/status` tells its state: none of them go through the model, and only the dashboard turns it on again. Any other text is a message to the agent, whose answers come back to the chat. The agent's questions about invoices arrive once, with two buttons (yes, pay it / no, hold it); a press from the linked chat, while the question is still open, is only a message to the agent, and what needs the owner's wallet stays in the dashboard. The bot token never reaches a log or an error | | `agent/tools` | What the agent can call. To read: the open invoices (with the last state of each payment), a search over them (supplier, due date, what the checks say), one invoice in full, a supplier's profile, an item's price history, the cash position, the state of a payment, and its own wallet (balance for network fees, whether the contract has it as its agent, whether the contract is paused). To act: pay now (sent at once, with the result back to the agent), schedule (on that day the agent is woken to pay), hold, ask the owner, set an alarm, keep a note, finish. To talk: reply to the owner's message, notify the owner of something to act on (at most 3 per session), and email a supplier from the shop's ERPNext, only after the owner agreed | | `agent/guardrails/gate.py` | Refuses a payment the rules do not allow (HOLD, ASK, already paid, over this week's room in the contract) before anything is sent, and the refusal goes back to the agent | | `agent/reflexes/wake.py` | When to wake the agent: an invoice new, changed or without its decision, a payment failed or blocked (not again for a result it already saw), the day it planned to pay something, a due date close with nothing planned, an owner answer or supplier signature, a limit change, an alarm it set, the daily round, a heartbeat after 4 hours. No reason, no model call | diff --git a/services/shops/agent.py b/services/shops/agent.py index f92b392..389b5ff 100644 --- a/services/shops/agent.py +++ b/services/shops/agent.py @@ -31,6 +31,7 @@ from services.shops import activity, chat from services.shops.inbox import take_answers, take_signatures from services.shops.owner_view import decision_view +from services.shops.questions import new_questions from services.shops.store import ShopStore from services.shops.supplier_view import supplier_view from services.telegram import links @@ -131,7 +132,7 @@ def not_woken(folder, thought, owner_waiting: bool, at: datetime) -> list[dict[s return [] key = f"{at.date().isoformat()}:{thought.skipped}" for entry in reversed(chat.read_chat(folder, limit=chat.MAX_ENTRIES)): - if entry.get("from") == chat.AGENT: + if entry.get("from") == chat.AGENT and not entry.get("asks"): break if entry.get("not_woken") == key: return [] @@ -315,7 +316,9 @@ def cycle() -> None: *activity.cycle_events(report, minutes)]) chat.append_chat(folder, on_telegram(folder, [ *chat_replies(report.thought.session if report.thought else None, bool(waiting)), - *not_woken(folder, report.thought, bool(waiting), report.at)])) + *not_woken(folder, report.thought, bool(waiting), report.at), + *(new_questions(folder, report.decisions, report.thought.agent, report.at) + if report.thought else [])])) last_check.update(decisions=list(report.decisions), done=set(report.already_paid), at=time.monotonic()) latest = {p.invoice: p for p in journal.latest_payments()} diff --git a/services/shops/chat.py b/services/shops/chat.py index ff4cbc9..d87b578 100644 --- a/services/shops/chat.py +++ b/services/shops/chat.py @@ -81,10 +81,11 @@ def read_chat(folder: Path, after: int = 0, limit: int = 100) -> list[dict[str, def unanswered(folder: Path) -> list[dict[str, Any]]: - """The owner's messages after the agent's last reply.""" + """The owner's messages after the agent's last reply (a question the agent sent about an + invoice is not a reply).""" pending: list[dict[str, Any]] = [] for entry in read_chat(folder, limit=MAX_ENTRIES): - if entry.get("from") == AGENT: + if entry.get("from") == AGENT and not entry.get("asks"): pending = [] elif entry.get("from") == OWNER: pending.append(entry) diff --git a/services/shops/questions.py b/services/shops/questions.py new file mode 100644 index 0000000..75604dd --- /dev/null +++ b/services/shops/questions.py @@ -0,0 +1,50 @@ +"""The agent's questions about invoices, sent to the owner's chat so they reach the phone too. + +When the agent asks the owner about an invoice, the question is written once into the chat (and +from there to Telegram) with two quick answers. An answer is only a message to the agent, the +same words the dashboard's buttons send: the agent reads it and decides, and the contract still +sets the limits. Nothing that needs the owner's wallet (approving a wallet, changing limits, +turning the agent on) can be answered this way. +""" + +from collections.abc import Iterable, Mapping +from datetime import datetime +from typing import Any + +from agent.guardrails.rules import Action, Decision, fingerprint +from agent.models import AgentDecision, Choice +from services.shops import chat + +YES, NO = "yes", "no" + + +def answer_text(invoice: str, answer: str) -> str: + """The message a quick answer sends to the agent (the dashboard sends the same words).""" + if answer == YES: + return f"About {invoice}: yes, go ahead and pay it." + return f"About {invoice}: no, hold it for now." + + +def question_key(agent: AgentDecision) -> str: + """One question: the same choice on the same decision is not sent twice.""" + return f"{agent.invoice}:{agent.fingerprint}" + + +def new_questions(folder, decisions: Iterable[Decision], agent: Mapping[str, AgentDecision], + at: datetime) -> list[dict[str, Any]]: + """Chat entries for the agent's questions that stand now and were not sent yet.""" + sent = {e.get("asks") for e in chat.read_chat(folder, limit=chat.MAX_ENTRIES)} + entries = [] + for decision in decisions: + choice = agent.get(decision.invoice) + if (choice is None or choice.choice is not Choice.ASK or decision.action is not Action.PAY + or choice.fingerprint != fingerprint(decision) + or question_key(choice) in sent): + continue + text = (f"Question about {decision.invoice} ({decision.supplier}, {decision.amount} " + f"USDC): {choice.question or choice.reason}") + if choice.recommendation: + text += f"\nI recommend: {choice.recommendation}" + entries.append({"from": chat.AGENT, "at": at.isoformat(), "text": text, + "asks": question_key(choice), "invoice": decision.invoice}) + return entries diff --git a/services/telegram/api.py b/services/telegram/api.py index a372d89..c5ef75a 100644 --- a/services/telegram/api.py +++ b/services/telegram/api.py @@ -1,4 +1,4 @@ -"""The Telegram Bot API, the two calls the bot needs. +"""The Telegram Bot API, the few calls the bot needs. The token is part of every request URL, so it must never reach a log or an error: errors are rebuilt without the URL, and the bot process keeps the HTTP library's request log quiet. @@ -24,14 +24,29 @@ def __repr__(self) -> str: return "TelegramApi(token=***)" def get_updates(self, offset: int | None, timeout: int = POLL_SECONDS) -> list[dict[str, Any]]: - params: dict[str, Any] = {"timeout": timeout, "allowed_updates": ["message"]} + params: dict[str, Any] = {"timeout": timeout, + "allowed_updates": ["message", "callback_query"]} if offset is not None: params["offset"] = offset return self._call("getUpdates", params) - def send(self, chat_id: int, text: str) -> None: - self._call("sendMessage", {"chat_id": chat_id, "text": text[:4000], - "disable_web_page_preview": True}) + def send(self, chat_id: int, text: str, + buttons: list[tuple[str, str]] | None = None) -> None: + """`buttons` are (label, callback data) shown under the message, in one row.""" + params: dict[str, Any] = {"chat_id": chat_id, "text": text[:4000], + "disable_web_page_preview": True} + if buttons: + params["reply_markup"] = {"inline_keyboard": [ + [{"text": label, "callback_data": data} for label, data in buttons]]} + self._call("sendMessage", params) + + def answer_button(self, callback_id: str, text: str) -> None: + """Tell Telegram the button press was handled, with a short note on the screen.""" + self._call("answerCallbackQuery", {"callback_query_id": callback_id, "text": text[:200]}) + + def drop_buttons(self, chat_id: int, message_id: int) -> None: + """Take the buttons off a message, so a question is answered once.""" + self._call("editMessageReplyMarkup", {"chat_id": chat_id, "message_id": message_id}) def _call(self, method: str, params: dict[str, Any]) -> Any: try: diff --git a/services/telegram/bot.py b/services/telegram/bot.py index b9bc6b9..82bcfda 100644 --- a/services/telegram/bot.py +++ b/services/telegram/bot.py @@ -6,7 +6,9 @@ /pause pause payments only: the contract is paused and the agent only observes; it still checks, decides and answers. /status the agent's state and what it is waiting for. -Any other text is a message to the agent; its answers come back to the chat. +Any other text is a message to the agent; its answers come back to the chat. The agent's +questions about invoices come with two buttons (yes, pay it / no, hold it): a press is only a +message to the agent, as if typed. What needs the owner's wallet stays in the dashboard. Only a private chat linked from the dashboard is heard: never a group, whose members could read the answers and stop the agent. Any other chat is told how to link and nothing else. A /link code @@ -27,6 +29,7 @@ from services.hub.units import UnitError, Units from services.shops import chat from services.shops.activity import append_activity, switched +from services.shops.questions import NO, YES, answer_text from services.shops.store import ShopError, ShopStore from services.telegram import links from services.telegram.api import TelegramApi, TelegramError @@ -35,6 +38,7 @@ PAUSE_CALL = keccak(text="pause()")[:4] LINK_TRIES = 5 # per chat and hour +ANSWERS = (("Yes, pay it", YES), ("No, hold it", NO)) # quick answers to the agent's questions HELP = ("Commands:\n/stop - turn the agent off now (no checks, no Claude, no payments)\n" "/pause - pause payments; the agent keeps watching and answering\n/status - how it is\n" "Anything else is a message to your agent.") @@ -68,6 +72,9 @@ class Bot: tries: dict[int, list[datetime]] = field(default_factory=dict) def handle(self, update: dict[str, Any]) -> None: + if update.get("callback_query"): + self._button(update["callback_query"]) + return message = update.get("message") or {} chat_id = (message.get("chat") or {}).get("id") text = (message.get("text") or "").strip() @@ -109,8 +116,12 @@ def forward(self) -> None: if e.get("from") in (chat.AGENT, chat.SYSTEM)] for entry in fresh: cost = f"\n({entry['cost']} est.)" if entry.get("cost") else "" + buttons = None + if entry.get("asks"): + buttons = [(label, f"{shop}:{entry['seq']}:{answer}") + for label, answer in ANSWERS] try: - self.api.send(int(link["chat_id"]), f"{entry['text']}{cost}") + self.api.send(int(link["chat_id"]), f"{entry['text']}{cost}", buttons) except TelegramError as error: log.warning("could not forward to the chat of %s: %s", shop, error) return @@ -187,6 +198,42 @@ def _status(self, shop: str) -> str: lines.append(f"Last word from the agent: {session['summary']}") return "\n".join(lines) + def _button(self, query: dict[str, Any]) -> None: + """A quick answer to one of the agent's questions: only from the linked private chat, + only while the question is still open, and it is only a message to the agent.""" + message = query.get("message") or {} + chat_id = (message.get("chat") or {}).get("id") + shop, _, rest = str(query.get("data") or "").partition(":") + seq, _, answer = rest.partition(":") + if (not isinstance(chat_id, int) or (message.get("chat") or {}).get("type") != "private" + or shop not in self.shops_of(chat_id) or answer not in (YES, NO) + or not seq.isdigit()): + self.api.answer_button(str(query.get("id")), "This button is not for this chat.") + return + asked = next((e for e in chat.read_chat(self.store.folder(shop), after=int(seq) - 1, + limit=chat.MAX_ENTRIES) + if e.get("seq") == int(seq) and e.get("asks")), None) + if asked is None or not self._still_asked(shop, asked["asks"]): + reply = "This question is no longer open." + else: + reply = self._to_agent(shop, answer_text(asked["invoice"], answer)) + self.api.answer_button(str(query.get("id")), reply) + try: + self.api.drop_buttons(chat_id, int(message.get("message_id", 0))) + except TelegramError: + log.warning("could not take the buttons off a question of %s", shop) + label = next(lbl for lbl, a in ANSWERS if a == answer) + self.api.send(chat_id, f"{label}: {reply}") + + def _still_asked(self, shop: str, key: str) -> bool: + """The agent still asks this question, as of its last check.""" + invoice, _, print_ = key.partition(":") + for d in (self.store.last_run(shop) or {}).get("decisions", []): + if d.get("invoice") == invoice: + agent = d.get("agent") or {} + return agent.get("choice") == "ASK" and d.get("fingerprint") == print_ + return False + def _to_agent(self, shop: str, text: str) -> str: if not self.store.agent_on(shop): return "The agent is off. Turn it on in the dashboard to talk to it." diff --git a/tests/test_questions.py b/tests/test_questions.py new file mode 100644 index 0000000..219e0c8 --- /dev/null +++ b/tests/test_questions.py @@ -0,0 +1,60 @@ +"""The agent's questions reach the owner's chat once, while they stand, and never count as an +answer to a message the owner is still waiting on.""" + +from datetime import UTC, date, datetime +from decimal import Decimal + +from agent.guardrails.controls.base import Finding, Outcome +from agent.guardrails.rules import Action, Decision, fingerprint +from agent.models import AgentDecision, Choice +from services.shops import chat +from services.shops.questions import NO, YES, answer_text, new_questions + +NOW = datetime(2026, 10, 13, 19, 0, tzinfo=UTC) + + +def decision(invoice="ACC-PINV-2026-00012", action=Action.PAY): + return Decision(invoice=invoice, supplier="Northern Grains", amount=Decimal("184"), + due_date=date(2026, 10, 19), action=action, reasons=("to weigh: receipt",), + findings=(Finding("three_way_match", Outcome.NOTE, "no receipt"),)) + + +def asks(d, question="Did the flour arrive?", fingerprint_of=None): + return AgentDecision(invoice=d.invoice, fingerprint=fingerprint(fingerprint_of or d), + choice=Choice.ASK, reason="no receipt", question=question, + recommendation="Pay it if it arrived.") + + +def test_a_question_is_sent_once_with_the_invoice_and_the_recommendation(tmp_path): + d = decision() + first = new_questions(tmp_path, [d], {d.invoice: asks(d)}, NOW) + assert len(first) == 1 + assert first[0]["text"].startswith( + "Question about ACC-PINV-2026-00012 (Northern Grains, 184 USDC): Did the flour arrive?") + assert "I recommend: Pay it if it arrived." in first[0]["text"] + chat.append_chat(tmp_path, first) + assert new_questions(tmp_path, [d], {d.invoice: asks(d, "Asked again?")}, NOW) == [] + + +def test_only_a_question_that_still_stands_is_sent(tmp_path): + d = decision() + other = decision(action=Action.HOLD) + assert new_questions(tmp_path, [d], {d.invoice: asks(d, fingerprint_of=other)}, NOW) == [] + assert new_questions(tmp_path, [other], {other.invoice: asks(other)}, NOW) == [] + held = AgentDecision(invoice=d.invoice, fingerprint=fingerprint(d), choice=Choice.HOLD, + reason="wait") + assert new_questions(tmp_path, [d], {d.invoice: held}, NOW) == [] + + +def test_a_question_is_not_an_answer_to_the_owner(tmp_path): + chat.append_chat(tmp_path, [{"from": chat.OWNER, "text": "What is due?"}]) + d = decision() + chat.append_chat(tmp_path, new_questions(tmp_path, [d], {d.invoice: asks(d)}, NOW)) + assert [m["text"] for m in chat.unanswered(tmp_path)] == ["What is due?"] + + +def test_the_quick_answers_say_what_the_dashboard_buttons_say(): + assert answer_text("ACC-PINV-2026-00012", YES) == ( + "About ACC-PINV-2026-00012: yes, go ahead and pay it.") + assert answer_text("ACC-PINV-2026-00012", NO) == ( + "About ACC-PINV-2026-00012: no, hold it for now.") diff --git a/tests/test_telegram.py b/tests/test_telegram.py index b174e50..69c5423 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -23,9 +23,20 @@ class FakeApi: def __init__(self): self.sent = [] + self.buttons = [] # (chat_id, buttons) of each message sent with buttons + self.pressed = [] # notes shown for button presses + self.dropped = [] # (chat_id, message_id) whose buttons were taken off - def send(self, chat_id, text): + def send(self, chat_id, text, buttons=None): self.sent.append((chat_id, text)) + if buttons: + self.buttons.append((chat_id, buttons)) + + def answer_button(self, callback_id, text): + self.pressed.append(text) + + def drop_buttons(self, chat_id, message_id): + self.dropped.append((chat_id, message_id)) def last(self, chat_id): return [t for c, t in self.sent if c == chat_id][-1] @@ -139,6 +150,54 @@ def test_a_message_goes_to_the_agent_and_only_its_answers_come_back(bot): assert len(bot.api.sent) == sent # never twice +INVOICE, PRINT = "ACC-PINV-2026-00012", "f1" + + +def asked(bot, choice="ASK"): + """The agent asked about INVOICE in its last check, and the bot forwarded the question.""" + folder = bot.store.folder(SHOP) + bot.store.write_last_run(SHOP, {"decisions": [ + {"invoice": INVOICE, "fingerprint": PRINT, "agent": {"choice": choice}}]}) + linked(bot) + chat.append_chat(folder, [{"from": chat.AGENT, "text": "Did the flour arrive?", + "asks": f"{INVOICE}:{PRINT}", "invoice": INVOICE}]) + bot.forward() + return bot.api.buttons[-1][1] + + +def press(chat_id, data, kind="private"): + return {"update_id": 2, "callback_query": { + "id": "cb1", "data": data, + "message": {"message_id": 77, "chat": {"id": chat_id, "type": kind}}}} + + +def test_a_question_comes_with_buttons_and_a_press_is_a_message_to_the_agent(bot): + buttons = asked(bot) + assert [label for label, _ in buttons] == ["Yes, pay it", "No, hold it"] + bot.handle(press(OWNER_CHAT, buttons[0][1])) + taken = chat.take_messages(bot.store.folder(SHOP)) + assert [(m["text"], m["by"]) for m in taken] == [ + ("About ACC-PINV-2026-00012: yes, go ahead and pay it.", "owner via Telegram")] + assert bot.api.dropped == [(OWNER_CHAT, 77)] + assert bot.api.last(OWNER_CHAT).startswith("Yes, pay it: Your agent is reading it") + + +def test_a_press_from_another_chat_or_a_group_does_nothing(bot): + buttons = asked(bot) + for chat_id, kind in ((STRANGER, "private"), (OWNER_CHAT, "group")): + bot.handle(press(chat_id, buttons[0][1], kind)) + bot.handle(press(OWNER_CHAT, f"0x{'1' * 40}:1:yes")) # a shop this chat is not linked to + assert not chat.messages_waiting(bot.store.folder(SHOP)) + assert bot.api.pressed == ["This button is not for this chat."] * 3 + + +def test_a_question_no_longer_open_is_not_answered(bot): + buttons = asked(bot, choice="PAY_NOW") + bot.handle(press(OWNER_CHAT, buttons[1][1])) + assert not chat.messages_waiting(bot.store.folder(SHOP)) + assert bot.api.pressed == ["This question is no longer open."] + + def test_an_agent_turned_off_is_not_disturbed(bot): linked(bot) bot.store.set_agent_on(SHOP, False, NOW.isoformat())