diff --git a/core/http/endpoints/studio/triggers.go b/core/http/endpoints/studio/triggers.go index 6d3db8e545..6b80179cd2 100644 --- a/core/http/endpoints/studio/triggers.go +++ b/core/http/endpoints/studio/triggers.go @@ -87,7 +87,7 @@ var triggerRules = []errRule{ verbatim(http.StatusConflict, typeConflict, "", studiosvc.ErrTooManyTriggers, studiosvc.ErrTooManyProposals), verbatim(http.StatusConflict, typeConflict, "studio: ", studiosvc.ErrTooManyChanges, studiosvc.ErrChangeStale), sentinelText(http.StatusConflict, typeConflict, studiosvc.ErrCannotUndo), - verbatim(http.StatusConflict, typeConflict, "studio: ", studiosvc.ErrProposalAnswered), + verbatim(http.StatusConflict, typeConflict, "studio: ", studiosvc.ErrProposalAnswered, studiosvc.ErrProposalExpired), verbatim(http.StatusConflict, typeTriggersOff, "", studiosvc.ErrTriggersOff), fixed(http.StatusServiceUnavailable, typeServerError, "the server is shutting down", studiosvc.ErrTriggersClosed), } diff --git a/core/http/react-ui/public/locales/de/chats.json b/core/http/react-ui/public/locales/de/chats.json index c7b1eaf1eb..d76ef761bc 100644 --- a/core/http/react-ui/public/locales/de/chats.json +++ b/core/http/react-ui/public/locales/de/chats.json @@ -22,6 +22,7 @@ "undone": "Rückgängig gemacht:", "again": "Erneut planen", "declined": "Abgelehnt:", + "expired": "Abgelaufen:", "restore": "Wiederherstellen", "blocked": { "ran": "Er ist schon einmal gelaufen; entferne ihn auf der Seite Trigger.", diff --git a/core/http/react-ui/public/locales/en/chats.json b/core/http/react-ui/public/locales/en/chats.json index bcad7db152..f6bd3330be 100644 --- a/core/http/react-ui/public/locales/en/chats.json +++ b/core/http/react-ui/public/locales/en/chats.json @@ -659,6 +659,7 @@ "undone": "Undone:", "again": "Schedule it again", "declined": "Declined:", + "expired": "Expired:", "restore": "Restore", "blocked": { "ran": "It already ran once; remove it from the Triggers page.", diff --git a/core/http/react-ui/public/locales/es/chats.json b/core/http/react-ui/public/locales/es/chats.json index 2568e440e6..db180c2163 100644 --- a/core/http/react-ui/public/locales/es/chats.json +++ b/core/http/react-ui/public/locales/es/chats.json @@ -22,6 +22,7 @@ "undone": "Deshecho:", "again": "Programarlo de nuevo", "declined": "Rechazado:", + "expired": "Caducada:", "restore": "Restaurar", "blocked": { "ran": "Ya se ejecutó una vez; elimínalo desde la página Disparadores.", diff --git a/core/http/react-ui/public/locales/fr/chats.json b/core/http/react-ui/public/locales/fr/chats.json index 3134fdc0b2..75e309e77d 100644 --- a/core/http/react-ui/public/locales/fr/chats.json +++ b/core/http/react-ui/public/locales/fr/chats.json @@ -22,6 +22,7 @@ "undone": "Annulé :", "again": "Le planifier de nouveau", "declined": "Refusé :", + "expired": "Expirée :", "restore": "Restaurer", "blocked": { "ran": "Il s'est déjà exécuté une fois ; supprimez-le depuis la page Déclencheurs.", diff --git a/core/http/react-ui/public/locales/id/chats.json b/core/http/react-ui/public/locales/id/chats.json index 724c4ebea3..866e628f81 100644 --- a/core/http/react-ui/public/locales/id/chats.json +++ b/core/http/react-ui/public/locales/id/chats.json @@ -22,6 +22,7 @@ "undone": "Diurungkan:", "again": "Jadwalkan lagi", "declined": "Ditolak:", + "expired": "Kedaluwarsa:", "restore": "Pulihkan", "blocked": { "ran": "Pemicu ini sudah pernah berjalan; hapus dari halaman Pemicu.", diff --git a/core/http/react-ui/public/locales/it/chats.json b/core/http/react-ui/public/locales/it/chats.json index 7f7a76f6b9..4c4893bb07 100644 --- a/core/http/react-ui/public/locales/it/chats.json +++ b/core/http/react-ui/public/locales/it/chats.json @@ -22,6 +22,7 @@ "undone": "Annullato:", "again": "Pianificalo di nuovo", "declined": "Rifiutato:", + "expired": "Scaduta:", "restore": "Ripristina", "blocked": { "ran": "È già stato eseguito una volta; rimuovilo dalla pagina Trigger.", diff --git a/core/http/react-ui/public/locales/ko/chats.json b/core/http/react-ui/public/locales/ko/chats.json index 17e236ccd8..acce99ae90 100644 --- a/core/http/react-ui/public/locales/ko/chats.json +++ b/core/http/react-ui/public/locales/ko/chats.json @@ -22,6 +22,7 @@ "undone": "취소됨:", "again": "다시 예약", "declined": "거절됨:", + "expired": "만료됨:", "restore": "복원", "blocked": { "ran": "이미 한 번 실행되었습니다. 트리거 페이지에서 삭제하세요.", diff --git a/core/http/react-ui/public/locales/pt-BR/chats.json b/core/http/react-ui/public/locales/pt-BR/chats.json index afe5d7aa34..b353fe9923 100644 --- a/core/http/react-ui/public/locales/pt-BR/chats.json +++ b/core/http/react-ui/public/locales/pt-BR/chats.json @@ -22,6 +22,7 @@ "undone": "Desfeito:", "again": "Agendar de novo", "declined": "Recusado:", + "expired": "Expirada:", "restore": "Restaurar", "blocked": { "ran": "Ele já foi executado uma vez; remova-o na página Gatilhos.", diff --git a/core/http/react-ui/public/locales/zh-CN/chats.json b/core/http/react-ui/public/locales/zh-CN/chats.json index 3bcfd38ed1..c7dbb60587 100644 --- a/core/http/react-ui/public/locales/zh-CN/chats.json +++ b/core/http/react-ui/public/locales/zh-CN/chats.json @@ -22,6 +22,7 @@ "undone": "已撤销：", "again": "重新安排", "declined": "已拒绝：", + "expired": "已过期：", "restore": "恢复", "blocked": { "ran": "它已经运行过一次；请在触发器页面中删除。", diff --git a/core/http/react-ui/src/components/chat/ProposalCard.jsx b/core/http/react-ui/src/components/chat/ProposalCard.jsx index 4c8cc03256..fe9d230ae2 100644 --- a/core/http/react-ui/src/components/chat/ProposalCard.jsx +++ b/core/http/react-ui/src/components/chat/ProposalCard.jsx @@ -154,13 +154,13 @@ export default function ProposalCard({ id, t, tick = 0 }) { ) } - if (view.kind === 'undone' || view.kind === 'declined') { + if (view.kind === 'undone' || view.kind === 'declined' || view.kind === 'expired') { const undone = view.kind === 'undone' return wrap(

- {t(undone ? 'proposal.receipt.undone' : 'proposal.receipt.declined')}{' '} + {t(undone ? 'proposal.receipt.undone' : view.kind === 'expired' ? 'proposal.receipt.expired' : 'proposal.receipt.declined')}{' '} {p.name}

{alert} diff --git a/core/http/react-ui/src/utils/api.js b/core/http/react-ui/src/utils/api.js index 4b477d07b2..57aea3581b 100644 --- a/core/http/react-ui/src/utils/api.js +++ b/core/http/react-ui/src/utils/api.js @@ -519,7 +519,7 @@ export const chatTriggersApi = { runs: (id) => fetchJSON(`/api/chats/triggers/${enc2(id)}/runs`), preview: (when, tz) => fetchJSON(`/api/chats/triggers/preview${q({ when, tz, n: 3 })}`), proposals: (chat) => fetchJSON(`/api/chats/triggers/proposals${q({ chat })}`), - // One proposal and what became of it: { state: pending|accepted|declined|undone, trigger, answered_at, can_undo, undo_blocked }. + // One proposal and what became of it: { state: pending|accepted|declined|expired|undone, trigger, answered_at, can_undo, undo_blocked }. proposal: (id) => fetchJSON(`/api/chats/triggers/proposals/${enc2(id)}`), undoProposal: (id) => postJSON(`/api/chats/triggers/proposals/${enc2(id)}/undo`, {}), reopenProposal: (id) => postJSON(`/api/chats/triggers/proposals/${enc2(id)}/reopen`, {}), diff --git a/core/http/react-ui/src/utils/chatProposal.js b/core/http/react-ui/src/utils/chatProposal.js index 6fb04e2576..d3892768d5 100644 --- a/core/http/react-ui/src/utils/chatProposal.js +++ b/core/http/react-ui/src/utils/chatProposal.js @@ -7,7 +7,7 @@ export const PROPOSAL_CARD = 'trigger_proposal' // What the card of a proposal shows, from the server's answer to // GET /api/chats/triggers/proposals/{id} (or the error that call raised): one of -// pending, accepted, declined or undone, as the server keeps it, removed when +// pending, accepted, declined, expired or undone, as the server keeps it, removed when // the trigger is gone from the Triggers page (404) and error otherwise. An // accepted proposal says whether Undo is still allowed and, if not, why // ('ran' or 'edited'). A state this code does not know is an error, not a guess. @@ -16,6 +16,7 @@ export function proposalView(res, err) { switch (res?.state) { case 'pending': case 'declined': + case 'expired': case 'undone': return { kind: res.state, trigger: res.trigger, at: res.answered_at || '' } case 'accepted': diff --git a/core/http/react-ui/tests/unit/chatfeatures.test.mjs b/core/http/react-ui/tests/unit/chatfeatures.test.mjs index 216df60973..711b7a6ab3 100644 --- a/core/http/react-ui/tests/unit/chatfeatures.test.mjs +++ b/core/http/react-ui/tests/unit/chatfeatures.test.mjs @@ -66,6 +66,7 @@ test('trigger proposals: the card says proposed, created (same id) or gone, and const ran = proposalView({ state: 'accepted', trigger: p, can_undo: false, undo_blocked: 'ran' }) assert.deepEqual([ran.canUndo, ran.blocked], [false, 'ran']) assert.equal(proposalView({ state: 'undone', trigger: p }).kind, 'undone') + assert.equal(proposalView({ state: 'expired', trigger: p }).kind, 'expired') assert.equal(proposalView({ state: 'declined', trigger: p }).kind, 'declined') assert.equal(proposalView(null, Object.assign(new Error('nope'), { status: 404 })).kind, 'removed') const failed = proposalView(null, Object.assign(new Error('boom'), { status: 500 })) diff --git a/core/services/studio/connectorcaps.go b/core/services/studio/connectorcaps.go new file mode 100644 index 0000000000..4323cd1b30 --- /dev/null +++ b/core/services/studio/connectorcaps.go @@ -0,0 +1,73 @@ +package studio + +import ( + "context" + + "github.com/mudler/LocalAI/pkg/addons/contract" +) + +// The capabilities the host asks about before it lets a connector chat do +// something on the owner's behalf. The names are the closed list of the +// contract, shared with the manifest and the binding. +const ( + CapTriggersPropose = contract.CapTriggersPropose + CapTriggersRun = contract.CapTriggersRun + CapProposalsConfirm = contract.CapProposalsConfirm +) + +// CapabilityRequest is one question to a CapabilityGate. Binding is the +// connector binding, Owner its owner and Agent its agent. Sender is the +// platform ID of the person who wrote, and is empty when the question is not +// about one message (the tools a turn gets, a run that is due). Kind is the +// kind of the proposal for CapProposalsConfirm, such as "trigger". Sensitive +// says that the thing asked for changes what the agent may do or reach, so a +// gate must not grant it on the word of a platform role alone. +type CapabilityRequest struct { + Capability string + Binding string + Owner string + Agent string + Sender string + Kind string + Sensitive bool +} + +// CapabilityGate decides whether a binding, and for a message its sender, may +// use a capability. It is the one seam between the features that act from a +// connector chat and the model of who may do what. An implementation reads +// the switches of the owner (the grants of the agent, the capabilities of the +// binding, pairing and roles) and answers from them. It must be fast and safe +// for concurrent use, and it must fail closed: any doubt is false. +type CapabilityGate interface { + Allow(ctx context.Context, r CapabilityRequest) bool +} + +// CapabilityGateFunc makes a function a CapabilityGate. +type CapabilityGateFunc func(ctx context.Context, r CapabilityRequest) bool + +// Allow calls f. +func (f CapabilityGateFunc) Allow(ctx context.Context, r CapabilityRequest) bool { return f(ctx, r) } + +// DenyCapabilities is the gate a service has until the application sets +// another: nothing is allowed. A connector chat therefore gets no trigger +// tool, no confirmation by text and no run from a trigger it proposed, until +// something enables them. +var DenyCapabilities CapabilityGate = CapabilityGateFunc(func(context.Context, CapabilityRequest) bool { return false }) + +// SetCapabilityGate gives the service the gate for the capabilities above. +// The application calls it once, before the service is shared. Nil restores +// DenyCapabilities. +func (s *Service) SetCapabilityGate(g CapabilityGate) { + if g == nil { + g = DenyCapabilities + } + s.capGate = g +} + +// allowCap asks the gate; a service without one denies. +func (s *Service) allowCap(ctx context.Context, r CapabilityRequest) bool { + if s.capGate == nil { + return false + } + return s.capGate.Allow(ctx, r) +} diff --git a/core/services/studio/connectorconfirm.go b/core/services/studio/connectorconfirm.go new file mode 100644 index 0000000000..02def0e795 --- /dev/null +++ b/core/services/studio/connectorconfirm.go @@ -0,0 +1,30 @@ +package studio + +import ( + "context" + + "github.com/mudler/LocalAI/core/services/studio/connectors" + "github.com/mudler/LocalAI/pkg/addons/contract" +) + +// answerPhrase answers a proposal when the message is exactly "confirm CODE" or +// "cancel CODE" and a waiting proposal of the binding has that code. It writes +// the answer back and reports true; for any other message it does nothing, so +// the text goes to the agent as usual. Who may answer is decided by the +// capability gate, not here. +func (c *Connectors) answerPhrase(ctx context.Context, b connectors.Binding, m contract.ConnectorMessage, say func(string)) bool { + h := c.svc.triggers + if h == nil { + return false + } + verb, code, ok := ParsePhrase(m.Text) + if !ok { + return false + } + reply, handled := h.AnswerByPhrase(ctx, PhraseFrom{Binding: b.ID, Owner: b.Owner, Agent: b.AgentID, + Chat: m.Chat, Thread: m.Thread, Sender: m.SenderID}, verb, code) + if handled { + say(reply) + } + return handled +} diff --git a/core/services/studio/connectoroutbox.go b/core/services/studio/connectoroutbox.go new file mode 100644 index 0000000000..75714fcf36 --- /dev/null +++ b/core/services/studio/connectoroutbox.go @@ -0,0 +1,168 @@ +package studio + +import ( + "context" + "errors" + "sync" + "time" + + "github.com/mudler/LocalAI/core/services/studio/connectors" + "github.com/mudler/LocalAI/pkg/addons/contract" +) + +// The outbox is the one door for text that the host writes to a platform on +// its own, not as the answer to a message: the report of a trigger run and the +// text of a proposal. A platform limits how fast a bot may write, so a binding +// sends at most outboxPerMinute of these in a minute. A message over the limit +// waits in a queue, in order, and goes when the minute allows; it is not +// dropped. +const ( + outboxPerMinute = 20 + outboxMaxQueue = 100 + outboxPollEvery = time.Second +) + +// Errors of the outbox. +var ( + // ErrBindingOff means the binding is missing or off, so nothing is sent. + ErrBindingOff = errors.New("studio: the connector binding is not on") + // ErrOutboxFull means a binding has the most messages it may have waiting. + ErrOutboxFull = errors.New("studio: too many connector messages are waiting to be sent") +) + +// Post is one message the host writes to a chat of a binding. +type Post struct { + Binding, Chat, Thread, Text string + // OnFail is called, from another goroutine, when a message that waited in + // the queue could not be sent. A message that goes at once reports its + // error as the result of Connectors.Post. + OnFail func(error) +} + +type outItem struct { + token string + Post +} + +type outbox struct { + sent []time.Time + queue []outItem + draining bool +} + +type outboxes struct { + mu sync.Mutex + by map[string]*outbox +} + +func (o *outboxes) get(id string) *outbox { + if o.by == nil { + o.by = map[string]*outbox{} + } + q := o.by[id] + if q == nil { + q = &outbox{} + o.by[id] = q + } + return q +} + +// prune forgets the sends of more than a minute ago. +func (q *outbox) prune(now time.Time) { + kept := q.sent[:0] + for _, t := range q.sent { + if now.Sub(t) < time.Minute { + kept = append(kept, t) + } + } + q.sent = kept +} + +// Post sends text to a chat of a binding. It sends at once when the binding is +// under its limit and nothing waits, and otherwise queues the message and +// reports queued. The binding must be on. +func (c *Connectors) Post(ctx context.Context, p Post) (queued bool, err error) { + b, err := c.o.Store.GetBinding(ctx, p.Binding) + if err != nil || !b.Enabled { + return false, ErrBindingOff + } + token, err := c.openToken(ctx, b) + if err != nil { + return false, err + } + it := outItem{token: token, Post: p} + o := &c.outbox + o.mu.Lock() + q := o.get(b.ID) + q.prune(c.o.Clock()) + if len(q.queue) == 0 && len(q.sent) < outboxPerMinute { + q.sent = append(q.sent, c.o.Clock()) + o.mu.Unlock() + return false, c.sendPost(ctx, b, it) + } + if len(q.queue) >= outboxMaxQueue { + o.mu.Unlock() + return false, ErrOutboxFull + } + q.queue = append(q.queue, it) + start := !q.draining + q.draining = true + o.mu.Unlock() + if start && !c.spawn(func() { c.drain(b) }) { + o.mu.Lock() + q.draining = false + o.mu.Unlock() + } + return true, nil +} + +func (c *Connectors) sendPost(ctx context.Context, b connectors.Binding, it outItem) error { + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + return c.send(ctx, b, contract.ConnectorRequest{Op: contract.ConnectorOpSend, Token: it.token, Chat: it.Chat, Thread: it.Thread, Text: it.Text}) +} + +// drain sends the queue of a binding as the limit allows, then stops. +func (c *Connectors) drain(b connectors.Binding) { + o := &c.outbox + poll := c.o.OutboxPoll + if poll <= 0 { + poll = outboxPollEvery + } + for { + o.mu.Lock() + q := o.get(b.ID) + q.prune(c.o.Clock()) + if len(q.queue) == 0 { + q.draining = false + o.mu.Unlock() + return + } + if len(q.sent) >= outboxPerMinute { + o.mu.Unlock() + select { + case <-c.done: + return + case <-time.After(poll): + } + continue + } + it := q.queue[0] + q.queue = q.queue[1:] + q.sent = append(q.sent, c.o.Clock()) + o.mu.Unlock() + if err := c.sendPost(context.Background(), b, it); err != nil { + c.log.Warn("studio: could not send a queued connector message", "binding", b.ID, "error", err) + if it.OnFail != nil { + it.OnFail(err) + } + } + } +} + +// Queued is how many messages of a binding wait for the limit. +func (c *Connectors) Queued(bindingID string) int { + c.outbox.mu.Lock() + defer c.outbox.mu.Unlock() + return len(c.outbox.get(bindingID).queue) +} diff --git a/core/services/studio/connectorplace.go b/core/services/studio/connectorplace.go new file mode 100644 index 0000000000..f255d9db92 --- /dev/null +++ b/core/services/studio/connectorplace.go @@ -0,0 +1,58 @@ +package studio + +import ( + "context" + "strings" + + "github.com/mudler/LocalAI/core/services/studio/connectors" + "github.com/mudler/LocalAI/pkg/addons/contract" +) + +// The place of a connector chat is where the platform shows it: the chat and +// the thread of the last message the bot answered. A tool that runs in the +// turn, and the host when a run reports back, need it to write to the same +// place. A connector chat is made for a sender or a thread, not for one +// platform chat, so the host keeps the place of each chat next to the chat +// mapping, under a key that cannot be the ID of a sender. Nothing in the +// store changes: the chat column holds "chat\x1fthread". +const ( + placeKeyPrefix = "place:" + placeSep = "\x1f" + // maxPlaceValue is the width of the chat column of the stores. + maxPlaceValue = 128 +) + +func placeKey(chatID string) string { return placeKeyPrefix + chatID } + +func encodePlace(chat, thread string) string { return chat + placeSep + thread } + +func decodePlace(v string) (chat, thread string, ok bool) { + chat, thread, ok = strings.Cut(v, placeSep) + return chat, thread, ok && chat != "" +} + +// rememberPlace keeps where the message m was written, for the chat chatID of +// binding b. A place that does not fit the store is not kept: the chat then +// has no place, and what needs one falls back to the Inbox. +func (c *Connectors) rememberPlace(ctx context.Context, b connectors.Binding, chatID string, m contract.ConnectorMessage) { + key, val := placeKey(chatID), encodePlace(m.Chat, m.Thread) + if len(key) > contract.MaxConnectorSender || len(val) > maxPlaceValue { + return + } + if cur, ok, err := c.o.Store.ChatOf(ctx, b.ID, key); err == nil && ok && cur == val { + return + } + if err := c.o.Store.SetChat(ctx, b.ID, key, val); err != nil { + c.log.Warn("studio: could not keep the place of a connector chat", "binding", b.ID, "error", err) + } +} + +// PlaceOf returns the platform chat and thread where the connector chat chatID +// of the binding was last written to. It is false when the chat has none. +func (c *Connectors) PlaceOf(ctx context.Context, bindingID, chatID string) (chat, thread string, ok bool) { + v, found, err := c.o.Store.ChatOf(ctx, bindingID, placeKey(chatID)) + if err != nil || !found { + return "", "", false + } + return decodePlace(v) +} diff --git a/core/services/studio/connectors.go b/core/services/studio/connectors.go index 0877410cfb..bef2da84e2 100644 --- a/core/services/studio/connectors.go +++ b/core/services/studio/connectors.go @@ -102,7 +102,10 @@ type ConnectorOptions struct { // FileCeiling reads the host ceiling of a connector file from the owner // policy, in bytes. Nil or zero: contract.DefaultFileCeilingBytes. FileCeiling func(ctx context.Context) (int64, error) - Logger *slog.Logger + // OutboxPoll is how often a queued message looks again for room under + // the limit of its binding; zero means one second. + OutboxPoll time.Duration + Logger *slog.Logger } // Connectors runs the bindings of the chats. A message becomes a chat of the @@ -140,6 +143,8 @@ type Connectors struct { emu sync.Mutex nextEdit map[string]time.Time // by binding: when its next edit may go + outbox outboxes // text the host writes on its own, by binding + amu sync.Mutex attached map[string][]string // by chat: files that attach_file marked for the answer of the turn @@ -729,6 +734,11 @@ func (c *Connectors) handle(ctx context.Context, b connectors.Binding, token str return } } + // "confirm 7K2Q" or "cancel 7K2Q" answers a proposal that this binding made; + // it is decided here, before any model turn, and the model never reads it. + if len(m.Media) == 0 && c.answerPhrase(ctx, b, m, say) { + return + } // A scoped agent whose folder is missing does not run, on any frontend // or worker: one plain reply, and the binding stays on, so it works // again once the folder is back. @@ -768,6 +778,7 @@ func (c *Connectors) handle(ctx context.Context, b connectors.Binding, token str say(sentUnavailable) return } + c.rememberPlace(ctx, b, chatID, m) // One reply goroutine serves a conversation; the rate limit stays the sender's. // The earlier messages of a thread go to the first turn of a chat that // the thread just started, once. diff --git a/core/services/studio/connectortriggers_test.go b/core/services/studio/connectortriggers_test.go new file mode 100644 index 0000000000..5c4382a3d9 --- /dev/null +++ b/core/services/studio/connectortriggers_test.go @@ -0,0 +1,624 @@ +package studio_test + +import ( + "context" + "errors" + "strings" + "sync" + "time" + + "github.com/mudler/LocalAI/core/services/studio" + "github.com/mudler/LocalAI/core/services/studio/connectors" + "github.com/mudler/LocalAI/core/services/studio/notify" + "github.com/mudler/LocalAI/core/services/studio/studiotest" + "github.com/mudler/LocalAI/core/services/studio/triggers" + "github.com/mudler/LocalAI/core/services/studio/workspace" + "github.com/mudler/LocalAI/pkg/addons/contract" + "github.com/mudler/LocalAI/tests/studiobot" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// postAddon is a connector add-on of a made-up platform. It records what the +// host writes, where, so the specs show what a person reads in a chat. +type postAddon struct { + mu sync.Mutex + sent []contract.ConnectorRequest +} + +func (a *postAddon) Connector(ctx context.Context, req contract.ConnectorRequest) (contract.ConnectorResponse, error) { + switch req.Op { + case contract.ConnectorOpPoll: + select { + case <-ctx.Done(): + case <-time.After(20 * time.Millisecond): + } + return contract.ConnectorResponse{NextOffset: req.Offset}, nil + case contract.ConnectorOpSend: + a.mu.Lock() + defer a.mu.Unlock() + a.sent = append(a.sent, req) + return contract.ConnectorResponse{MessageID: "out"}, nil + } + return contract.ConnectorResponse{}, nil +} + +// texts are the messages written to a chat and thread, oldest first. +func (a *postAddon) texts(chat, thread string) []string { + a.mu.Lock() + defer a.mu.Unlock() + var out []string + for _, r := range a.sent { + if r.Chat == chat && r.Thread == thread { + out = append(out, r.Text) + } + } + return out +} + +func (a *postAddon) count() int { + a.mu.Lock() + defer a.mu.Unlock() + return len(a.sent) +} + +// capSwitch is a CapabilityGate that a spec turns on and off by name. +type capSwitch struct { + mu sync.Mutex + on map[string]bool + ask []studio.CapabilityRequest +} + +func (g *capSwitch) Allow(_ context.Context, r studio.CapabilityRequest) bool { + g.mu.Lock() + defer g.mu.Unlock() + g.ask = append(g.ask, r) + return g.on[r.Capability] +} + +func (g *capSwitch) set(name string, on bool) { + g.mu.Lock() + defer g.mu.Unlock() + if g.on == nil { + g.on = map[string]bool{} + } + g.on[name] = on +} + +func (g *capSwitch) asked(name string) []studio.CapabilityRequest { + g.mu.Lock() + defer g.mu.Unlock() + var out []studio.CapabilityRequest + for _, r := range g.ask { + if r.Capability == name { + out = append(out, r) + } + } + return out +} + +var _ = Describe("triggers proposed from a connector chat", func() { + ctx := context.Background() + const ( + chatID = "C1" + thread = "T1" + ) + var ( + fake *postAddon + gate *capSwitch + r *connRig + b connectors.Binding + args string // the arguments of the next propose_trigger + seq int + ) + script := func(_ workspace.Actor, text string) []studiotest.ToolStep { + switch { + case text == "propose": + return []studiotest.ToolStep{{Tool: "propose_trigger", Args: args}, {Say: "I proposed it."}} + case text == "list": + return []studiotest.ToolStep{{Tool: "list_triggers", Args: `{}`}, {Say: "listed"}} + case strings.Contains(text, "RUN-SILENT"): + return []studiotest.ToolStep{{Say: "[SILENT]"}} + case strings.Contains(text, "RUN-ERR"): + return []studiotest.ToolStep{{Err: errors.New("the model broke")}} + case strings.Contains(text, "RUN-FAIL"): + return []studiotest.ToolStep{{Say: "@channel done: 42"}} + case strings.Contains(text, "RUN-LONG"): + return []studiotest.ToolStep{{Say: strings.Repeat("x", 5000)}} + case strings.Contains(text, "RUN"): + return []studiotest.ToolStep{{Say: "result is 42"}} + } + return []studiotest.ToolStep{{Say: "ok"}} + } + proposal := func(name, delivery string) string { + return `{"name":"` + name + `","prompt":"RUN do the thing","when":"daily 08:00","time_zone":"UTC"` + delivery + `}` + } + origin := `,"delivery":{"target":"origin"}` + + setup := func() { + fake = &postAddon{} + gate = &capSwitch{} + tr := studiobot.NewTransport(studiobot.Addon{ID: "chat-addon", Kind: contract.ConnectorKind{ID: "chatty", Title: "Chatty", Delivery: contract.DeliveryPull}, Runtime: fake}) + r = newConnRigOpts(newShared(), connStore(), tr, script, nil, func(o *studio.ConnectorOptions) { + o.OutboxPoll = 10 * time.Millisecond + o.RatePerMinute = 100 + }) + r.svc.SetCapabilityGate(gate) + b = r.bind(studio.ConnectorInput{Kind: "chatty", Token: "tok", Allow: []string{"11"}}) + args = proposal("Digest", origin) + seq = 0 + } + say := func(text string) { + seq++ + id := "m" + string(rune('a'+seq)) + Expect(r.conns.Ingest(ctx, "chat-addon", contract.ConnectorInbound{BindingID: b.ID, ConnectorMessage: contract.ConnectorMessage{ + Chat: chatID, Thread: thread, Mention: true, MessageID: id, SenderID: "11", SenderName: "Cy", Text: text}})).To(Succeed()) + } + answered := func(n int) { + waitFor("the answer", func() bool { return len(fake.texts(chatID, thread)) >= n }) + } + proposals := func() []triggers.Trigger { + ps, err := r.host.Proposals(ctx, alice) + Expect(err).NotTo(HaveOccurred()) + return ps + } + // propose runs a turn that proposes and waits for the code message and the answer. + propose := func() triggers.Trigger { + before := len(fake.texts(chatID, thread)) + say("propose") + answered(before + 2) + ps := proposals() + Expect(ps).NotTo(BeEmpty()) + return ps[len(ps)-1] + } + chatTools := func() []string { + l, err := r.svc.List(ctx, alice, 10) + Expect(err).NotTo(HaveOccurred()) + Expect(l).NotTo(BeEmpty()) + return r.h.ToolNames(l[0].ID) + } + + Describe("the tools", func() { + It("are not offered until the capability is on", func() { + setup() + say("hello") + answered(1) + names := chatTools() + Expect(names).NotTo(ContainElement("propose_trigger")) + Expect(names).NotTo(ContainElement("list_triggers")) + asked := gate.asked(studio.CapTriggersPropose) + Expect(asked).NotTo(BeEmpty()) + Expect(asked[0].Binding).To(Equal(b.ID)) + Expect(asked[0].Agent).To(Equal(r.agent.ID)) + Expect(asked[0].Owner).To(Equal(alice)) + }) + + It("are propose and list once it is on, and never change or delete", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + say("hello") + answered(1) + names := chatTools() + Expect(names).To(ContainElement("propose_trigger")) + Expect(names).To(ContainElement("list_triggers")) + Expect(names).NotTo(ContainElement("update_trigger")) + Expect(names).NotTo(ContainElement("delete_trigger")) + }) + + It("are not offered to the run of a trigger", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, true) + tr := r.schedule(studio.TriggerInput{AgentID: r.agent.ID, Prompt: "RUN go"}) + r.due() + run := r.waitEnded(tr.ID) + Expect(r.h.ToolNames(run.ChatID)).NotTo(ContainElement("propose_trigger")) + }) + }) + + Describe("a proposal", func() { + It("is shown in the chat with its code and waits for an answer", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + p := propose() + Expect(p.Enabled).To(BeFalse()) + Expect(p.Route.Binding).To(Equal(b.ID)) + Expect(p.Route.Chat).To(Equal(chatID)) + Expect(p.Route.Thread).To(Equal(thread)) + Expect(p.Route.Target).To(Equal(triggers.TargetOrigin)) + Expect(p.Route.Code).To(MatchRegexp(`^[A-Z2-9]{4}$`)) + Expect(p.Route.Expires.Sub(p.Route.ProposedAt)).To(Equal(24 * time.Hour)) + Expect(p.AgentID).To(Equal(r.agent.ID)) + texts := fake.texts(chatID, thread) + Expect(texts[0]).To(ContainSubstring("Proposal " + p.Route.Code)) + Expect(texts[0]).To(ContainSubstring(`confirm ` + p.Route.Code)) + Expect(texts[0]).To(ContainSubstring("this chat")) + Expect(texts[0]).To(ContainSubstring("RUN do the thing"), "the person reads what the run will do") + res := r.h.ResultsFor("propose_trigger") + Expect(res).To(HaveLen(1)) + Expect(res[0].Out).To(ContainSubstring("not active")) + Expect(res[0].Out).NotTo(ContainSubstring(p.Route.Code), "the model does not repeat the code") + }) + + It("defaults to the Inbox", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + args = proposal("Quiet", "") + p := propose() + Expect(p.Route.TargetOrDefault()).To(Equal(triggers.TargetInbox)) + Expect(fake.texts(chatID, thread)[0]).To(ContainSubstring("the Inbox")) + }) + + It("refuses a fourth proposal in an hour from one binding", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + for i, n := range []string{"One", "Two", "Three"} { + args = proposal(n, origin) + propose() + Expect(proposals()).To(HaveLen(i + 1)) + } + args = proposal("Four", origin) + before := len(fake.texts(chatID, thread)) + say("propose") + answered(before + 1) + Expect(proposals()).To(HaveLen(3)) + res := r.h.ResultsFor("propose_trigger") + Expect(res[len(res)-1].Err).To(MatchError(ContainSubstring("at most 3"))) + r.sh.clock.Advance(61 * time.Minute) + args = proposal("Five", origin) + propose() + Expect(proposals()).To(HaveLen(4)) + }) + + It("lapses after 24 hours and cannot be confirmed", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapProposalsConfirm, true) + p := propose() + r.sh.clock.Advance(24*time.Hour + time.Minute) + Expect(proposals()).To(BeEmpty(), "an expired proposal is not listed as waiting") + _, err := r.host.ConfirmProposal(ctx, alice, p.ID, "UTC") + Expect(err).To(MatchError(studio.ErrProposalExpired)) + v, err := r.host.ProposalStatus(ctx, alice, p.ID) + Expect(err).NotTo(HaveOccurred()) + Expect(v.State).To(Equal(studio.ProposalExpired)) + before := len(fake.texts(chatID, thread)) + say("confirm " + p.Route.Code) + Expect(fake.texts(chatID, thread)).To(HaveLen(before), "a code that no waiting proposal has goes to the agent: it answers") + }) + + It("is swept to expired by a pass", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + p := propose() + r.sh.clock.Advance(25 * time.Hour) + r.pass() + got, err := r.sh.trg.GetTrigger(ctx, p.ID) + Expect(err).NotTo(HaveOccurred()) + Expect(got.Outcome).To(Equal(triggers.OutcomeExpired)) + }) + }) + + Describe("an answer by text", func() { + It("from an allowed sender in the same chat and thread turns the proposal into a trigger, before any model turn", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapProposalsConfirm, true) + p := propose() + chats, _ := r.svc.List(ctx, alice, 10) + sentBefore := len(r.h.Sent(chats[0].ID)) + n := len(fake.texts(chatID, thread)) + say("Confirm " + strings.ToLower(p.Route.Code)) + answered(n + 1) + last := fake.texts(chatID, thread) + Expect(last[len(last)-1]).To(ContainSubstring("Confirmed")) + Expect(r.h.Sent(chats[0].ID)).To(HaveLen(sentBefore), "the model never read the phrase") + list, err := r.host.ListTriggers(ctx, alice) + Expect(err).NotTo(HaveOccurred()) + Expect(list).To(HaveLen(1)) + Expect(list[0].Enabled).To(BeTrue()) + Expect(list[0].Route.Target).To(Equal(triggers.TargetOrigin)) + asked := gate.asked(studio.CapProposalsConfirm) + Expect(asked).NotTo(BeEmpty()) + Expect(asked[0]).To(Equal(studio.CapabilityRequest{Capability: studio.CapProposalsConfirm, Binding: b.ID, Owner: alice, Agent: r.agent.ID, Sender: "11", Kind: "trigger"})) + // Single use: the code is spent. + n = len(fake.texts(chatID, thread)) + say("confirm " + p.Route.Code) + answered(n + 1) + Expect(r.host.Proposals(ctx, alice)).To(BeEmpty()) + list, _ = r.host.ListTriggers(ctx, alice) + Expect(list).To(HaveLen(1)) + }) + + It("cancel declines the proposal", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapProposalsConfirm, true) + p := propose() + n := len(fake.texts(chatID, thread)) + say("cancel " + p.Route.Code) + answered(n + 1) + v, err := r.host.ProposalStatus(ctx, alice, p.ID) + Expect(err).NotTo(HaveOccurred()) + Expect(v.State).To(Equal(studio.ProposalDeclined)) + }) + + It("is refused when the gate does not allow the sender", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + p := propose() + n := len(fake.texts(chatID, thread)) + say("confirm " + p.Route.Code) + answered(n + 1) + last := fake.texts(chatID, thread) + Expect(last[len(last)-1]).To(ContainSubstring("Studio")) + Expect(proposals()).To(HaveLen(1), "it still waits") + }) + + It("is refused from another thread", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapProposalsConfirm, true) + p := propose() + Expect(r.conns.Ingest(ctx, "chat-addon", contract.ConnectorInbound{BindingID: b.ID, ConnectorMessage: contract.ConnectorMessage{ + Chat: chatID, Thread: "T2", Mention: true, MessageID: "x1", SenderID: "11", Text: "confirm " + p.Route.Code}})).To(Succeed()) + waitFor("the refusal", func() bool { return len(fake.texts(chatID, "T2")) == 1 }) + Expect(fake.texts(chatID, "T2")[0]).To(ContainSubstring("same place")) + Expect(proposals()).To(HaveLen(1)) + }) + + It("with an unknown code is an ordinary message for the agent", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapProposalsConfirm, true) + propose() + chats, _ := r.svc.List(ctx, alice, 10) + before := len(r.h.Sent(chats[0].ID)) + say("confirm ZZZZ") + waitFor("the agent", func() bool { return len(r.h.Sent(chats[0].ID)) == before+1 }) + Expect(proposals()).To(HaveLen(1)) + }) + + It("is audited without any text of the prompt", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapProposalsConfirm, true) + p := propose() + n := len(fake.texts(chatID, thread)) + say("confirm " + p.Route.Code) + answered(n + 1) + evs, err := r.svc.Changes().Events(ctx, alice, 20) + Expect(err).NotTo(HaveOccurred()) + var actions []string + for _, e := range evs { + if e.Kind == "trigger" { + actions = append(actions, e.Action) + Expect(e.Hash).NotTo(BeEmpty()) + Expect(e.Detail).NotTo(ContainSubstring("RUN do the thing")) + Expect(e.Source.ID).To(Equal(b.ID)) + } + } + Expect(actions).To(ContainElements("propose", "confirm")) + }) + }) + + Describe("the report of a run", func() { + // confirmed makes a confirmed trigger that reports as asked, and + // returns it. The clock stands before its first due time. + confirmed := func(name, delivery string) triggers.Trigger { + args = proposal(name, delivery) + p := propose() + t, err := r.host.ConfirmProposal(ctx, alice, p.ID, "UTC") + Expect(err).NotTo(HaveOccurred()) + return t + } + runOnce := func(t triggers.Trigger) triggers.Run { + r.sh.clock.Advance(25 * time.Hour) + r.pass() + return r.waitEnded(t.ID) + } + reports := func() []string { + var out []string + for _, s := range fake.texts(chatID, thread) { + if strings.HasPrefix(s, "Digest\n\n") { + out = append(out, s) + } + } + return out + } + + It("is written back to the chat and thread of the proposal", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, true) + t := confirmed("Digest", origin) + run := runOnce(t) + Expect(run.Status).To(Equal(triggers.StatusDone)) + waitFor("the report", func() bool { return len(reports()) == 1 }) + Expect(reports()[0]).To(Equal("Digest\n\nresult is 42")) + Expect(r.notes()).To(BeEmpty(), "a report that reached the chat is not repeated in the Inbox") + }) + + It("goes to the Inbox by default", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, true) + t := confirmed("Digest", "") + runOnce(t) + waitFor("the notice", func() bool { return len(r.notes()) == 1 }) + Expect(reports()).To(BeEmpty()) + }) + + It("says nothing for a silent run, and logs it", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, true) + args = `{"name":"Digest","prompt":"RUN-SILENT","when":"daily 08:00","time_zone":"UTC","delivery":{"target":"origin"}}` + p := propose() + t, err := r.host.ConfirmProposal(ctx, alice, p.ID, "UTC") + Expect(err).NotTo(HaveOccurred()) + run := runOnce(t) + Expect(run.Status).To(Equal(triggers.StatusDone)) + time.Sleep(100 * time.Millisecond) + Expect(reports()).To(BeEmpty()) + Expect(r.notes()).To(BeEmpty()) + }) + + It("breaks mentions of everyone and cuts a long text", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, true) + args = `{"name":"Digest","prompt":"RUN-FAIL","when":"daily 08:00","time_zone":"UTC","delivery":{"target":"origin"}}` + p := propose() + t, _ := r.host.ConfirmProposal(ctx, alice, p.ID, "UTC") + runOnce(t) + waitFor("the report", func() bool { return len(reports()) == 1 }) + Expect(reports()[0]).NotTo(ContainSubstring("@channel")) + Expect(reports()[0]).To(ContainSubstring("done: 42")) + }) + + It("cuts a long answer", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, true) + args = `{"name":"Digest","prompt":"RUN-LONG","when":"daily 08:00","time_zone":"UTC","delivery":{"target":"origin"}}` + p := propose() + t, _ := r.host.ConfirmProposal(ctx, alice, p.ID, "UTC") + runOnce(t) + waitFor("the report", func() bool { return len(reports()) == 1 }) + Expect([]rune(reports()[0])).To(HaveLen(len("Digest\n\n") + studio.MaxReportRunes + 1)) + }) + + It("on_problem stays silent for a good run", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, true) + t := confirmed("Digest", `,"delivery":{"target":"origin","mode":"on_problem"}`) + runOnce(t) + time.Sleep(100 * time.Millisecond) + Expect(reports()).To(BeEmpty()) + Expect(r.notes()).To(BeEmpty()) + }) + + It("on_change reports the first run and then only a different answer", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, true) + t := confirmed("Digest", `,"delivery":{"target":"origin","mode":"on_change"}`) + runOnce(t) + waitFor("the first report", func() bool { return len(reports()) == 1 }) + runOnce(t) + time.Sleep(150 * time.Millisecond) + Expect(reports()).To(HaveLen(1), "same answer, nothing new") + }) + + It("falls back to the Inbox when the binding is off", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, true) + t := confirmed("Digest", origin) + _, err := r.conns.SetConnectorEnabled(ctx, alice, b.ID, false) + Expect(err).NotTo(HaveOccurred()) + runOnce(t) + waitFor("the notice", func() bool { return len(r.notes()) == 1 }) + Expect(reports()).To(BeEmpty()) + Expect(r.notes()[0].Kind).To(Equal(notify.KindRunEnded)) + }) + + It("pauses with the capability off, keeps the record, and resumes without replay", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, false) + t := confirmed("Digest", origin) + r.sh.clock.Advance(25 * time.Hour) + r.pass() + run := r.waitEnded(t.ID) + Expect(run.Status).To(Equal(triggers.StatusSkipped)) + Expect(run.Reason).To(Equal("paused: capability off")) + got, err := r.host.GetTrigger(ctx, alice, t.ID) + Expect(err).NotTo(HaveOccurred()) + Expect(got.Enabled).To(BeTrue(), "the record is unchanged") + Expect(r.notes()).To(BeEmpty(), "a pause is not a failure") + // Three more days pass while it is off: nothing is replayed once it is on. + r.sh.clock.Advance(72 * time.Hour) + gate.set(studio.CapTriggersRun, true) + r.pass() + runs, _ := r.host.TriggerRuns(ctx, alice, t.ID, 10) + ran := 0 + for _, x := range runs { + if x.Status == triggers.StatusDone || x.Status == triggers.StatusRunning { + ran++ + } + } + Expect(ran).To(BeNumerically("<=", 1), "runs = %+v", runs) + }) + + It("alerts once after three failures in a row and not again within six hours", func() { + setup() + gate.set(studio.CapTriggersPropose, true) + gate.set(studio.CapTriggersRun, true) + args = `{"name":"Digest","prompt":"RUN-ERR","when":"0 * * * *","time_zone":"UTC","delivery":{"target":"origin"}}` + p := propose() + t, err := r.host.ConfirmProposal(ctx, alice, p.ID, "UTC") + Expect(err).NotTo(HaveOccurred()) + alerts := func() int { + n := 0 + for _, s := range fake.texts(chatID, thread) { + if strings.Contains(s, "failed 3 times in a row") { + n++ + } + } + return n + } + failRun := func(advance time.Duration, want int) { + r.sh.clock.Advance(advance) + r.pass() + until("the run to fail", func() bool { + runs, _ := r.host.TriggerRuns(ctx, alice, t.ID, 20) + failed := 0 + for _, x := range runs { + if x.Status == triggers.StatusFailed { + failed++ + } + } + return failed >= want + }) + } + failRun(61*time.Minute, 1) + failRun(60*time.Minute, 2) + Expect(alerts()).To(Equal(0)) + failRun(60*time.Minute, 3) + waitFor("the alert", func() bool { return alerts() == 1 }) + failRun(60*time.Minute, 4) + time.Sleep(150 * time.Millisecond) + Expect(alerts()).To(Equal(1), "throttled") + failRun(6*time.Hour, 5) + waitFor("the second alert", func() bool { return alerts() == 2 }) + }) + }) + + Describe("the outbox", func() { + It("sends 20 messages a minute for a binding and queues the rest in order", func() { + setup() + for i := 0; i < 25; i++ { + _, err := r.conns.Post(ctx, studio.Post{Binding: b.ID, Chat: "Q", Text: "m" + string(rune('A'+i))}) + Expect(err).NotTo(HaveOccurred()) + } + Expect(fake.texts("Q", "")).To(HaveLen(20)) + Expect(r.conns.Queued(b.ID)).To(Equal(5)) + r.sh.clock.Advance(61 * time.Second) + waitFor("the queue to drain", func() bool { return len(fake.texts("Q", "")) == 25 }) + got := fake.texts("Q", "") + Expect(got[20:]).To(Equal([]string{"mU", "mV", "mW", "mX", "mY"})) + Expect(r.conns.Queued(b.ID)).To(Equal(0)) + }) + + It("refuses a binding that is off", func() { + setup() + _, err := r.conns.SetConnectorEnabled(ctx, alice, b.ID, false) + Expect(err).NotTo(HaveOccurred()) + _, err = r.conns.Post(ctx, studio.Post{Binding: b.ID, Chat: "Q", Text: "x"}) + Expect(err).To(MatchError(studio.ErrBindingOff)) + }) + }) +}) diff --git a/core/services/studio/proposals/store.go b/core/services/studio/proposals/store.go index f0985dc937..e4694f198e 100644 --- a/core/services/studio/proposals/store.go +++ b/core/services/studio/proposals/store.go @@ -60,6 +60,12 @@ const ( ActionExpire = "expire" ActionStale = "stale" ActionRevert = "revert" + // The actions of the reports that a trigger made from a connector chat + // sends: a run delivered, a trigger paused because its capability is + // off, and resumed. + ActionDeliver = "deliver" + ActionPause = "pause" + ActionResume = "resume" ) // Store keeps proposals and their audit. It is safe for concurrent use. diff --git a/core/services/studio/proposalstate.go b/core/services/studio/proposalstate.go index 8277bd346b..6e4b27835d 100644 --- a/core/services/studio/proposalstate.go +++ b/core/services/studio/proposalstate.go @@ -20,6 +20,8 @@ const ( ProposalAccepted ProposalState = "accepted" ProposalDeclined ProposalState = "declined" ProposalUndone ProposalState = "undone" + // ProposalExpired is a proposal that nobody answered in time. + ProposalExpired ProposalState = "expired" ) // Why an accepted trigger cannot be undone. The codes are for the interface, @@ -57,6 +59,8 @@ func stateOf(t triggers.Trigger) ProposalState { return ProposalDeclined case t.Outcome == triggers.OutcomeUndone: return ProposalUndone + case t.Outcome == triggers.OutcomeExpired: + return ProposalExpired } return ProposalAccepted } @@ -98,6 +102,7 @@ func (h *Triggers) ProposalStatus(ctx context.Context, user, id string) (Proposa if !t.Proposed && t.Outcome != triggers.OutcomeAccepted { return ProposalView{}, triggers.ErrNotFound // a trigger nobody proposed } + t = h.lapse(ctx, t) v := ProposalView{State: stateOf(t), Trigger: t, At: t.OutcomeAt} if v.State == ProposalAccepted { block, err := h.undoBlock(ctx, t) @@ -173,8 +178,9 @@ func (h *Triggers) ReopenProposal(ctx context.Context, user, id string) (trigger return triggers.Trigger{}, err } pending := 0 + now := h.o.Clock() for _, m := range mine { - if m.Pending() { + if m.Pending() && !m.Expired(now) { pending++ } } @@ -182,6 +188,10 @@ func (h *Triggers) ReopenProposal(ctx context.Context, user, id string) (trigger return triggers.Trigger{}, ErrTooManyProposals } t.Outcome, t.OutcomeAt, t.Touched = "", time.Time{}, false + if !t.Route.Expires.IsZero() { + // The owner asked for it again: it waits a full time once more. + t.Route.Expires = h.o.Clock().UTC().Add(ProposalTTL) + } if err := h.o.Store.PutTrigger(ctx, t); err != nil { return triggers.Trigger{}, err } @@ -238,6 +248,8 @@ func (h *Triggers) ProposalNote(ctx context.Context, user, chatID string) string case triggers.OutcomeUndone: parts = append(parts, fmt.Sprintf("%q undone by the user after they scheduled it, so it is not active", name)) again = true + case triggers.OutcomeExpired: + parts = append(parts, fmt.Sprintf("%q expired: nobody confirmed it in time, so it is not active", name)) } } line := "Trigger proposals and changes in this chat: " + strings.Join(parts, "; ") + "." diff --git a/core/services/studio/service.go b/core/services/studio/service.go index 923ec8b1de..76bb07db9e 100644 --- a/core/services/studio/service.go +++ b/core/services/studio/service.go @@ -316,11 +316,12 @@ type Service struct { admit sync.Mutex // Sub-agents; see subagent.go. - changes *Changes // the change proposals - triggers *Triggers // set once by SetTriggers - loops *Loops // set once by SetLoops - jobs *Jobs // set once by SetJobs - connectors *Connectors // set once by SetConnectors + changes *Changes // the change proposals + triggers *Triggers // set once by SetTriggers + loops *Loops // set once by SetLoops + jobs *Jobs // set once by SetJobs + connectors *Connectors // set once by SetConnectors + capGate CapabilityGate // set once by SetCapabilityGate; nil denies ctxfiles *contextfiles.Loader conversion ConversionSource // set once by SetConversion diff --git a/core/services/studio/triggerroute.go b/core/services/studio/triggerroute.go new file mode 100644 index 0000000000..8b0420f9e4 --- /dev/null +++ b/core/services/studio/triggerroute.go @@ -0,0 +1,411 @@ +package studio + +import ( + "context" + "crypto/rand" + "errors" + "fmt" + "regexp" + "strings" + "time" + "unicode/utf8" + + "github.com/mudler/LocalAI/core/services/studio/notify" + "github.com/mudler/LocalAI/core/services/studio/proposals" + "github.com/mudler/LocalAI/core/services/studio/triggers" + "github.com/mudler/LocalAI/pkg/addons/contract" +) + +// What a trigger that a connector chat proposed is bound by. +const ( + // ProposalTTL is how long such a proposal waits for an answer. + ProposalTTL = 24 * time.Hour + // MaxProposalsPerHour is how many proposals one binding may make in an + // hour, whoever writes to it. + MaxProposalsPerHour = 3 + // MaxReportRunes bounds the text of a run that is written to a chat. + MaxReportRunes = 4000 + // SilentMark is the whole final message of a run that has nothing to + // say: it is logged and delivers nothing. + SilentMark = "[SILENT]" + + // failureAlertAfter is how many failed runs in a row raise an alert, and + // failureAlertEvery how often one trigger may raise it. + failureAlertAfter = 3 + failureAlertEvery = 6 * time.Hour + + reasonCapOff = "paused: capability off" + codeLen = 4 + // codeAlphabet leaves out the letters and digits that look alike. + codeAlphabet = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" +) + +// auditKindTrigger is the Kind of the audit lines about triggers. +const auditKindTrigger contract.ChangeKind = "trigger" + +// Errors of the proposals that connector chats make. +var ( + // ErrProposalExpired means nobody answered the proposal in time. + ErrProposalExpired = errors.New("studio: this proposal expired") + // ErrProposalRate means a binding made the most proposals of the hour. + ErrProposalRate = fmt.Errorf("studio: at most %d trigger proposals an hour can come from one chat connection; try again later", MaxProposalsPerHour) +) + +// proposalsInHour counts the proposals of a binding made in the hour before +// now, answered or not: the limit is about asking, not about waiting. +func proposalsInHour(all []triggers.Trigger, binding string, now time.Time) int { + n := 0 + for _, t := range all { + if t.Proposed && !t.IsChange() && t.Route.Binding == binding && now.Sub(t.Route.ProposedAt) < time.Hour { + n++ + } + } + return n +} + +// freshCode returns a code that no waiting proposal of the binding has. +func freshCode(all []triggers.Trigger, binding string) string { + used := map[string]bool{} + for _, t := range all { + if t.Route.Binding == binding && t.Route.Code != "" { + used[t.Route.Code] = true + } + } + for { + var b [codeLen]byte + _, _ = rand.Read(b[:]) + for i := range b { + b[i] = codeAlphabet[int(b[i])%len(codeAlphabet)] + } + if c := string(b[:]); !used[c] { + return c + } + } +} + +// audit writes one line about a trigger that a connector chat is part of. The +// line has a hash of the prompt, never text. +func (h *Triggers) audit(ctx context.Context, action string, t triggers.Trigger, detail string) { + ch := h.svc.changes + if ch == nil { + return + } + ev := proposals.Event{At: h.o.Clock().UTC(), Owner: t.Owner, Action: action, Proposal: t.ID, Kind: auditKindTrigger, + Target: t.Name, Agent: t.AgentID, Hash: hashOf(t.Prompt), Detail: detail, + Source: contract.ChangeSource{Kind: "connector", ID: t.Route.Binding, Chat: t.Route.Chat, Thread: t.Route.Thread}} + if err := ch.store.AddEvent(ctx, ev); err != nil { + h.log.Warn("studio: could not write the trigger audit", "action", action, "trigger", t.ID, "error", err) + } +} + +// lapse marks a waiting proposal that is past its time as expired, and returns +// the record as it is after. Another frontend may have done it first; either +// way the record that is returned is the stored one. +func (h *Triggers) lapse(ctx context.Context, t triggers.Trigger) triggers.Trigger { + now := h.o.Clock().UTC() + if !t.Expired(now) { + return t + } + ok, err := h.o.Store.SetOutcome(ctx, t.ID, "", triggers.OutcomeExpired, now) + if err != nil { + h.log.Warn("studio: could not expire a trigger proposal", "trigger", t.ID, "error", err) + return t + } + if cur, gerr := h.o.Store.GetTrigger(ctx, t.ID); gerr == nil { + t = cur + } + if ok { + h.audit(ctx, proposals.ActionExpire, t, "") + } + return t +} + +// expireProposals marks every waiting proposal past its time as expired. The +// leader calls it in a pass, at most once a minute. +func (h *Triggers) expireProposals(ctx context.Context) { + h.mu.Lock() + if now := h.o.Clock(); now.Sub(h.swept) < time.Minute { + h.mu.Unlock() + return + } else { + h.swept = now + } + h.mu.Unlock() + all, err := h.o.Store.AllTriggers(ctx) + if err != nil { + h.log.Warn("studio: could not list triggers to expire proposals", "error", err) + return + } + now := h.o.Clock() + for _, t := range all { + if t.Expired(now) { + h.lapse(ctx, t) + } + } +} + +// ---- Confirmation by text ---- + +// Sentences the bot says about a proposal. +const ( + sentProposalGone = "That proposal expired. Ask again if you still want it." + sentProposalPlace = "Answer in the same place where it was proposed, or open Studio." + sentProposalNoRight = "Only a sender the owner trusts can answer a proposal from here. Open Studio to answer it." + sentProposalFailed = "I could not do that: %s" + sentProposalCancel = "Cancelled." +) + +var phraseRE = regexp.MustCompile(`(?i)^\s*(confirm|cancel)\s+([a-z0-9]{4})\s*$`) + +// ParsePhrase reads "confirm 7K2Q" or "cancel 7K2Q" from the whole text of a +// message. The code comes back in upper case. +func ParsePhrase(text string) (verb, code string, ok bool) { + m := phraseRE.FindStringSubmatch(text) + if m == nil { + return "", "", false + } + return strings.ToLower(m[1]), strings.ToUpper(m[2]), true +} + +// PhraseFrom is who wrote a confirm or cancel phrase, and where. +type PhraseFrom struct { + Binding, Owner, Agent string + Chat, Thread, Sender string +} + +// AnswerByPhrase answers the waiting proposal that has the code, when the +// phrase comes from the place it was made, and the sender holds the +// capability to confirm. It returns the sentence to write back, and handled +// false when no proposal of the binding has that code: the message is then an +// ordinary one. A proposal is answered once; every attempt is audited. +func (h *Triggers) AnswerByPhrase(ctx context.Context, from PhraseFrom, verb, code string) (reply string, handled bool) { + all, err := h.o.Store.ListTriggers(ctx, principal(from.Owner)) + if err != nil { + return "", false + } + var p triggers.Trigger + found := false + for _, t := range all { + if t.Pending() && t.Route.Binding == from.Binding && t.Route.Code == code { + p, found = t, true + break + } + } + if !found { + return "", false + } + detail := "by text from sender " + from.Sender + if p.Expired(h.o.Clock()) { + h.lapse(ctx, p) + return sentProposalGone, true + } + if p.Route.Chat != from.Chat || p.Route.Thread != from.Thread { + h.audit(ctx, proposals.ActionRefuse, p, "wrong place; "+detail) + return sentProposalPlace, true + } + if !h.svc.allowCap(ctx, CapabilityRequest{Capability: CapProposalsConfirm, Binding: from.Binding, Owner: p.Owner, Agent: from.Agent, + Sender: from.Sender, Kind: "trigger"}) { + h.audit(ctx, proposals.ActionRefuse, p, "not allowed; "+detail) + return sentProposalNoRight, true + } + if verb == "cancel" { + if err := h.DismissProposal(ctx, p.Owner, p.ID); err != nil { + return fmt.Sprintf(sentProposalFailed, toolErr(err)), true + } + h.audit(ctx, proposals.ActionDismiss, p, detail) + return sentProposalCancel, true + } + t, err := h.ConfirmProposal(ctx, p.Owner, p.ID, "") + if err != nil { + h.audit(ctx, proposals.ActionRefuse, p, "confirm failed; "+detail) + return fmt.Sprintf(sentProposalFailed, toolErr(err)), true + } + h.audit(ctx, proposals.ActionConfirm, t, detail) + return fmt.Sprintf("Confirmed. %q is on: %s.", t.Name, h.summary(t)), true +} + +// proposalText is what the host writes in the chat where a proposal was made. +// It shows what the run will do, so a person can answer it from there, and +// how to answer. +func (h *Triggers) proposalText(t triggers.Trigger) string { + prompt := strings.Join(strings.Fields(t.Prompt), " ") + if utf8.RuneCountInString(prompt) > 280 { + prompt = string([]rune(prompt)[:280]) + "…" + } + where := "the Inbox in Studio" + if t.Route.TargetOrDefault() == triggers.TargetOrigin { + where = "this chat" + } + var mode string + switch t.Route.ModeOrDefault() { + case triggers.ModeOnChange: + mode = ", only when the result changes" + case triggers.ModeOnProblem: + mode = ", only when a run fails" + } + return fmt.Sprintf("Proposal %s: %q, %s. It will: %s\nResults go to %s%s.\nNothing runs until it is confirmed. Reply \"confirm %s\" or \"cancel %s\" here within 24 hours, or open Studio.", + t.Route.Code, t.Name, strings.ToLower(h.summary(t)[:1])+h.summary(t)[1:], prompt, where, mode, t.Route.Code, t.Route.Code) +} + +// Announce writes the proposal, with its code, in the chat where it was made. +// It reports whether the text was sent or queued. +func (h *Triggers) Announce(ctx context.Context, t triggers.Trigger) bool { + c := h.svc.connectors + if c == nil || !t.Route.FromConnector() || t.Route.Chat == "" { + return false + } + _, err := c.Post(ctx, Post{Binding: t.Route.Binding, Chat: t.Route.Chat, Thread: t.Route.Thread, Text: h.proposalText(t)}) + if err != nil { + h.log.Warn("studio: could not write a proposal to its chat", "trigger", t.ID, "error", err) + return false + } + return true +} + +// ---- Pause ---- + +// capOff says whether the runs of a trigger that a connector chat proposed +// are paused because its capability is off now. A trigger made in Studio never +// is. It writes the audit line when the state changes. +func (h *Triggers) capOff(ctx context.Context, t triggers.Trigger) bool { + if !t.Route.FromConnector() { + return false + } + off := !h.svc.allowCap(ctx, CapabilityRequest{Capability: CapTriggersRun, Binding: t.Route.Binding, Owner: t.Owner, Agent: t.AgentID}) + h.mu.Lock() + was := h.paused[t.ID] + if h.paused == nil { + h.paused = map[string]bool{} + } + h.paused[t.ID] = off + h.mu.Unlock() + switch { + case off && !was: + h.audit(ctx, proposals.ActionPause, t, reasonCapOff) + case !off && was: + h.audit(ctx, proposals.ActionResume, t, "") + } + return off +} + +// ---- Delivery ---- + +// neutralMentions breaks the forms that notify a whole channel, so a run's +// text cannot ping everyone: a zero-width space after the opening sign. +var neutralMentions = regexp.MustCompile(`(?i)(@| MaxReportRunes { + s = string([]rune(s)[:MaxReportRunes]) + "…" + } + return neutralMentions.ReplaceAllString(s, "${1}​${2}") +} + +// report delivers the end of a successful run as its trigger asks, and says +// whether the Inbox notice is left out: it is when the run is silent, when its +// mode does not call for it, or when it was written to the chat. When the +// delivery fails it returns false so that the Inbox gets the notice. +func (h *Triggers) report(ctx context.Context, t triggers.Trigger, run triggers.Run, text string) bool { + text = strings.TrimSpace(text) + if text == SilentMark { + h.audit(ctx, proposals.ActionDeliver, t, "silent") + return true + } + switch t.Route.ModeOrDefault() { + case triggers.ModeOnProblem: + return true + case triggers.ModeOnChange: + if !h.changed(ctx, t, run, text) { + return true + } + } + if t.Route.TargetOrDefault() != triggers.TargetOrigin { + return false + } + body := t.Name + "\n\n" + reportText(text) + return h.deliverOrigin(ctx, t, body, func() { + h.tell(context.WithoutCancel(ctx), t, run.ChatID, notify.KindRunEnded, notify.StatusOK, t.Name+" finished", text) + }) +} + +// deliverOrigin writes body in the chat where the trigger was proposed. It +// checks, now and not when the trigger was made, that the binding is on and +// the capability is held. fallback is what runs when a queued message cannot +// be sent later. It reports whether the message was sent or queued. +func (h *Triggers) deliverOrigin(ctx context.Context, t triggers.Trigger, body string, fallback func()) bool { + c := h.svc.connectors + switch { + case c == nil: + h.audit(ctx, proposals.ActionDeliver, t, "refused: connectors are off") + return false + case !h.svc.allowCap(ctx, CapabilityRequest{Capability: CapTriggersRun, Binding: t.Route.Binding, Owner: t.Owner, Agent: t.AgentID}): + h.audit(ctx, proposals.ActionDeliver, t, "refused: capability off") + return false + } + queued, err := c.Post(ctx, Post{Binding: t.Route.Binding, Chat: t.Route.Chat, Thread: t.Route.Thread, Text: body, + OnFail: func(error) { + h.audit(context.Background(), proposals.ActionDeliver, t, "failed; sent to the Inbox") + fallback() + }}) + if err != nil { + h.audit(ctx, proposals.ActionDeliver, t, "failed; sent to the Inbox: "+err.Error()) + return false + } + if queued { + h.audit(ctx, proposals.ActionDeliver, t, "origin, queued") + } else { + h.audit(ctx, proposals.ActionDeliver, t, "origin, sent") + } + return true +} + +// changed says whether the answer of a run differs from the one of the run +// before it. A run with no earlier answer is a change. +func (h *Triggers) changed(ctx context.Context, t triggers.Trigger, run triggers.Run, text string) bool { + runs, err := h.o.Store.ListRuns(ctx, t.ID, 5) + if err != nil { + return true + } + for _, r := range runs { + if r.ID == run.ID || r.Status != triggers.StatusDone || r.ChatID == "" { + continue + } + prev := strings.TrimSpace(h.svc.finalText(ctx, r.ChatID)) + return prev == "" || prev != text + } + return true +} + +// reportFailure raises the alert for a trigger whose last runs all failed: one +// message to the chat of a trigger that reports there, at most once in +// failureAlertEvery for each trigger. The Inbox gets every failure already. +func (h *Triggers) reportFailure(ctx context.Context, t triggers.Trigger, reason string) { + if t.Route.TargetOrDefault() != triggers.TargetOrigin { + return + } + runs, err := h.o.Store.ListRuns(ctx, t.ID, failureAlertAfter) + if err != nil || len(runs) < failureAlertAfter { + return + } + for _, r := range runs { + if r.Status != triggers.StatusFailed { + return + } + } + now := h.o.Clock() + h.mu.Lock() + if h.alerted == nil { + h.alerted = map[string]time.Time{} + } + if last, ok := h.alerted[t.ID]; ok && now.Sub(last) < failureAlertEvery { + h.mu.Unlock() + return + } + h.alerted[t.ID] = now + h.mu.Unlock() + body := fmt.Sprintf("%s failed %d times in a row: %s\nFix or pause it in Studio.", t.Name, failureAlertAfter, reportText(reason)) + h.deliverOrigin(ctx, t, body, func() {}) +} diff --git a/core/services/studio/triggers/gorm.go b/core/services/studio/triggers/gorm.go index 2f0e6d2200..5e63c018c3 100644 --- a/core/services/studio/triggers/gorm.go +++ b/core/services/studio/triggers/gorm.go @@ -41,6 +41,8 @@ type triggerRow struct { // Change is the JSON of Trigger.Change, empty on a record that is not a // change. Change string `gorm:"size:1024"` + // Route is the JSON of Trigger.Route, empty on a trigger made in Studio. + Route string `gorm:"size:1024"` } func (triggerRow) TableName() string { return "studio_triggers" } @@ -96,7 +98,7 @@ func (r triggerRow) trigger() Trigger { Schedule: r.Schedule, TZ: r.TZ, Folder: r.Folder, Quiet: time.Duration(r.QuietNs), DailyCap: r.DailyCap, Since: utc(r.Since), AgentID: r.AgentID, SourceChatID: r.SourceChatID, Model: r.Model, Proposed: r.Proposed, MinGap: time.Duration(r.MinGapNs), Reads: r.Reads, Writes: r.Writes, Widget: r.Widget, EstTokens: r.EstTokens, - Outcome: r.Outcome, OutcomeAt: utc(r.OutcomeAt), Touched: r.Touched, Change: changeOf(r.Change)} + Outcome: r.Outcome, OutcomeAt: utc(r.OutcomeAt), Touched: r.Touched, Change: changeOf(r.Change), Route: routeOf(r.Route)} } func changeOf(raw string) Change { @@ -115,12 +117,28 @@ func changeText(c Change) string { return string(raw) } +func routeOf(raw string) Route { + var r Route + if raw != "" { + _ = json.Unmarshal([]byte(raw), &r) // a broken value reads as a trigger made in Studio + } + return r +} + +func routeText(r Route) string { + if r == (Route{}) { + return "" + } + raw, _ := json.Marshal(r) + return string(raw) +} + func (s *gormStore) PutTrigger(ctx context.Context, t Trigger) error { row := triggerRow{ID: t.ID, Owner: t.Owner, Name: t.Name, Prompt: t.Prompt, Enabled: t.Enabled, Kind: string(t.Kind), Schedule: t.Schedule, TZ: t.TZ, Folder: t.Folder, QuietNs: int64(t.Quiet), DailyCap: t.DailyCap, Since: utc(t.Since), AgentID: t.AgentID, SourceChatID: t.SourceChatID, Model: t.Model, Proposed: t.Proposed, MinGapNs: int64(t.MinGap), Reads: t.Reads, Writes: t.Writes, Widget: t.Widget, EstTokens: t.EstTokens, - Outcome: t.Outcome, OutcomeAt: utc(t.OutcomeAt), Touched: t.Touched, Change: changeText(t.Change)} + Outcome: t.Outcome, OutcomeAt: utc(t.OutcomeAt), Touched: t.Touched, Change: changeText(t.Change), Route: routeText(t.Route)} return s.db.WithContext(ctx).Save(&row).Error } diff --git a/core/services/studio/triggers/json.go b/core/services/studio/triggers/json.go index 8480744a2d..30f7eb669d 100644 --- a/core/services/studio/triggers/json.go +++ b/core/services/studio/triggers/json.go @@ -75,7 +75,11 @@ func utc(t time.Time) time.Time { return t.UTC() } -func normTrigger(t Trigger) Trigger { t.Since, t.OutcomeAt = utc(t.Since), utc(t.OutcomeAt); return t } +func normTrigger(t Trigger) Trigger { + t.Since, t.OutcomeAt = utc(t.Since), utc(t.OutcomeAt) + t.Route.ProposedAt, t.Route.Expires = utc(t.Route.ProposedAt), utc(t.Route.Expires) + return t +} func normRun(r Run) Run { r.Due, r.Started, r.Ended = utc(r.Due), utc(r.Started), utc(r.Ended) diff --git a/core/services/studio/triggers/triggerstest/suite.go b/core/services/studio/triggers/triggerstest/suite.go index 616e6f00f5..acb43c7114 100644 --- a/core/services/studio/triggers/triggerstest/suite.go +++ b/core/services/studio/triggers/triggerstest/suite.go @@ -54,7 +54,12 @@ func same(a, b triggers.Trigger) bool { a.Enabled == b.Enabled && a.Kind == b.Kind && a.Schedule == b.Schedule && a.TZ == b.TZ && a.Folder == b.Folder && a.Quiet == b.Quiet && a.DailyCap == b.DailyCap && a.Since.Equal(b.Since) && a.AgentID == b.AgentID && a.SourceChatID == b.SourceChatID && a.Model == b.Model && a.Proposed == b.Proposed && - a.Outcome == b.Outcome && a.OutcomeAt.Equal(b.OutcomeAt) && a.Touched == b.Touched && a.Change == b.Change + a.Outcome == b.Outcome && a.OutcomeAt.Equal(b.OutcomeAt) && a.Touched == b.Touched && a.Change == b.Change && sameRoute(a.Route, b.Route) +} + +func sameRoute(a, b triggers.Route) bool { + return a.Binding == b.Binding && a.Chat == b.Chat && a.Thread == b.Thread && a.Code == b.Code && + a.ProposedAt.Equal(b.ProposedAt) && a.Expires.Equal(b.Expires) && a.Target == b.Target && a.Mode == b.Mode } func sameRun(a, b triggers.Run) bool { @@ -83,6 +88,22 @@ func RunStoreSuite(newStore NewStore, opts ...Option) { Expect(same(got, tr)).To(BeTrue(), "GetTrigger(%s) = %+v; want %+v", tr.ID, got, tr) } }) + It("stores the connector place and the delivery of a trigger", func() { + s := newStore() + tr := sample("r1", "1", "Morning") + tr.Enabled, tr.Proposed = false, true + tr.Route = triggers.Route{Binding: "b1", Chat: "C1", Thread: "T1", Code: "7K2Q", ProposedAt: base, Expires: base.Add(24 * time.Hour), + Target: triggers.TargetOrigin, Mode: triggers.ModeOnChange} + Expect(s.PutTrigger(ctx, tr)).To(Succeed()) + got, err := s.GetTrigger(ctx, tr.ID) + Expect(err).NotTo(HaveOccurred()) + Expect(same(got, tr)).To(BeTrue(), "GetTrigger = %+v; want %+v", got.Route, tr.Route) + plain := sample("r2", "1", "Plain") + Expect(s.PutTrigger(ctx, plain)).To(Succeed()) + got, err = s.GetTrigger(ctx, plain.ID) + Expect(err).NotTo(HaveOccurred()) + Expect(got.Route).To(Equal(triggers.Route{}), "a trigger made in Studio has no place") + }) It("stores a change record with its change", func() { s := newStore() tr := sample("c1", "1", "Morning") diff --git a/core/services/studio/triggers/types.go b/core/services/studio/triggers/types.go index fc985b669a..3b435ff4ae 100644 --- a/core/services/studio/triggers/types.go +++ b/core/services/studio/triggers/types.go @@ -135,6 +135,98 @@ type Trigger struct { // that a model asked for in a chat (see Change). Such a record is also // Proposed and never enabled, so it never runs. The package only stores it. Change Change + // Route says where a trigger that a connector chat proposed came from and + // where its runs report. The zero value is a trigger made in Studio, whose + // runs go to the Inbox. The package only stores it. + Route Route +} + +// Where the runs of a trigger report. Inbox is the default: the owner's Inbox +// and the chat the owner chose for notices. Origin is the chat and thread +// where the trigger was proposed. +const ( + TargetInbox = "inbox" + TargetOrigin = "origin" +) + +// When a run is reported. Always reports every run, OnChange a run whose +// answer differs from the one before, and OnProblem only a run that failed. +const ( + ModeAlways = "always" + ModeOnChange = "on_change" + ModeOnProblem = "on_problem" +) + +// MaxRefBytes bounds the binding, chat and thread of a Route: the width of +// the sender column of the connector stores. +const MaxRefBytes = 64 + +// Route is the connector side of a trigger: the place it was proposed from +// and the delivery of its runs. Binding, Chat and Thread name the place in +// the host's own words, with no platform in them; they are set only for a +// trigger that a connector chat proposed. Code is what a person types in that +// place to confirm or cancel the proposal while it waits. Expires is when a +// waiting proposal lapses; zero means it never does. +type Route struct { + Binding string `json:",omitempty"` + Chat string `json:",omitempty"` + Thread string `json:",omitempty"` + Code string `json:",omitempty"` + // ProposedAt is when the proposal was made. + ProposedAt time.Time `json:",omitzero"` + Expires time.Time `json:",omitzero"` + // Target and Mode are the delivery; empty means TargetInbox and + // ModeAlways. + Target string `json:",omitempty"` + Mode string `json:",omitempty"` +} + +// FromConnector says whether a connector chat proposed the trigger. +func (r Route) FromConnector() bool { return r.Binding != "" } + +// TargetOrDefault gives Target, or TargetInbox when it is empty. +func (r Route) TargetOrDefault() string { + if r.Target == "" { + return TargetInbox + } + return r.Target +} + +// ModeOrDefault gives Mode, or ModeAlways when it is empty. +func (r Route) ModeOrDefault() string { + if r.Mode == "" { + return ModeAlways + } + return r.Mode +} + +// Validate checks the delivery and the place of a Route. +func (r Route) Validate() error { + switch r.Target { + case "", TargetInbox: + case TargetOrigin: + if r.Binding == "" || r.Chat == "" { + return errors.New("a run can report to the chat of the proposal only when a connector chat proposed it") + } + default: + return fmt.Errorf("delivery target must be %q or %q", TargetInbox, TargetOrigin) + } + switch r.Mode { + case "", ModeAlways, ModeOnChange, ModeOnProblem: + default: + return fmt.Errorf("delivery mode must be %q, %q or %q", ModeAlways, ModeOnChange, ModeOnProblem) + } + for _, v := range []string{r.Binding, r.Chat, r.Thread, r.Code} { + if len(v) > MaxRefBytes || hasControl(v, false) { + return errors.New("the connector place of a trigger is too long or has control characters") + } + } + return nil +} + +// Expired says whether t is a proposal that waited past its time at now. +func (t Trigger) Expired(now time.Time) bool { + return t.Pending() && !t.Route.Expires.IsZero() && !now.Before(t.Route.Expires) } // Change says what a change record does to the trigger it names. While it @@ -175,6 +267,8 @@ const ( OutcomeAccepted = "accepted" OutcomeDeclined = "declined" OutcomeUndone = "undone" + // OutcomeExpired is a proposal that nobody answered in time. + OutcomeExpired = "expired" ) // Pending says whether t is a proposal that waits for an answer. @@ -262,6 +356,9 @@ func (t Trigger) Validate() error { case t.EstTokens < 0 || t.EstTokens > MaxEstTokens: return fmt.Errorf("estimated tokens must be 0 to %d", MaxEstTokens) } + if err := t.Route.Validate(); err != nil { + return err + } switch t.Kind { case KindSchedule: if _, err := ParseSchedule(t.Schedule); err != nil { diff --git a/core/services/studio/triggers/types_test.go b/core/services/studio/triggers/types_test.go index 5c18b2cd06..7c98a2c86f 100644 --- a/core/services/studio/triggers/types_test.go +++ b/core/services/studio/triggers/types_test.go @@ -196,3 +196,28 @@ var _ = Describe("Trigger types", func() { } }) }) + +var _ = Describe("Route", func() { + It("reads the empty value as the Inbox, always", func() { + var r Route + Expect(r.TargetOrDefault()).To(Equal(TargetInbox)) + Expect(r.ModeOrDefault()).To(Equal(ModeAlways)) + Expect(r.FromConnector()).To(BeFalse()) + Expect(r.Validate()).To(Succeed()) + }) + It("lets a run report to the origin only when a connector chat proposed it", func() { + Expect(Route{Target: TargetOrigin}.Validate()).NotTo(Succeed()) + Expect(Route{Binding: "b", Chat: "C", Target: TargetOrigin, Mode: ModeOnProblem}.Validate()).To(Succeed()) + Expect(Route{Target: "elsewhere"}.Validate()).NotTo(Succeed()) + Expect(Route{Mode: "sometimes"}.Validate()).NotTo(Succeed()) + }) + It("expires a waiting proposal that has a time, and no other", func() { + now := time.Date(2026, 11, 3, 12, 0, 0, 0, time.UTC) + p := Trigger{Proposed: true, Route: Route{Expires: now.Add(time.Hour)}} + Expect(p.Expired(now)).To(BeFalse()) + Expect(p.Expired(now.Add(time.Hour))).To(BeTrue()) + p.Outcome = OutcomeDeclined + Expect(p.Expired(now.Add(2*time.Hour))).To(BeFalse(), "an answered proposal does not expire") + Expect(Trigger{Proposed: true}.Expired(now.Add(1000*time.Hour))).To(BeFalse(), "a proposal without a time never does") + }) +}) diff --git a/core/services/studio/triggerscrud.go b/core/services/studio/triggerscrud.go index 6493d0a3bf..5631ca4d8f 100644 --- a/core/services/studio/triggerscrud.go +++ b/core/services/studio/triggerscrud.go @@ -10,6 +10,7 @@ import ( "github.com/mudler/LocalAI/core/services/studio/agents" "github.com/mudler/LocalAI/core/services/studio/files" + "github.com/mudler/LocalAI/core/services/studio/proposals" "github.com/mudler/LocalAI/core/services/studio/triggers" "github.com/mudler/LocalAI/core/services/studio/wellknown" "github.com/mudler/LocalAI/core/services/studio/workspace" @@ -54,6 +55,9 @@ type TriggerInput struct { // file, and about how many tokens one run uses. All optional. Reads, Writes, Widget string EstTokens int + // Route is the connector place a proposal came from and where its runs + // report. A person's form leaves it empty and an edit keeps the old one. + Route triggers.Route } // keepPlan fills the fields of the plan and the gap that in leaves empty from @@ -117,6 +121,10 @@ func (h *Triggers) pendingProposal(ctx context.Context, user, id string) (trigge if err != nil { return t, err } + if t.Expired(h.o.Clock()) { + h.lapse(ctx, t) + return triggers.Trigger{}, ErrProposalExpired + } if !t.Pending() { return triggers.Trigger{}, ErrProposalAnswered } @@ -140,7 +148,7 @@ func countLive(ts []triggers.Trigger) int { func (h *Triggers) build(ctx context.Context, owner string, in TriggerInput) (triggers.Trigger, error) { t := triggers.Trigger{Owner: owner, Name: strings.TrimSpace(in.Name), Prompt: in.Prompt, Enabled: in.Enabled, Kind: in.Kind, TZ: in.TZ, Quiet: in.Quiet, DailyCap: in.DailyCap, AgentID: in.AgentID, Model: in.Model, SourceChatID: in.SourceChatID, - MinGap: in.MinGap, Reads: strings.TrimSpace(in.Reads), Widget: in.Widget, EstTokens: in.EstTokens} + MinGap: in.MinGap, Reads: strings.TrimSpace(in.Reads), Widget: in.Widget, EstTokens: in.EstTokens, Route: in.Route} if w := strings.TrimSpace(in.Writes); w != "" { // A path of the home as the person writes it, stored as they see it: // the widget footer and the proposal card show it back. @@ -269,7 +277,7 @@ func (h *Triggers) UpdateTrigger(ctx context.Context, user, id string, in Trigge if err != nil { return triggers.Trigger{}, err } - t.ID = old.ID + t.ID, t.Route = old.ID, old.Route // The answer to the proposal stays, and the edit is the reason the card // can no longer undo it. t.Outcome, t.OutcomeAt, t.Touched = old.Outcome, old.OutcomeAt, old.Touched || old.Outcome != "" @@ -323,7 +331,7 @@ func (h *Triggers) HasProposal(ctx context.Context, user, chatID string) bool { return true } return slices.ContainsFunc(all, func(t triggers.Trigger) bool { - return (t.Pending() && t.SourceChatID == chatID) || (t.IsChange() && t.Outcome == "" && t.Change.Chat == chatID) + return (t.Pending() && !t.Expired(h.o.Clock()) && t.SourceChatID == chatID) || (t.IsChange() && t.Outcome == "" && t.Change.Chat == chatID) }) } @@ -340,14 +348,23 @@ func (h *Triggers) Proposals(ctx context.Context, user string) ([]triggers.Trigg if err != nil { return nil, err } - return slices.DeleteFunc(all, func(t triggers.Trigger) bool { return !t.Pending() }), nil + now := h.o.Clock() + return slices.DeleteFunc(all, func(t triggers.Trigger) bool { return !t.Pending() || t.Expired(now) }), nil } // ProposeTrigger stores a trigger that a model proposed. It is never enabled. -// Nothing runs until the owner confirms it. +// Nothing runs until the owner confirms it. A proposal that a connector chat +// made (in.Route names the place) also gets a code to confirm it by text and a +// time after which it lapses, and counts against the hourly limit of its +// binding. func (h *Triggers) ProposeTrigger(ctx context.Context, user string, in TriggerInput) (triggers.Trigger, error) { owner := principal(user) in.Enabled = false + now := h.o.Clock() + if in.Route.FromConnector() { + in.Route.ProposedAt, in.Route.Expires = now.UTC(), now.UTC().Add(ProposalTTL) + in.Route.Code = "" + } t, err := h.build(ctx, owner, in) if err != nil { return triggers.Trigger{}, err @@ -363,7 +380,7 @@ func (h *Triggers) ProposeTrigger(ctx context.Context, user string, in TriggerIn } pending := 0 for _, m := range mine { - if !m.Pending() { + if !m.Pending() || m.Expired(now) { continue } // A model that proposes the same thing again, in this chat or the next, @@ -380,10 +397,20 @@ func (h *Triggers) ProposeTrigger(ctx context.Context, user string, in TriggerIn case countLive(mine) >= maxTriggersPerUser: return triggers.Trigger{}, ErrTooManyTriggers } + if t.Route.FromConnector() { + if proposalsInHour(mine, t.Route.Binding, now) >= MaxProposalsPerHour { + h.audit(ctx, proposals.ActionRefuse, t, "too many proposals from this binding in the last hour") + return triggers.Trigger{}, ErrProposalRate + } + t.Route.Code = freshCode(mine, t.Route.Binding) + } t.ID, t.Proposed = newID(), true if err := h.o.Store.PutTrigger(ctx, t); err != nil { return triggers.Trigger{}, err } + if t.Route.FromConnector() { + h.audit(ctx, proposals.ActionPropose, t, "") + } return t, nil } @@ -497,7 +524,7 @@ func (h *Triggers) EditProposal(ctx context.Context, user, id string, in Trigger if err := h.checkCadence(t); err != nil { return triggers.Trigger{}, err } - t.ID, t.Proposed = old.ID, true + t.ID, t.Proposed, t.Route = old.ID, true, old.Route if t.SourceChatID == "" { t.SourceChatID = old.SourceChatID } diff --git a/core/services/studio/triggersvc.go b/core/services/studio/triggersvc.go index 3648bf164d..45fe44e24a 100644 --- a/core/services/studio/triggersvc.go +++ b/core/services/studio/triggersvc.go @@ -86,6 +86,13 @@ type Triggers struct { wmu sync.Mutex // serialises writes of triggers + // Guarded by mu: when the proposals past their time were last swept, the + // triggers whose capability to run is off now, and when a trigger last + // raised its failure alert. + swept time.Time + paused map[string]bool + alerted map[string]time.Time + wake chan struct{} done chan struct{} once sync.Once @@ -224,6 +231,7 @@ func (h *Triggers) Pass(ctx context.Context) error { } err = r.Sync(ctx) h.reconcile(ctx) + h.expireProposals(ctx) return err } @@ -269,7 +277,7 @@ func (h *Triggers) start(ctx context.Context, t triggers.Trigger, run triggers.R if err != nil && ctx.Err() != nil && !errors.As(err, &skip) { return "", ctx.Err() // the host stops: the Runner puts the run back } - if err != nil { + if err != nil && !(errors.As(err, &skip) && skip.Reason == reasonCapOff) { h.tell(context.WithoutCancel(ctx), t, "", notify.KindRunFailed, notify.StatusFailed, t.Name+" did not run", err.Error()) } return id, err @@ -280,6 +288,12 @@ func (h *Triggers) startRun(ctx context.Context, t triggers.Trigger, run trigger if err := h.checkOwner(ctx, t); err != nil { return "", err } + // A trigger that a connector chat proposed runs only while its capability + // is on. The record stays and so does Enabled; a run that falls due while + // the capability is off is skipped and is not made up for later. + if h.capOff(ctx, t) { + return "", &triggers.SkipError{Reason: reasonCapOff} + } // The folder of a scoped agent is checked before anything is read for // the run: a run never falls back to the whole home. var scope *folderScope @@ -639,12 +653,16 @@ func (h *Triggers) finishRun(run triggers.Run, t triggers.Trigger, chatID string } if status == triggers.StatusFailed { h.tell(ctx, t, chatID, notify.KindRunFailed, notify.StatusFailed, t.Name+" failed", reason) + h.reportFailure(ctx, t, reason) return } body := reason if body == "" { body = h.svc.finalText(ctx, chatID) } + if h.report(ctx, t, cur, body) { + return + } h.tell(ctx, t, chatID, notify.KindRunEnded, notify.StatusOK, t.Name+" finished", body) } diff --git a/core/services/studio/triggertools.go b/core/services/studio/triggertools.go index d02c548d11..2355672846 100644 --- a/core/services/studio/triggertools.go +++ b/core/services/studio/triggertools.go @@ -54,6 +54,32 @@ const proposeTriggerSchema = `{"type":"object","properties":{` + `"estimated_tokens":{"type":"integer","minimum":0,"maximum":10000000,"description":"About how many tokens one run uses, prompt and answer."}},` + `"required":["name","prompt","when"]}` +// proposeTriggerConnectorSchema is the schema of propose_trigger in a chat of +// a messaging app: the same, with the delivery of the runs. +const proposeTriggerConnectorSchema = `{"type":"object","properties":{` + + `"name":{"type":"string","minLength":1,"maxLength":80,"description":"A short name for the trigger."},` + + `"prompt":{"type":"string","minLength":1,"description":"What the agent must do at each run."},` + + `"when":{"description":"A schedule text, or an object with the folder to watch.","oneOf":[` + + `{"type":"string"},` + + `{"type":"object","properties":{"file_arrives_in":{"type":"string","description":"A folder inside the user's files."}},"required":["file_arrives_in"]}]},` + + `"time_zone":{"type":"string","description":"An IANA time zone name such as Europe/Rome; when absent, UTC is used."},` + + `"daily_cap":{"type":"integer","minimum":0,"maximum":1000,"description":"The most runs in one day; 0 is the default limit."},` + + `"reads":{"type":"string","maxLength":200,"description":"What a run reads, in a few words."},` + + `"writes":{"type":"string","maxLength":256,"description":"The file in the user's home that a run rewrites."},` + + `"widget":{"type":"string","pattern":"^[a-z0-9][a-z0-9-]{0,39}$","description":"The id of the Home widget that shows the file."},` + + `"estimated_tokens":{"type":"integer","minimum":0,"maximum":10000000,"description":"About how many tokens one run uses."},` + + `"delivery":{"type":"object","description":"Where the answer of each run goes. Default: the owner's Inbox.","properties":{` + + `"target":{"type":"string","enum":["inbox","origin"],"description":"origin writes the answer in this chat and thread, inbox in the owner's Inbox."},` + + `"mode":{"type":"string","enum":["always","on_change","on_problem"],"description":"always: every run. on_change: only when the answer differs from the run before. on_problem: only when a run fails."}}}},` + + `"required":["name","prompt","when"]}` + +// connectorProposeNote is added to the description of propose_trigger in a +// chat of a messaging app. +const connectorProposeNote = " This chat is in a messaging app. The host writes the proposal and its confirmation code in the chat itself, " + + "and the person answers there with confirm or cancel and the code, so do not repeat the code. " + + "Use delivery.target origin when the person wants the result in this chat, and a mode that fits (on_change for a watch, on_problem for a check). " + + "A proposal lapses after 24 hours, and at most 3 can be made in an hour." + const listTriggersDescription = "List the triggers of the user: a ref, the name, when it runs or which folder it watches, whether it is on or off, and how its last run ended. " + "It lists at most 50 and shows no prompts. Use the ref to name a trigger in update_trigger and delete_trigger. " + "To read the prompt and the other settings of ONE trigger before you change it, pass its ref and include_prompt: true." @@ -106,13 +132,13 @@ func (s *Service) registerTriggerTools() error { {proposeTriggerName, "Propose a trigger", proposeTriggerDescription, proposeTriggerSchema, false, func(h *Triggers, a workspace.Actor, c builtins.Chat) func(context.Context, json.RawMessage) (string, error) { return func(ctx context.Context, raw json.RawMessage) (string, error) { - return h.proposeTool(ctx, a, c.AgentID, c.Model, raw) + return h.proposeTool(ctx, a, c, raw) } }}, {listTriggersName, "List triggers", listTriggersDescription, listTriggersSchema, true, func(h *Triggers, a workspace.Actor, c builtins.Chat) func(context.Context, json.RawMessage) (string, error) { return func(ctx context.Context, raw json.RawMessage) (string, error) { - return h.listTool(ctx, a, c.AgentID, raw) + return h.listTool(ctx, a, c.AgentID, c.Origin == session.OriginConnector, raw) } }}, {updateTriggerName, "Change a trigger", updateTriggerDescription, updateTriggerSchema, false, @@ -136,18 +162,35 @@ func (s *Service) registerTriggerTools() error { Enabled: func() bool { return s.triggers != nil }, Build: func(ctx context.Context, c builtins.Chat) (session.Tool, bool, string) { h := s.triggers + connector := c.Origin == session.OriginConnector switch { case h == nil: return session.Tool{}, false, "triggers are off on this server" case c.SubAgent: return session.Tool{}, false, "a sub-agent cannot set triggers" - case c.Origin != "": + case c.Origin != "" && !connector: return session.Tool{}, false, "a run that started by itself cannot set triggers" } if err := s.mayUseTriggers(ctx, c.Actor.Principal); err != nil { return session.Tool{}, false, err.Error() } - return session.Tool{Description: d.desc, Schema: json.RawMessage(d.schema), Run: d.run(h, c.Actor, c)}, true, "" + desc, schema := d.desc, d.schema + if connector { + // A chat of a messaging app proposes and lists. A change or a + // delete needs a card that only Studio shows, so it stays there. + switch { + case d.id == updateTriggerName || d.id == deleteTriggerName: + return session.Tool{}, false, "change or delete a trigger in Studio" + case c.AgentID == "": + return session.Tool{}, false, "a trigger from a messaging app needs the agent of the connection" + case !s.allowCap(ctx, CapabilityRequest{Capability: CapTriggersPropose, Binding: c.OriginID, Owner: c.Actor.Principal, Agent: c.AgentID}): + return session.Tool{}, false, "triggers are off for this connection" + } + if d.id == proposeTriggerName { + desc, schema = desc+connectorProposeNote, proposeTriggerConnectorSchema + } + } + return session.Tool{Description: desc, Schema: json.RawMessage(schema), Run: d.run(h, c.Actor, c)}, true, "" }, }) if err != nil { @@ -242,7 +285,8 @@ func (h *Triggers) toolTriggers(ctx context.Context, a workspace.Actor, agentID return slices.DeleteFunc(all, func(t triggers.Trigger) bool { return t.AgentID != agentID }), nil } -func (h *Triggers) proposeTool(ctx context.Context, a workspace.Actor, agentID, model string, raw json.RawMessage) (string, error) { +func (h *Triggers) proposeTool(ctx context.Context, a workspace.Actor, c builtins.Chat, raw json.RawMessage) (string, error) { + agentID, model := c.AgentID, c.Model var args struct { Name string `json:"name"` Prompt string `json:"prompt"` @@ -253,6 +297,10 @@ func (h *Triggers) proposeTool(ctx context.Context, a workspace.Actor, agentID, Writes string `json:"writes"` Widget string `json:"widget"` Tokens int `json:"estimated_tokens"` + Delivery *struct { + Target string `json:"target"` + Mode string `json:"mode"` + } `json:"delivery"` } if err := json.Unmarshal(raw, &args); err != nil { return "", fmt.Errorf("bad arguments: %w", err) @@ -272,6 +320,21 @@ func (h *Triggers) proposeTool(ctx context.Context, a workspace.Actor, agentID, default: return "", errors.New("when must be a schedule text or {\"file_arrives_in\": folder}") } + var err error + connector := c.Origin == session.OriginConnector + if connector { + if in.Route, err = h.connectorRoute(ctx, c); err != nil { + return "", err + } + if d := args.Delivery; d != nil { + in.Route.Target, in.Route.Mode = d.Target, d.Mode + } + if in.Route.Target == "" { + in.Route.Target = triggers.TargetInbox + } + } else if args.Delivery != nil { + return "", errors.New("delivery is for a chat in a messaging app; the answer of a run goes to the Inbox") + } sc, err := h.toolScope(ctx, a, agentID) if err != nil { return "", toolErr(err) @@ -285,6 +348,9 @@ func (h *Triggers) proposeTool(ctx context.Context, a workspace.Actor, agentID, if err != nil { return "", errors.New(strings.TrimPrefix(err.Error(), ErrBadTrigger.Error()+": ")) } + if connector { + return h.announceProposal(ctx, a, t, in.Route) + } // The card lets the interface show Create, Edit and Dismiss on this tool // line, also after a reload; the event tells a client that is watching. session.SetToolCard(ctx, session.ToolCard{Kind: ProposalCardKind, ID: t.ID}) @@ -292,7 +358,7 @@ func (h *Triggers) proposeTool(ctx context.Context, a workspace.Actor, agentID, return fmt.Sprintf("Proposed %q (%s). It is not active: the user must confirm it in the chat or on the Triggers page. Do not say it will run until they do.", t.Name, h.summary(t)), nil } -func (h *Triggers) listTool(ctx context.Context, a workspace.Actor, agentID string, raw json.RawMessage) (string, error) { +func (h *Triggers) listTool(ctx context.Context, a workspace.Actor, agentID string, onlyAgent bool, raw json.RawMessage) (string, error) { var args struct { Ref string `json:"ref"` IncludePrompt bool `json:"include_prompt"` @@ -332,6 +398,9 @@ func (h *Triggers) listTool(ctx context.Context, a workspace.Actor, agentID stri if err != nil { return "", errors.New(strings.TrimPrefix(err.Error(), "studio: ")) } + if onlyAgent && t.AgentID != agentID { + return "", errors.New("no such trigger") + } all = []triggers.Trigger{t} case args.IncludePrompt: return "", errors.New("include_prompt needs the ref of one trigger") @@ -340,6 +409,11 @@ func (h *Triggers) listTool(ctx context.Context, a workspace.Actor, agentID stri return "", err } } + if onlyAgent { + // A chat of a messaging app is open to more people than the owner: it + // lists only what its own agent runs. + all = slices.DeleteFunc(all, func(t triggers.Trigger) bool { return t.AgentID != agentID }) + } sort.Slice(all, func(i, j int) bool { return all[i].Name < all[j].Name }) if len(all) > listTriggersMax { all = all[:listTriggersMax] @@ -483,3 +557,34 @@ func (h *Triggers) summary(t triggers.Trigger) string { } return s } + +// connectorRoute is the place a trigger proposed in the connector chat c comes +// from: the binding, and the chat and thread where the person wrote. A chat +// with no known place cannot propose, since the proposal could not be shown. +func (h *Triggers) connectorRoute(ctx context.Context, c builtins.Chat) (triggers.Route, error) { + conns := h.svc.connectors + if conns == nil { + return triggers.Route{}, errors.New("connections to messaging apps are off on this server") + } + chat, thread, ok := conns.PlaceOf(ctx, c.OriginID, c.Actor.ChatID) + if !ok { + return triggers.Route{}, errors.New("I cannot tell where this conversation is written, so I cannot show a proposal here; ask the owner to propose it in Studio") + } + return triggers.Route{Binding: c.OriginID, Chat: chat, Thread: thread}, nil +} + +// announceProposal writes the proposal in the chat it came from, with its code, +// and words the result for the model. +func (h *Triggers) announceProposal(ctx context.Context, a workspace.Actor, t triggers.Trigger, here triggers.Route) (string, error) { + h.svc.bus.Publish(key(a.Principal, a.ChatID), session.Event{Kind: session.KindTriggerProposed, IDs: []string{t.ID}, Text: t.Name}) + if t.Route.Binding != here.Binding || t.Route.Chat != here.Chat || t.Route.Thread != here.Thread { + // The same proposal was made from another place: its code works there. + return fmt.Sprintf("A proposal like %q waits already, from another conversation. It is not active. Tell the person to confirm it in Studio.", t.Name), nil + } + if h.Announce(ctx, t) { + return fmt.Sprintf("Proposed %q (%s). It is not active. The host wrote the proposal and its code in this chat; the person answers there with confirm or cancel and the code, or in Studio. "+ + "Say in one short sentence that you proposed it; do not repeat the code and do not say it will run until they confirm.", t.Name, h.summary(t)), nil + } + return fmt.Sprintf("Proposed %q (%s). It is not active. The host could not write the code in this chat. Tell the person to reply \"confirm %s\" or \"cancel %s\" here, or to open Studio. "+ + "Do not say it will run until they confirm.", t.Name, h.summary(t), t.Route.Code, t.Route.Code), nil +} diff --git a/docs/content/features/chats-connectors.md b/docs/content/features/chats-connectors.md index fccf8af088..cced82df69 100644 --- a/docs/content/features/chats-connectors.md +++ b/docs/content/features/chats-connectors.md @@ -251,8 +251,10 @@ gives the same events without one. - **The agent's tools only.** A message gets exactly the tools the agent grants, intersected with what you may use now. It never gets the tool set of a chat, and - a message cannot change the agent. A run started by a connector never gets - `propose_trigger`. + a message cannot change the agent. A run that a trigger started never gets + `propose_trigger`. A connector chat gets `propose_trigger` and `list_triggers` + only while the binding holds `triggers.propose` (see + [Triggers from a connector chat](#triggers-from-a-connector-chat)). - **Your account.** The chat acts as you, so the feature gate, the model allowlist and your quota apply. If the model refuses the request, the sender gets one fixed sentence, for example that your quota is used up. @@ -479,6 +481,33 @@ The host stores and checks these values and shows them to the add-on pages. The features that a capability switches on or off (triggers from chat, talk-back, and so on) read them as they land. +## Triggers from a connector chat + +When the binding holds the capability `triggers.propose`, the agent can propose a +trigger from a chat in a messaging app. The agent gets `propose_trigger` and +`list_triggers` (the list shows only the triggers of that agent). Changing or deleting a +trigger stays in Studio. Nothing runs before the owner confirms. + +- **The proposal is shown in the chat.** LocalAI writes the name, the schedule, what the + run will do, where the results go and a four-character code in the chat and thread + where the person wrote. The model does not write it. +- **Confirm or cancel by text.** A sender writes `confirm CODE` or `cancel CODE`, as the + whole message, in the same chat and thread. LocalAI handles it before any model turn. + It needs the capability `proposals.confirm` for that sender (a paired sender, or a role + with the grant). A code is used once, and a proposal lapses after 24 hours. Studio + always works as well. +- **Limits.** A binding can make 3 proposals an hour, and an owner can have 10 waiting. +- **Where a run reports.** The proposal sets `delivery`: `inbox` (the default) or + `origin`, the chat and thread of the proposal, and a mode: `always`, `on_change` or + `on_problem`. LocalAI writes the last message of the run, cut to 4,000 characters, + with mentions that notify everyone broken. A final message of exactly `[SILENT]` + writes nothing. After three failed runs in a row the chat gets one alert, and not + more than one every 6 hours. If the binding is off or the capability is off when a run + ends, the Inbox gets the notice. A binding sends 20 of these messages a minute; more + wait in a queue and go in order. +- **Pause.** While `triggers.run` is off for the binding, its triggers do not run. The + records stay and are not replayed when the capability comes back. + ## Notifications in Telegram Set a notification chat ID on a binding to receive your run notifications there