Help us shape the future of Go! Take the 2026 Go Developer Survey

runtime

package
v0.2.4 Latest Latest
Warning

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

Go to latest
Published: Aug 9, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Overview

Package runtime is the shared, cross-cutting machinery a generated tool-pack uses at call time: HTTP request assembly, auth injection, retries/backoff, pagination hints, jq-style response trimming, and a secret-redaction hook. Generators emit declarative metadata (the IR); this package executes it, so the runtime is shared rather than regenerated per API.

Index

Constants

View Source
const (
	DefaultStreamIdleTTL     = 5 * time.Minute  // cancelled when not polled for this long
	DefaultStreamMaxLifetime = 30 * time.Minute // cancelled this long after start, regardless
	DefaultStreamMaxSessions = 8                // live sessions per manager
	DefaultStreamMaxFrames   = 256              // ring capacity, in frames
	DefaultStreamMaxBytes    = 4 << 20          // ring capacity, in bytes (matches Options.MaxBody default)
	DefaultPollMaxFrames     = 50               // frames returned per poll
	DefaultPollMaxBytes      = 256 << 10        // bytes returned per poll

)

Stream lifecycle defaults.

View Source
const (
	AuthNone              = ""                          // no token flow (static apiKey/basic/bearer handled by InjectAuth)
	AuthStatic            = "static"                    // read token(s) from env; no login, still refreshable
	AuthJWTLogin          = "jwt_login"                 // custom JSON login -> {access,refresh,expires_in}
	AuthOAuth2Password    = "oauth2_password"           // OAuth2 Resource Owner Password Credentials
	AuthOAuth2ClientCreds = "oauth2_client_credentials" // OAuth2 Client Credentials
)

Auth grant types accepted by AuthConfig.Type.

Variables

This section is empty.

Functions

func ApplyJQ

func ApplyJQ(data []byte, filter string) (string, error)

ApplyJQ trims a JSON document using a deliberately small subset of jq path syntax — enough for the common "trim a large response before it enters the model's context" case, without embedding a full jq engine:

.            identity (whole document)
.field       object field access
.a.b.c       nested field access
.arr[0]      array index
.arr[]       iterate an array (maps the rest of the path over each element)
.data[].id   project one field from each element of an array

It returns the selected value rendered as indented JSON.

func DeriveWSURL

func DeriveWSURL(httpURL string) string

DeriveWSURL maps an HTTP(S) URL to its WS(S) twin. Non-http schemes are returned unchanged (already ws://, or malformed — the dial will say so).

func InjectAuth

func InjectAuth(req *http.Request, schemes []ir.AuthScheme, cred Credential)

InjectAuth applies the API's auth schemes to an HTTP request with the given credentials. Shared by the REST and GraphQL executors. apiKey schemes set a header or query param; http bearer/basic and oauth2 set Authorization.

func RenderPoll

func RenderPoll(id string, res *PollResult) string

RenderPoll formats a PollResult the way the session tools present it — one frame per block, with status/drop/pending annotations — so gRPC streams and GraphQL subscriptions read identically to the model.

func WithBearer

func WithBearer(ctx context.Context, token string) context.Context

WithBearer returns a context that overrides the bearer token for a single executor call. Used by the owner to run a delegated invocation under the subscriber's identity without persisting their token.

Types

type AuthConfig

type AuthConfig struct {
	Type        string `json:"type" yaml:"type"`                                     // see Auth* constants
	TokenURL    string `json:"token_url,omitempty" yaml:"token_url,omitempty"`       // login / token endpoint
	RefreshURL  string `json:"refresh_url,omitempty" yaml:"refresh_url,omitempty"`   // defaults to TokenURL
	ContentType string `json:"content_type,omitempty" yaml:"content_type,omitempty"` // "form" | "json"

	// Secrets are ENV-VAR NAMES only — resolved from the environment at construction.
	UsernameEnv     string `json:"username_env,omitempty" yaml:"username_env,omitempty"`
	PasswordEnv     string `json:"password_env,omitempty" yaml:"password_env,omitempty"`
	ClientIDEnv     string `json:"client_id_env,omitempty" yaml:"client_id_env,omitempty"`
	ClientSecretEnv string `json:"client_secret_env,omitempty" yaml:"client_secret_env,omitempty"`
	AccessTokenEnv  string `json:"access_token_env,omitempty" yaml:"access_token_env,omitempty"`   // static
	RefreshTokenEnv string `json:"refresh_token_env,omitempty" yaml:"refresh_token_env,omitempty"` // static (optional)

	Scope string `json:"scope,omitempty" yaml:"scope,omitempty"`

	// Response field paths (dotted) for jwt_login / non-standard token responses.
	AccessTokenField  string `json:"access_token_field,omitempty" yaml:"access_token_field,omitempty"`
	RefreshTokenField string `json:"refresh_token_field,omitempty" yaml:"refresh_token_field,omitempty"`
	ExpiresInField    string `json:"expires_in_field,omitempty" yaml:"expires_in_field,omitempty"`

	// Injection target. Defaults: header "Authorization", scheme "Bearer".
	Header string `json:"header,omitempty" yaml:"header,omitempty"`
	Scheme string `json:"scheme,omitempty" yaml:"scheme,omitempty"`

	ExpiryMargin string `json:"expiry_margin,omitempty" yaml:"expiry_margin,omitempty"` // e.g. "30s"
}

AuthConfig declares how an integration (owner, Model A) or a subscription (subscriber, Model B) acquires and refreshes a bearer token. It is persisted in the integration store and the YAML config, so it holds ONLY env-var names and URLs — never secret values.

func (AuthConfig) IsTokenFlow

func (c AuthConfig) IsTokenFlow() bool

IsTokenFlow reports whether this config drives a TokenManager (a login/refresh flow) rather than the static apiKey/basic/fixed-bearer path handled by InjectAuth. An empty/unknown type, "apikey", "basic" or "bearer" return false.

type Credential

type Credential struct {
	APIKey   string // apiKey value, or bearer token
	Username string // http basic
	Password string // http basic
}

Credential carries resolved auth material for an integration. Which fields are used depends on the matched AuthScheme.

type GraphQLExecutor

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

GraphQLExecutor executes GraphQL operations against a single endpoint. Unlike the REST HTTPExecutor (which maps one operation per tool), GraphQL is exposed to agents through a small set of generic tools (query + schema): the agent writes the GraphQL query and this executor POSTs {query, variables} to the endpoint, applying the same auth, retries, jq-filtering, and redaction.

func NewGraphQLExecutor

func NewGraphQLExecutor(endpoint string, auth []ir.AuthScheme, cred Credential, opts Options) *GraphQLExecutor

NewGraphQLExecutor builds a GraphQL executor for one endpoint.

func (*GraphQLExecutor) Endpoint

func (e *GraphQLExecutor) Endpoint() string

Endpoint returns the configured GraphQL endpoint URL.

func (*GraphQLExecutor) Query

func (e *GraphQLExecutor) Query(ctx context.Context, query string, variables map[string]any, jqFilter string) (string, error)

Query executes a GraphQL document with optional variables, returning the (optionally jq-filtered, optionally redacted) JSON response. A GraphQL "errors" array in a 200 response is surfaced as an error.

func (*GraphQLExecutor) SetWSEndpoint

func (e *GraphQLExecutor) SetWSEndpoint(u string)

SetWSEndpoint overrides the derived websocket URL for subscriptions.

func (*GraphQLExecutor) Subscribe

func (e *GraphQLExecutor) Subscribe(sm *StreamManager, query string, variables map[string]any) (string, error)

Subscribe starts a GraphQL subscription as a streaming session (design 015 §2.1): it dials the websocket, performs the connection_init/ack handshake, sends one subscribe, and pumps "next" payloads into the session's ring buffer until complete / error / cancellation. Frames are redacted with Options.Redact before entering the ring. The returned session id is what the model polls/stops with.

func (*GraphQLExecutor) WSEndpoint

func (e *GraphQLExecutor) WSEndpoint() string

WSEndpoint returns the websocket URL subscriptions will dial: the explicit override when set, else the HTTP endpoint with its scheme swapped per convention (http→ws, https→wss).

type HTTPExecutor

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

HTTPExecutor performs REST calls described by IR operations.

func NewHTTPExecutor

func NewHTTPExecutor(baseURL string, auth []ir.AuthScheme, cred Credential, opts Options) *HTTPExecutor

NewHTTPExecutor builds an executor for one API.

func (*HTTPExecutor) Call

func (e *HTTPExecutor) Call(ctx context.Context, op *ir.Operation, args map[string]any, jqFilter string) (string, error)

Call executes op with a flat argument object (path/query/header params plus body properties, as produced by Operation.ToolInputSchema). jqFilter, when non-empty, trims the JSON response before it is returned (and before it would enter a model's context). The returned string is the (optionally filtered, optionally redacted) response body.

func (*HTTPExecutor) StartServerStream

func (e *HTTPExecutor) StartServerStream(sm *StreamManager, op *ir.Operation, args map[string]any) (string, error)

StartServerStream opens a gRPC server-streaming call through a grpc-gateway-style JSON-over-HTTP bridge and pumps its frames into a streaming session (design 015 §2.2).

Wire contract: the request message is POSTed as JSON to the method's path; the response is newline-delimited JSON where each line is either

{"result": <response message>}    — one stream message
{"error":  {"message": …, …}}     — a terminal error

(grpc-gateway's streaming envelope). Some bridges emit bare message objects per line; those are passed through as-is. Connect's enveloped binary framing is NOT supported — this consumer only speaks NDJSON.

The HTTP client used here deliberately has NO overall timeout (a stream is long-lived by design); cancellation comes from the session context — Stop, idle expiry, max lifetime, or manager CloseAll. Each frame is capped at Options.MaxBody and redacted with Options.Redact before it enters the ring.

type Options

type Options struct {
	MaxRetries   int                 // retry count for 5xx/429 (default 2)
	Timeout      time.Duration       // per-request timeout (default 30s)
	Redact       func(string) string // applied to response text before return
	MaxBody      int64               // response read cap (default 4 MiB)
	ExtraHeaders map[string]string   // always-applied headers

	// Token auth (forged-API auth plan, Phase A). When TokenProvider is set, the
	// executor fetches a fresh bearer before each request and injects it
	// (superseding the static Credential bearer). On a 401, it calls
	// TokenRefresher once and retries the request a single extra time. Both come
	// from a runtime.TokenManager (TokenProvider=tm.Token, TokenRefresher=tm.Refresh).
	TokenProvider  func(context.Context) (string, error)
	TokenRefresher func(context.Context) (string, error)
	AuthHeader     string // injection header (default "Authorization")
	AuthScheme     string // injection scheme (default "Bearer"); "" ⇒ raw token
}

Options configure an HTTPExecutor.

type PollResult

type PollResult struct {
	Frames  []string
	Dropped int    // frames lost to ring eviction since the previous poll
	Pending int    // frames still buffered after this poll (poll again now)
	Done    bool   // the stream has ended (after draining, stop the session)
	Reason  string // why it ended ("" while live)
}

PollResult is what one Poll call returns.

type StreamInfo

type StreamInfo struct {
	ID        string
	Kind      string
	Label     string
	StartedAt time.Time
	Frames    uint64 // total frames received so far
	Pending   uint64 // frames buffered but not yet polled
	Done      bool
	EndReason string
}

StreamInfo is a snapshot of one session for listings.

type StreamManager

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

StreamManager owns the streaming sessions of one integration instance.

func NewStreamManager

func NewStreamManager(opts StreamOptions) *StreamManager

NewStreamManager builds a manager and (unless opts.NoSweeper) starts its background sweeper, which enforces idle/lifetime expiry every 30s.

func (*StreamManager) CloseAll

func (m *StreamManager) CloseAll()

CloseAll cancels every session and stops the sweeper. Called when the owning integration is disabled / its scope is torn down / the workspace shuts down.

func (*StreamManager) List

func (m *StreamManager) List() []StreamInfo

List returns a snapshot of all sessions, oldest first.

func (*StreamManager) Poll

func (m *StreamManager) Poll(id string, maxFrames int) (*PollResult, error)

Poll returns the frames received since the previous poll, bounded to maxFrames (default DefaultPollMaxFrames) and DefaultPollMaxBytes per call. It refreshes the session's idle timer.

func (*StreamManager) Start

func (m *StreamManager) Start(kind, label string, run func(ctx context.Context, push func(frame string)) error) (string, error)

Start registers a new session and runs `run` on its own goroutine. `run` receives the session context (cancelled by Stop / expiry / CloseAll) and a push callback for frames; when it returns, the session is marked done with the returned error (nil ⇒ "stream ended"). The returned id is what the model polls/stops with.

func (*StreamManager) Stop

func (m *StreamManager) Stop(id string) (*PollResult, error)

Stop cancels a session's underlying request and removes it, returning any frames that had not been polled yet (bounded like a poll).

func (*StreamManager) Sweep

func (m *StreamManager) Sweep()

Sweep enforces expiry: live sessions past their idle TTL or max lifetime are cancelled (kept, with the reason, for one more idle period so a poll can report why); done sessions idle past the TTL are removed.

type StreamOptions

type StreamOptions struct {
	IdleTTL       time.Duration
	MaxLifetime   time.Duration
	MaxSessions   int
	MaxFrames     int   // ring capacity in frames
	MaxBytes      int64 // ring capacity in bytes
	MaxFrameBytes int64 // per-frame cap; oversize frames are truncated with a marker (default: executor MaxBody default)
	Now           func() time.Time
	// NoSweeper disables the background sweeper goroutine (tests drive expiry
	// via Sweep with a fake clock instead).
	NoSweeper bool
}

StreamOptions configure a StreamManager. Zero values take the defaults above.

type StreamSession

type StreamSession struct {
	ID    string
	Kind  string // "grpc-stream" | "graphql-subscription"
	Label string // method id / subscription summary, for listings
	// contains filtered or unexported fields
}

StreamSession is one live (or recently finished) stream.

type TokenManager

type TokenManager interface {
	// Token returns a valid bearer, refreshing proactively if it is at/near expiry.
	Token(ctx context.Context) (string, error)
	// Refresh forces a refresh (used reactively on a 401) and returns the new bearer.
	Refresh(ctx context.Context) (string, error)
}

TokenManager returns valid bearer tokens, refreshing as needed. Both methods are safe for concurrent use; refresh is single-flight.

func NewTokenManager

func NewTokenManager(cfg AuthConfig, client *http.Client) (TokenManager, error)

NewTokenManager builds a TokenManager from cfg, or returns (nil, nil) when cfg does not describe a token flow (the caller then relies on InjectAuth's static path). client may be nil (a default client with a sane timeout is used).

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL