diff --git a/CLAUDE.md b/CLAUDE.md index f65026b..141db1f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -305,9 +305,20 @@ not a transcript. The transcript still expires. The gesture that writes one is two buttons beside the reply on `/chat`, reached over `ipc.CorrectTurn` and the trace id that now rides back on `ipc.ChatReply`. A turn marked wrong with no target is a usable negative, so naming the intent is never required. The target -is one of the seven intents and never free text. Only `/chat` offers it: the wire -op assumes no browser, but telegram and voice do not call it yet, and -`docs/plans/22-correcting-a-turn.md` says why voice is the hard one. Adding a rung to the ladder +is one of the seven intents and never free text. **All three reaches offer it as +of 06-08-2026**, and this section used to say only `/chat` did. Voice is the +`repair` rung, which has read spoken corrections since V-455 and now writes the +durable label beside the classifier seed it always wrote; a spoken negative with +no target is its own rung, `repair-negative` (V-636, `docs/plans/22-correcting-a-turn.md`). +Telegram is an inline keyboard under the reply, and it needed the chat to become +readable first — **telegram is no longer outbound only** (V-637, +`docs/plans/23-inbound-telegram.md`). The poller is dark unless the `telegram` +block says `intake`, it long-polls because the box takes no inbound connections, +it accepts `chat_id` and no other sender, and it drops whatever queued while the +daemon was down. It reaches the daemon through `ipc.CoreAPI` alone, so a chat +turn takes the path `POST /api/chat` takes. Note that the turn source is still +`tap:text` for both, so provenance cannot tell a chat turn from a typed one. +Adding a rung to the ladder in `runTurn` means adding its name to `preRouteLadder` in `cmd/mavend/decisiontrace.go`, or that rung is silently missing from the record. diff --git a/cmd/mavend/actions_act.go b/cmd/mavend/actions_act.go index 024f1b6..ed9321c 100644 --- a/cmd/mavend/actions_act.go +++ b/cmd/mavend/actions_act.go @@ -104,4 +104,3 @@ func (h *reactiveHandler) actionAct(ctx context.Context, dec router.Decision) st } return phraser.A(phraser.ActDone, nil) } - diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go index 8408d76..391bd71 100644 --- a/cmd/mavend/main.go +++ b/cmd/mavend/main.go @@ -364,6 +364,9 @@ func run(args []string) error { if !locked { wireMailIntake(srv, st, phr, cfg, evBus) wireModelSwap(srv, phr, cfg) + // Inbound telegram (V-637). Dark unless the telegram block says intake, + // and it reads one chat. + wireTelegramIntake(ctx, &wg, coreAPI, cfg) // Vision + the media blob store (Vikunja #252). Both stay dark without a // media block; MethodDescribeImage answers ErrUnknownMethod then. keeper := wireVision(ctx, &wg, srv, st, embedderOf(voiceW), cfg) @@ -511,6 +514,9 @@ func run(args []string) error { srv.Check = (&auth.Gate{Enrollment: auth.NewFloorEnrollment(), Session: passkeySess}).Check wireMailIntake(srv, st, phr, cfg, evBus) wireModelSwap(srv, phr, cfg) + // Same on the unlock path, with the API that has just replaced the + // locked placeholder (V-637). + wireTelegramIntake(ctx, &wg, newAPI, cfg) keeper := wireVision(ctx, &wg, srv, st, embedderOf(voiceW), cfg) wireCapture(ctx, &wg, srv, keeper, st, voiceW, phr, cfg) // Voice identification (Vikunja #255). Enrolment plumbing only until a diff --git a/cmd/mavend/telegramintake.go b/cmd/mavend/telegramintake.go new file mode 100644 index 0000000..ed63bba --- /dev/null +++ b/cmd/mavend/telegramintake.go @@ -0,0 +1,59 @@ +// mavend/telegramintake.go — wiring the inbound telegram poller (V-637). +// +// The poller reaches the daemon through ipc.CoreAPI and nothing else, so a +// telegram turn takes exactly the path the web's POST /api/chat takes: Chat +// returns the reply and the persisted trace id, and CorrectTurn writes the +// label. Nothing in internal/delivery knows what a handler is. +package main + +import ( + "context" + "log" + "sync" + + "github.com/kami/maven/internal/config" + "github.com/kami/maven/internal/delivery/telegramsink" + "github.com/kami/maven/internal/ipc" +) + +// wireTelegramIntake starts the poller, or returns having done nothing. It is +// nil-safe in every argument, because it is called from both boot paths — the +// unlocked start and the passkey unlock — and telegram must behave the same on +// either. +// +// A sink that will not build is logged rather than fatal here. The push half +// already failed the boot in wireDispatcher for the same config, so a second +// hard failure would only lose that message. +func wireTelegramIntake(ctx context.Context, wg *sync.WaitGroup, api ipc.CoreAPI, cfg *config.Config) { + if cfg == nil || cfg.Telegram == nil || !cfg.Telegram.Intake || api == nil { + return + } + sink, err := telegramsink.New(*cfg.Telegram) + if err != nil { + log.Printf("telegram intake: %v", err) + return + } + poller, err := telegramsink.NewPoller(sink, chatTurnFn(api), api.CorrectTurn) + if err != nil { + log.Printf("telegram intake: %v", err) + return + } + wg.Add(1) + go func() { + defer wg.Done() + poller.Run(ctx) + }() +} + +// chatTurnFn adapts ipc.Chat to the poller's Turn. The trace id comes back on +// the reply because the daemon's Chat collects it off the context (V-630), so +// the chat can offer the same correction the web does without a second op. +func chatTurnFn(api ipc.CoreAPI) telegramsink.Turn { + return func(ctx context.Context, conversation, text string) (string, int64, error) { + reply, err := api.Chat(ctx, conversation, text) + if err != nil { + return "", 0, err + } + return reply.Reply, reply.TraceID, nil + } +} diff --git a/docs/plans/23-inbound-telegram.md b/docs/plans/23-inbound-telegram.md new file mode 100644 index 0000000..e5c3cee --- /dev/null +++ b/docs/plans/23-inbound-telegram.md @@ -0,0 +1,70 @@ +# Inbound telegram + +Last verified: 06-08-2026 @ c61b0b3 + +V-637, under V-628. Reads with `22-correcting-a-turn.md`. + +## What was missing + +Telegram was a reach and nothing else. `telegramsink` pushed an away message and the chat +had no way to answer, so the correction gesture reached the web and voice only. + +That skews the labels. V-546 fits routing heads on them, and a sample drawn from wherever +the owner happens to be sitting is the wrong sample. + +## Long-poll, not a webhook + +The box takes no inbound connections and reaches api.telegram.org through a relay, so the +connection has to open outward. `getUpdates` with a 25 second hold, one goroutine in the +daemon's WaitGroup. + +A failed poll waits 15 seconds and retries without escalating. The relay going down is the +normal cause and it comes back on its own. + +## The backlog is dropped on start + +Telegram keeps undelivered updates for 24 hours. A daemon that was down overnight would +otherwise wake and answer every queued message in order. + +That is worse than missing them. A question asked eight hours ago has been answered +already. A reminder set from it lands at the wrong time. So the first call moves the offset +past whatever is queued and acts on none of it. + +## One chat + +`ChatID` is the only accepted sender, and it is the same chat the push half already sends +to. A message from anywhere else is dropped with no reply, because a reply confirms the bot +exists and whose it is. + +Chat ids are not guessable. They are also not secret, since they travel in every forwarded +message. So this is the whole authorisation and it is an allowlist of one. + +## The gesture + +Two taps at most. The reply carries one button, `не то`. Tapping it writes nothing and opens +the seven intents plus `просто неверно`. The untargeted negative stays reachable, because he +may have opened the row without meaning to name anything. + +Callback data carries the trace id and the target, under telegram's 64 byte cap. It comes +off the wire. So an id that will not parse is dropped, and so is a target that is not one of +the seven. A label nothing can score is worse than no label. + +A failed write says so on the button and leaves the keyboard up. A successful one takes the +keyboard off, because a live keyboard on an answered turn invites correcting it twice. + +## The seam + +`NewPoller` takes two functions and no daemon type. `cmd/mavend/telegramintake.go` fills +them from `ipc.CoreAPI`: `Chat` returns the reply and the trace id it collected off the +context, and `CorrectTurn` writes the label. So a chat turn takes the path +`POST /api/chat` already takes, and nothing in `internal/delivery` knows what a handler is. + +## What is not done + +The turn source is still `tap:text`, which telegram shares with the web. Provenance cannot +tell a chat turn from a typed one, so a label's `source` column cannot either. +That matters the first time someone asks whether corrections given in the chat differ from +corrections given at the desk. + +Voice messages are ignored. The poller reads `message.text` and nothing else, so a voice +note in the chat does not reach `mavsttd`. diff --git a/internal/delivery/telegramsink/botapi.go b/internal/delivery/telegramsink/botapi.go new file mode 100644 index 0000000..c4cbb79 --- /dev/null +++ b/internal/delivery/telegramsink/botapi.go @@ -0,0 +1,175 @@ +// botapi.go — the telegram bot API calls the intake half makes, and the inbound +// shapes it reads (V-637). Split out of intake.go so the poller reads as the +// policy it is, with the wire in one place under it. +package telegramsink + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "strings" +) + +// getUpdates long-polls. The offset is telegram's own acknowledgement: asking +// for lastSeen+1 is what drops everything before it from the queue, so an +// update is handled once even across a restart. +func (p *Poller) getUpdates(ctx context.Context, timeoutSec int) ([]update, error) { + body, err := json.Marshal(map[string]any{ + "offset": p.offset, + "timeout": timeoutSec, + "allowed_updates": []string{"message", "callback_query"}, + }) + if err != nil { + return nil, err + } + var env struct { + telegramResp + Result []update `json:"result"` + } + if err := p.call(ctx, "getUpdates", body, &env); err != nil { + return nil, err + } + for _, u := range env.Result { + if u.UpdateID >= p.offset { + p.offset = u.UpdateID + 1 + } + } + return env.Result, nil +} + +func (p *Poller) send(ctx context.Context, text string, kb *inlineKeyboard) error { + body, err := json.Marshal(sendMessageReq{ + ChatID: p.cfgChatID(), + Text: text, + // A reply to something he just typed is not an alarm, but it is still his + // own data in a third party's chat, so it stays unforwardable like the + // away messages the sink pushes. + ProtectContent: true, + ReplyMarkup: kb, + }) + if err != nil { + return err + } + return p.call(ctx, "sendMessage", body, nil) +} + +// answerCallback stops the clock on the tapped button. text empty is a silent +// acknowledgement; anything else shows as a toast. +func (p *Poller) answerCallback(ctx context.Context, id, text string) { + body, err := json.Marshal(map[string]any{"callback_query_id": id, "text": text}) + if err != nil { + return + } + if err := p.call(ctx, "answerCallbackQuery", body, nil); err != nil { + log.Printf("telegram intake: answer callback: %v", err) + } +} + +// editKeyboard replaces the buttons under a message the bot sent. kb nil takes +// them off. +func (p *Poller) editKeyboard(ctx context.Context, chatID string, messageID int64, kb *inlineKeyboard) error { + payload := map[string]any{"chat_id": chatID, "message_id": messageID} + if kb != nil { + payload["reply_markup"] = kb + } else { + payload["reply_markup"] = inlineKeyboard{Rows: [][]inlineButton{}} + } + body, err := json.Marshal(payload) + if err != nil { + return err + } + return p.call(ctx, "editMessageReplyMarkup", body, nil) +} + +// call posts one bot API method and checks the envelope. out may be nil when +// only the ok flag matters. Every error goes through the sink's redaction: the +// token is in the URL path because telegram accepts it nowhere else, and +// net/http prints that URL in transport errors. +func (p *Poller) call(ctx context.Context, method string, body []byte, out any) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + p.sink.base+"/bot"+p.sink.cfg.BotToken+"/"+method, bytes.NewReader(body)) + if err != nil { + return p.sink.redact(err) + } + req.Header.Set("Content-Type", "application/json") + resp, err := p.hc.Do(req) + if err != nil { + return fmt.Errorf("telegramsink: %s: %w", method, p.sink.redact(err)) + } + defer resp.Body.Close() + rb, _ := io.ReadAll(io.LimitReader(resp.Body, maxIntakeRespBytes)) + + var tr telegramResp + if err := json.Unmarshal(rb, &tr); err != nil { + return fmt.Errorf("telegramsink: %s: %d with a body that is not the bot API envelope: %s", + method, resp.StatusCode, snippet(rb)) + } + if !tr.Ok { + return fmt.Errorf("telegramsink: %s: telegram returned error %d: %s", + method, tr.ErrorCode, strings.TrimSpace(tr.Description)) + } + if out == nil { + return nil + } + if err := json.Unmarshal(rb, out); err != nil { + return fmt.Errorf("telegramsink: %s: decode result: %w", method, err) + } + return nil +} + +// maxIntakeRespBytes — a getUpdates batch carries up to 100 messages, so the +// send path's cap is too small here. Still bounded: the body is wire-controlled +// and a relay sits in front of it. +const maxIntakeRespBytes = 4 << 20 + +// The inbound shapes, cut to what the poller reads. +type update struct { + UpdateID int64 `json:"update_id"` + Message *message `json:"message,omitempty"` + CallbackQuery *callbackQuery `json:"callback_query,omitempty"` +} + +type message struct { + MessageID int64 `json:"message_id"` + Chat chat `json:"chat"` + Text string `json:"text"` +} + +type callbackQuery struct { + ID string `json:"id"` + Data string `json:"data"` + Message message `json:"message"` +} + +// chat — the id arrives as a JSON number for a user and a string for a channel, +// and the config holds whichever was written. json.Number keeps both without +// choosing. +type chat struct { + ID json.Number `json:"id"` + Username string `json:"username,omitempty"` +} + +func (c chat) idString() string { + if s := c.ID.String(); s != "" { + return s + } + if c.Username != "" { + return "@" + c.Username + } + return "" +} + +// inlineKeyboard — the reply_markup shape. Rows of buttons, each carrying +// callback data. +type inlineKeyboard struct { + Rows [][]inlineButton `json:"inline_keyboard"` +} + +type inlineButton struct { + Text string `json:"text"` + Data string `json:"callback_data"` +} diff --git a/internal/delivery/telegramsink/correction.go b/internal/delivery/telegramsink/correction.go new file mode 100644 index 0000000..c09d6de --- /dev/null +++ b/internal/delivery/telegramsink/correction.go @@ -0,0 +1,101 @@ +// correction.go — the correction gesture as it appears in the chat (V-637). +// Two taps at most: "не то" opens the seven intents, and one of them writes the +// label. The web's version of the same gesture is cmd/mavweb/chat.go. +package telegramsink + +import ( + "fmt" + "strconv" + "strings" +) + +// CorrectionTargets — the intents a correction may name, in the order the +// buttons are drawn. It mirrors the seven the web offers, and it is a closed +// list for the same reason: V-632 fits prototypes from the label table, and a +// label nothing can score is worse than no label. +var CorrectionTargets = []string{"fact", "note", "reminder", "query", "act", "chat", "system"} + +// correctionKeyboard — the one gesture beside the reply. Nothing when the turn +// did not persist: a button that cannot name a row would report a failure the +// owner cannot act on. +func (p *Poller) correctionKeyboard(traceID int64) *inlineKeyboard { + if traceID <= 0 || p.correct == nil { + return nil + } + return &inlineKeyboard{Rows: [][]inlineButton{{ + {Text: "не то", Data: fmt.Sprintf("%s%d", prefixAsk, traceID)}, + }}} +} + +// targetKeyboard — the seven intents, plus the cheap half kept reachable. He +// opened the row without knowing he had to name something, and closing it with +// no way out would price the negative he was willing to give. +func targetKeyboard(traceID int64) *inlineKeyboard { + var rows [][]inlineButton + row := []inlineButton{} + for _, t := range CorrectionTargets { + row = append(row, inlineButton{Text: t, Data: fmt.Sprintf("%s%d:%s", prefixTarget, traceID, t)}) + if len(row) == 4 { + rows, row = append(rows, row), nil + } + } + if len(row) > 0 { + rows = append(rows, row) + } + return &inlineKeyboard{Rows: append(rows, []inlineButton{ + {Text: "просто неверно", Data: fmt.Sprintf("%s%d:", prefixTarget, traceID)}, + })} +} + +// Callback data is capped at 64 bytes by telegram, so it carries the trace id +// and the target and nothing else. +const ( + prefixAsk = "w:" + prefixTarget = "t:" +) + +type callbackKind int + +const ( + callbackUnknown callbackKind = iota + callbackAskTarget + callbackTarget +) + +// parseCallback reads button data. An unparseable id, or a target that is not +// one of the seven, is callbackUnknown — the data came off the wire, and a +// label the fitting code cannot score is worse than no label. +func parseCallback(data string) (traceID int64, target string, kind callbackKind) { + switch { + case strings.HasPrefix(data, prefixAsk): + id, err := strconv.ParseInt(strings.TrimPrefix(data, prefixAsk), 10, 64) + if err != nil || id <= 0 { + return 0, "", callbackUnknown + } + return id, "", callbackAskTarget + case strings.HasPrefix(data, prefixTarget): + rest := strings.TrimPrefix(data, prefixTarget) + idPart, target, ok := strings.Cut(rest, ":") + if !ok { + return 0, "", callbackUnknown + } + id, err := strconv.ParseInt(idPart, 10, 64) + if err != nil || id <= 0 { + return 0, "", callbackUnknown + } + if target != "" && !isCorrectionTarget(target) { + return 0, "", callbackUnknown + } + return id, target, callbackTarget + } + return 0, "", callbackUnknown +} + +func isCorrectionTarget(s string) bool { + for _, t := range CorrectionTargets { + if t == s { + return true + } + } + return false +} diff --git a/internal/delivery/telegramsink/correction_test.go b/internal/delivery/telegramsink/correction_test.go new file mode 100644 index 0000000..b0c89f0 --- /dev/null +++ b/internal/delivery/telegramsink/correction_test.go @@ -0,0 +1,30 @@ +package telegramsink + +import "testing" + +// Button data comes off the wire. An unparseable id or an intent that is not one +// of the seven must not reach the label table V-632 fits prototypes from. +func TestParseCallbackRejectsWhatCannotBeALabel(t *testing.T) { + for _, data := range []string{ + "", "nonsense", "w:", "w:0", "w:-3", "w:abc", + "t:77", "t:0:note", "t:abc:note", "t:77:погода", "t:77:fact:extra", + } { + if _, _, kind := parseCallback(data); kind != callbackUnknown { + t.Errorf("%q was accepted, want callbackUnknown", data) + } + } + if id, target, kind := parseCallback("t:77:reminder"); id != 77 || target != "reminder" || kind != callbackTarget { + t.Errorf("got %d %q %v, want the reminder correction", id, target, kind) + } +} + +// Every intent the web offers has a button here, so a new intent cannot exist +// with no way to correct a chat turn into it. +func TestIntakeTargetsAreTheSeven(t *testing.T) { + if len(CorrectionTargets) != 7 { + t.Fatalf("%d targets, want the seven public intents", len(CorrectionTargets)) + } + if isCorrectionTarget("") { + t.Error("empty is the absence of a target, not one of them") + } +} diff --git a/internal/delivery/telegramsink/intake.go b/internal/delivery/telegramsink/intake.go new file mode 100644 index 0000000..09e6035 --- /dev/null +++ b/internal/delivery/telegramsink/intake.go @@ -0,0 +1,210 @@ +// intake.go — the inbound half of the telegram channel (V-637). +// +// Until this file, telegram was a reach and nothing else: the sink pushes an +// away message and the chat has no way to answer. That made the correction +// gesture (V-630) reachable from the web and from voice only, and the sample of +// labels skews to wherever the owner happens to be standing. +// +// Long-poll getUpdates, not a webhook. The box takes no inbound connections and +// it reaches api.telegram.org through a relay, so the direction of the +// connection has to stay outbound. The poller is off unless the telegram block +// says intake, and it accepts messages from exactly one chat. +package telegramsink + +import ( + "context" + "errors" + "fmt" + "log" + "net/http" + "strings" + "time" +) + +// longPollSeconds — how long telegram holds an empty getUpdates open. The HTTP +// client's own timeout has to sit above it or every poll ends as a transport +// error, which is why the poller does not reuse the sink's client. +const longPollSeconds = 25 + +// pollBackoff — the wait after a failed poll. The relay going down is the +// normal cause and it comes back on its own, so this is a quiet retry rather +// than an escalation. +const pollBackoff = 15 * time.Second + +// Turn runs one utterance as a turn and reports the reply and the persisted +// trace id. traceID 0 means nothing persisted, and then the reply carries no +// correction buttons — there is no row for them to point at. +type Turn func(ctx context.Context, conversation, text string) (reply string, traceID int64, err error) + +// Correct records the owner's correction of one turn. shouldBe empty is the +// cheap half of the gesture: wrong, target unstated. +type Correct func(ctx context.Context, traceID int64, shouldBe string) error + +// Poller reads the configured chat and answers in it. One per daemon. +type Poller struct { + sink *Sink + turn Turn + correct Correct + hc *http.Client + offset int64 +} + +// NewPoller builds the intake half around an already-validated sink, so the +// token, the base URL and the relay are resolved in one place. turn is +// required; correct may be nil, and then the reply carries no buttons. +func NewPoller(s *Sink, turn Turn, correct Correct) (*Poller, error) { + if s == nil { + return nil, errors.New("telegramsink: intake needs a sink") + } + if turn == nil { + return nil, errors.New("telegramsink: intake needs a turn handler") + } + // The push half accepts @channelusername as a destination. The intake half + // cannot: an inbound update names its chat by numeric id, so an @-name would + // match nothing and the poller would read the chat and answer none of it. + // Refusing here is the difference between a boot error and a dead reach. + if strings.HasPrefix(strings.TrimSpace(s.cfg.ChatID), "@") { + return nil, fmt.Errorf("telegramsink: intake needs the numeric chat id, not %s", s.cfg.ChatID) + } + // The sink's transport already carries the relay. Only the timeout differs, + // and it has to clear the long poll. + hc := &http.Client{ + Timeout: (longPollSeconds + 10) * time.Second, + Transport: s.hc.Transport, + } + return &Poller{sink: s, turn: turn, correct: correct, hc: hc}, nil +} + +// Run polls until the context ends. It never returns an error: a chat that +// cannot be read is a degraded reach, not a reason to stop the daemon. +func (p *Poller) Run(ctx context.Context) { + p.discardBacklog(ctx) + log.Printf("telegram intake: reading chat %s", p.sink.cfg.ChatID) + for ctx.Err() == nil { + updates, err := p.getUpdates(ctx, longPollSeconds) + if err != nil { + if ctx.Err() != nil { + return + } + log.Printf("telegram intake: poll: %v", err) + select { + case <-ctx.Done(): + return + case <-time.After(pollBackoff): + } + continue + } + for _, u := range updates { + p.handle(ctx, u) + } + } +} + +// discardBacklog moves the offset past whatever is already queued, without +// acting on any of it. +// +// Telegram holds undelivered updates for 24 hours, so a daemon that was down +// overnight would otherwise wake up and answer every question in order. A +// question asked eight hours ago has been answered by the owner himself or has +// stopped mattering, and a reminder set from it would land at the wrong time. +// Missing it is the safe direction. +func (p *Poller) discardBacklog(ctx context.Context) { + // getUpdates returns at most 100 per call, so one call is not the queue. The + // loop is bounded rather than "until empty": the timeout is 0, so an instance + // that keeps handing back a full batch would spin, and a thousand skipped + // messages is already a box that was down for a long time. + skipped := 0 + for range 10 { + updates, err := p.getUpdates(ctx, 0) + if err != nil { + // Not fatal. The offset stays where it was, so the first real poll sees + // what is left and answers it late. Say so rather than hide it. + log.Printf("telegram intake: could not skip the backlog, old messages may be answered: %v", err) + return + } + skipped += len(updates) + if len(updates) == 0 { + break + } + } + if skipped > 0 { + log.Printf("telegram intake: skipped %d message(s) queued while the daemon was down", skipped) + } +} + +// handle dispatches one update. Anything that is neither a message from the +// owner's chat nor a callback on one of Maven's own keyboards is dropped in +// silence: a reply to a stranger confirms the bot exists and who it belongs to. +func (p *Poller) handle(ctx context.Context, u update) { + switch { + case u.CallbackQuery != nil: + p.onCallback(ctx, u.CallbackQuery) + case u.Message != nil: + p.onMessage(ctx, u.Message) + } +} + +func (p *Poller) onMessage(ctx context.Context, m *message) { + text := strings.TrimSpace(m.Text) + if text == "" || !p.fromOwner(m.Chat.idString()) { + return + } + // The conversation id keys the dialogue, so a clarify question asked in the + // chat is not answered by an utterance typed on the web. + reply, traceID, err := p.turn(ctx, "telegram:"+m.Chat.idString(), text) + if err != nil { + log.Printf("telegram intake: turn: %v", err) + return + } + if strings.TrimSpace(reply) == "" { + return + } + if err := p.send(ctx, reply, p.correctionKeyboard(traceID)); err != nil { + log.Printf("telegram intake: reply: %v", err) + } +} + +// onCallback handles a tap on a correction button. Every path answers the +// callback: telegram spins a clock on the button until it is answered, and an +// unanswered tap reads as a gesture that was dropped. +func (p *Poller) onCallback(ctx context.Context, cb *callbackQuery) { + if !p.fromOwner(cb.Message.Chat.idString()) { + return + } + traceID, target, kind := parseCallback(cb.Data) + if kind == callbackUnknown || p.correct == nil { + p.answerCallback(ctx, cb.ID, "") + return + } + // A tap on "не то" only opens the second row. Nothing is written yet: the + // target is worth much more than the negative, so he gets the chance to name + // it before the gesture is spent. + if kind == callbackAskTarget { + p.answerCallback(ctx, cb.ID, "") + if err := p.editKeyboard(ctx, cb.Message.Chat.idString(), cb.Message.MessageID, targetKeyboard(traceID)); err != nil { + log.Printf("telegram intake: open the target row: %v", err) + } + return + } + if err := p.correct(ctx, traceID, target); err != nil { + log.Printf("telegram intake: correct turn %d: %v", traceID, err) + p.answerCallback(ctx, cb.ID, "не записалось") + return + } + p.answerCallback(ctx, cb.ID, "записала") + // The buttons come off, because the correction is given and a live keyboard + // on an answered turn invites correcting it twice. + if err := p.editKeyboard(ctx, cb.Message.Chat.idString(), cb.Message.MessageID, nil); err != nil { + log.Printf("telegram intake: clear the keyboard: %v", err) + } +} + +// fromOwner — one chat, and it is the one the sink already sends to. Telegram +// chat ids are not guessable, but they are also not secret: they travel in +// every forwarded message. So this is the whole authorisation and it is an +// allowlist of one. +func (p *Poller) fromOwner(chatID string) bool { + return chatID != "" && chatID == p.cfgChatID() +} + +func (p *Poller) cfgChatID() string { return strings.TrimSpace(p.sink.cfg.ChatID) } diff --git a/internal/delivery/telegramsink/intake_test.go b/internal/delivery/telegramsink/intake_test.go new file mode 100644 index 0000000..780336d --- /dev/null +++ b/internal/delivery/telegramsink/intake_test.go @@ -0,0 +1,201 @@ +package telegramsink + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" +) + +// The turn he types in the chat is the turn the web would run, and the reply +// carries the one gesture beside it. +func TestIntakeRunsTheTurnAndOffersTheCorrection(t *testing.T) { + b := newFakeBot(t) + rec := &recorder{reply: "поняла", traceID: 91} + p := newTestPoller(t, b, rec) + + p.handle(context.Background(), msg(ownerChat, " поужинал ")) + + if got := rec.took(); len(got) != 1 || got[0] != "поужинал" { + t.Fatalf("turns %q, want the trimmed utterance once", got) + } + // The dialogue is keyed per chat, so a clarify asked here is not answered on + // the web. + if rec.conversation != "telegram:"+ownerChat { + t.Errorf("conversation %q does not name the chat", rec.conversation) + } + sends := b.called("sendMessage") + if len(sends) != 1 { + t.Fatalf("%d sends, want 1", len(sends)) + } + if sends[0].body["text"] != "поняла" { + t.Errorf("sent %v, want the reply", sends[0].body["text"]) + } + if sends[0].body["protect_content"] != true { + t.Error("his own data went out forwardable") + } + kb, _ := json.Marshal(sends[0].body["reply_markup"]) + if !strings.Contains(string(kb), "w:91") { + t.Errorf("keyboard %s does not point at the turn's trace", kb) + } +} + +// A turn nothing persisted has no row to correct, and a button that would name +// one reports a failure he cannot act on. +func TestIntakeSkipsTheGestureWithNoTrace(t *testing.T) { + b := newFakeBot(t) + p := newTestPoller(t, b, &recorder{reply: "поняла", traceID: 0}) + + p.handle(context.Background(), msg(ownerChat, "привет")) + + sends := b.called("sendMessage") + if len(sends) != 1 { + t.Fatalf("%d sends, want 1", len(sends)) + } + if _, ok := sends[0].body["reply_markup"]; ok { + t.Error("offered a correction on a turn with no trace") + } +} + +// One chat, and a stranger is not answered at all: a reply confirms the bot +// exists and whose it is. +func TestIntakeIgnoresAnyOtherChat(t *testing.T) { + b := newFakeBot(t) + rec := &recorder{reply: "поняла", traceID: 5} + p := newTestPoller(t, b, rec) + + p.handle(context.Background(), msg("9999", "включи свет")) + p.handle(context.Background(), update{UpdateID: 8, CallbackQuery: &callbackQuery{ + ID: "cb", Data: "t:5:note", Message: message{Chat: chat{ID: json.Number("9999")}}, + }}) + + if got := rec.took(); len(got) != 0 { + t.Errorf("ran %q for a chat that is not the owner's", got) + } + if len(rec.corrections) != 0 { + t.Errorf("wrote %v from a chat that is not the owner's", rec.corrections) + } + if len(b.calls) != 0 { + t.Errorf("answered a stranger: %v", b.calls) + } +} + +// Tapping "не то" opens the seven and writes nothing yet. The target is worth +// much more than the negative, so it must not be spent before he can name it. +func TestIntakeFirstTapOnlyOpensTheTargets(t *testing.T) { + b := newFakeBot(t) + rec := &recorder{} + p := newTestPoller(t, b, rec) + + p.handle(context.Background(), update{UpdateID: 9, CallbackQuery: &callbackQuery{ + ID: "cb", Data: "w:77", Message: message{MessageID: 11, Chat: chat{ID: json.Number(ownerChat)}}, + }}) + + if len(rec.corrections) != 0 { + t.Fatalf("wrote %v before he named a target", rec.corrections) + } + if len(b.called("answerCallbackQuery")) != 1 { + t.Error("left the clock spinning on the button") + } + edits := b.called("editMessageReplyMarkup") + if len(edits) != 1 { + t.Fatalf("%d edits, want the target row", len(edits)) + } + kb, _ := json.Marshal(edits[0].body["reply_markup"]) + for _, want := range CorrectionTargets { + if !strings.Contains(string(kb), `"`+want+`"`) { + t.Errorf("target row %s is missing %s", kb, want) + } + } + // And the way out, because he opened the row without knowing he had to name + // anything. + if !strings.Contains(string(kb), `"t:77:"`) { + t.Errorf("target row %s prices out the untargeted negative", kb) + } +} + +func TestIntakeWritesTheCorrection(t *testing.T) { + for _, tc := range []struct { + name, data, want string + }{ + {"with a target", "t:77:note", "note"}, + {"untargeted", "t:77:", ""}, + } { + t.Run(tc.name, func(t *testing.T) { + b := newFakeBot(t) + rec := &recorder{} + p := newTestPoller(t, b, rec) + + p.handle(context.Background(), update{UpdateID: 9, CallbackQuery: &callbackQuery{ + ID: "cb", Data: tc.data, Message: message{MessageID: 11, Chat: chat{ID: json.Number(ownerChat)}}, + }}) + + if len(rec.corrections) != 1 || rec.corrections[0] != (correction{77, tc.want}) { + t.Fatalf("corrections %v, want trace 77 → %q", rec.corrections, tc.want) + } + // The buttons come off once the gesture is given. + edits := b.called("editMessageReplyMarkup") + if len(edits) != 1 { + t.Fatalf("%d edits, want the keyboard cleared", len(edits)) + } + kb, _ := json.Marshal(edits[0].body["reply_markup"]) + if strings.Contains(string(kb), "t:77") { + t.Errorf("keyboard %s still invites a second correction", kb) + } + }) + } +} + +// A write that failed says so on the button. Silence would read as recorded. +func TestIntakeSaysWhenTheLabelDidNotLand(t *testing.T) { + b := newFakeBot(t) + rec := &recorder{correctErr: errors.New("no such routing trace")} + p := newTestPoller(t, b, rec) + + p.handle(context.Background(), update{UpdateID: 9, CallbackQuery: &callbackQuery{ + ID: "cb", Data: "t:77:fact", Message: message{MessageID: 11, Chat: chat{ID: json.Number(ownerChat)}}, + }}) + + answers := b.called("answerCallbackQuery") + if len(answers) != 1 || answers[0].body["text"] == "" { + t.Fatalf("answers %v, want a toast saying it did not land", answers) + } + if len(b.called("editMessageReplyMarkup")) != 0 { + t.Error("cleared the buttons after a failed write, so he cannot try again") + } +} + +// A question asked while the daemon was down has been answered by him or has +// stopped mattering, and a reminder set from it would land at the wrong time. +func TestIntakeDiscardsTheBacklog(t *testing.T) { + b := newFakeBot(t, []update{msg(ownerChat, "напомни в 7 позвонить маме")}) + rec := &recorder{reply: "поняла", traceID: 3} + p := newTestPoller(t, b, rec) + + p.discardBacklog(context.Background()) + + if got := rec.took(); len(got) != 0 { + t.Errorf("answered %q from the overnight queue", got) + } + // And the offset moved past it, so the next poll does not see it again. + if p.offset != 8 { + t.Errorf("offset %d, want the skipped update acknowledged", p.offset) + } +} + +// The poller does not start without somewhere to send the turn. +func TestNewPollerNeedsATurn(t *testing.T) { + sink, err := New(Config{BotToken: "t", ChatID: ownerChat}) + if err != nil { + t.Fatal(err) + } + if _, err := NewPoller(sink, nil, nil); err == nil { + t.Error("built a poller that reads the chat and answers nothing") + } + if _, err := NewPoller(nil, func(context.Context, string, string) (string, int64, error) { + return "", 0, nil + }, nil); err == nil { + t.Error("built a poller with no sink to answer through") + } +} diff --git a/internal/delivery/telegramsink/intakeharness_test.go b/internal/delivery/telegramsink/intakeharness_test.go new file mode 100644 index 0000000..b255f4b --- /dev/null +++ b/internal/delivery/telegramsink/intakeharness_test.go @@ -0,0 +1,123 @@ +// intakeharness_test.go — a fake bot API and a recorder for what the poller +// asked the daemon to do. Shared by the intake tests beside it. +package telegramsink + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +// fakeBot stands in for the bot API. It hands out queued updates once, records +// every other call, and answers the ok=true envelope the poller checks. +type fakeBot struct { + mu sync.Mutex + updates [][]update // one batch per getUpdates call, then empty + calls []botCall + srv *httptest.Server +} + +type botCall struct { + method string + body map[string]any +} + +func newFakeBot(t *testing.T, batches ...[]update) *fakeBot { + t.Helper() + b := &fakeBot{updates: batches} + b.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + method := r.URL.Path[strings.LastIndex(r.URL.Path, "/")+1:] + raw, _ := io.ReadAll(r.Body) + var body map[string]any + _ = json.Unmarshal(raw, &body) + + b.mu.Lock() + b.calls = append(b.calls, botCall{method: method, body: body}) + var batch []update + if method == "getUpdates" && len(b.updates) > 0 { + batch, b.updates = b.updates[0], b.updates[1:] + } + b.mu.Unlock() + + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "result": batch}) + })) + t.Cleanup(b.srv.Close) + return b +} + +func (b *fakeBot) called(method string) []botCall { + b.mu.Lock() + defer b.mu.Unlock() + var out []botCall + for _, c := range b.calls { + if c.method == method { + out = append(out, c) + } + } + return out +} + +// recorder collects what the poller asked the daemon to do. +type recorder struct { + mu sync.Mutex + turns []string + conversation string + traceID int64 + corrections []correction + reply string + err error + correctErr error +} + +type correction struct { + traceID int64 + shouldBe string +} + +func (r *recorder) turn(_ context.Context, conversation, text string) (string, int64, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.turns = append(r.turns, text) + r.conversation = conversation + return r.reply, r.traceID, r.err +} + +func (r *recorder) correct(_ context.Context, traceID int64, shouldBe string) error { + r.mu.Lock() + defer r.mu.Unlock() + r.corrections = append(r.corrections, correction{traceID, shouldBe}) + return r.correctErr +} + +func (r *recorder) took() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.turns...) +} + +const ownerChat = "4242" + +func newTestPoller(t *testing.T, b *fakeBot, rec *recorder) *Poller { + t.Helper() + sink, err := New(Config{BotToken: "secret-token", ChatID: ownerChat, BaseURL: b.srv.URL}) + if err != nil { + t.Fatal(err) + } + p, err := NewPoller(sink, rec.turn, rec.correct) + if err != nil { + t.Fatal(err) + } + return p +} + +func msg(chatID, text string) update { + return update{UpdateID: 7, Message: &message{ + MessageID: 11, Text: text, Chat: chat{ID: json.Number(chatID)}, + }} +} diff --git a/internal/delivery/telegramsink/telegramsink.go b/internal/delivery/telegramsink/telegramsink.go index bbf157b..b4016ee 100644 --- a/internal/delivery/telegramsink/telegramsink.go +++ b/internal/delivery/telegramsink/telegramsink.go @@ -72,6 +72,13 @@ type Config struct { // Timeout — per-request; 0 = DefaultTimeout. a dead relay can't hang the // tick loop. Timeout time.Duration + + // Intake — read the chat as well as write to it (V-637). Off by default, + // like the search and weather blocks: a bot that only pushes cannot be + // talked into anything, and turning that off has to stay a deletion. When + // set, a message from ChatID becomes a turn and its reply carries the + // correction gesture. ChatID is the only accepted sender. + Intake bool `json:"intake,omitempty"` } // Sink — implements delivery.Sink via the telegram bot sendMessage API. one @@ -130,6 +137,11 @@ type sendMessageReq struct { Text string `json:"text"` DisableNotification bool `json:"disable_notification"` // false = ring (always — these are alarms) ProtectContent bool `json:"protect_content"` // true = no forwarding out of chat + + // ReplyMarkup — the inline keyboard, used only by the intake half (V-637): + // a reply to a turn he typed carries the correction gesture. nil on every + // push the sink sends, and omitted from the wire when nil. + ReplyMarkup *inlineKeyboard `json:"reply_markup,omitempty"` } // telegramResp — the shape telegram returns. ok=false on logical error with diff --git a/internal/tool/tool_test.go b/internal/tool/tool_test.go index ac98d4c..b69fb74 100644 --- a/internal/tool/tool_test.go +++ b/internal/tool/tool_test.go @@ -328,7 +328,7 @@ func TestExecEmptyCmdRefuses(t *testing.T) { func TestExecRefusesATargetTheSystemCannotHave(t *testing.T) { api := fakeAPI{tools: map[string]ipc.Tool{ "restart": {Name: "restart", Cmd: []string{"systemctl", "restart"}, Status: "enabled"}, - "drop": {Name: "drop", Cmd: []string{"dropdb"}, Destructive: true, Status: "enabled"}, + "drop": {Name: "drop", Cmd: []string{"dropdb"}, Destructive: true, Status: "enabled"}, }} ran := false e := NewExecutor(api, 0)