trigger

package
v0.4.0-rc.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 13, 2026 License: AGPL-3.0 Imports: 54 Imported by: 0

Documentation

Overview

Package trigger provides services that trigger agent containers in response to external events: webhooks, cron schedules, and channel messages.

trigger/manager_bot.go houses the singleton Telegram Managed Bots poller — the bot whose can_manage_bots flag is true and that the platform's operators use to create new bots on behalf of their users. The poller shares the same getUpdates long-poll shape as the per-bridge bridge pollers in bridge.go, so cohabitation in trigger/ keeps the goroutine lifecycle conventions consistent — only the update-type handlers differ.

Flow (Bot API 9.6):

  1. Airlock UI POST /bridges/managed/sessions inserts a session row + returns deep_link https://t.me/newbot/<manager_username>/<suggested_username>?name=<...>.
  2. User opens the link in Telegram. Telegram's native bot-creation UI walks them through.
  3. On completion Telegram delivers a ManagedBotCreated{bot} service message into the manager-bot's chat with the creator. We match bot.username against the session nonce (the suggested_username we set on the deeplink — Telegram preserves it through the flow), fetch the token via getManagedBotToken{user_id: bot.id}, call bridges.Service.CreateFromManagedSession, and delete the session. The complementary ManagedBotUpdated event isn't consumed: rotation/owner-change is rare for managed bots in v1, and the Created path is sufficient for creation.

Index

Constants

View Source 
const (
	// BridgeReferenceReply — the user replied to another message (Discord
	// reply, Telegram reply_to_message, Slack thread parent).
	BridgeReferenceReply = "reply"
	// BridgeReferenceForward — the user forwarded a message authored
	// elsewhere (Discord message snapshot, Telegram forward_origin).
	BridgeReferenceForward = "forward"
)

BridgeReferenceKind distinguishes the platform mechanism that produced the reference so the prompt builder can label it for the LLM.

View Source 
const CancelButtonAfter = 20 * time.Second

CancelButtonAfter is how long a bridge run can stream before the driver posts a "Still working… Tap to stop" message with a cancel button. The message is deleted when the run ends (naturally or via the user tap).

View Source 
const FilesManifestSource = "llm"

FilesManifestSource tags the attached-files manifest message: a user-role message that IS sent to the model (non-ephemeral, so SessionStore.Load returns it) but is hidden from the human UI (the frontend drops source=="llm"). The human already sees the attachments via the separate upload echo / their chat platform, so rendering the manifest too would be noise.

View Source 
const ManagerBotTokenScope = "system/telegram_manager_bot_token"

ManagerBotTokenScope is the encryptor scope under which the admin-configured manager bot token is stored. system_settings holds the ciphertext; the poller decrypts at Start time. Exported so the settings handler can encrypt under the same scope when persisting a new token.

View Source 
const PromptHTTPCeiling = 30 * time.Minute

PromptHTTPCeiling is the absolute cap on a prompt run's outbound HTTP request. Generous on purpose: prompt runs may legitimately stream for many minutes (long tool chains, slow LLMs); the user cancels manually via DELETE /api/v1/runs/{runID} when they want to stop earlier. Cron and webhook callers pass their own (typically shorter) timeout to ForwardCron/Webhook.

Variables

View Source 
var (
	// ErrAgentStopped — the agent is parked via /stop and only a manual
	// /start resumes it; EnsureRunning refuses to auto-start it.
	ErrAgentStopped = errors.New("agent is stopped")
	// ErrAgentNoImage — the agent has never finished a build, so there is
	// no container image to run.
	ErrAgentNoImage = errors.New("agent has no image")
)

Sentinel errors from EnsureRunning for agents that exist but aren't in a runnable state. Callers map these to a surface-appropriate response (409 on HTTP, an in-chat notice on bridges, a JSON-RPC error on A2A) instead of a generic 500. Both are expected operator states, not faults.

View Source 
var ErrTranscriptionNotConfigured = errors.New("transcription not configured")

ErrTranscriptionNotConfigured is returned when no system-wide transcription model is set. Callers handle this non-fatally (degrade to file-only delivery).

View Source 
var Registry = []SlashCommand{
	{Name: "auth", Description: "Link your Airlock account", Access: agentsdk.AccessPublic},
	{Name: "cancel", Description: "Stop the current run", Access: agentsdk.AccessPublic},
	{Name: "clear", Description: "Clear conversation context", Access: agentsdk.AccessUser},
	{Name: "compact", Description: "Summarize and compact context", Access: agentsdk.AccessUser},
	{Name: "echo", Description: "Toggle tool output bubbles (on / off / blank=flip)", Access: agentsdk.AccessUser},
}

Registry is the canonical list of commands. Order is preserved when rendered into platform command menus.

Functions

func PostFilesManifest added in v0.4.0 

func PostFilesManifest(ctx context.Context, q *dbq.Queries, convID pgtype.UUID, files []agentsdk.FileInfo) error

PostFilesManifest writes the attached-files manifest as its own conversation message — the single, canonical way file attachments are described to the model. agentsdk no longer inlines a manifest into the prompt; every files-bearing ingress (web / bridge / A2A) calls this instead, BEFORE dispatch, so the row is in history when the agent's SessionStore loads. No-op when there are no files.

Persist-only (q.CreateMessage) — deliberately NOT postToConversation: the manifest is model-only and must never reach a human WS or bridge channel. Sol's same-role coalescer folds it together with the user's actual message for providers that reject consecutive user turns.

func ResolveEcho 

func ResolveEcho(settingsJSON []byte, driverDefault bool) bool

ResolveEcho returns the effective echo flag for a conversation given its raw settings JSON and the driver's default. Explicit conversation settings win over the driver default; missing JSON or missing key falls through to the default.

func StartPublicSweeper added in v0.2.8 

func StartPublicSweeper(ctx context.Context, database *db.DB, mgr *BridgeManager, interval time.Duration, logger *zap.Logger)

StartPublicSweeper kicks off a goroutine that runs SweepExpiredPublicConversations every interval until ctx is done. Single-replica only — for multi-replica deployments this needs a Postgres advisory lock so exactly one airlock holds the sweeper at any moment.

func StreamNDJSONResponse 

func StreamNDJSONResponse(body io.Reader, runID string, events chan<- ResponseEvent, suppressConfirmation bool) (string, []message.Message, *usageInfo, error)

StreamNDJSONResponse reads NDJSON events from a response stream, forwards ResponseEvents to the events channel for real-time delivery, and collects the full text response. Closes the events channel when done. runID is stamped onto confirmation_required events so drivers can build callback-bound UI (e.g. Telegram inline keyboards).

suppressConfirmation drops confirmation_required events instead of forwarding them: a one-shot (public, single-turn) session has no second turn in which to answer, so rendering approve/deny buttons would only produce dead UI. The caller detects the resulting suspended run and auto-denies it.

func SweepExpiredPublicConversations added in v0.2.8 

func SweepExpiredPublicConversations(ctx context.Context, database *db.DB, mgr *BridgeManager, logger *zap.Logger)

SweepExpiredPublicConversations runs one pass: list public bridge conversations whose updated_at is older than the bridge's configured TTL, send a final "Conversation completed" message, then delete the conversation row (which cascade-drops messages via the FK).

Sends are best-effort — a failure to deliver the final message (network blip, revoked bot token, deleted DM channel) does not block deletion. Local state is the source of truth; the user just doesn't see the goodbye.

func ValidateManagerBotToken 

func ValidateManagerBotToken(ctx context.Context, token string) (string, bool, error)

ValidateManagerBotToken does a getMe round-trip against Telegram and returns (username, can_manage_bots, error). Used by the settings handler when an admin pastes a new manager-bot token, so validation happens against the same wire shape the poller uses.

Types

type AgentSlashConv added in v0.4.0 

type AgentSlashConv struct {
	// contains filtered or unexported fields
}

AgentSlashConv is the SlashConv implementation for agent-bridge and web conversations stored in agent_conversations / agent_messages. /compact forwards to the agent container (sets ForwardAsCompact); every other command operates locally against the airlock DB.

func NewAgentSlashConv added in v0.4.0 

func NewAgentSlashConv(q *dbq.Queries, canceler RunCanceler, logger *zap.Logger) *AgentSlashConv

NewAgentSlashConv builds an adapter for the agent-conversation path. canceler dispatches /cancel into the agent dispatcher; nil yields the "Nothing to cancel" path on restart (no live in-memory state).

func (*AgentSlashConv) Cancel added in v0.4.0 

func (a *AgentSlashConv) Cancel(ctx context.Context, convID pgtype.UUID) bool

Cancel — q.GetLatestRunningPromptRun then dispatcher.CancelRun. The HTTP-request abort is what flips the run row out of 'running' (via the agent's r.Complete or the stuck-run sweeper as a backstop).

func (*AgentSlashConv) Clear added in v0.4.0 

func (a *AgentSlashConv) Clear(ctx context.Context, convID pgtype.UUID) (bool, error)

Clear writes a checkpoint marker row and advances the conversation's context_checkpoint_message_id. Also best-effort resolves any suspended run for THIS conversation (not a sibling-delegated suspension that happens to be the agent's latest).

func (*AgentSlashConv) Compact added in v0.4.0 

func (a *AgentSlashConv) Compact(_ context.Context, _ pgtype.UUID) (string, bool, error)

Compact is a no-op locally: the agent container runs the actual summarization via Sol.Runner.Compact, so we just signal "forward as compact" and let the proxy set ForceCompact=true on the forwarded request.

func (*AgentSlashConv) Echo added in v0.4.0 

func (a *AgentSlashConv) Echo(ctx context.Context, convID pgtype.UUID, args string) (bool, error)

Echo flips the conversation's settings.echo flag. Toggle treats unset as off so the first /echo in a chat that's quiet by default always turns echo on.

type BridgeCallback 

type BridgeCallback struct {
	Data      string // opaque payload, e.g. "approve:<runID>"
	AckID     string // platform-specific ack handle (Telegram callback_query.id)
	MessageID string // ID of the message the button is attached to (so we

}

BridgeCallback represents an interactive UI acknowledgement — a button tap on an inline keyboard or similar platform-native affordance. Drivers that don't support rich UI leave this nil.

type BridgeDriver 

type BridgeDriver interface {
	// Init is called once when a bridge is first created.
	// Uses pointer so the driver can set initial config (e.g. poll offset).
	Init(ctx context.Context, br *dbq.Bridge) error

	// Activate is called on every startup for active bridges.
	Activate(ctx context.Context, br dbq.Bridge) error

	// Teardown is called when a bridge is deleted or disabled.
	Teardown(ctx context.Context, br dbq.Bridge) error

	// Poll fetches new events from the platform.
	// Uses pointer so the driver can update br.Config (e.g. poll offset).
	Poll(ctx context.Context, br *dbq.Bridge) ([]BridgeEvent, error)

	// SendStream delivers a response, streaming text deltas as they arrive.
	// echo controls whether tool-call / tool-result bubbles are rendered;
	// drivers that collapse tool output some other way may ignore it.
	// Returns the final assembled text.
	SendStream(ctx context.Context, br dbq.Bridge, externalID string, echo bool, events <-chan ResponseEvent) (string, error)

	// DefaultEcho returns whether tool bubbles render by default on this
	// platform. Drivers that display each tool-call/tool-result as its own
	// chat message (Telegram) should return false; drivers whose UI can
	// collapse tool output inline (web) should return true.
	// Used when a conversation has no explicit settings.echo override.
	DefaultEcho() bool

	// RemoveButtons strips the inline keyboard / component buttons from a
	// previously sent message, leaving its text intact. Called after the
	// user taps an approve/deny button so the resolved confirmation can't
	// be tapped again. Best-effort: errors are logged but not propagated.
	RemoveButtons(ctx context.Context, br dbq.Bridge, externalID, messageID string) error
}

BridgeDriver handles platform-specific message parsing and delivery.

type BridgeEvent 

type BridgeEvent struct {
	BridgeID          uuid.UUID
	ExternalID        string // platform chat_id (Telegram chat ID, etc.)
	SenderID          string // platform user ID of sender (for identity lookup)
	SenderName        string
	Text              string
	Files             []BridgeFile // attached files (photos, documents)
	Callback          *BridgeCallback
	ReferencedMessage *BridgeReferencedMessage // reply target / forward source (driver-populated)
	RawPayload        []byte
}

BridgeEvent represents a normalized incoming event from any platform. Either Text/Files (new user message) or Callback (button tap) is populated.

type BridgeFile 

type BridgeFile struct {
	FileID      string // platform file ID (e.g. Telegram file_id)
	Filename    string
	ContentType string
	Size        int64
	Data        []byte // file content (downloaded by driver)

	// IsVoiceNote marks a short voice recording (e.g. Telegram "voice")
	// that the bridge layer should auto-transcribe before forwarding to
	// the agent. Plain audio/video/document attachments leave this false.
	IsVoiceNote bool
}

BridgeFile is a file attached to a bridge message.

type BridgeManager 

type BridgeManager struct {
	// contains filtered or unexported fields
}

BridgeManager manages bridge drivers and routes events to agents.

func NewBridgeManager 

func NewBridgeManager(drivers map[string]BridgeDriver, prompter *PromptProxy, database *db.DB, encryptor secrets.Store, hmacSecret, publicURL string, agentBaseURL func(slug string) string, logger *zap.Logger) *BridgeManager

NewBridgeManager creates a BridgeManager. agentBaseURL builds the external URL for an agent's subdomain ({scheme}://{slug}.{domain}[:port]); the Telegram driver needs it to register the web-app menu button.

func (*BridgeManager) AddBridge 

func (m *BridgeManager) AddBridge(bridgeID uuid.UUID)

AddBridge activates a newly created bridge and starts its poller. Idempotent: if a poller is already running for this bridge ID (e.g. the bridge was re-registered after a config change), the existing one is cancelled first so only one poller hits the platform at a time.

func (*BridgeManager) AttachSysagent added in v0.4.0 

func (m *BridgeManager) AttachSysagent(s SysagentRuntime)

AttachSysagent wires the sysagent runtime after the router has built it. Idempotent; the last set wins.

func (*BridgeManager) BotTokenForBridge added in v0.4.0 

func (m *BridgeManager) BotTokenForBridge(ctx context.Context, agentID, bridgeID uuid.UUID) (string, error)

BotTokenForBridge returns the decrypted Telegram bot token for (agentID, bridgeID) — only if the bridge belongs to that agent and is of type "telegram". The agent guard is the cross-agent boundary at the lookup layer; HMAC verification of the caller's initData then gates the actual auth. Returns an error on any mismatch so the caller can 401 without leaking which constraint failed.

Used by the airlock proxy's Telegram Web App auth handler.

func (*BridgeManager) HandleEvent 

func (m *BridgeManager) HandleEvent(ctx context.Context, event BridgeEvent) error

HandleEvent processes a parsed BridgeEvent — routes to agent via PromptProxy.

func (*BridgeManager) RemoveBridge 

func (m *BridgeManager) RemoveBridge(bridgeID uuid.UUID)

RemoveBridge stops the poller for a bridge. Safe to call for an unknown bridge ID — it's a no-op. The DB row is NOT touched here; callers that want full deletion do the DB work separately (typically by calling q.DeleteBridge alongside this).

func (*BridgeManager) RemoveBridgesByOwner added in v0.4.0 

func (m *BridgeManager) RemoveBridgesByOwner(ctx context.Context, ownerID uuid.UUID) error

RemoveBridgesByOwner stops every poller for bridges owned by a specific user. Called from service/users.Delete BEFORE the DB CASCADE removes the bridge rows — otherwise the poller goroutines would keep calling getUpdates against now-deleted bridges until their next transient failure, racing on the bot token with any replacement bridge that happened to land on the same row id.

func (*BridgeManager) SendMessage 

func (m *BridgeManager) SendMessage(ctx context.Context, bridgeID uuid.UUID, externalID, text string) error

SendMessage sends a text message to a bridge conversation. Convenience wrapper.

func (*BridgeManager) SendParts 

func (m *BridgeManager) SendParts(ctx context.Context, bridgeID uuid.UUID, externalID string, parts []agentsdk.DisplayPart) error

SendParts sends display parts to a bridge conversation. Looks up the bridge, decrypts the token, and delegates to the driver.

func (*BridgeManager) Start 

func (m *BridgeManager) Start(ctx context.Context) error

Start sets up all active bridges and starts pollers.

func (*BridgeManager) Stop 

func (m *BridgeManager) Stop()

Stop gracefully shuts down all pollers.

type BridgeReferencedMessage added in v0.2.8 

type BridgeReferencedMessage struct {
	Kind       string // BridgeReferenceReply | BridgeReferenceForward
	SenderName string // author of the referenced content
	Text       string
	AuthoredAt time.Time
	FromBot    bool // true when the referenced message was authored by our bot (replies only)
}

BridgeReferencedMessage describes a message the current event points at — either a reply target or a forwarded message. Surfaced to the LLM as a wrapped context block regardless of session mode, so the model has the referenced content even when it's outside the active conversation history (or when there is no history at all).

type CommandRegistrar 

type CommandRegistrar interface {
	RegisterCommands(ctx context.Context, br dbq.Bridge, cmds []SlashCommand) error
}

CommandRegistrar is an optional BridgeDriver capability: platforms with a native command menu (Telegram setMyCommands, Discord app commands, ...) implement it to receive the slash-command registry on activation. Drivers without such a menu simply don't implement it.

type DiscordDriver added in v0.2.8 

type DiscordDriver struct {
	// contains filtered or unexported fields
}

DiscordDriver speaks the Discord Gateway WebSocket for inbound events and the v10 REST API for outbound message delivery. Single-replica only today: the Gateway connection holds session state and only one process can identify per token, so multi-replica needs a per-bridge advisory lock added later.

func NewDiscordDriver added in v0.2.8 

func NewDiscordDriver(logger *zap.Logger) *DiscordDriver

NewDiscordDriver creates a DiscordDriver. logger surfaces gateway and REST errors. Pass zap.NewNop() in tests.

func NewDiscordDriverWithBase added in v0.2.8 

func NewDiscordDriverWithBase(apiBase string, client *http.Client) *DiscordDriver

NewDiscordDriverWithBase creates a DiscordDriver pointed at an alternate REST base URL. The Gateway URL is fetched via GET /gateway/bot and inherits the same prefix for tests.

func (*DiscordDriver) Activate added in v0.2.8 

func (d *DiscordDriver) Activate(ctx context.Context, br dbq.Bridge) error

func (*DiscordDriver) CreateDMChannel added in v0.2.8 

func (d *DiscordDriver) CreateDMChannel(ctx context.Context, token, recipientID string) (string, error)

CreateDMChannel opens (or fetches) a DM channel with a recipient and returns its channel ID. Idempotent — Discord returns the same channel every time for a given (bot, user) pair.

func (*DiscordDriver) DefaultEcho added in v0.2.8 

func (d *DiscordDriver) DefaultEcho() bool

DefaultEcho mirrors Telegram: tool-call/tool-result bubbles are off by default. Each tool call would render as a separate message and a chatty agent quickly fills the DM.

func (*DiscordDriver) FetchUser added in v0.2.8 

func (d *DiscordDriver) FetchUser(ctx context.Context, token, userID string) (DiscordUserInfo, error)

FetchUser hits GET /users/{id} for a snowflake. Bot tokens have permission to look up any user by ID, returning the public-profile fields (username, global_name, avatar). Used by the identity-linking preview so the user can verify which Discord account is about to bind.

func (*DiscordDriver) GetMe added in v0.2.8 

func (d *DiscordDriver) GetMe(ctx context.Context, token string) (string, error)

GetMe is the public-facing token validator used by the bridge create handler. Returns the bot's username for storage on the bridge row.

func (*DiscordDriver) Init added in v0.2.8 

func (d *DiscordDriver) Init(ctx context.Context, br *dbq.Bridge) error

func (*DiscordDriver) Poll added in v0.2.8 

func (d *DiscordDriver) Poll(ctx context.Context, br *dbq.Bridge) ([]BridgeEvent, error)

Poll blocks until the gateway goroutine has buffered events or the timeout fires. The driver handles its own reconnect, so Poll never returns errors — a dead WS just yields no events until reconnect.

func (*DiscordDriver) RegisterCommands added in v0.2.8 

func (d *DiscordDriver) RegisterCommands(ctx context.Context, br dbq.Bridge, cmds []SlashCommand) error

RegisterCommands bulk-overwrites the global slash commands for this bot's application. Global commands take up to ~1 hour to propagate across Discord's edge — first-time install is fine, name changes lag.

func (*DiscordDriver) RemoveButtons added in v0.2.11 

func (d *DiscordDriver) RemoveButtons(ctx context.Context, br dbq.Bridge, externalID, messageID string) error

RemoveButtons strips the action-row components from a previously sent message via PATCH /channels/{ch}/messages/{msg} with components=[]. The message text is left intact.

func (*DiscordDriver) SendDM added in v0.2.8 

func (d *DiscordDriver) SendDM(ctx context.Context, token, recipientID, content string) error

SendDM is a convenience wrapper that opens (or reuses) a DM channel with a user and posts a plain-text message. Used by the bridge layer for the identity-linking prompt so the link is never posted publicly.

func (*DiscordDriver) SendParts added in v0.2.8 

func (d *DiscordDriver) SendParts(ctx context.Context, token, channelID string, parts []agentsdk.DisplayPart) error

SendParts sends display parts to a Discord channel — text, images, files.

func (*DiscordDriver) SendStream added in v0.2.8 

func (d *DiscordDriver) SendStream(ctx context.Context, br dbq.Bridge, externalID string, echo bool, events <-chan ResponseEvent) (string, error)

func (*DiscordDriver) Teardown added in v0.2.8 

func (d *DiscordDriver) Teardown(_ context.Context, br dbq.Bridge) error

type DiscordUserInfo added in v0.2.8 

type DiscordUserInfo struct {
	ID         string
	Username   string // legacy @handle (no discriminator)
	GlobalName string // newer "display name" (Discord rolled out 2023)
	AvatarURL  string
}

DiscordUserInfo is the slice of /users/{id} we surface in identity linking. AvatarURL is the resolved CDN URL ("" if the user has no avatar set) so callers don't have to know the snowflake-hash format.

type Dispatcher 

type Dispatcher struct {
	// contains filtered or unexported fields
}

Dispatcher ensures agent containers are running and forwards HTTP requests to them.

func NewDispatcher 

func NewDispatcher(cfg *config.Config, db *db.DB, containers container.ContainerManager, enc secrets.Store, logger *zap.Logger) *Dispatcher

NewDispatcher creates a Dispatcher.

func (*Dispatcher) CancelRun added in v0.2.8 

func (d *Dispatcher) CancelRun(runID uuid.UUID) bool

CancelRun aborts the in-flight outbound request for the given run, if any. Returns true if a cancel was fired. Idempotent — repeat calls and calls for runs that already finished are no-ops.

A2A cascade: the cancel also walks runs.parent_run_id downward and fires the cancel hook on every still-in-flight descendant. The HTTP disconnect chain already cascades cancels (parent's outbound HTTP closing → child's ctx.Done() → child's CancelRun), but this gives an explicit best-effort kick for runs on the same replica when the user cancels mid-chain.

func (*Dispatcher) EnsureRunning 

func (d *Dispatcher) EnsureRunning(ctx context.Context, agentID uuid.UUID) (*container.Container, error)

EnsureRunning looks up the agent, decrypts its DB credentials, and starts (or reconnects to) the agent container. Returns the running container.

func (*Dispatcher) ForwardA2APrompt added in v0.4.0 

func (d *Dispatcher) ForwardA2APrompt(ctx context.Context, agentID uuid.UUID, parentRunID uuid.UUID, callerAccess agentsdk.Access, userID *uuid.UUID, input agentsdk.PromptInput) (io.ReadCloser, uuid.UUID, error)

ForwardA2APrompt is ForwardPrompt for the sibling-agent code path: the caller is another agent's run, parentRunID is its run.id, and the new run's parent_run_id and trigger_type/_ref are wired accordingly. callerAccess is the access level Airlock pre-resolved against the target agent (see api/access.computeA2ACallerAccess). userID is the original user (the human at the top of the chain — propagated through every A2A hop via the conversation's user_id), used for both the new run's VisibleSiblings computation and audit.

func (*Dispatcher) ForwardCron 

func (d *Dispatcher) ForwardCron(ctx context.Context, agentID uuid.UUID, cronName string, timeout time.Duration) (io.ReadCloser, uuid.UUID, error)

ForwardCron ensures the agent is running, creates a run record, and POSTs to the agent's cron endpoint. Returns the response body stream and the run ID. The timeout parameter controls the HTTP client timeout.

func (*Dispatcher) ForwardPrompt 

func (d *Dispatcher) ForwardPrompt(ctx context.Context, agentID uuid.UUID, input agentsdk.PromptInput, bridgeID *uuid.UUID, userID *uuid.UUID) (io.ReadCloser, uuid.UUID, error)

ForwardPrompt ensures the agent is running, creates a run record, and POSTs the prompt input to the agent container. Returns the response body stream (NDJSON) and the run ID. userID is the prompting user (anchor for A2A VisibleSiblings); pass nil for anonymous/system runs.

func (*Dispatcher) ForwardWebhook 

func (d *Dispatcher) ForwardWebhook(ctx context.Context, agentID uuid.UUID, path string, body []byte, bridgeID *uuid.UUID, timeout time.Duration) (io.ReadCloser, uuid.UUID, error)

ForwardWebhook ensures the agent is running, creates a run record, and POSTs the webhook payload to the agent container. Returns the response body stream and the run ID. The timeout parameter controls the HTTP client timeout.

func (*Dispatcher) InFlightIDs added in v0.2.8 

func (d *Dispatcher) InFlightIDs() []uuid.UUID

InFlightIDs returns a snapshot of currently-tracked run IDs. Used by the stuck-run sweeper so it doesn't race the dispatcher and prematurely terminate a still-live run.

func (*Dispatcher) RefreshAgent 

func (d *Dispatcher) RefreshAgent(ctx context.Context, agentID uuid.UUID) error

RefreshAgent triggers a synchronous re-sync on the agent container. Used after server-side state changes the cached system prompt depends on (typically MCP OAuth completion) so the running agent picks up new tools without a restart. If the container isn't running, returns nil — there's nothing to refresh; the agent will sync fresh on its next startup.

type ManagerBot 

type ManagerBot struct {
	// contains filtered or unexported fields
}

ManagerBot is the singleton Telegram Managed Bots poller. One per airlock instance; configured via system_settings.telegram_manager_bot_*. Start spawns the poll goroutine; Stop cancels it; Reload re-reads the stored token (called after a settings update).

The poller is *separate* from the per-bridge bridge pollers — those hold a bridge row's token, this one holds the admin-level manager bot's. Failures route through the system_settings.telegram_manager_bot_error column (visible inline on the settings page) rather than panicking — invalid token or revoked can_manage_bots shouldn't take down airlock.

func NewManagerBot 

func NewManagerBot(database *db.DB, encryptor secrets.Store, bridges *bridgessvc.Service, bridgeMgr *BridgeManager, logger *zap.Logger) *ManagerBot

NewManagerBot wires the poller. Start must be called separately so the caller controls lifecycle (e.g. defer Stop on shutdown).

func (*ManagerBot) Reload 

func (mb *ManagerBot) Reload(ctx context.Context) error

Reload re-reads the configured token and (re)starts the poller. Called by the settings handler after PUT /settings/telegram-manager-bot. Idempotent.

func (*ManagerBot) Start 

func (mb *ManagerBot) Start(ctx context.Context) error

Start reads the configured token, calls getMe to validate, and spawns the poll goroutine if validation succeeds. Validation failures (no token, network error, can_manage_bots=false, revoked) are recorded in system_settings.telegram_manager_bot_error and surfaced inline in the settings UI — the poller silently does nothing.

Returns nil on both success ("poller spawned") and the "no token configured" case (legitimate empty state). Returns a real error only when something the operator can't fix from the UI failed (DB write of the error string itself).

func (*ManagerBot) Stop 

func (mb *ManagerBot) Stop()

Stop cancels the poll goroutine. Safe to call when not running.

func (*ManagerBot) Username 

func (mb *ManagerBot) Username() string

Username returns the manager bot's @handle resolved by getMe at Start time (empty when the poller isn't running or hasn't yet validated the token). The managedbots service reads this via a callback to template deep links — keeping the value an atomic makes a Reload-during-CreateSession race observable as "stale-but-correct" rather than torn.

type PromptProxy 

type PromptProxy struct {
	// contains filtered or unexported fields
}

PromptProxy manages conversation history and forwards prompts to agent containers.

func NewPromptProxy 

func NewPromptProxy(dispatcher *Dispatcher, database *db.DB, s3 *storage.S3Client, resolveTranscription TranscriptionResolver, logger *zap.Logger) *PromptProxy

NewPromptProxy creates a PromptProxy. The resolver is invoked to obtain the admin-configured transcription model when a voice note arrives; pass nil to disable auto-transcription (voice notes flow through as plain attachments).

func (*PromptProxy) HandleCallback 

func (p *PromptProxy) HandleCallback(
	ctx context.Context,
	agentID, bridgeID, userID uuid.UUID,
	externalID, data string,
	events chan<- ResponseEvent,
) (staleRun bool, err error)

HandleCallback resolves a suspended run based on a bridge UI callback (inline-keyboard tap). data is the opaque platform payload — expected format: 

"approve:<runID>"  — resume with Approved=true, no prompt
"deny:<runID>"     — resume with Approved=false + a "Rejected by user." prompt

If the referenced run is no longer suspended (e.g. already resolved via web), emits a single "info" event and returns — the driver's AnswerCallbackQuery should still fire to clear the spinner.

func (*PromptProxy) HandleMessage 

func (p *PromptProxy) HandleMessage(
	ctx context.Context,
	agentID, bridgeID, userID uuid.UUID,
	externalID string,
	storeHistory bool,
	oneShot bool,
	userMessage string,
	files []BridgeFile,
	referenced *BridgeReferencedMessage,
	events chan<- ResponseEvent,
) (string, error)

HandleMessage processes an incoming DM for an agent via a bridge. Manages conversation history, forwards to agent, streams response events to the provided channel, stores response when complete. The events channel is closed when streaming completes.

func (*PromptProxy) TranscribeVoicePlain added in v0.4.0 

func (p *PromptProxy) TranscribeVoicePlain(ctx context.Context, files []BridgeFile) (text string, hasVoice bool, hasNonVoice bool)

TranscribeVoicePlain runs each voice-note file through the configured transcription model and returns the concatenated plain text — used by the sysagent-bridge path where there's no agent container, no per-file S3 key, and tagging transcripts with source keys would be meaningless. hasNonVoice signals that at least one non-voice file was attached so the caller can reject with a "files not supported" reply. Transcription failures degrade gracefully: the bool stays true if any voice file existed, the returned text just omits the failing entries.

type ResponseEvent 

type ResponseEvent struct {
	Type       string // "run_started", "text-delta", "tool-call", "tool-result", "confirmation_required", "info"
	Text       string // for text-delta / info: the delta text or info message
	ToolCallID string // for tool_call/tool_result
	ToolName   string // for tool_call/tool_result
	ToolInput  string // for tool_call: the tool arguments
	ToolOutput string // for tool_result: the tool output
	ToolError  string // for tool_result: error message if failed
	Raw        []byte // full NDJSON line (for non-text events drivers may need)

	// Populated when Type == "run_started" or "confirmation_required":
	RunID      string
	Permission string
	Patterns   []string
	Code       string
}

ResponseEvent represents an NDJSON event from the agent response stream, forwarded to the bridge driver for progressive delivery.

type RunCanceler added in v0.2.11 

type RunCanceler interface {
	CancelRun(runID uuid.UUID) bool
}

RunCanceler is the slice of *Dispatcher that the agent-path slash conv adapter uses to fire /cancel. Defined as an interface so tests can stub without standing up containers / DB / encryptor.

type Scheduler 

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler manages cron schedules for all active agents.

func NewScheduler 

func NewScheduler(dispatcher *Dispatcher, db *db.DB, logger *zap.Logger) *Scheduler

NewScheduler creates a Scheduler.

func (*Scheduler) ReloadAgent 

func (s *Scheduler) ReloadAgent(ctx context.Context, agentID uuid.UUID) error

ReloadAgent reloads cron entries for a specific agent after a sync.

func (*Scheduler) Start 

func (s *Scheduler) Start(ctx context.Context) error

Start loads all enabled crons from the database and starts the scheduler.

func (*Scheduler) Stop 

func (s *Scheduler) Stop()

Stop gracefully stops the scheduler.

type SlashCommand 

type SlashCommand struct {
	Name        string
	Description string
	Access      agentsdk.Access
}

SlashCommand describes a user-typable command. Registry is the single source of truth: TrySlashCommand dispatches from it, and bridge drivers that support native command menus (Telegram setMyCommands, etc.) expose it verbatim. Name is stored without the leading slash.

type SlashCommandResult 

type SlashCommandResult struct {
	Handled          bool
	Reply            string
	ForwardAsCompact bool
}

SlashCommandResult is the outcome of handling a slash command. Handled=false means the message was not a slash command — callers should forward it to the agent as usual. Handled=true means the command was recognized and processed. Reply is the short message to show the user, unless ForwardAsCompact is set — in that case the caller should invoke the agent with PromptInput.ForceCompact=true so Sol's summarization produces the user-visible text via a normal run. ForwardAsCompact is agent-path only; the sysagent path runs compact inline and returns the summary in Reply.

func TrySlashCommand 

func TrySlashCommand(
	ctx context.Context,
	conv SlashConv,
	convID pgtype.UUID,
	access agentsdk.Access,
	message string,
) (SlashCommandResult, error)

TrySlashCommand parses message for a leading slash command and, if recognized, dispatches it against conv. Returns Handled=false when the message is a plain user prompt — callers forward it to the agent (or sysagent) as usual.

access is the caller's resolved agent access; commands gate against SlashCommand.Access. /clear, /echo, /compact, /cancel route into SlashConv; /auth replies with the already-linked acknowledgement (the un-linked path is handled by the bridge before identity lookup).

type SlashConv added in v0.4.0 

type SlashConv interface {
	// Cancel aborts the latest live run on this conversation. Returns
	// true if a cancel was dispatched, false if there was nothing in
	// flight (or the cancel target couldn't be reached, e.g. process
	// restart cleared the in-memory dispatcher state).
	Cancel(ctx context.Context, convID pgtype.UUID) bool

	// Clear advances the conversation's context checkpoint to a fresh
	// marker message and resolves any lingering suspended run. Returns
	// whether a suspension was cleared so the reply can mention it.
	Clear(ctx context.Context, convID pgtype.UUID) (suspensionCleared bool, err error)

	// Compact compacts the conversation's history. Agent path: returns
	// (forward=true) so TrySlashCommand sets ForwardAsCompact and the
	// agent container runs Sol.Runner.Compact, emitting the summary as
	// a normal assistant message. Sysagent path: runs compaction
	// in-process and returns (summary, forward=false).
	Compact(ctx context.Context, convID pgtype.UUID) (reply string, forward bool, err error)

	// Echo updates the "show tool output bubbles" setting and returns
	// the resulting state. args is the raw post-/echo token ("on",
	// "off", or empty for toggle).
	Echo(ctx context.Context, convID pgtype.UUID, args string) (on bool, err error)
}

SlashConv is the per-conversation backing the slash-command dispatcher operates against. Two implementations: agentConvAdapter (agent bridges, web chat) and sysagentConvAdapter (system bridges → sysagent). Each method is the full compound operation the corresponding /command performs, not a primitive query — the per-conversation table choice stays inside the adapter.

type SysagentRuntime added in v0.4.0 

type SysagentRuntime interface {
	CancelRun(runID uuid.UUID) bool
	Compact(ctx context.Context, p authz.Principal, conversationID uuid.UUID) (string, error)
	RunPromptInline(
		ctx context.Context,
		p authz.Principal,
		conversationID uuid.UUID,
		text, platform string,
		approved *bool,
		resumeRunID string,
		sink eventstream.Sink,
		onStart func(runID uuid.UUID),
	) (uuid.UUID, error)
}

SysagentRuntime is the narrow surface trigger uses from the sysagent package. Declared as an interface here so trigger doesn't import sysagent directly (sysagent → service/agents → trigger would cycle). *sysagent.Service satisfies this directly; api/router.go passes it in.

RunPromptInline handles three turn shapes — fresh user message (text non-empty), Approve/Reject of a previously-suspended run (approved non-nil + resumeRunID set), or auto-resume after a system event (all zero). Same shape as the web RunPrompt path.

type SysagentSlashConv added in v0.4.0 

type SysagentSlashConv struct {
	// contains filtered or unexported fields
}

SysagentSlashConv is the SlashConv implementation for system bridges, which route inbound DMs into the in-airlock sysagent. Each method operates against system_conversations / system_messages / system_runs instead of the agent_* tables. /compact runs sol.Runner.Compact locally and returns the summary text (there's no agent container to forward to).

func NewSysagentSlashConv added in v0.4.0 

func NewSysagentSlashConv(svc SysagentRuntime, q *dbq.Queries, p authz.Principal, logger *zap.Logger) *SysagentSlashConv

NewSysagentSlashConv builds an adapter bound to a specific resolved caller. Per-request (not long-lived) because the principal changes per inbound DM.

func (*SysagentSlashConv) Cancel added in v0.4.0 

func (s *SysagentSlashConv) Cancel(ctx context.Context, convID pgtype.UUID) bool

Cancel — pick the latest running-or-suspended sysagent run on this conversation and ask the Service to interrupt its goroutine.

func (*SysagentSlashConv) Clear added in v0.4.0 

func (s *SysagentSlashConv) Clear(ctx context.Context, convID pgtype.UUID) (bool, error)

Clear writes a sysagent checkpoint marker via AppendSystemMessage and advances system_conversations.context_checkpoint_message_id. Also flips any suspended sysagent run to cancelled so the pending- confirmation dialog doesn't linger past the clear.

func (*SysagentSlashConv) Compact added in v0.4.0 

func (s *SysagentSlashConv) Compact(ctx context.Context, convID pgtype.UUID) (string, bool, error)

Compact runs sysagent's in-process summarize-and-checkpoint and returns the summary text. Unlike the agent path there's no container to forward to.

func (*SysagentSlashConv) Echo added in v0.4.0 

func (s *SysagentSlashConv) Echo(ctx context.Context, convID pgtype.UUID, args string) (bool, error)

Echo flips system_conversations.settings.echo via the JSONB merge query. Toggle treats unset as off so the first /echo always turns it on. Mirrors the agent-path semantics.

type TelegramChatInfo 

type TelegramChatInfo struct {
	Username  string
	FirstName string
	LastName  string
}

TelegramChatInfo holds the subset of getChat fields we expose.

type TelegramDriver 

type TelegramDriver struct {
	// contains filtered or unexported fields
}

TelegramDriver handles Telegram bridges via long-polling.

func NewTelegramDriver 

func NewTelegramDriver(logger *zap.Logger) *TelegramDriver

NewTelegramDriver creates a TelegramDriver. logger is used to surface Telegram API errors (notably the silent-failure Markdown/HTML parse rejections that otherwise vanish). Pass zap.NewNop() in tests.

The HTTP client's Timeout caps the entire request — connect, TLS, upload, server processing, response body read. Sized to comfortably cover the long-poll getUpdates window (timeout=30s server-side) plus TLS/network slack, while still releasing a goroutine if a connection stalls on a short call (sendMessage, getMe, …). Without a ceiling here those short calls can hang indefinitely on a half-open TCP socket — the kind of failure that happens on a local network blip (WG/VPN reconnect, NIC cycle) when the OS hasn't yet noticed the peer is gone.

func NewTelegramDriverWithBaseURL 

func NewTelegramDriverWithBaseURL(baseURL string, client *http.Client) *TelegramDriver

NewTelegramDriverWithBaseURL creates a TelegramDriver with a custom base URL and HTTP client (for testing).

func (*TelegramDriver) Activate 

func (d *TelegramDriver) Activate(ctx context.Context, br dbq.Bridge) error

func (*TelegramDriver) AnswerCallbackQuery 

func (d *TelegramDriver) AnswerCallbackQuery(ctx context.Context, token, callbackID, text string) error

AnswerCallbackQuery clears the loading spinner on a tapped inline-keyboard button. If text is non-empty, Telegram shows it as a transient toast.

func (*TelegramDriver) DefaultEcho 

func (d *TelegramDriver) DefaultEcho() bool

DefaultEcho reports that Telegram defaults to hiding tool bubbles: each tool-call / tool-result is rendered as its own chat message and a chatty agent quickly swamps the conversation. Users can opt in per-chat with `/echo on`.

func (*TelegramDriver) GetChat 

func (d *TelegramDriver) GetChat(ctx context.Context, token, chatID string) (TelegramChatInfo, error)

GetChat calls the Telegram getChat API for a private chat. Because private chat IDs equal the user ID, this resolves a telegram user's public username and display name as long as that user has DM'd the bot.

func (*TelegramDriver) GetMe 

func (d *TelegramDriver) GetMe(ctx context.Context, token string) (string, error)

GetMe calls the Telegram getMe API and returns the bot username.

func (*TelegramDriver) Init 

func (d *TelegramDriver) Init(ctx context.Context, br *dbq.Bridge) error

func (*TelegramDriver) Poll 

func (d *TelegramDriver) Poll(ctx context.Context, br *dbq.Bridge) ([]BridgeEvent, error)

func (*TelegramDriver) RegisterCommands 

func (d *TelegramDriver) RegisterCommands(ctx context.Context, br dbq.Bridge, cmds []SlashCommand) error

RegisterCommands publishes the slash-command registry to Telegram's global command menu via setMyCommands. Telegram stores names without the leading slash.

func (*TelegramDriver) RemoveButtons added in v0.2.11 

func (d *TelegramDriver) RemoveButtons(ctx context.Context, br dbq.Bridge, externalID, messageID string) error

RemoveButtons strips the inline keyboard from a previously sent message via editMessageReplyMarkup. The message text is left intact so the conversation history still shows what was being confirmed.

func (*TelegramDriver) SendMessage 

func (d *TelegramDriver) SendMessage(ctx context.Context, token string, chatID int64, text string) error

SendMessage sends a text message to a Telegram chat.

func (*TelegramDriver) SendParts 

func (d *TelegramDriver) SendParts(ctx context.Context, token string, chatID int64, parts []agentsdk.DisplayPart) error

SendParts sends display parts to a Telegram chat. Renders each part appropriately: text → sendMessage, image → sendPhoto, file → sendDocument.

func (*TelegramDriver) SendStream 

func (d *TelegramDriver) SendStream(ctx context.Context, br dbq.Bridge, externalID string, echo bool, events <-chan ResponseEvent) (string, error)

func (*TelegramDriver) SetMenuButton added in v0.4.0 

func (d *TelegramDriver) SetMenuButton(ctx context.Context, token, url string) error

SetMenuButton configures the bot's default chat menu button to launch a Telegram Web App at the given URL. The button is persistent — it shows for every private chat the bot is in, opens the URL in Telegram's in-app browser, and exposes initData to the page so airlock can authenticate the user automatically.

Passing url=="" clears the bot back to Telegram's default menu (the commands list). Called once on bridge activation / token refresh; not idempotent at the Telegram side (each call re-publishes the button) but cheap and safe to call repeatedly.

func (*TelegramDriver) Teardown 

func (d *TelegramDriver) Teardown(_ context.Context, _ dbq.Bridge) error

type TranscriptionResolver 

type TranscriptionResolver func(ctx context.Context) (model.TranscriptionModel, error)

TranscriptionResolver returns the admin-configured transcription model or ErrTranscriptionNotConfigured when no model is set.

func NewTranscriptionResolver 

func NewTranscriptionResolver(database *db.DB, encryptor secrets.Store) TranscriptionResolver

NewTranscriptionResolver returns a resolver that reads system_settings.default_stt_provider_id + default_stt_model and looks up the associated provider row credentials.

Source Files

View all Source files

Directories

Path Synopsis
Package tgwebapp verifies Telegram Web App initData payloads.
 Package tgwebapp verifies Telegram Web App initData payloads.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.