fiber

README

fiber

Define agents, run them durably, and stream answers across your tools.

Report Bug · Go Docs

     

Real output from the offline example. The model and destination services are simulated.

Status

fiber is v0 alpha software. It is pre-1.0 and not recommended for production.

Interfaces are still moving and may change between minor versions, and a minor release may require code changes on your side.

What that means concretely today:

• Works and is tested. The in-process composition (fiber.New) on the in-memory backends, the topic and delivery model, durable runs with replay, and the Slack, GitHub, Jira, cron and webhook connectors.
• New and lightly exercised. The Postgres backends for every store and for the bus, the knowledge layer, the built-in author agent, and history compaction. Each backend passes the same conformance suite as its in-memory counterpart in CI's integration job; none of it has seen production load.

See Limitations for the full list.

fiber is a Go library for agents and a low-latency pub/sub bus. Define an agent with a spec, tag it from Slack, GitHub, Jira or a schedule, and its events stream to any subscriber while being delivered to any destination you name. It is a library you compose in Go: there is no service, no configuration file and no environment variable. fiber.New composes the packages under pkg/, every backend is a constructed value passed as an option, and anything not passed is the in-memory implementation. One small command, fiber run, runs a spec once on that in-memory composition so you can check it locally (Validate a spec locally). Humagen is the platform built on it.

Where a tag is raised and where its answer lands are independent. A GitHub request can stream into Slack while a webhook receives only its completion event. Each run pins an agent revision, so retries use the same definition.

Features

• Agents are documents. Create one by submitting a spec. No redeploy, no code, no restart.
• Subscribe once. agent:docs-bot receives current and future runs of that agent through one topic.
• Route by event. Send one run to multiple destinations; each chooses text, citations, lifecycle events, or all of them.
• Connectors act, not just talk. The Actor face contributes native verbs (transition a Jira ticket, request a review) as agent tools, sandboxed by NGAC, each emitting an auditable action.taken event.
• Backends are values. fiber.WithPostgres(pool) puts the registry, run ledger and queue, knowledge, thread journals and tool providers on one pool; fiber.WithBus takes the in-memory, Redis Streams or Postgres bus. Anything not given is memory, which is a complete implementation, not a stub.
• Revisions are pinned per run. A run accepted under revision 6 executes as revision 6 forever, even after a revise, because replay demands it.
• Citations are first-class events, carrying the exact quoted source and resolvable provenance.
• Every run reports what it spent. Each model call is a usage.call event (tokens, cache share, model, context window, and cost where saige's price catalog has the model), and the run's total, delegations included and each call counted once, is on its terminal event and App.Status.
• A run can wait for a person. A tool listed in a spec's approvals, or the built-in ask_person tool, parks the run on a question (approve or deny a call, pick an option, or reply). The run frees its worker, survives restarts, publishes the question on lifecycle.parked, and resumes with the first answer (App.Answer). An unanswered question takes its default when it expires.

Installation

go get github.com/urmzd/fiber

Requires the Go version in go.mod. Real inference needs provider credentials, a local Ollama, or an OpenAI-compatible server. The Postgres and Redis backends need those services (make up starts both with Docker).

Quick Start

Run the complete offline example from a checkout. After Go downloads its dependencies, the demo needs no API key or external services:

git clone https://github.com/urmzd/fiber.git
cd fiber
go run ./examples/local

It creates an agent, revises it, tags it from a simulated GitHub trigger with delivery routed to Slack and a webhook, shows a third party listening to the agent's topic, and demonstrates that a replayed run does not pay for the model twice.

Compose it

pool, err := pgxpool.New(ctx, dsn) // your pool, from your configuration
pg := bus.NewPostgres(pool)        // or bus.NewRedis(client), or omit for memory
if err := pg.Migrate(ctx); err != nil {
    return err
}
app, err := fiber.New(ctx,
    fiber.WithTenant("acme"),
    fiber.WithPostgres(pool), // every store family, migrated by New
    fiber.WithBus(pg),
)
if err != nil {
    return err
}
defer app.Close()
go app.Run(ctx) // delivery and execution, in this process

app.Registry().Create(ctx, "acme", agentspec.AgentSpec{
    Name: "docs-bot", Provider: "anthropic", Model: "claude-haiku-5-5",
    SystemPrompt: "You answer questions about the deploy pipeline. Cite your sources.",
}, "ada")
acc, err := app.Invoke(ctx, envelope.Tag{
    ID: "req-1", Tenant: "acme", Agent: "docs-bot", Origin: "api",
    Text: "how does a release get out the door?",
    Deliver: []envelope.Route{{Target: address.MustParse("webhook://acme/deploys"),
        Kinds: []string{"lifecycle.completed"}}},
})

App.Bus().Subscribe streams any topic (agent:docs-bot sees every run of that agent); App.Status, Stop, Compact and Answer act on a run; App.ToolsFor lists what a run of a tag would be granted, without running it; App.WebhookHandler() is an http.Handler for the connectors that receive webhooks, for your program to mount. examples/downstream is a complete program.

Use a model

Set ANTHROPIC_API_KEY for "provider": "anthropic". To use OpenAI, or any server that speaks its Chat Completions API (vLLM, LM Studio, llama.cpp, OpenRouter, Groq), set "provider": "openai" and the model that server hosts. For a local server, point fiber at it instead of setting a key:

export OPENAI_BASE_URL="http://localhost:8000/v1"   # omit for OpenAI itself
export OPENAI_API_KEY="your-api-key"                # optional for a local server

Models saige's catalog does not know report their token usage and no cost.

An agent can name fallback models, tried in order when a call to its primary model fails with an error another model could fix (an outage, a rate limit, a refused credential):

"fallbacks": [
  {"provider": "openai", "model": "gpt-5"},
  {"provider": "ollama", "model": "llama4"}
]

Each fallback names a provider the deployment knows and a model, none repeats the primary or another fallback, and there are at most four. By default any error another model could fix falls through. "fallback_on" narrows that to named classes: unavailable (an outage), rate_limit, transient (a timeout or reset connection, and also a rate limit or outage), auth, context_length and content_filter, or any alone, which is the default. For example "fallback_on": ["rate_limit", "unavailable"]. It is a set, and naming it without fallbacks is refused. Whatever it names, an invalid request, a stopped run or an exhausted budget does not fall through, and a stream that already produced output is not replayed on another model. Every fallback is built with the agent, so one whose provider is not configured (say, OPENAI_API_KEY is unset) fails the build like the primary would. Each call is priced on the rate card of the model that answered it. Order is part of the revision: reordering fallbacks is an edit. A spec without fallbacks hashes exactly as it did before they existed.

An agent can be granted an MCP provider's tools as one name: "tools": ["mcp:urmzd"] grants the tools an admin accepted on the provider urmzd as read, and "mcp:urmzd:write" its write tools as well. Read only is the default, so leaving the class off can only grant less, and the class is checked again on every call. The model is shown each member tool (urmzd__create_issue), never the group, and approvals name members: "approvals": ["urmzd__create_issue"]. The group is expanded for the run's tenant each time a run builds the agent, so a newly accepted tool reaches the agent without a new revision; tools that are not accepted, or that changed since, are never offered. An App expands groups against its tool provider store (Postgres with fiber.WithPostgres) and resolves env:NAME credential references from its environment. A host embedding pkg/runtime does the same with runtime.WithToolGroupSource and an mcptools.Grants; one embedding pkg/agentrt alone supplies groups with agentrt.WithToolGroups or from its tool catalog.

A provider is reached with one shared credential, its CredentialRef, unless it is person-scoped (CredentialScope: toolprovider.ScopePerson): then each run reaches it with the credential of the person the run acts for, the actor on its tag, which fiber.WithPersonCredentials looks up. That actor is the one recorded when the run was accepted, so nothing a model says can change it. Each person gets a session of their own, opened with their own credential. A person with none gets no tools from the provider, the model is told they have not connected one, and the operator log names them; a shared credential is never used instead, and a person-scoped provider cannot even name one.

fiber.WithPersonCredentials(mcptools.PersonResolverFunc(
    func(ctx context.Context, tenant string, p toolprovider.Provider, who envelope.Actor) (mcptools.PersonCredential, error) {
        tok, ok := vault.Token(ctx, tenant, who.User, p.Name) // your own store, keyed by your account id
        if !ok {
            return mcptools.PersonCredential{}, mcptools.ErrNoPersonCredential
        }
        return mcptools.PersonCredential{Secret: mcptools.NewSecret(tok)}, nil
    }))

Validate a spec locally

fiber run runs one agent spec once, in process, and prints what the run does. It is cmd/fiber, a program built on the library like any other: it composes fiber.New with the in-memory backends and a temporary workspace, registers the spec, raises one tag with your message, and prints the run's events as they stream. It listens on nothing, reads no configuration file and no environment variable of its own, and keeps nothing once it exits. It is not the server binary earlier releases had (see Upgrading).

Releases publish the Go module, not a prebuilt binary. Install the command at the version of fiber you build against, or run it from a checkout with go run ./cmd/fiber:

go install github.com/urmzd/fiber/cmd/fiber@latest
fiber run -offline docs-bot.json "how does a release get out the door?"

[lifecycle.accepted] docs-bot revision 1, run r22e035b6d38cc721ba356a8a0290f805
[lifecycle.started] attempt 1
docs-bot is running with the offline provider, so no model was called. ...
[usage.call] MODEL: 0 tokens in (0 cached), 0 out
[lifecycle.completed] attempt 1, 1 model calls, 0 tokens in, 0 out

Each line is one event under its kind, the name a route's kinds selects, except the answer and any reasoning, which stream inline as they arrive. Warnings from the operator log, such as a granted tool this composition cannot resolve, go to standard error.

Flag	Does
-offline	Runs on the offline provider: no model is called and no key is needed. The spec's fallbacks are dropped with its provider.
-provider NAME, -model NAME	Replace the spec's provider and model. A real provider reads its vendor's variables when the agent is built (ANTHROPIC_API_KEY, OPENAI_API_KEY and OPENAI_BASE_URL, OLLAMA_HOST).
-tools	Lists the tools the agent would be granted, marks those whose calls wait for approval, and names each grant left out and why, then exits without running anything. It needs no key and no MESSAGE.
-json	Prints one JSON object per event, per line: the event as the bus carries it, with its payload as JSON. With -tools, the listing as one object.
-timeout D	Gives up after D, such as 30s (default 5m; 0 waits forever).

Flags go before SPEC. SPEC is a file holding one agent spec as JSON (the fields of agentspec.AgentSpec), or - for standard input. A field the format does not have is an error rather than ignored, so a misspelt system-prompt is caught.

Exit status	Means
0	The run completed, or the listing was printed.
1	The run did not complete: it failed, was stopped, parked on a question, or ran out of time. A failed attempt is retried as in any deployment, and the run counts as failed only once the run ledger says so.
2	The command line or the spec is wrong, and nothing ran.

A local run has what this composition has: the built-in providers, the ask_person tool, the author_* tools every App carries, and a tag_<agent> tool per agent the spec may spawn (only the one spec is registered, so a delegation finds no agent to tag). No connector and no MCP tool provider is configured, so a connector's actions (jira_transition) and mcp: grants are unavailable here: the run goes ahead without them, as a deployment's would, and -tools says which and why. Nothing here can answer a question, so a run that parks on an approval or on ask_person exits 1. The knowledge store is empty, so sources retrieve nothing. To list what a run would be granted in your own composition, call App.ToolsFor(ctx, tag) there.

Examples

Example	What it demonstrates
local	Offline agent revisions, selective delivery, citations, and replay, composed by fiber.New
downstream	A program composing fiber with its own model provider and tool

See the example index for more.

Documentation

Document	Covers
docs/architecture/overview.md	Topics, connector faces, N:M delivery, revision pinning, package DAG, limitations
AGENTS.md	AI-facing conventions, commands, rules, extension guide
examples/README.md	Runnable example index
CONTRIBUTING.md	Development workflow and commit convention
SECURITY.md	Security boundaries and reporting

API reference: pkg.go.dev/github.com/urmzd/fiber

Where this sits

fiber is layer four of five: Humagen > fiber > saige > playscape > anchor. It builds on saige's agent loop, playscape's sandbox and anchor's durability, and adds addressing, distribution, connectors and the control plane. Humagen is the only thing that configures or serves it: fiber is composed in Go and reads no configuration of its own. See the ecosystem guide for who owns what and the limitations that span the stack.

For existing Warp deployments of Humagen, see the upgrade guide.

Architecture

fiber builds on three libraries and adds addressing, distribution and the control plane:

Concern	Owner
Agent loop, typed deltas, RAG, citations	saige
Durability, replay, per-run journal	anchor
Sandboxed execution nodes, NGAC	playscape
Topics, bus, routing, connectors, specs	fiber

See the architecture guide for delivery, authorization, and replay guarantees.

Composition

Every choice is an option to fiber.New. Configuration from files or the environment belongs to the program composing fiber (Humagen is one), which turns it into these.

Option	Chooses	Default
WithPostgres(pool)	Registry, run ledger and queue, knowledge, threads, tool providers on one pool	memory
WithBus(b)	bus.NewMemory, bus.NewRedis(client), bus.NewPostgres(pool)	memory
WithRegistry, WithDurable(ledger, queue), WithThreads, WithKnowledge, WithBlobs, WithToolProviders	One store, built by the caller	memory, or the WithPostgres one
WithEmbedder(e)	The embedder of the knowledge store New builds	offline hashing
WithExecutor, WithRuntimeOptions, WithWorkerOptions	The sandbox, and options for the runtime and worker	playscape sandbox
WithWorkspace(dir)	Where the sandbox writes agent artifacts	a temporary directory
WithTenant, WithConcurrency, WithCompaction, WithAuthor, WithLogger, WithMetrics	Deployment	none, one, off, offline, discard, none
WithSlack, WithGitHub, WithJira, WithWebhooks, WithSchedules	Built-in connectors, typed Go values	off
WithModelProvider, WithOpenAICompatible, WithTools, WithToolSource, WithConnector, WithToolProvider	Extensions	none
WithPersonCredentials	Each person's own credential for person-scoped MCP providers	none, so those providers offer no tools

Model providers read their own credentials when an agent is built: ANTHROPIC_API_KEY, OPENAI_API_KEY and OPENAI_BASE_URL, OLLAMA_HOST, and the variable an OpenAICompatible.KeyEnv names. An MCP provider's env:NAME credential reference resolves the same way, as does a reference a WithPersonCredentials resolver returns for a person. Connector setup is covered in the GitHub and Slack guides.

Extending fiber

WithModelProvider, WithOpenAICompatible, WithTools, WithToolSource, WithConnector and WithToolProvider add to the composition:

app, err := fiber.New(ctx,
    fiber.WithModelProvider("house", houseProvider),
    fiber.WithOpenAICompatible(agentrt.OpenAICompatible{
        Name: "fireworks", BaseURL: "https://api.fireworks.ai/inference/v1", KeyEnv: "FIREWORKS_API_KEY"}),
    fiber.WithTools(runbookTool{}),
    fiber.WithToolProvider(toolprovider.Provider{Name: "issues", Transport: toolprovider.TransportHTTP,
        Endpoint: "https://mcp.example.com/mcp", CredentialRef: "env:ISSUES_TOKEN"},
        toolprovider.Tool{Name: "search", SchemaHash: "4f1c…", Class: toolprovider.ClassRead}),
)

A declared MCP tool is accepted only while the server's schema for it matches its SchemaHash, and its description may not change once accepted. Anything the server adds or changes stays pending, so a restart never accepts what the program does not declare. fiber.New logs every pending tool with its schema hash; review it, then pin it. A server that cannot be reached at startup is logged and its tools are unavailable to runs. A person-scoped provider is not dialled at startup, since it has no credential of its own: each tool declared for it carries the Description and InputSchema the server offers, and is accepted in that form when its SchemaHash matches. Review one with a credential of your own (mcptools.Connect(ctx, p, mcptools.WithPersonCredential(mine))).

A new backend implements the package's interface and passes its conformance suite: bustest, registrytest, knowledgetest, threadtest, toolprovidertest, blobtest.

Upgrading

fiber is a library from this release. The fiber binary, its commands, the HTTP and Connect API and its protobuf contract, registry.ServerStore, the providers file and every FIBER_* variable are gone. Compose fiber.New with options instead: FIBER_POSTGRES_URL is WithPostgres(pool), FIBER_REDIS_URL is WithBus(bus.NewRedis(client)) (size the client with bus.RedisPoolSize), a providers file's entries are WithOpenAICompatible and WithToolProvider, and the connector variables are WithSlack, WithGitHub, WithJira and WithWebhooks. GetRun is App.Status, the stop and compact routes are App.Stop and App.Compact, InvokeService.Answer is App.Answer, and fiber index is App.Index.

The binary's defaults are not the library's. The binary served tenant default unless FIBER_TENANT said otherwise, so the agents, threads and runs it wrote live under that tenant: pass WithTenant("default") (or the FIBER_TENANT you had) to keep them. WithSlack, WithGitHub and WithJira are ErrConfig without a tenant. The binary also kept files on disk under its workspace; the library keeps them in memory unless you pass WithBlobs(blob.NewFS(dir)).

Persisted names moved to fiber. Postgres objects, Redis keys, anchor activity and record names and the agent-spec hash domain no longer use the old name. Drain runs before upgrading, including runs parked on a question, which wait until they are answered, stopped or expire: a run still pending from the old version is failed with a reason when the new one first claims it, and is not resumed. Stop every old process before the first new one starts. Postgres migrates itself and keeps its rows. See docs/renaming.md.

Limitations

• The bus is chosen independently of the stores: WithPostgres does not move the bus, WithBus does.
• The Postgres bus serializes publishes on one advisory lock per schema, so its throughput is one short transaction per event; Redis is the better bus for a chatty deployment.
• The Postgres bus stores every event it carries, body included, as a row in the fiber_bus_events table of the database it is given. A row stays until its run's retention trim drops it or a sweep finds it older than bus.WithIndexTTL (24 hours by default; the sweep runs from publishes, so with none nothing is removed). Those rows reach that database's WAL, replicas and backups like any other, and tool events carry their call's arguments. Where arguments can hold secrets, scrub them with agentrt.WithToolArguments, shorten the TTL, or use the Redis bus.
• Every store is in memory unless WithPostgres or its own option is given. On Postgres they survive a restart, but the run queue fiber builds has no DeltaLog, so anchor's per-run delta journal is not written; the bus still carries every event.
• Files (blob) are in memory unless WithBlobs(blob.NewFS(dir)) puts them on disk; there is no Postgres blob store.
• fiber authenticates nothing. An embedding program that accepts tags or stop and answer requests from callers sets the tenant and actor from the caller's credential.
• playscape's queue is at-most-once in beta, so durability comes from anchor rather than from playscape.
• Citations are derived from RAG context blocks. saige's CitationDelta, which reports what a model or tool cited, is not translated into events yet.
• The bus is at-least-once with no consumer groups. If durable competing consumers become a requirement, NATS JetStream behind the Bus interface is a better answer than growing a broker here.
• Agent specs load from JSON, not YAML.
• App.Status reports a stopped run as stopped. A stop is recorded in the run ledger before it is announced on the bus, so once it is acknowledged the run ends stopped unless its turn had already returned its answer. A route that selects only lifecycle.completed is not told about a stop.
• A request to compact a running turn (App.Compact) is not recorded: it reaches only the attempt running when it is made, before that turn's next model call. See Compacting a run on request.
• A program declares MCP tool providers with fiber.WithToolProvider, or writes them to the tool provider store (App.ToolProviders). A group grant that cannot be expanded (no such provider for the tenant, the server unreachable, nothing the grant's class admits) is dropped like a single unavailable tool: the run continues without it, the model is told it is unavailable, the operator is logged why, and Runner.Unavailable lists it. agentrt.WithStrictTools fails the build with agentrt.ErrNoSuchToolGroup instead.
• A person-scoped MCP provider's tools are expanded per person, so two people in one thread can be offered different tools, and the thread's prompt cache misses when the person changes. At most 128 person sessions are kept open (mcptools.WithPersonSessions on a hand-built Grants); past that the one used longest ago is closed and a turn still holding its tools sees a tool error. A stdio provider runs a server per open person session. The resolver is asked on every turn, and a new value behind an unchanged reference is not noticed until the reference, the secret or the provider record changes.
• Only MCP tool providers take a person's credential. The Slack, GitHub and Jira connectors' actions and sinks still act with the credential the deployment configured for them.
• A granted tool this deployment cannot resolve (no connector contributes it, or an MCP tool is held back) is dropped rather than failing the run. The model is told it is unavailable and the server logs a warning naming the tool and the reason; agentrt.WithStrictTools restores the old refusal.
• Tool events carry the call's arguments, capped at 8 KiB; scrub sensitive ones with agentrt.WithToolArguments.
• A parked run waits on a person through the run ledger. An answer or a stop wakes it at once; an expired question is noticed by the worker's janitor poll (30 seconds by default), so its default applies up to that late. The requester cannot approve their own calls unless the spec sets self_approval, and approvers/answerers restrict who may answer. A tool implementing agentrt.Approving decides per call whether it needs approval and may name that call's approvers, which then replace the spec's. A host Delegator that blocks on a child run holds a worker and can exhaust them; runtime.WithChildRuns parks the parent on the child instead. Approvals inside an in-turn sub-task (a tag_<agent> call without child runs or a Delegator) are refused, not asked. See Parking a run on a person.
• Cost is reported only for models saige's price catalog prices, in integer micro-units with the catalog's as_of; anything else reports tokens and no cost. Usage is delivered only to routes that select usage. A turn that crashes mid-call is on the bus but not in the run's total. In-turn delegation is one level deep (a sub-task holds no spawn grant), and a host that runs delegations as runs of their own rolls them up with runtime.ReportChildUsage. See Usage.

Reproduce the Screenshot

Install teasr, then run from this checkout:

go run ./examples/local
teasr run --config teasr.toml

The first run warms Go's build cache. The capture runs the real demo and writes showcase/durable-replay.png.

License

Apache 2.0. See LICENSE.

Documentation

Overview

Package fiber is an agent library you compose in Go. New assembles the agent registry, the bus, the router, 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)
acc, err := app.Invoke(ctx, tag)

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 payload.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) Index(ctx context.Context, tenant, source string, paths ...string) (IndexResult, error)
  • func (a *App) Invoke(ctx context.Context, tag envelope.Tag) (runtime.Acceptance, error)
  • func (a *App) Knowledge() knowledge.Store
  • func (a *App) Providers() []string
  • func (a *App) Registry() registry.Store
  • func (a *App) Router() *router.Router
  • 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, tag envelope.Tag) (agentrt.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 agentrt.Compaction) Option
  • func WithConcurrency(n int) Option
  • func WithConnector(c connector.Connector) Option
  • func WithDurable(l Ledger, q dqueue.Queue) Option
  • func WithEmbedder(e knowledge.Embedder) Option
  • func WithExecutor(x runtime.Executor) Option
  • func WithGitHub(c GitHub) Option
  • func WithJira(c Jira) Option
  • func WithKnowledge(s knowledge.Store) Option
  • func WithLogger(log *slog.Logger) Option
  • func WithMetrics(rec metrics.Recorder) Option
  • func WithModelProvider(name string, fn agentrt.ProviderFunc) Option
  • func WithOpenAICompatible(ep agentrt.OpenAICompatible) Option
  • func WithPersonCredentials(r mcptools.PersonResolver) Option
  • func WithPostgres(pool *pgxpool.Pool) Option
  • func WithRegistry(s registry.Store) 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 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 raise tags 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.

registry  →  runtime  →  bus  →  router  →  connector sinks
                ↑                              ↓
            anchor worker              a surface somewhere

Nothing here is a stub standing in for a deployment. The in-memory registry, 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 payload.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) 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) Invoke

func (a *App) Invoke(ctx context.Context, tag envelope.Tag) (runtime.Acceptance, error)

Invoke accepts a tag: it registers the tag's routes with the router and then accepts the run durably. It returns before any inference.

The order matters: routes are registered before the run is accepted, because accept publishes the run's first event and an event for a run the router has no routes for has nowhere to go. RunID is a pure function of the tenant and tag id, so the routes can be registered before the run exists.

The tag's Tenant and Actor are trusted as given. An embedding program that accepts tags from callers must set both from the caller's credential, never from a request body.

func (*App) Knowledge

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

Knowledge is the indexed corpus agents are grounded in.

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) Registry

func (a *App) Registry() registry.Store

Registry is the agent store.

func (*App) Router

func (a *App) Router() *router.Router

Router is the delivery router.

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.

func (*App) RunDelivery

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

RunDelivery runs only the router: 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 and the triggers: it executes runs and accepts what connectors raise, 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 lifecycle.completed, failed or stopped 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, tag envelope.Tag) (agentrt.Granted, error)

ToolsFor reports what a run of tag would be granted, without accepting it: the tools the current revision of tag's agent in tag'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 tag'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 tag is trusted as given, as Invoke trusts it, and needs only Tenant and Agent.

It asks the App's sandbox. An executor passed by WithExecutor is asked too if it has a method ToolsFor(context.Context, runtime.Request) (agentrt.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 answered by DefaultAgent, and with none it raises nothing.
	Login, DefaultAgent 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 agentrt.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 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 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 WithExecutor

func WithExecutor(x runtime.Executor) Option

WithExecutor replaces the playscape sandbox a turn runs in. Model providers, tool sources, compaction and knowledge are wired into the sandbox, so an executor passed here 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 WithMetrics

func WithMetrics(rec metrics.Recorder) Option

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

func WithModelProvider

func WithModelProvider(name string, fn agentrt.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 agentrt.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 tag 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 registry, the run ledger and queue, knowledge, thread journals 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 WithRegistry

func WithRegistry(s registry.Store) Option

WithRegistry sets the agent store. The store decides which providers it accepts (registry.WithProviders); App.Providers lists the names this App can run.

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 tag a connector raises (a trigger has no credential to derive one from), mentions on Slack, GitHub and Jira resolve agents in it, 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 raise 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 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
	Rev     int
	Origin  string
	Topic   topic.Topic
	Status  Status
	Tag     envelope.Tag
	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 *payload.RunUsage
	// Question is what the run is waiting on a person for, set while
	// Status is StatusParked.
	Question *payload.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
	// DefaultAgent answers a mention of the app that names no agent.
	DefaultAgent string
	// Bots maps a Slack user id to the agent it stands for.
	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 lifecycle.* event kinds: 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.