runtime

package
v0.7.0-rc.7 Latest Latest
Warning

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

 Go to latest Published: Sep 18, 2026 License: AGPL-3.0 Imports: 64 Imported by: 0

Documentation

Overview

Package runtime implements the backing operations used by hosted chat and the authenticated agent protocol. The capability broker authorizes hosted calls; this package never depends on HTTP handlers or accepts browser attribution.

Index

Constants

View Source 
const (
	HttpMaxTimeout     = 120 // seconds
	HttpDefaultTimeout = 30  // seconds
	// httpAutoSaveThreshold — text responses larger than this are streamed
	// to S3 automatically instead of returned inline. Binary responses are
	// always auto-saved regardless of size. Keeps httpRequest tool results
	// well below agentsdk.maxToolOutputLen (16 KB) so they don't burn the
	// LLM's context window on a single call.
	HttpAutoSaveThreshold = 8 * 1024
	// httpMaxHTMLBytes caps the raw HTML we'll buffer for markdown
	// conversion. Pages bigger than this get treated as normal text and
	// either inlined (if somehow under the threshold) or streamed to S3.
	HttpMaxHTMLBytes = 10 * 1024 * 1024 // 10 MB
)
View Source 
const (
	RpcErrParse          = -32700
	RpcErrInvalidRequest = -32600
	RpcErrMethodNotFound = -32601
	RpcErrInvalidParams  = -32602
	RpcErrInternal       = -32603
	RpcErrServerError    = -32000
)
View Source 
const MaxBufferedResponseBytes = 20 << 20 // 20 MiB

MaxBufferedResponseBytes is the hard cap for any response a buffered SDK method accumulates fully in memory before returning to the caller. Overflow surfaces as agentsdk.ErrOutputTooLarge with no partial result.

The cap also applies Airlock-side as defense in depth on integration proxies, so a misbehaving SDK cannot exhaust Airlock memory.

Sized so structured small responses (JSON API replies, HTML pages, CLI tool summaries) pass through and larger data uses streaming storage.

Variables

View Source 
var ErrNoSearchProvider = errors.New("no search provider configured")

errNoSearchProvider is returned when the cascade can't resolve any search backend — neither the agent's exec-model provider nor any dedicated search-capable provider row is usable.

Functions

func ApplyHeaderMap 

func ApplyHeaderMap(h http.Header, m map[string]string)

applyHeaderMap merges m into h, using the empty-string-as-delete rule: a key whose value is "" removes any header of that name set by a lower layer. Non-empty values overwrite per-key.

func CallMCPTool 

func CallMCPTool(ctx context.Context, httpClient *http.Client, serverURL string, authInjection []byte, creds string, req wire.MCPToolCallRequest) (*wire.MCPToolCallResponse, error)

callMCPTool does a stateless MCP interaction: connect → initialize → tools/call → disconnect.

func ConnectionUpstreamURL 

func ConnectionUpstreamURL(policy *networkpolicy.Policy, baseURL, path string) (*url.URL, error)

func CurateHeaders 

func CurateHeaders(h http.Header, all bool) map[string]string

curateHeaders returns response headers. Default: only the few an agent reasons about (content negotiation, redirects, caching, pagination, rate limits). all=true returns every header verbatim.

func DbMessageToSession 

func DbMessageToSession(m dbq.AgentMessage) session.Message

dbMessageToSession converts a DB row to a session.Message.

func DecodeConnHeaders 

func DecodeConnHeaders(raw []byte) map[string]string

decodeConnHeaders unmarshals the connection's headers jsonb column. A malformed value is treated as "no overrides" — the platform baseline and per-call layers still apply — and logged at the call site rather than here so we don't pull a logger into a pure helper.

func DecodeToolOutput 

func DecodeToolOutput(raw json.RawMessage) (text, outcome, errText string)

decodeToolOutput resolves a stream tool-result/error "output" payload (the discriminated ToolResultOutput union) to (display text, outcome, error text). outcome is "success" | "error" | "denied". A nil/empty output is treated as an empty success; a malformed one yields the raw bytes as text with a success outcome (no legacy-shape handling — the data migration converts all history to the new shape).

func ExtractCanonicalKeys 

func ExtractCanonicalKeys(partsJSON []byte, agentID string) []string

ExtractCanonicalKeys reads `s3ref:K` sentinels from the stored goai-shaped parts JSON (image.image / file.data fields) and returns the canonical `llm/agents/<agentID>/K` keys. The sentinel survives the goai.Content marshal roundtrip since it's just a string in Image/Data.

func ExtractTextSummary 

func ExtractTextSummary(parts []wire.DisplayPart) string

ExtractTextSummary builds a text summary from display parts for the content column.

func GenerateAutoSaveKey 

func GenerateAutoSaveKey(rawURL, contentType, contentDisposition string) string

generateAutoSaveKey builds a tmp/ key with a stable basename derived from Content-Disposition, the URL path, or the content type's default extension.

func InjectAuth 

func InjectAuth(req *http.Request, authInjectionJSON []byte, creds string)

InjectAuth adds credentials to the upstream request based on the auth injection config.

func IsBinaryContentType 

func IsBinaryContentType(ct string) bool

isBinaryContentType returns true if the content type represents binary data that should be base64-encoded rather than returned as a string.

func IsHTMLContentType 

func IsHTMLContentType(ct string) bool

isHTMLContentType matches text/html and application/xhtml+xml (ignoring charset/params).

func NewCompactionFinishedEvent 

func NewCompactionFinishedEvent(runID string, raw json.RawMessage) *airlockv1.CompactionFinishedEvent

func NormalizeToolOrdering 

func NormalizeToolOrdering(convID pgtype.UUID, msgs []dbq.AgentMessage) ([]dbq.AgentMessage, []OrphanPair, []OrphanPair, error)

normalizeToolOrdering returns a provider-valid model history. Every real result is moved directly behind its originating assistant turn, missing results are synthesized there, and result rows with no call get a synthetic assistant turn at their original position. Provider-valid canonical histories are returned unchanged.

func ParseMCPSSE 

func ParseMCPSSE(reader ResponseLimitReader, requestID int64) (json.RawMessage, error)

func ParseSchema 

func ParseSchema(raw []byte) (map[string]any, error)

parseSchema returns the schema as a parsed map. Empty / unparseable schemas yield (nil, nil) so callers can fast-path through.

func ParseToolParts 

func ParseToolParts(parts []byte, wantType string) ([]rawToolPart, bool)

func PostToConversation 

func PostToConversation(ctx context.Context, deps PostDeps, opts PostOpts) error

PostToConversation stores a message, delivers it via the appropriate channel (WebSocket or bridge), and optionally triggers an LLM turn.

func PreviewText 

func PreviewText(b []byte) string

previewText returns a valid-UTF-8 head of b, capped at previewMaxBytes, for use as HTTPResponse.BodyPreview.

func PublishRunEvents 

func PublishRunEvents(
	ctx context.Context,
	body io.ReadCloser,
	pubsub *realtime.PubSub,
	db *pgxpool.Pool,
	agentID, runID uuid.UUID,
	logger *zap.Logger,
) (responseText string, newMessages []message.Message, tokensIn, tokensOut int32)

PublishRunEvents reads NDJSON from body, publishes typed proto events to WS, accumulates the assistant response text, and returns it along with token usage. This runs in a goroutine after the HTTP response has been sent.

User and conversation routing are loaded from the admitted run's immutable origin. Neither transport arguments nor streamed payloads can select recipients.

func PublishRunTerminal 

func PublishRunTerminal(ctx context.Context, q *dbq.Queries, pubsub *realtime.PubSub, agentID, runID uuid.UUID, status, errMsg string)

PublishRunTerminal publishes the appropriate WebSocket event for a run's terminal state. Mirrors what PublishRunEvents emits from the agent's NDJSON stream so frontends and bridges see one event regardless of which path the run took to terminal — happy path (NDJSON), cancel (CancelRun closes the stream early), the agent's detached r.Complete POST after the stream died, or the sweeper for stuck rows.

Maps airlock-side status strings to WS event types:

  • "error" / "failed" / "timeout" → run.error
  • everything else (success, tool_errors, cancelled) -> run.complete

Idempotent at the client: chat.ts ignores events for runIDs already finalized locally, so a duplicate from the happy-path NDJSON + this helper is harmless.

func ResolveMediaPartsJSON 

func ResolveMediaPartsJSON(ctx context.Context, s3Client *storage.S3Client, logger *zap.Logger, partsJSON []byte) []byte

ResolveMediaPartsJSON walks a JSON array of parts and presigns S3 source keys on media entries. Non-media parts (tool-call, tool-result) are passed through verbatim — deserializing into DisplayPart would strip unknown fields like toolCallId, toolName, and args.

func ResolveSearchClient 

func ResolveSearchClient(
	ctx context.Context,
	database *db.DB,
	enc secrets.Store,
	logger *zap.Logger,
	agentID string,
	slug string,
) (websearch.Client, error)

resolveSearchClient creates a websearch.Client using a cascade:

  1. The per-slug binding: when the request names a registered CapSearch slot bound to a provider, use that provider (+ optional model). An empty or unbound slug drops through, exactly like an unbound model slot.
  2. The agent's configured search slot (or the tenant default when unset): the chosen provider's overlay SearchBackend + the chosen model threaded into Options.Model (empty → the backend's default model).
  3. The agent's exec-model provider, if its overlay entry declares a SearchBackend. Reuses the LLM provider's stored API key.
  4. Any enabled provider row whose overlay entry declares a SearchBackend. Catalog-only (brave/perplexity) rows are preferred over LLM providers that happen to offer search on the side.
  5. errNoSearchProvider.

Decrypt errors are reported as hard errors, not silently swallowed: a key we can't decrypt is a misconfiguration the admin needs to see.

func SchemaHasAgentMarker 

func SchemaHasAgentMarker(schema map[string]any) bool

schemaHasAgentMarker walks the schema tree once looking for any format=agent-file or agent-dir marker. Lets us skip the full args/result walk for the common no-FilePath tool.

func StoreSessionMessage 

func StoreSessionMessage(ctx context.Context, q *dbq.Queries, convID pgtype.UUID, runID pgtype.UUID, source string, msg session.Message) error

storeSessionMessage persists a session.Message to the DB.

func StoreSessionMessageReturningID 

func StoreSessionMessageReturningID(ctx context.Context, q *dbq.Queries, convID pgtype.UUID, runID pgtype.UUID, source string, msg session.Message) (pgtype.UUID, error)

storeSessionMessageReturningID persists a session.Message and returns the ID of the first row inserted. A single session.Message may expand into multiple DB rows when its parts contain tool calls + results — callers needing the checkpoint anchor use the first row.

func StreamSaveToS3 

func StreamSaveToS3(ctx context.Context, s3 *storage.S3Client, s3Key string, r io.Reader, binary bool) (int, string, error)

streamSaveToS3 streams r into S3 at s3Key, returning the exact number of bytes written and — unless binary — a short UTF-8 preview of the head. The head is buffered once and re-prepended so the upload is still a single pass with no full-body buffering.

func SynthesizeOrphanToolResults 

func SynthesizeOrphanToolResults(ctx context.Context, q *dbq.Queries, runID uuid.UUID, status string, logger *zap.Logger)

SynthesizeOrphanToolResults inserts a synthetic role=tool message for every tool-call this run emitted that doesn't have a paired tool-result row. Required for the next LLM turn: provider APIs (Anthropic, OpenAI) reject inputs where an assistant's tool_use isn't followed by a tool_result with the matching id. Common after cancel, deadline-exceeded, panic mid-tool, or any path where the agent didn't get to write the tool's result before terminating.

The synthesized output text is derived from the run's terminal status so the LLM has some signal about why the tool didn't complete:

  • "cancelled" → "Cancelled by user."
  • "timeout" → "Tool timed out."
  • else → "Tool execution failed."

Best-effort: failures are logged but don't block the caller. The SessionLoad lazy-synthesis path is the safety net if this misses.

func ToolOrderingValid 

func ToolOrderingValid(msgs []dbq.AgentMessage) (bool, error)

func TryConfiguredSearch 

func TryConfiguredSearch(
	ctx context.Context,
	q *dbq.Queries,
	enc secrets.Store,
	agentID string,
) (websearch.Client, error)

tryConfiguredSearch builds a client from the agent's configured search slot, falling back to the tenant default when the agent leaves it unset. It honors both the chosen provider (its overlay SearchBackend) and the chosen model (threaded into Options.Model; empty falls back to the backend default). Returns (nil, nil) when nothing is configured or the provider isn't usable, so the caller drops to the exec/any cascade. Hard error only on decrypt.

func TrySlotSearch 

func TrySlotSearch(
	ctx context.Context,
	q *dbq.Queries,
	enc secrets.Store,
	agentID string,
	slug string,
) (websearch.Client, error)

trySlotSearch builds a client from a registered CapSearch model slot's binding. Returns (nil, nil) when the slug is empty, the slot is missing or unbound, or its provider isn't usable — the caller then drops to the agent/system search default, mirroring how an unbound model slot resolves. Hard error only on decrypt failure.

func ValidMediaFilename 

func ValidMediaFilename(filename string) bool

func ValidateSuspendedCheckpoint 

func ValidateSuspendedCheckpoint(checkpoint []byte) error

ValidateSuspendedCheckpoint validates the wire fields Airlock needs before a suspended run can become resumable. It intentionally does not depend on Sol.

Types

type BridgePartsDeliverer 

type BridgePartsDeliverer interface {
	SendParts(ctx context.Context, bridgeID uuid.UUID, externalID string, parts []wire.DisplayPart) error
}

BridgePartsDeliverer is the subset of trigger.BridgeManager needed for message delivery.

type Config 

type Config struct {
	DB                     *db.DB
	Encryptor              secrets.Store
	OAuthClient            *oauth.Client
	S3                     *storage.S3Client
	Files                  *agentstorage.Service
	Builder                UpgradeRunner
	PubSub                 *realtime.PubSub
	BridgeMgr              BridgePartsDeliverer
	HTTPNetwork            *networkpolicy.Policy
	Logger                 *zap.Logger
	LLMProxyURL            string
	PublicURL              string
	AgentBaseURL           func(slug string) string
	ForceInlineAttachments bool
}

type CountingReader 

type CountingReader struct {
	R io.Reader
	N int
}

countingReader tallies bytes read so a streamed-to-S3 body can report an exact Size even when the upstream Content-Length is unknown (chunked transfer).

func (*CountingReader) Read 

func (c *CountingReader) Read(p []byte) (int, error)

type JsonrpcError 

type JsonrpcError struct {
	Code    int    `json:"code"`
	Message string `json:"message"`
}

type JsonrpcMessage 

type JsonrpcMessage struct {
	JSONRPC string          `json:"jsonrpc"`
	ID      json.RawMessage `json:"id,omitempty"`
	Method  string          `json:"method,omitempty"`
	Params  json.RawMessage `json:"params,omitempty"`
	Result  json.RawMessage `json:"result,omitempty"`
	Error   *JsonrpcError   `json:"error,omitempty"`
}

type LlmUsageCapture 

type LlmUsageCapture struct {
	ProviderCatalogID string
	ProviderSlug      string
	Model             string
	Capability        string
	Slug              string

	TokensIn        int64
	TokensOut       int64
	TokensCached    int64
	TokensReasoning int64

	Units    float64
	UnitKind string // "" | "image" | "character" | "second"

	FinishReason        string
	Errored             bool
	Latency             time.Duration
	TaskTokensAccounted bool
	TaskOwnerToken      uuid.UUID
	TaskUsageID         uuid.UUID
}

llmUsageCapture is the per-call observation the proxy hands to the ledger. Token fields come from stream.FinishEvent.Usage (streaming) or the model result's Usage (non-streaming); unit fields carry image/audio quantities the token catalog cannot price.

type MaterializeError 

type MaterializeError struct {
	Code    int
	Message string
}

materializeError carries a JSON-RPC error code + human-readable message back to the dispatcher. Caller wraps via writeJSONRPCError.

func WalkSchema 

func WalkSchema(value any, schema map[string]any, ptr string, fn func(format string, v any, ptr string) (any, *MaterializeError)) (any, *MaterializeError)

walkSchema traverses value in parallel with schema. At each leaf where schema declares format=agent-file or agent-dir, fn is called. ptr is a dotted JSON pointer for error messages.

type McpHTTPClient 

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

func (*McpHTTPClient) CallTool 

func (c *McpHTTPClient) CallTool(ctx context.Context, name string, arguments json.RawMessage) (*wire.MCPToolCallResponse, error)

func (*McpHTTPClient) Close 

func (c *McpHTTPClient) Close() error

func (*McpHTTPClient) ListTools 

func (c *McpHTTPClient) ListTools(ctx context.Context) error

type McpToolInfo 

type McpToolInfo struct {
	Name        string          `json:"name"`
	Description string          `json:"description"`
	InputSchema json.RawMessage `json:"inputSchema"`
}

mcpToolInfo is the internal representation of a discovered MCP tool.

func DiscoverMCPTools 

func DiscoverMCPTools(ctx context.Context, httpClient *http.Client, serverURL string, authInjection []byte, creds string) ([]McpToolInfo, string, error)

DiscoverMCPTools connects to a remote MCP server and returns its tool schemas plus the server-level `instructions` it advertised in the initialize result (empty when the server set none).

type OrphanPair 

type OrphanPair struct {
	ToolCallID string
	ToolName   string
}

type PostDeps 

type PostDeps struct {
	DB         *db.DB
	PubSub     *realtime.PubSub
	BridgeMgr  BridgePartsDeliverer // nil if no bridge configured
	Dispatcher *trigger.Dispatcher  // nil if TriggerLLM not used
	S3         *storage.S3Client    // for resolving presigned URLs
	Logger     *zap.Logger
}

PostDeps holds shared dependencies for PostToConversation.

type PostOpts 

type PostOpts struct {
	AgentID        uuid.UUID
	ConversationID uuid.UUID
	RunID          uuid.UUID          // zero = no run linkage
	Role           string             // "assistant", "system"
	Text           string             // plain text content
	Parts          []wire.DisplayPart // rich content (optional)
	Source         string             // "notification", "system", etc.
	Ephemeral      bool               // stored for UI but excluded from LLM context
	BridgeOnly     bool               // reject loss of a bridge route instead of emitting a web event
	TriggerLLM     bool               // forward to agent for a response turn
	LLMMessage     string             // message text for the LLM turn (if TriggerLLM)
}

PostOpts configures a message post to a conversation.

type ResolvedModel 

type ResolvedModel struct {
	Limits                    session.ModelLimits
	ProviderID                string
	ProviderSlug              string
	ModelID                   string
	ApiKey                    string
	BaseURL                   string
	IncludeUsage              *bool
	SupportsStructuredOutputs *bool
}

resolveModel determines which provider row + model name to use for a request, then loads the row by FK and decrypts its API key.

Precedence:

  1. A non-empty slug names a registered agent_model_slots row: - bound (assigned_provider_id + assigned_model) ⇒ use it directly - unbound ⇒ resolve the default for the SLOT's declared capability, not the request-supplied one (the slot owns the capability — it is what the operator sees and binds in the UI) An unregistered non-empty slug is a loud error: the agentsdk getters require RegisterModel, so a missing row means a stale/typo'd slug, not something to silently route to a default.
  2. An empty slug is the capability-routed path used by the built-in media tools (transcribe/vision/image/speech/embedding): resolve the default for the request-supplied capability.

Steps 1(unbound) and 2 then walk the agent's per-capability override pair, then the system_settings capability default pair (modelForCapability). Empty FK at every tier ⇒ "no model configured" error.

type ResponseLimitReader 

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

type Service 

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

func New 

func New(c Config) *Service

func (*Service) CallMCPTool 

func (h *Service) CallMCPTool(ctx context.Context, agentID uuid.UUID, slug string, req wire.MCPToolCallRequest) (wire.MCPToolCallResponse, error)

func (*Service) CleanupOrphanedAttachments 

func (h *Service) CleanupOrphanedAttachments(ctx context.Context, agentID string, convID pgtype.UUID, checkpointID pgtype.UUID)

cleanupOrphanedAttachments scans the conversation's messages and deletes llm/ blobs referenced only in pre-checkpoint messages. Safe to call repeatedly — S3 DeleteObject is idempotent on missing keys. Runs synchronously for the scan, hands deletes to attachref.ScheduleDelete (which detaches to its own goroutine).

func (*Service) InvokePlatform 

func (h *Service) InvokePlatform(ctx context.Context, scope wire.RuntimeContext, definition capability.Definition, input json.RawMessage) (tool.Result, error)

InvokePlatform receives only broker-authorized definitions and host attribution. Resource credentials and MCP negotiation stay in their existing backing services.

func (*Service) LanguageModelOptions 

func (h *Service) LanguageModelOptions(resolved ResolvedModel) solprovider.Options

func (*Service) ListMCPTools 

func (h *Service) ListMCPTools(ctx context.Context, agentID uuid.UUID, slug string) (integrationservice.MCPTools, error)

func (*Service) ModelForCapability 

func (h *Service) ModelForCapability(ctx context.Context, q *dbq.Queries, agentID pgtype.UUID, capability string) (pgtype.UUID, string, error)

modelForCapability picks the model for a capability using the tier-2 and tier-3 fallbacks: per-agent override pair, then system default pair. Returns invalid FK + empty name when both tiers are empty so the caller can produce a single clear error.

func (*Service) PrepareAgent 

func (h *Service) PrepareAgent(ctx context.Context, scope capabilities.Scope, d wire.AgentDefinition) (agentruntime.Input, error)

func (*Service) PrepareChat 

func (h *Service) PrepareChat(ctx context.Context, scope capabilities.Scope, input wire.PromptInput) (chatruntime.Input, error)

func (*Service) RecordLLMUsage 

func (h *Service) RecordLLMUsage(agentID uuid.UUID, runIDHeader string, c LlmUsageCapture) error

RecordLLMUsage writes best-effort analytics and settles task token budgets. A supplied run must resolve exactly; missing attribution cannot bypass a task budget. Budget failures are returned to the caller. An independent bounded context records incurred spend when a client disconnects.

func (*Service) RequestConnection 

func (h *Service) RequestConnection(ctx context.Context, agentID uuid.UUID, slug string, req wire.ProxyRequest) (integrationservice.ConnectionResult, error)

func (*Service) ResolveModel 

func (h *Service) ResolveModel(ctx context.Context, agentID, slug, capability string) (ResolvedModel, error)

func (*Service) RuntimeHTTP 

func (h *Service) RuntimeHTTP(ctx context.Context, scope wire.RuntimeContext, input wire.HTTPRequest) (wire.HTTPResponse, error)

func (*Service) RuntimeMedia 

func (h *Service) RuntimeMedia(ctx context.Context, scope wire.RuntimeContext, operation string, input json.RawMessage) (result tool.Result, err error)

func (*Service) RuntimeModel 

func (h *Service) RuntimeModel(ctx context.Context, agentID, runID uuid.UUID, slug, capability string) (stream.Model, error)

RuntimeModel uses the same resolution, attachment policy and ledger as the agent model proxy. Each Stream call, including compaction, records its usage.

func (*Service) RuntimeSession 

func (h *Service) RuntimeSession(agentID, conversationID, runID, ownerToken uuid.UUID, source string) session.SessionStore

RuntimeSession shares the canonical history codec with the agent session API.

func (*Service) SettleChat 

func (h *Service) SettleChat(ctx context.Context, q *dbq.Queries, runID uuid.UUID, status string) error

func (*Service) ValidateAppFiles 

func (h *Service) ValidateAppFiles(ctx context.Context, scope wire.RuntimeContext, input, schemaJSON json.RawMessage) error

type UpgradeRunner 

type UpgradeRunner interface {
	AcquireUpgradeLock(context.Context, string) error
	RunUpgrade(context.Context, builder.UpgradeInput)
}

Source Files

View all Source files

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.