host

package
v0.12.0 Latest Latest
Warning

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

 Go to latest Published: Aug 6, 2026 License: MIT Imports: 40 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	EnvGatewayURL   = "NEUBLOX_GATEWAY_URL"
	EnvGatewayToken = "NEUBLOX_GATEWAY_TOKEN"
)

The pair the app hands the daemon, from the Neublox account token. These are PRODUCT credentials on a SHARED paid account — they must never reach a vendor CLI child, whose agent has a shell and is prompt-injectable.

View Source 
const (
	EnvRoutePolicy  = "HERRSCHER_ROUTE_POLICY"
	EnvDefaultModel = "HERRSCHER_DEFAULT_MODEL"
)

EnvRoutePolicy selects which models are offered; EnvDefaultModel names the one a session gets when it creates without --model. Both are written by `herrscher init` and read here, so the wizard and the daemon cannot drift.

View Source 
const GatewayRetryWindow = 2 * time.Minute

GatewayRetryWindow bounds how long the daemon waits at startup for a gateway that did not build. Long enough for a machine that has just booted to finish bringing its network up, short enough that a stack whose credential is simply wrong still gets on with serving the gateways that do work.

Variables

This section is empty.

Functions

func ApplyOrchestratorScope 

func ApplyOrchestratorScope(cfg *contracts.PluginConfig, session, project, agent, extractor, journal string, consolidateEvery int)

ApplyOrchestratorScope threads a session's runtime scope into an orchestrator plugin's config bag. It is the single source of truth for these Settings keys, shared by the live bridge (bridge.go) and the one-shot seed so the two paths cannot drift when a scope key is added or renamed. Empty optional values are omitted so a plain/unconfigured run's config stays byte-for-byte unchanged.

func BuildBackendFor added in v0.3.4 

func BuildBackendFor(ctx context.Context, req BackendRequest) (contracts.Backend, error)

BuildBackendFor selects and constructs a backend. A remote resolver backend wins when configured; otherwise the matching registered plugin is built with the invocation, kind, working directory — and, if a ModelID is supplied, the environment variables its route requires.

func BuildFirstMemory 

func BuildFirstMemory(ctx context.Context) (contracts.Memory, error)

BuildFirstMemory builds the first registered memory plugin without its manifest — the form the CLI verbs need. The caller closes the memory.

func BuildFirstMemoryWithManifest 

func BuildFirstMemoryWithManifest(ctx context.Context) (contracts.Memory, contracts.Manifest, error)

BuildFirstMemoryWithManifest builds the first registered memory plugin from its resolved config, returning it with its manifest for announcement. It is the single source both the CLI verbs (via BuildFirstMemory) and the plugin-host (via firstMemory) build memory through.

func CaptureGatewayCreds added in v0.3.4 

func CaptureGatewayCreds()

CaptureGatewayCreds reads the gateway pair from the process environment, stores it in-process, and UNSETS both variables.

Without this the pair propagates daemon → bridge → vendor CLI unconditionally on EVERY route, because the backends spawn with MergeEnv(os.Environ(), env). A coding agent inside any session could then read the product's shared paid credential out of its own environment, from a session never routed to the gateway at all.

Idempotent: a second call keeps the first capture, so a process that captures at more than one entry point (main + runBridge) does not lose the value.

func CommandSocketPath 

func CommandSocketPath(instanceID string) string

CommandSocketPath is the daemon-level operator command socket.

func DaemonDispatch added in v0.11.0 

func DaemonDispatch(ctx context.Context, instanceID string, argv []string) (string, error)

DaemonDispatch runs one operator command inside the running daemon and returns its output. It reports an error when no daemon is listening, unlike the operator CLI's own path, which falls back to running the command locally: a client that meant to speak to the daemon must hear that it could not, rather than silently acting on a second copy of the state.

func DefaultAgentsRoot 

func DefaultAgentsRoot() string

DefaultAgentsRoot returns the directory holding agent homes, derived the same way the daemon derives it (the "agents" dir beside the state file), so the separate bridge process resolves the identical roster the coordinator sees.

func DefaultStatePath 

func DefaultStatePath() string

DefaultStatePath returns the default path to the daemon state file.

func EventsSocketPath 

func EventsSocketPath(instanceID string) string

EventsSocketPath is the daemon-level per-session events fan-out socket: a sibling of the command socket (herrscher-command → herrscher-events). It is the path Neublox's HerrscherEventSource connects to, derived there the same way.

func GatewayEnvPairs added in v0.3.4 

func GatewayEnvPairs() []string

GatewayEnvPairs returns the captured pair as KEY=VALUE entries, for the environment of a TRUSTED child only — `herrscher bridge`, which is this same binary and captures-and-unsets at its own startup before building a backend. Never for a vendor CLI, and never on argv: /proc/<pid>/cmdline is world readable. Empty when nothing was captured.

func Interrupt 

func Interrupt(session string) bool

Interrupt cancels the in-flight turn of the named session, returning false when no live session by that name is driving (mirror of Pick).

func LockState added in v0.10.0 

func LockState(statePath string) (func(), error)

LockState claims the right to serve for one state file and returns the release.

Two daemons sharing a state file is not a smaller version of one daemon: both connect to the same gateways with the same credentials, so every message is delivered and answered twice, and both supervise the same sessions down to the same control socket, where the second bridge overwrites the first. The symptom is a bot that replies twice and bills twice, which reads as a routing bug and is not one — so the second daemon is refused here, at the one place that can still explain why.

The lock is advisory and held by the process, not the file: it is released by the kernel when the daemon exits, however it exits, so a crash never leaves a lock nobody can clear. The pid is written for the error message alone; nothing reads it to make a decision.

func Logger 

func Logger(verbose bool) *slog.Logger

Logger builds the operator logger for composition roots outside core (the root `main` package can't reach core/internal/obs directly). The level follows the -v flag and HERRSCHER_LOG, matching the daemon's own logger.

func ModelsCommands added in v0.3.4 

func ModelsCommands() []contracts.Cmd

ModelsCommands returns the model catalog verbs.

func NewRegistry 

func NewRegistry(ctx context.Context, d Deps, o Options) (*cli.Registry, error)

NewRegistry builds the operator CLI registry: it loads its own state + supervisor (the operator invocation is a short-lived process) and registers the session/service handler's commands. The returned registry dispatches argv (see core/cli).

func NewRoster 

func NewRoster(root string) contracts.RosterProvider

NewRoster builds a RosterProvider over the agent homes under root.

func Pick 

func Pick(session, value string) bool

Pick routes a select-menu value to the named session's driver, returning false when no live session by that name is driving.

func ResolvePolicy added in v0.3.4 

func ResolvePolicy(getenv func(string) string) contracts.RoutePolicy

ResolvePolicy reads the route policy from the environment. Missing or unrecognized, it falls back to PolicyAll — the behavior that predates this change. This setting is set by the app when it launches the daemon, not by the user: hardening it into a fatal error would break a daemon started by hand.

func RunHub 

func RunHub(ctx context.Context, gws []Deps, o Options) error

RunHub is the always-on multi-gateway daemon: it supervises one pure-runner bridge per persisted session, drives each session's turns over a control Acceptor (fanning events out to every bound gateway), and serves health/liveness. gws are the gateway sets the daemon owns (built from the registry by the caller). Command dispatch no longer runs here — session/service commands run through the operator CLI (see NewRegistry).

func RunSession 

func RunSession(ctx context.Context, name, channel string, gws []contracts.GatewaySet, acc *control.Acceptor, participants string, m *metrics.Registry, coord contracts.Coordinator, persistResume func(string), record func(state.TranscriptEntry), gate budgetGate)

func Seed 

func Seed(session, task string) bool

Seed routes an opening task to the named session's driver, returning false when no live session by that name is driving (mirror of Pick).

func ServingPID added in v0.11.0 

func ServingPID(statePath string) (int, bool)

ServingPID reports the pid of the daemon currently serving statePath, and false when nobody is. It answers the question the lock error raises — "which process has it, and is it still alive?" — without claiming the lock: the check takes the lock only long enough to find out it was free, then drops it.

func StateLockPath added in v0.10.0 

func StateLockPath(statePath string) string

StateLockPath is the lock a daemon holds for the whole time it serves. It sits beside the state file rather than being the state file, so a stale lock can be reasoned about without risking the state itself.

func Submit added in v0.3.2 

func Submit(session string, in contracts.Inbound) bool

Submit injects one inbound message into the named session's turn queue, returning false when no live session by that name is driving (mirror of Pick). It is the push counterpart of the driver's own poll loop: a gateway that receives messages by push calls this instead of being polled.

func SubscribeDaemonEvents added in v0.11.0 

func SubscribeDaemonEvents(ctx context.Context, instanceID string) (<-chan DaemonEvent, error)

SubscribeDaemonEvents dials the daemon's events socket and streams what it publishes until ctx ends or the daemon goes away. The channel is closed on either, so a frontend can tell "the daemon stopped" from "I stopped".

The daemon drops a subscriber's lines rather than block a turn on it, so a consumer that falls far enough behind loses telemetry — read promptly.

func SupportedRemoteCategory 

func SupportedRemoteCategory(c contracts.Category) bool

SupportedRemoteCategory reports whether c can be hosted or resolved remotely. An unsupported category stays in-process (the host warns and skips it).

Types

type BackendRequest added in v0.3.4 

type BackendRequest struct {
	Vendor  string
	Cmd     string
	Kind    string
	Dir     string
	Resume  string
	ModelID string // empty = session predates the catalog, legacy path
}

BackendRequest is everything building a backend needs. It used to be six positional parameters; routing would have added a seventh, which made call sites unreadable.

type CatalogEntry added in v0.3.4 

type CatalogEntry struct {
	Vendor string
	contracts.ModelSpec
}

CatalogEntry is an offered model plus the backend that serves it. The vendor is never shown to the user: it lets the host know which plugin to instantiate once the model is chosen.

func Catalog added in v0.3.4 

func Catalog(plugins []contracts.Plugin, policy contracts.RoutePolicy) ([]CatalogEntry, error)

Catalog aggregates the Models of every registered backend and applies the route policy. The aggregation works off the Manifests, so it never instantiates a single backend — that's what lets the app populate its selector before any session exists.

An inconsistent catalog (empty ID, duplicate within one backend, duplicate across two backends) is an error, not a warning: the daemon must refuse to start rather than silently serve a wrong selector.

func LookupModel added in v0.3.4 

func LookupModel(plugins []contracts.Plugin, policy contracts.RoutePolicy, modelID string) (CatalogEntry, error)

LookupModel resolves a model ID to its catalog entry, under the given policy. A model the policy excludes is NOT FOUND, not merely hidden: a persisted session naming it fails to resume, which is the intended behavior when a public build inherits internal state.

type CoordinationView 

type CoordinationView struct {
	Role     string // "lead" | "worker"
	Lead     string // cohort identity: a worker's Parent, or a lead's own name
	Reported int    // workers that delivered (lead only; 0 for a worker)
	Expected int    // cohort size once sealed (lead only; 0 if unsealed)
	Complete bool   // Expected>0 && Reported>=Expected
}

CoordinationView is a read-only snapshot of a session's join state, projected from the coordinator's in-memory maps for observability (Neublox reads it via session list --json). It exposes only state the coordinator actually holds — no invented phase: reported/expected are the join counters, complete is the deterministic barrier (the same done>=N Report uses).

type DaemonEvent added in v0.11.0 

type DaemonEvent struct {
	Session string `json:"session"`
	contracts.Event
}

DaemonEvent is one line off the events socket: a turn event tagged with the session it belongs to, since one socket carries every session's stream.

type Deps 

type Deps = contracts.GatewaySet

Deps is the channel the daemon drives. It is the neutral contracts.GatewaySet the registry produces, so serve carries no platform-specific imports and works with any gateway plugin.

type GatewayHub 

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

GatewayHub holds every registered gateway plugin instantiated into a GatewaySet, keyed by Manifest.Kind. It is the core's multi-gateway port: the daemon resolves a session's bound gateways through it instead of hand-wiring a single gateway. Kinds() preserves registration order.

func BuildHub 

func BuildHub(ctx context.Context, plugins []contracts.Plugin, getenv func(string) string) (*GatewayHub, error)

BuildHub instantiates each gateway plugin in plugins. A plugin whose config can't resolve, or whose factory errors, is skipped (its required vars are absent — e.g. a missing gateway token — which must not stop other gateways from running). If NO gateway builds, the aggregated per-gateway reasons are returned so a single-gateway stack still fails fast with a clear message. Every skip is also recorded on the hub (Failures) so the daemon can report a gateway that dropped out — a stale token silently costing you a whole edge is worse than a noisy line at boot.

func (*GatewayHub) AwaitPending added in v0.6.4 

func (h *GatewayHub) AwaitPending(ctx context.Context, getenv func(string) string, window time.Duration, note func(failures []string, in time.Duration)) int

AwaitPending retries the gateways that did not build, backing off between attempts, until every one is up or the window closes, and reports how many joined. It belongs to the daemon at startup, where nothing is supervising sessions yet and waiting costs nothing — never to a one-shot operator command, which must not hang because a name will not resolve.

note, if given, is called before each wait with the reasons still standing, so the caller decides how a retry is reported.

func (*GatewayHub) Failures added in v0.3.2 

func (h *GatewayHub) Failures() []string

Failures returns one "kind: reason" line per gateway that did not build, for the caller to log. It is empty when every registered gateway came up.

func (*GatewayHub) First 

func (h *GatewayHub) First() (contracts.GatewaySet, bool)

First returns the first built gateway set (registration order) and whether the hub has one. It preserves the pre-hub "first registered gateway" behavior for callers not yet gateway-aware.

func (*GatewayHub) Get 

func (h *GatewayHub) Get(kind string) (contracts.GatewaySet, bool)

Get returns the GatewaySet for a kind and whether it was built.

func (*GatewayHub) Kinds 

func (h *GatewayHub) Kinds() []string

Kinds returns the built gateway kinds in registration order.

func (*GatewayHub) Pending added in v0.6.4 

func (h *GatewayHub) Pending() bool

Pending reports whether any registered gateway is still missing.

func (*GatewayHub) Retry added in v0.6.4 

func (h *GatewayHub) Retry(ctx context.Context, getenv func(string) string) int

Retry rebuilds the gateways that did not come up and reports how many joined. A factory can fail for a reason that fixes itself: at boot the daemon races the network, and a name that does not resolve yet is a few seconds of waiting rather than a broken install — while the process stayed up, so nothing restarted it and the edge it serves was simply gone. A genuinely bad credential keeps failing, which is why the waiting is bounded by the caller rather than endless.

It is a no-op once everything is up, and safe to call again: whatever is still pending stays pending, with its latest reason.

type HomeRef 

type HomeRef struct {
	ID   string
	Type string
}

HomeRef is the seed home channel from config.json: a channel id and its kind ("category" | "forum"). Kept platform-neutral so the caller need not import the internal state package.

type Options 

type Options struct {
	StatePath     string
	DefaultCmd    string
	HealthAddr    string
	StatusChannel string

	// InstanceID is the explicit per-daemon namespace (-instance flag /
	// HERRSCHER_INSTANCE_ID). Empty falls back to HERRSCHER_OWNER_ID, then legacy
	// mode.
	InstanceID string

	// Declarative config.json defaults. Owner is the per-daemon instance-id
	// fallback (env HERRSCHER_OWNER_ID takes precedence, resolved by the caller).
	// Home, Workspace and Source seed state in-memory only if unset, so a live
	// /set always wins (see state.ApplyDefaults).
	Owner     string
	Home      *HomeRef
	Workspace string
	Source    string

	// RemoteCategories lists plugin categories served out-of-process; RunHub
	// spawns and supervises a plugin-host child per entry. Empty => all in-proc.
	RemoteCategories map[contracts.Category]bool

	// ForegroundBound is true when a foreground (TUI) gateway is bound to the
	// process. Set by the caller (serve.go runServe) where fg != nil is computed.
	ForegroundBound bool

	// DefaultGateways is the primary gateway set a new session binds to when it
	// names none. The caller derives it from the built gateways (the concrete
	// platform kinds), so the manager package never names a gateway itself.
	DefaultGateways []string
}

Options holds the parsed flags for the serve daemon.

type RemoteResolveError 

type RemoteResolveError struct {
	Category contracts.Category
	Attempts int
	Elapsed  time.Duration
	Err      error
}

RemoteResolveError reports that a remote category could not be resolved within the retry budget. It carries the category, attempts made, and elapsed time so the caller can degrade cleanly; Unwrap exposes the last underlying error.

func (*RemoteResolveError) Error 

func (e *RemoteResolveError) Error() string

func (*RemoteResolveError) Unwrap 

func (e *RemoteResolveError) Unwrap() error

type Resolver 

type Resolver struct {
	NatsURL string // "" => nats.DefaultURL; consulted only on the remote path
	// contains filtered or unexported fields
}

Resolver turns registered plugins into live port objects, choosing local (in-proc factory) or remote (gRPC proxy) per category. With a nil/empty remote set every category resolves local — today's behaviour.

func NewResolver 

func NewResolver(remote map[contracts.Category]bool, natsURL string) *Resolver

func (*Resolver) Backend 

func (r *Resolver) Backend(ctx context.Context, plugins []contracts.Plugin, desired ...string) (contracts.Backend, error)

Backend resolves a REMOTE backend proxy when HERRSCHER_REMOTE names "backend", reusing the same retry/timeout/metrics harness. Like Orchestrator it returns (nil, nil) when the category is not remote, leaving the caller to build the local backend (which closes over model config the resolver does not hold). The remote proxy streams turn events over gRPC and surfaces a stream loss as a Respond error so the turn loop abandons the in-flight turn.

func (*Resolver) Memory 

func (r *Resolver) Memory(ctx context.Context, plugins []contracts.Plugin, getenv func(string) string) (contracts.Memory, error)

Memory resolves the first registered memory plugin. Local: call the factory. Remote: dial a gRPC proxy via NATS announcements, retried within a deadline. Returns nil (no error) when none is registered — memory stays optional, matching buildMemory's contract.

func (*Resolver) Orchestrator 

func (r *Resolver) Orchestrator(ctx context.Context, plugins []contracts.Plugin) (contracts.Orchestrator, error)

Orchestrator resolves a REMOTE orchestrator proxy when HERRSCHER_REMOTE names "orchestrator", reusing the same retry/timeout/metrics harness as Memory. It returns (nil, nil) when the category is not remote: the local orchestrator needs runtime state the resolver does not hold (the in-process Memory plus the session/scope/learn config bag), so the caller builds the local one itself.

func (*Resolver) SetCredentials 

func (r *Resolver) SetCredentials(c credentials.TransportCredentials)

SetCredentials installs the transport credentials the remote dial authenticates with. nil keeps plaintext loopback (the default); an mTLS credential is required to dial a category running on another host.

func (*Resolver) SetLogger 

func (r *Resolver) SetLogger(l *slog.Logger)

SetLogger installs the operator logger remote-resolve diagnostics flow through (component=resolver is attached for filtering).

func (*Resolver) SetMetrics 

func (r *Resolver) SetMetrics(m *metrics.Registry)

SetMetrics installs the registry remote-resolve attempts/failures/latency are recorded into.

type TurnIDError 

type TurnIDError struct {
	Reason string
}

TurnIDError reports a supplied seed turn_id that cannot be carried safely through argv and JSON-line protocol surfaces.

func (*TurnIDError) Error 

func (e *TurnIDError) Error() string

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.