inprocess

package
v1.0.6 Latest Latest
Warning

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

 Go to latest Published: Mar 21, 2026 License: MIT Imports: 40 Imported by: 0

Documentation

Overview

Package inprocess provides an in-process plugin router that dispatches tool calls, storage operations, and prompt requests via direct Go function calls instead of QUIC network round-trips. It implements the Sender interface used by StdioTransport and the StorageClient interface used by plugins.

Package inprocess — tunnel_token provides one-time registration tokens for the web-gate tunnel system. When a machine starts `orchestra serve --web-gate`, it generates a token containing machine metadata. The user copies this token to the web app to register the tunnel.

Package inprocess — webgate adds a WebSocket JSON-RPC 2.0 gateway to orchestra serve, allowing browser clients to call MCP tools directly via the in-process Router. This turns each machine running `orchestra serve` into a remotely accessible tunnel.

Unlike the QUIC bridge's wsbridge (which connects to an orchestrator over QUIC), the WebGateServer uses the Router directly — no network hop.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ClaimTunnel added in v1.0.4 

func ClaimTunnel(ctx context.Context, cloudURL, nonce string) (tunnelID, connectionToken, teamID, authToken string, err error)

ClaimTunnel polls the cloud server's claim endpoint until the user registers the tunnel in the web app. Once registered, it returns the tunnel ID and connection token needed to establish the reverse tunnel.

The nonce is the secret from the registration token that the CLI generated. The cloud server stores a nonce→credentials mapping when the user pastes the token in the web app.

func ClientTLSConfigForBridge 

func ClientTLSConfigForBridge(certsDir string) (*tls.Config, error)

ClientTLSConfigForBridge returns a TLS config suitable for the orchestra host to connect to child plugin QUIC servers. This is a convenience wrapper around the SDK's ClientTLSConfig with the correct ALPN protocol.

func FormatTokenDisplay added in v1.0.4 

func FormatTokenDisplay(raw string, token *TunnelToken) string

FormatTokenDisplay returns a terminal-friendly display of the tunnel token for the user to copy to the web app. Uses \r\n line endings and plain ASCII so it renders correctly whether the terminal is in raw mode or not.

func TunnelLog added in v1.0.4 

func TunnelLog(color int, format string, args ...any)

func VerifyAPIKeyHash added in v1.0.4 

func VerifyAPIKeyHash(apiKey, hash string) bool

VerifyAPIKeyHash checks if a plaintext API key matches the hash stored in a token.

Types

type AutoRegisterRequest added in v1.0.5 

type AutoRegisterRequest struct {
	Hostname     string `json:"hostname"`
	OS           string `json:"os"`
	Architecture string `json:"architecture"`
	LocalIP      string `json:"local_ip"`
	GateAddress  string `json:"gate_address"`
	Workspace    string `json:"workspace"`
	ToolCount    int    `json:"tool_count"`
	Version      string `json:"version"`
}

AutoRegisterRequest is the JSON body sent to POST /api/tunnels/auto-register.

type AutoRegisterResponse added in v1.0.5 

type AutoRegisterResponse struct {
	TunnelID        string `json:"tunnel_id"`
	ConnectionToken string `json:"connection_token"`
	TeamID          string `json:"team_id"`
	AuthToken       string `json:"auth_token"`
	Workspace       string `json:"workspace"`
	Reconnected     bool   `json:"reconnected"`
}

AutoRegisterResponse is the JSON response from POST /api/tunnels/auto-register.

func AutoRegisterTunnel added in v1.0.5 

func AutoRegisterTunnel(cloudURL, authToken, gateAddress, workspace string, toolCount int) (*AutoRegisterResponse, error)

AutoRegisterTunnel calls POST /api/tunnels/auto-register on the cloud server using the user's stored JWT. Returns the tunnel credentials immediately.

type CatalogEntry added in v1.0.5 

type CatalogEntry = plugin.CatalogEntry

CatalogEntry is an alias for the sdk-go plugin.CatalogEntry type.

type DualStorage added in v1.0.6 

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

DualStorage implements plugin.StorageHandler by dispatching to two backends simultaneously. The primary backend is authoritative for reads and versioning. The secondary backend receives a best-effort synchronous mirror of every write and delete. If the mirror operation fails, a warning is logged but the operation still succeeds (the primary result is returned).

Reads and list operations are served from the primary only. Mirror writes always use ExpectedVersion = -1 (upsert) to avoid CAS version conflicts, since the primary and secondary version counters diverge.

func NewDualStorage added in v1.0.6 

func NewDualStorage(primary, secondary plugin.StorageHandler) *DualStorage

NewDualStorage creates a DualStorage that reads from primary and mirrors writes to secondary.

func (*DualStorage) Delete added in v1.0.6 

func (d *DualStorage) Delete(ctx context.Context, req *pluginv1.StorageDeleteRequest) (*pluginv1.StorageDeleteResponse, error)

Delete deletes from the primary backend first. On success, it mirrors the delete to the secondary backend. A secondary failure is logged but does not fail the overall operation.

func (*DualStorage) List added in v1.0.6 

func (d *DualStorage) List(ctx context.Context, req *pluginv1.StorageListRequest) (*pluginv1.StorageListResponse, error)

List delegates entirely to the primary backend.

func (*DualStorage) Read added in v1.0.6 

func (d *DualStorage) Read(ctx context.Context, req *pluginv1.StorageReadRequest) (*pluginv1.StorageReadResponse, error)

Read delegates entirely to the primary backend.

func (*DualStorage) Write added in v1.0.6 

func (d *DualStorage) Write(ctx context.Context, req *pluginv1.StorageWriteRequest) (*pluginv1.StorageWriteResponse, error)

Write writes to the primary backend first. On success, it mirrors the write to the secondary backend using ExpectedVersion = -1 (unconditional upsert). A secondary failure is logged but does not fail the overall operation.

type EventBus added in v1.0.5 

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

EventBus is an in-memory pub/sub event dispatcher. Plugins publish events to named topics, and subscribers receive matching events via channels. Wildcard subscribers (created via SubscribeAll) receive events on all topics.

func NewEventBus added in v1.0.5 

func NewEventBus() *EventBus

NewEventBus creates a new EventBus ready for use.

func (*EventBus) Close added in v1.0.5 

func (eb *EventBus) Close()

Close removes all subscriptions and closes their channels. After Close is called, the EventBus should not be used.

func (*EventBus) Publish added in v1.0.5 

func (eb *EventBus) Publish(topic, eventType string, payload *structpb.Struct, sourcePlugin string)

Publish sends an event to all subscriptions whose topic matches or that are wildcard subscribers. Delivery is non-blocking: if a subscriber's channel is full, the event is dropped and a warning is logged.

func (*EventBus) Subscribe added in v1.0.5 

func (eb *EventBus) Subscribe(topic string) (string, <-chan *pluginv1.EventDelivery)

Subscribe creates a subscription for the given topic. It returns a unique subscription ID and a read-only channel that receives matching events. The channel is buffered (capacity 64) and the caller must consume it to avoid dropped events.

func (*EventBus) SubscribeAll added in v1.0.5 

func (eb *EventBus) SubscribeAll() (string, <-chan *pluginv1.EventDelivery)

SubscribeAll creates a wildcard subscription that receives events on all topics. It returns a unique subscription ID and a read-only channel.

func (*EventBus) Unsubscribe added in v1.0.5 

func (eb *EventBus) Unsubscribe(id string)

Unsubscribe removes a subscription by ID and closes its channel. It is safe to call with an ID that does not exist.

type ExternalPlugin 

type ExternalPlugin struct {
	ID         string
	ProvidesAI []string
	ToolDefs   []*pluginv1.ToolDefinition
	PromptDefs []*pluginv1.PromptDefinition
	Client     Sender
}

ExternalPlugin represents a QUIC-connected plugin that runs as a separate process (e.g. engine-rag written in Rust, or third-party plugins). The router forwards requests to it via its Send method.

func (*ExternalPlugin) Send 

func (ep *ExternalPlugin) Send(ctx context.Context, req *pluginv1.PluginRequest) (*pluginv1.PluginResponse, error)

Send forwards a request to the external plugin via its QUIC client.

type Metrics added in v1.0.5 

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

Metrics collects per-tool call metrics: count, errors, and latency percentiles. Thread-safe for concurrent use from the router.

func NewMetrics added in v1.0.5 

func NewMetrics() *Metrics

NewMetrics creates a new metrics collector.

func (*Metrics) Record added in v1.0.5 

func (m *Metrics) Record(toolName string, duration time.Duration, isError bool, inputSize int)

Record records a tool call with its duration, error status, and input size.

func (*Metrics) Snapshot added in v1.0.5 

func (m *Metrics) Snapshot() []ToolStats

Snapshot returns current metrics for all tools.

type QUICBridge 

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

QUICBridge is a QUIC listener that child plugin processes connect to for storage access and cross-plugin tool calls. The router proxies all requests through its in-process handlers.

type ReverseTunnelClient added in v1.0.4 

type ReverseTunnelClient struct {
	OnConnect func(ctx context.Context) // called after each successful connect
	// contains filtered or unexported fields
}

ReverseTunnelClient maintains a persistent outbound WebSocket connection from the local machine to the cloud server. It receives relay envelopes containing JSON-RPC requests from browser sessions, dispatches them through the in-process Router, and sends responses back.

func NewReverseTunnelClient added in v1.0.4 

func NewReverseTunnelClient(cloudURL, tunnelID, connectionToken string, router *Router) *ReverseTunnelClient

NewReverseTunnelClient creates a new reverse tunnel client.

func (*ReverseTunnelClient) ReconnectLoop added in v1.0.4 

func (rt *ReverseTunnelClient) ReconnectLoop(ctx context.Context)

ReconnectLoop connects to the cloud server and automatically reconnects with exponential backoff on failure. Blocks until ctx is cancelled.

type Router 

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

Router dispatches PluginRequests to in-process tool handlers, storage handlers, and prompt handlers. It replaces the QUIC-based orchestrator router for local IDE use. External plugins (e.g. engine-rag) are supported via ExternalPlugin entries that forward requests over QUIC.

Router implements the Sender interface: 

Send(ctx, *PluginRequest) (*PluginResponse, error)

This means it can be passed directly to StdioTransport and also used as the clientAdapter for plugins that need cross-plugin storage calls.

func NewRouter 

func NewRouter() *Router

NewRouter creates a new in-process Router.

func (*Router) CatalogCount added in v1.0.5 

func (r *Router) CatalogCount(pluginFilter string) int

CatalogCount returns the total number of catalog entries (optionally filtered by plugin).

func (*Router) EventBus added in v1.0.5 

func (r *Router) EventBus() *EventBus

EventBus returns the router's event bus for subscribing to events.

func (*Router) GetCatalogEntry added in v1.0.5 

func (r *Router) GetCatalogEntry(toolName string) *CatalogEntry

GetCatalogEntry returns a single tool's CatalogEntry by name, or nil if not found.

func (*Router) GetMetrics added in v1.0.5 

func (r *Router) GetMetrics() *Metrics

GetMetrics returns the metrics collector for the router.

func (*Router) GetStreamHandler added in v1.0.2 

func (r *Router) GetStreamHandler(toolName string) (plugin.StreamingToolHandler, bool)

GetStreamHandler returns the streaming tool handler for the given tool name.

func (*Router) HealthCheck added in v1.0.5 

func (r *Router) HealthCheck() map[string]any

HealthCheck verifies that the router has storage and at least one tool registered.

func (*Router) ListCatalog added in v1.0.5 

func (r *Router) ListCatalog(pluginFilter string, offset, limit int) []CatalogEntry

ListCatalog returns all registered tools as CatalogEntry values, sorted by name. If pluginFilter is non-empty, only tools from that plugin are returned.

func (*Router) ListPluginIDs added in v1.0.5 

func (r *Router) ListPluginIDs() []string

ListPluginIDs returns the unique set of plugin IDs that have registered tools.

func (*Router) ListToolNames 

func (r *Router) ListToolNames() []string

ListToolNames returns a list of all registered tool names (for logging).

func (*Router) ListenAndServeQUIC 

func (r *Router) ListenAndServeQUIC(ctx context.Context, addr string, certsDir string) (string, error)

ListenAndServeQUIC starts a QUIC listener using mTLS certificates from the given certsDir. Child plugins connect to this address (passed as --orchestrator-addr) to make storage and cross-plugin RPC calls.

Returns the actual bound address (e.g. "127.0.0.1:56789") which should be passed to spawnPlugin when starting child processes.

func (*Router) OnToolsChanged added in v1.0.5 

func (r *Router) OnToolsChanged(fn func())

OnToolsChanged registers a callback invoked when tools are added or removed. Used by transports to send notifications/tools/list_changed to clients.

func (*Router) RegisterExternal 

func (r *Router) RegisterExternal(ep *ExternalPlugin)

RegisterExternal adds an external QUIC-connected plugin to the router. External plugins override in-process handlers for same-named tools, allowing installable plugins to supersede bundled core tool definitions.

func (*Router) RegisterPlugin 

func (r *Router) RegisterPlugin(ep *plugin.ExportedPlugin)

RegisterPlugin registers all tools, streaming tools, and prompts from an ExportedPlugin. If the plugin's manifest declares ProvidesAI, tools are indexed under the AI routing table instead of the generic tool table.

func (*Router) SearchCatalog added in v1.0.5 

func (r *Router) SearchCatalog(query string) []CatalogEntry

SearchCatalog searches tool names and descriptions for the query string. Results are sorted: exact name matches first, then name-contains, then description-contains.

func (*Router) Send 

func (r *Router) Send(ctx context.Context, req *pluginv1.PluginRequest) (*pluginv1.PluginResponse, error)

Send implements the Sender interface. It dispatches a PluginRequest to the appropriate in-process handler based on the request type.

func (*Router) SetStorageHandler 

func (r *Router) SetStorageHandler(h plugin.StorageHandler)

SetStorageHandler sets the storage backend. Pass a DualStorage for dual-write (both Markdown and SQLite simultaneously).

type Sender 

type Sender interface {
	Send(ctx context.Context, req *pluginv1.PluginRequest) (*pluginv1.PluginResponse, error)
}

Sender abstracts the QUIC client for external plugins.

type TCPSender added in v1.0.4 

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

TCPSender implements the Sender interface by forwarding Protobuf requests to an existing orchestra instance over a persistent TCP connection. This is used by proxy mode: when a new `orchestra serve` is spawned and finds a healthy instance, it creates a TCPSender connected to that instance and passes it to StdioTransport, bridging stdin/stdout ↔ TCP.

If the connection drops (e.g. orchestrator restart), Send automatically reconnects with exponential backoff and retries the request. During reconnect, the info file is re-read in case the primary restarted on a different port. A background health check detects primary death and exits the proxy process so the IDE can restart it as a new primary instance.

func NewTCPSender added in v1.0.4 

func NewTCPSender(addr, infoFile string) (*TCPSender, error)

NewTCPSender connects to an existing orchestra instance's TCP server. infoFile is the path to the instance's .info JSON file — it's re-read during reconnection in case the primary restarted on a different port. It starts a background health check that exits the process if the primary instance becomes permanently unreachable.

func (*TCPSender) Close added in v1.0.4 

func (s *TCPSender) Close() error

Close closes the TCP connection and stops the health check.

func (*TCPSender) Send added in v1.0.4 

func (s *TCPSender) Send(ctx context.Context, req *pluginv1.PluginRequest) (*pluginv1.PluginResponse, error)

Send forwards a PluginRequest to the remote instance and reads the response. It implements the Sender interface used by StdioTransport.

On connection errors (broken pipe, connection reset, EOF), it automatically reconnects with exponential backoff and retries the request.

type TCPServer 

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

TCPServer listens for desktop app connections (Swift, Windows, Linux) and proxy connections from other orchestra serve instances. Uses the same length-delimited Protobuf protocol as the orchestrator's TCP bridge. Each TCP connection is persistent — multiple requests can be sent over a single connection (the connection stays open until the client disconnects).

func NewTCPServer 

func NewTCPServer(addr string, router *Router) *TCPServer

NewTCPServer creates a TCP server bound to the given address.

func (*TCPServer) Addr 

func (s *TCPServer) Addr() string

Addr returns the actual listening address. Only valid after Listen.

func (*TCPServer) Listen added in v1.0.5 

func (s *TCPServer) Listen() error

Listen binds the TCP socket without accepting connections. Call Serve to start processing. Splitting listen from serve lets the caller detect port conflicts and fall back before committing.

func (*TCPServer) ListenAndServe 

func (s *TCPServer) ListenAndServe(ctx context.Context) error

ListenAndServe binds and serves in one call (convenience wrapper).

func (*TCPServer) Serve added in v1.0.5 

func (s *TCPServer) Serve(ctx context.Context) error

Serve accepts connections on the already-bound listener until the context is cancelled. Must call Listen first.

type TerminalManager added in v1.0.5 

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

TerminalManager manages PTY terminal sessions for remote access via MCP tools. Sessions can be created directly (Go PTY) or relayed from the desktop Flutter app.

func NewTerminalManager added in v1.0.5 

func NewTerminalManager() *TerminalManager

NewTerminalManager creates a new TerminalManager.

func (*TerminalManager) RegisterTools added in v1.0.5 

func (tm *TerminalManager) RegisterTools(builder *plugin.PluginBuilder)

RegisterTools registers all 6 terminal PTY tools with the plugin builder.

type ToolCategory added in v1.0.5 

type ToolCategory = string

ToolCategory classifies how a tool is registered in the router. 

const (
	ToolCategoryGeneric  ToolCategory = "tool"
	ToolCategoryAI       ToolCategory = "ai"
	ToolCategoryStream   ToolCategory = "stream"
	ToolCategoryExternal ToolCategory = "external"
)

type ToolStats added in v1.0.5 

type ToolStats struct {
	Name      string  `json:"name"`
	Calls     int64   `json:"calls"`
	Errors    int64   `json:"errors"`
	ErrorRate float64 `json:"error_rate"`
	P50Ms     float64 `json:"p50_ms"`
	P95Ms     float64 `json:"p95_ms"`
	P99Ms     float64 `json:"p99_ms"`
}

ToolStats holds aggregated statistics for a single tool.

type TunnelToken added in v1.0.4 

type TunnelToken struct {
	// Machine identification
	Hostname string `json:"hostname"`
	OS       string `json:"os"`
	Arch     string `json:"arch"`

	// Network
	LocalIP     string `json:"local_ip"`
	GateAddress string `json:"gate_address"`
	CloudURL    string `json:"cloud_url,omitempty"` // cloud server URL for reverse tunnel

	// Security
	APIKeyHash string `json:"api_key_hash,omitempty"` // SHA-256 of the API key (empty if no auth)
	Nonce      string `json:"nonce"`                  // random nonce for uniqueness

	// Metadata
	ToolCount int    `json:"tool_count"`
	CreatedAt string `json:"created_at"`          // ISO 8601
	Workspace string `json:"workspace,omitempty"` // absolute workspace path (for multi-workspace tunnels)
}

TunnelToken contains the machine metadata embedded in a registration token. The web app decodes this to verify the tunnel and store connection info.

func DecodeToken added in v1.0.4 

func DecodeToken(raw string) (*TunnelToken, error)

DecodeToken decodes a raw base64url token string into a TunnelToken. This is used by the web backend to inspect the token contents.

type TunnelTokenManager added in v1.0.4 

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

TunnelTokenManager handles token generation, storage, and one-time verification.

func NewTunnelTokenManager added in v1.0.4 

func NewTunnelTokenManager() *TunnelTokenManager

NewTunnelTokenManager creates a new token manager.

func (*TunnelTokenManager) GenerateToken added in v1.0.4 

func (m *TunnelTokenManager) GenerateToken(gateAddress, apiKey, cloudURL, workspace string, toolCount int) (string, error)

GenerateToken creates a new registration token with machine metadata. Only one token is active at a time — generating a new one invalidates the previous.

func (*TunnelTokenManager) HasPendingToken added in v1.0.4 

func (m *TunnelTokenManager) HasPendingToken() bool

HasPendingToken returns true if there is an unconsumed registration token.

func (*TunnelTokenManager) RevokeToken added in v1.0.4 

func (m *TunnelTokenManager) RevokeToken()

RevokeToken invalidates any pending token without consuming it.

func (*TunnelTokenManager) VerifyToken added in v1.0.4 

func (m *TunnelTokenManager) VerifyToken(rawToken string) (*TunnelToken, error)

VerifyToken checks if the provided token matches the current pending token. On success, the token is consumed (one-time use) and the decoded token is returned. Returns an error if no token is pending, or the token doesn't match.

type WebGateServer added in v1.0.4 

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

WebGateServer exposes the in-process Router over WebSocket for remote browser clients. Each WebSocket connection is a persistent session that can send multiple JSON-RPC 2.0 requests over the same connection.

func NewWebGateServer added in v1.0.4 

func NewWebGateServer(router *Router, apiKey, cloudURL, workspace string, corsOrigins []string) *WebGateServer

NewWebGateServer creates a WebGateServer. If apiKey is empty, authentication is disabled. corsOrigins controls which browser origins can connect (empty = allow all).

func (*WebGateServer) Addr added in v1.0.4 

func (wg *WebGateServer) Addr() string

Addr returns the actual listening address. Only valid after ListenAndServe.

func (*WebGateServer) BroadcastToolsListChanged added in v1.0.5 

func (wg *WebGateServer) BroadcastToolsListChanged()

BroadcastToolsListChanged sends a notifications/tools/list_changed JSON-RPC notification to all connected WebSocket clients.

func (*WebGateServer) GenerateRegistrationToken added in v1.0.4 

func (wg *WebGateServer) GenerateRegistrationToken() (string, *TunnelToken, error)

GenerateRegistrationToken creates a new tunnel registration token using the current gate address and tool count. Returns the raw base64 token string.

func (*WebGateServer) ListenAndServe added in v1.0.4 

func (wg *WebGateServer) ListenAndServe(ctx context.Context, addr string) error

ListenAndServe starts the HTTP server with WebSocket upgrade support. It blocks until ctx is cancelled or a fatal error occurs.

func (*WebGateServer) SetServerInfo added in v1.0.6 

func (wg *WebGateServer) SetServerInfo(info protocol.MCPServerInfo)

SetServerInfo sets the server name and version for MCP initialize responses.

func (*WebGateServer) StartDataChangeBroadcaster added in v1.0.6 

func (wg *WebGateServer) StartDataChangeBroadcaster(ctx context.Context)

StartDataChangeBroadcaster subscribes to all EventBus topics and broadcasts notifications/data JSON-RPC messages to all connected WebSocket clients whenever data changes (storage writes/deletes or mutating tool calls).

func (*WebGateServer) StartEventPoller added in v1.0.4 

func (wg *WebGateServer) StartEventPoller(ctx context.Context)

StartEventPoller starts a background goroutine that polls drain_session_events every 200ms and pushes any pending events to all connected WebSocket clients as server-initiated notifications. This enables real-time streaming of tool cards and text in the web copilot.

func (*WebGateServer) StartPermissionPoller added in v1.0.4 

func (wg *WebGateServer) StartPermissionPoller(ctx context.Context)

StartPermissionPoller starts a background goroutine that polls get_pending_permission every second and pushes any pending requests to all connected WebSocket clients as server-initiated notifications. This bridges the gap where bridge-claude (external QUIC plugin) captures permission events from Claude CLI processes but has no way to push them to the browser.

func (*WebGateServer) TokenManager added in v1.0.4 

func (wg *WebGateServer) TokenManager() *TunnelTokenManager

TokenManager returns the tunnel token manager for external use (e.g. serve.go generates a token after startup and displays it in the terminal).

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.