fiber

README

fiber

Moved. fiber is now flow in urmzd/orchagent. This repository is archived. Its last release only marks the Go module deprecated.

Agents on a bus: the event union, channels, connectors, the durable turn. It lived in the Humagen repository as fiber/ before moving to orchagent.

Migrating

go get github.com/urmzd/orchagent@latest

Then rewrite imports:

Before	After
github.com/urmzd/fiber	github.com/urmzd/orchagent/flow
github.com/urmzd/fiber/pkg/<pkg>	github.com/urmzd/orchagent/flow/<pkg>
github.com/urmzd/fiber/pkg/durable	github.com/urmzd/orchagent/flow/runschema

Package names follow the directory: fiber.X becomes flow.X. Names stored in databases or on the wire (fiber_* tables, fiber:// URIs, fiber.* event kinds) are unchanged.

orchagent carries the later history from humagen/fiber; this repository keeps the history before the fold.

Documentation

Overview

Package fiber is an agent library you compose in Go. New assembles the agent definitions, the bus, dispatch, egress, the durable runtime, the sandbox, knowledge, thread journals, files, MCP tool providers, the author agent and the connectors into an App, and Run drives it in process. It is a library: there is no service, no configuration file and no environment variable, every choice is a value passed to an Option, and anything not passed is the in-memory implementation, which is complete rather than a stub. The one command in the module, cmd/fiber, is a program built on it like any other, for running a spec locally (fiber run).

pool, _ := pgxpool.New(ctx, dsn) // the program's own pool
pg := bus.NewPostgres(pool)
_ = pg.Migrate(ctx)
app, err := fiber.New(ctx,
	fiber.WithTenant("acme"),
	fiber.WithPostgres(pool), // every store family on this pool
	fiber.WithBus(pg),        // and the bus
	fiber.WithModelProvider("house", newHouseProvider),
)
if err != nil { ... }
defer app.Close()
go app.Run(ctx)
_ = app.Memberships().Join(ctx, "acme", "slack://T01/C02", "docs-bot",
	membership.Rule{Types: []string{"message"}})
err = app.Publish(ctx, msg) // a message.posted.v1 naming docs-bot

Code-shaped extensions are options too: a model provider, a tool source, a connector, an MCP tool provider. Configuration from files or the environment belongs to the program embedding fiber (Humagen is one), which turns it into these options.

Where it sits

The package is the module root because it composes everything under pkg/, and nothing under pkg/ may depend on it. An App listens on nothing: connectors that receive webhooks are offered as http.Handlers for the embedding program to mount.

Index

• Constants
• Variables
• func ControlStreams(compactOnRequest bool) int
• type App
  • func New(ctx context.Context, opts ...Option) (*App, error)
  • func (a *App) Answer(ctx context.Context, req runtime.AnswerRequest) (answer event.Answer, first bool, err error)
  • func (a *App) Blobs() blob.Store
  • func (a *App) Bus() bus.Bus
  • func (a *App) Close() error
  • func (a *App) Compact(ctx context.Context, req runtime.CompactRequest) error
  • func (a *App) Connectors() *connector.Registry
  • func (a *App) Definitions() definition.Source
  • func (a *App) Dispatcher() *dispatch.Dispatcher
  • func (a *App) Egress() *egress.Engine
  • func (a *App) Index(ctx context.Context, tenant, source string, paths ...string) (IndexResult, error)
  • func (a *App) Knowledge() knowledge.Store
  • func (a *App) Memberships() membership.Store
  • func (a *App) Providers() []string
  • func (a *App) Publish(ctx context.Context, e event.Envelope) error
  • func (a *App) Run(ctx context.Context) error
  • func (a *App) RunDelivery(ctx context.Context) error
  • func (a *App) RunExecution(ctx context.Context) error
  • func (a *App) Runtime() *runtime.Runtime
  • func (a *App) Status(ctx context.Context, tenant, runID string) (Run, error)
  • func (a *App) Stop(ctx context.Context, req runtime.StopRequest) error
  • func (a *App) Threads() thread.Store
  • func (a *App) ToolProviders() toolprovider.Store
  • func (a *App) ToolsFor(ctx context.Context, trigger event.Envelope, agent string) (native.Granted, error)
  • func (a *App) WebhookHandler() http.Handler
  • func (a *App) Webhooks() map[string]http.Handler
• type GitHub
• type IndexResult
• type Jira
• type Ledger
• type Option
  • func WithAuthor(provider, model string) Option
  • func WithBlobs(s blob.Store) Option
  • func WithBus(b bus.Bus) Option
  • func WithCompaction(c native.Compaction) Option
  • func WithConcurrency(n int) Option
  • func WithConnector(c connector.Connector) Option
  • func WithDefinitions(s definition.Source) Option
  • func WithDurable(l Ledger, q dqueue.Queue) Option
  • func WithEgressOptions(opts ...egress.Option) Option
  • func WithEmbedder(e knowledge.Embedder) Option
  • func WithGitHub(c GitHub) Option
  • func WithJira(c Jira) Option
  • func WithKnowledge(s knowledge.Store) Option
  • func WithLogger(log *slog.Logger) Option
  • func WithMemberships(s membership.Store) Option
  • func WithMetrics(rec metrics.Recorder) Option
  • func WithModelProvider(name string, fn native.ProviderFunc) Option
  • func WithOpenAICompatible(ep native.OpenAICompatible) Option
  • func WithPersonCredentials(r mcptools.PersonResolver) Option
  • func WithPostgres(pool *pgxpool.Pool) Option
  • func WithRuntimeOptions(opts ...runtime.Option) Option
  • func WithSchedules(s ...cron.Schedule) Option
  • func WithSlack(c Slack) Option
  • func WithTenant(tenant string) Option
  • func WithThreads(s thread.Store) Option
  • func WithToolProvider(p toolprovider.Provider, accept ...toolprovider.Tool) Option
  • func WithToolProviders(s toolprovider.Store) Option
  • func WithToolSource(c runtime.Catalog) Option
  • func WithTools(tools ...saigetypes.Tool) Option
  • func WithTurnRuntime(rt turn.Runtime) Option
  • func WithWebhooks(endpoints map[string]webhook.Endpoint) Option
  • func WithWorkerOptions(opts ...worker.Option) Option
  • func WithWorkspace(dir string) Option
• type Run
• type Slack
• type Status
  • func (s Status) Terminal() bool

Constants

const HeaderJiraSecret = "X-Fiber-Secret"

HeaderJiraSecret carries the shared secret a Jira webhook is registered with. Jira signs nothing, so the secret is whatever the person who configured the URL put in a header; see jiraSecret.

Variables

var ErrConfig = errors.New("fiber: invalid configuration")

ErrConfig reports a composition that cannot be what its author meant: a half-configured connector, a trigger connector with no tenant to publish events in, a tool provider declared twice, a tool pinned with no schema hash. New fails with it before anything runs.

var ErrNotFound = errors.New("fiber: not found")

ErrNotFound reports a run that does not exist or belongs to another tenant. The two are one error on purpose: telling them apart would confirm existence across a tenant boundary.

var ErrUnavailable = errors.New("fiber: backend unavailable")

ErrUnavailable reports a backend New could not migrate. Retrying unchanged is the right response.

Functions

func ControlStreams

func ControlStreams(compactOnRequest bool) int

ControlStreams is how many control subscriptions a running turn holds for its whole turn: one for stop requests, and one more for requests to compact when compactOnRequest. On Redis each is a pooled connection, so a program sizing a Redis client for an App wants bus.RedisPoolSize(concurrency, ControlStreams(compactOnRequest)).

Types

type App

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

App is fiber assembled: every long-lived component a process owns, in the order data flows through them.

triggers  →  bus  →  dispatch  →  runtime  →  bus  →  egress  →  sinks
                                     ↑                             ↓
                               anchor worker              a surface somewhere

Nothing here is a stub standing in for a deployment. The in-memory definitions, bus, ledger and queue are complete implementations of the same interfaces the shared backends implement; which one runs is an Option that passes a constructed value, never a change to anything below it.

An App listens on nothing. Connectors that receive webhooks are exposed as http.Handlers (Webhooks, WebhookHandler) for the embedding program to mount.

func New

func New(ctx context.Context, opts ...Option) (*App, error)

New assembles the system from opts and returns before anything runs: call Run to start delivery and execution, and Close to release what New made. Every backend not passed by an option is the in-memory implementation; WithPostgres puts every store family on one pool, migrated here. New opens no connection of its own and reads no environment or file.

It is deliberately linear: read it top to bottom and you have read how fiber fits together.

func (*App) Answer

func (a *App) Answer(ctx context.Context, req runtime.AnswerRequest) (answer event.Answer, first bool, err error)

Answer answers a question a run is parked on, and returns the answer the question has, which is this one only when first is true. See runtime.Runtime.Answer for the errors.

func (*App) Blobs

func (a *App) Blobs() blob.Store

Blobs is the file store.

func (*App) Bus

func (a *App) Bus() bus.Bus

Bus is the broker runs publish to.

func (*App) Close

func (a *App) Close() error

Close releases what New made: MCP sessions and a workspace New created. Every backend passed by an option, the Postgres pool included, is the caller's and is left alone. It is idempotent. Stop Run first.

func (*App) Compact

func (a *App) Compact(ctx context.Context, req runtime.CompactRequest) error

Compact asks a run's running turn to compact its history. It needs Config.Compaction.OnRequest; see runtime.Runtime.Compact.

func (*App) Connectors

func (a *App) Connectors() *connector.Registry

Connectors is the connector registry.

func (*App) Definitions

func (a *App) Definitions() definition.Source

Definitions is the agent store.

func (*App) Dispatcher

func (a *App) Dispatcher() *dispatch.Dispatcher

Dispatcher wakes channel members for the events published on the bus. Run starts it for the deployment's tenant; a program serving more tenants runs it once per tenant.

func (*App) Egress

func (a *App) Egress() *egress.Engine

Egress delivers run events to connector sinks. Run starts it for the deployment's tenant; a program serving more tenants runs it once per tenant.

func (*App) Index

func (a *App) Index(ctx context.Context, tenant, source string, paths ...string) (IndexResult, error)

Index writes text files into the knowledge store under source, for tenant. An agent whose spec lists source under "sources" is then grounded in them, and its answers carry citations. A directory is walked for Markdown and plain text (hidden directories skipped); a file named explicitly is indexed whatever its extension. Each document's URI is its path as given.

func (*App) Knowledge

func (a *App) Knowledge() knowledge.Store

Knowledge is the indexed corpus agents are grounded in.

func (*App) Memberships

func (a *App) Memberships() membership.Store

Memberships is who is a member of which channel.

func (*App) Providers

func (a *App) Providers() []string

Providers is the model provider names specs may select: the built-ins and every extension's.

func (*App) Publish

func (a *App) Publish(ctx context.Context, e event.Envelope) error

Publish puts an ingress event on the bus. The dispatcher started by Run (or RunExecution) wakes every member of the event's channel whose rule matches it, each in a run of its own. An event with no tenant is the deployment's (WithTenant).

The event's Tenant and actor are trusted as given. An embedding program that publishes events from callers sets the tenant from the caller's credential, never from a request body.

func (*App) Run

func (a *App) Run(ctx context.Context) error

Run starts delivery and execution and blocks until ctx is done or one of them fails, then stops both. A cancelled ctx is not a failure. Dispatch and egress run for the deployment's tenant (WithTenant); a program serving more tenants runs Dispatcher and Egress once per tenant.

func (*App) RunDelivery

func (a *App) RunDelivery(ctx context.Context) error

RunDelivery runs only egress: it reads the bus and renders events onto connector sinks, accepting and executing nothing, so delivery can scale apart from execution. It needs a shared bus to be useful.

func (*App) RunExecution

func (a *App) RunExecution(ctx context.Context) error

RunExecution runs only the durable worker, dispatch and the triggers: it executes runs and accepts what connectors publish, and delivers nothing.

func (*App) Runtime

func (a *App) Runtime() *runtime.Runtime

Runtime is the durable runtime.

func (*App) Status

func (a *App) Status(ctx context.Context, tenant, runID string) (Run, error)

Status reads a run's durable record, scoped to tenant. A run of another tenant is ErrNotFound, exactly like one that does not exist.

The record trails the run's terminal event briefly. A run publishes run.completed.v1, run.failed.v1 or run.stopped.v1 from inside its turn, and the worker records how it ended only once the turn has returned, so a Status read the moment that event arrives can still report accepted. The ledger is the authority: a worker that dies in between leaves the run pending, and the next attempt replays it. A caller that needs the two to agree polls until the status is Terminal.

func (*App) Stop

func (a *App) Stop(ctx context.Context, req runtime.StopRequest) error

Stop asks a run to stop. See runtime.Runtime.Stop for what a nil error does and does not mean; errors are the runtime's (runtime.ErrInvalid, runtime.ErrStopUnavailable).

func (*App) Threads

func (a *App) Threads() thread.Store

Threads is the conversation journal.

func (*App) ToolProviders

func (a *App) ToolProviders() toolprovider.Store

ToolProviders is the MCP tool provider store.

func (*App) ToolsFor

func (a *App) ToolsFor(ctx context.Context, trigger event.Envelope, agent string) (native.Granted, error)

ToolsFor reports what a run of agent answering trigger would be granted, without accepting it: the tools the current version of agent in the trigger's tenant would show its model, and the grants it would drop, each with its reason. The tools are resolved as a turn of that run would resolve them, for its tenant, its surface and its actor, so a person-scoped MCP grant expands for the trigger's actor, and an MCP server may be reached to expand a group. No tool is called and no model client is built, so it needs no model credential. The trigger is trusted as given, as Publish trusts it, and needs only a tenant (or the App's, WithTenant).

It asks the App's sandbox. A turn runtime passed by WithTurnRuntime is asked too if it has a method ToolsFor(context.Context, runtime.Request) (native.Granted, error); otherwise the error is errors.ErrUnsupported.

func (*App) WebhookHandler

func (a *App) WebhookHandler() http.Handler

WebhookHandler serves every webhook connector under /connectors/{name}, for a program that wants one mount point.

func (*App) Webhooks

func (a *App) Webhooks() map[string]http.Handler

Webhooks returns the connectors that receive their own webhooks, by connector name. Each verifies its own signature, so no unverified body reaches anything else. Mount them wherever the embedding program serves HTTP; the path a third party POSTs to is that connector's contract.

type GitHub

type GitHub struct {
	Token, WebhookSecret string
	Owners               []string
	ReadRepositories     []string
	Bots                 map[string]string
	Checks               bool
	// Login is the app's own login. A comment that mentions it and names no
	// agent is published with no mentions, and the members of the issue's
	// channel decide whether anyone answers.
	Login string
}

GitHub configures the built-in GitHub connector (WithGitHub). Token and WebhookSecret together turn it on, and it needs WithTenant.

type IndexResult

type IndexResult struct {
	// Files is how many files were read; Skipped how many were not text.
	Files, Skipped int
	// Changed is how many documents were new or different, and Chunks how
	// many chunks they were written as. Re-indexing unchanged files
	// changes nothing.
	Changed, Chunks int
}

IndexResult is what Index did.

type Jira

type Jira struct {
	Sites                       map[string]string
	Email, Token, WebhookSecret string
	Bots                        map[string]string
}

Jira configures the built-in Jira connector (WithJira). Sites (short key to base URL; the key is what a jira:// address names) turns it on, and then Email, Token, WebhookSecret and WithTenant are required.

type Ledger

type Ledger interface {
	ledger.Ledger
	ledger.RunGetter
}

Ledger is the run ledger an App executes from: anchor's ledger plus the point read Status polls with. ledger.NewMemory and durable.Open's ledger both satisfy it.

type Option

type Option func(*options)

Option is one choice New is given.

func WithAuthor

func WithAuthor(provider, model string) Option

WithAuthor chooses what the built-in author agent runs on. Without it the agent is seeded on the offline provider.

func WithBlobs

func WithBlobs(s blob.Store) Option

WithBlobs sets the file store, for example blob.NewFS(dir).

func WithBus

func WithBus(b bus.Bus) Option

WithBus sets the broker: bus.NewMemory (the default), bus.NewRedis or bus.NewPostgres, or any implementation that passes bustest.

func WithCompaction

func WithCompaction(c native.Compaction) Option

WithCompaction sets the history compaction every agent runs with. The zero value never compacts; OnRequest lets a host ask a running turn to compact (App.Compact).

func WithConcurrency

func WithConcurrency(n int) Option

WithConcurrency sets how many runs the worker executes at once (default one).

func WithConnector

func WithConnector(c connector.Connector) Option

WithConnector registers a connector beside the built-in ones. Its roles (trigger, sink, actor) are derived from the interfaces it implements, as every connector's are.

func WithDefinitions

func WithDefinitions(s definition.Source) Option

WithDefinitions sets the agent store. The store decides which providers it accepts (definition.WithProviders or WithChecks); App.Providers lists the names this App can run.

func WithDurable

func WithDurable(l Ledger, q dqueue.Queue) Option

WithDurable sets the run ledger and queue runs execute from, for example the two durable.Open returns. Both are required.

func WithEgressOptions

func WithEgressOptions(opts ...egress.Option) Option

WithEgressOptions adds options to the egress engine, after the App's own.

func WithEmbedder

func WithEmbedder(e knowledge.Embedder) Option

WithEmbedder sets the embedder of the knowledge store New builds (memory, or Postgres with WithPostgres). Without it the offline hashing embedder is used. A store passed by WithKnowledge brings its own.

func WithGitHub

func WithGitHub(c GitHub) Option

WithGitHub registers the GitHub connector.

func WithJira

func WithJira(c Jira) Option

WithJira registers the Jira connector.

func WithKnowledge

func WithKnowledge(s knowledge.Store) Option

WithKnowledge sets the knowledge store agents are grounded in.

func WithLogger

func WithLogger(log *slog.Logger) Option

WithLogger sends operational output to log. Without it it is discarded.

func WithMemberships

func WithMemberships(s membership.Store) Option

WithMemberships sets which agents are members of which channels, which is what decides who an event published with App.Publish wakes. Without it, and without WithPostgres, memberships live in memory and start empty.

func WithMetrics

func WithMetrics(rec metrics.Recorder) Option

WithMetrics records counters to rec. Without it nothing is recorded.

func WithModelProvider

func WithModelProvider(name string, fn native.ProviderFunc) Option

WithModelProvider makes a model provider selectable by name: a spec whose provider is name runs on what fn builds. The registry New builds accepts specs naming it, and a built-in of the same name is replaced.

func WithOpenAICompatible

func WithOpenAICompatible(ep native.OpenAICompatible) Option

WithOpenAICompatible registers an OpenAI-compatible endpoint under ep.Name. It is WithModelProvider with ep.Provider(); the key is read from the environment variable ep.KeyEnv names when an agent is built.

func WithPersonCredentials

func WithPersonCredentials(r mcptools.PersonResolver) Option

WithPersonCredentials sets how a person-scoped MCP tool provider (toolprovider.ScopePerson) finds the credential of the person a run acts for. That person is the actor on the run's trigger as it was accepted, which the program composing fiber sets from the caller's credential (a surface marked ActAs names the person it authenticated) and which no model output can change. A reference the resolver returns is resolved like a provider's own ("env:NAME"); a resolver that holds the secrets returns them. A person it has nothing for gets no tools from such a provider, the model is told they have not connected a credential, and the shared credential is never used instead. Without this option every person-scoped grant is unavailable.

func WithPostgres

func WithPostgres(pool *pgxpool.Pool) Option

WithPostgres puts every store family on pool: the agent definitions, the run ledger and queue, knowledge, thread journals, memberships and MCP tool providers. New migrates each. A store passed by its own option wins over this one. The bus is chosen separately (WithBus with bus.NewPostgres or bus.NewRedis).

func WithRuntimeOptions

func WithRuntimeOptions(opts ...runtime.Option) Option

WithRuntimeOptions adds options to the durable runtime, after the App's own, so they can override them.

func WithSchedules

func WithSchedules(s ...cron.Schedule) Option

WithSchedules registers the cron connector with its standing requests.

func WithSlack

func WithSlack(c Slack) Option

WithSlack registers the Slack connector.

func WithTenant

func WithTenant(tenant string) Option

WithTenant sets the authorization scope this deployment serves. It is stamped onto every event a connector publishes (a trigger has no credential to derive one from), mentions on GitHub and Jira resolve agents in it, Run dispatches and delivers its events, the author agent is seeded for it, and tool providers declared without a tenant are recorded under it. There is no default: without it nothing is seeded, and WithSlack, WithGitHub and WithJira are ErrConfig, since they would publish nothing a run could be accepted for. A schedule carries its own tenant.

func WithThreads

func WithThreads(s thread.Store) Option

WithThreads sets the conversation journal.

func WithToolProvider

func WithToolProvider(p toolprovider.Provider, accept ...toolprovider.Tool) Option

WithToolProvider records an MCP tool provider at startup and accepts the tools in accept from it, each pinned by Name and SchemaHash and accepted with its Class (write when empty). Only an offered tool whose schema hash matches its pin is accepted, and its description may not change once it has been; anything the server adds or changes stays pending and is logged with its hash. A tool with no SchemaHash, or a provider declared twice, is ErrConfig. An empty p.Tenant is the deployment's tenant. Agents reach the provider with an "mcp:<name>" or "mcp:<name>:write" grant.

A person-scoped provider is not dialled at startup, since it has no credential of its own: each tool in accept must then carry the Description and InputSchema the server offers, and is accepted in that form when its SchemaHash matches the schema. Its runs reach it with each person's own credential (WithPersonCredentials).

func WithToolProviders

func WithToolProviders(s toolprovider.Store) Option

WithToolProviders sets the MCP tool provider store.

func WithToolSource

func WithToolSource(c runtime.Catalog) Option

WithToolSource adds a catalog that spec tool names resolve against, after the connectors' actions. It is built per turn, so it can bind tools to the run's tenant and surface; action reports a call as an action.taken event.

func WithTools

func WithTools(tools ...saigetypes.Tool) Option

WithTools adds a fixed set of tools, resolved by name for every turn. It is WithToolSource for tools that need nothing from the run.

func WithTurnRuntime

func WithTurnRuntime(rt turn.Runtime) Option

WithTurnRuntime replaces the playscape sandbox a turn runs in. Model providers, tool sources, compaction and knowledge are wired into the sandbox, so a turn runtime passed here brings its own.

func WithWebhooks

func WithWebhooks(endpoints map[string]webhook.Endpoint) Option

WithWebhooks registers the outbound webhook sink with its endpoints, by the workspace a webhook:// address names. Each needs a secret.

func WithWorkerOptions

func WithWorkerOptions(opts ...worker.Option) Option

WithWorkerOptions adds options to the durable worker Run starts, after the App's own.

func WithWorkspace

func WithWorkspace(dir string) Option

WithWorkspace sets the directory the sandbox writes agent artifacts under. Without it New makes a temporary directory and Close removes it.

type Run

type Run struct {
	ID     string
	Tenant string
	Agent  string
	// Version is the definition version the run pinned.
	Version string
	Origin  string
	// Channel is where the run's events are published.
	Channel string
	Status  Status
	// Trigger is the event the run answers.
	Trigger event.Envelope
	LastSeq uint64
	Error   string
	// StopReason and StoppedBy say why a stopped run was stopped and who
	// asked. Empty unless Status is StatusStopped.
	StopReason string
	StoppedBy  string
	// Usage is what the run spent, nil while nothing is recorded.
	Usage *event.RunUsage
	// Question is what the run is waiting on a person for, set while
	// Status is StatusParked.
	Question *event.Question

	CreatedAt time.Time
	EndedAt   time.Time
}

Run is the durable record of one invocation. It is a projection of the ledger, so reading one never touches a worker and polling it is cheap.

type Slack

type Slack struct {
	Token, SigningSecret string
	// Team is the workspace id deliveries are confined to.
	Team string
	// AppID marks messages this deployment posted when a thread is
	// reconciled.
	AppID string
	// Bots maps a Slack user id to the agent it stands for. A mention of one
	// names that agent on the event; the members of the channel decide who
	// answers.
	Bots map[string]string
}

Slack configures the built-in Slack connector (WithSlack). Token and SigningSecret together turn it on; one without the other is ErrConfig, and so is either without WithTenant.

type Status

type Status string

Status is where a run is in its lifecycle. It mirrors the run lifecycle types (run.completed.v1 and the rest): status answers "where is it now", the events answer "how did it get there".

const (
	StatusAccepted  Status = "accepted"
	StatusParked    Status = "parked"
	StatusCompleted Status = "completed"
	StatusFailed    Status = "failed"
	// StatusStopped is a run that ended because it was asked to. It is
	// terminal, and it is neither completed nor failed.
	StatusStopped Status = "stopped"
)

Run statuses.

func (Status) Terminal

func (s Status) Terminal() bool

Terminal reports whether a run in s has ended.