fiber

README

fiber

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

Install · Report Bug · Go Docs

   

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

fiber is a Go library for agents and a low-latency pub/sub bus. Store an agent as a saige .agent.md definition, make it a member of a channel, and publish a typed event there: a Slack message, a GitHub comment, a Jira issue or a schedule. The run's events stream to any subscriber and are delivered wherever the channel leads. You compose it in Go: there is no service, no configuration file and no environment variable. Every backend is a value passed to fiber.New, and anything you do not pass is the in-memory implementation.

Where an event arrives and where its answer lands are independent. A GitHub comment can stream into Slack while a webhook receives only its completion event. Each run pins its definition's version and digest, so a retry never runs a newer prompt.

Contents

• Concepts
• Features
• Installation
• Quick Start
• Examples
• Usage
• Configuration
• API
• Development
• Documentation
• License

Concepts

Concept	What it is
Event	One event.Envelope (id, source, type, subject, channel, time, hops, plus tenant, run, seq on run events) whose data is a typed body. type is the only discriminator and carries the version: message.posted.v1, github.pull_request.opened.v1, cron.tick.v1, run.delta.text.v1, run.completed.v1.
Catalog	Every type is registered once with its decoder, a renderer that turns it into model input, and a sensitivity flag. A type this build does not know decodes to event.Unknown and keeps its bytes.
Channel	Where an event is published, a URI: slack://T01/C02, github://acme/handbook/issues/42, cron://acme/nightly. A run publishes on the channel of the event that started it.
Subscription	A tenant, channel prefixes and type prefixes. Default deny: an empty list matches nothing. run.usage and run.context arrive only when named.
Definition	A saige .agent.md file plus a fiber Policy sidecar, stored insert only. A run pins its name, version, saige digest, sub-agent versions and policy hash, and fails terminally if any of them moved.
Membership	Which agents a channel's events wake, and for which types. Default deny, and by default only for events that name the agent. An agent's own output never wakes it, and a hop limit stops two agents answering each other forever.
Runtime	turn.Runtime, the one seam that runs a turn: a pinned definition and an input event in, catalog events out. saige is the default; bring your own with fiber.WithTurnRuntime.
Adapters	Ingress adapters (Slack, GitHub, Jira, cron) publish typed events and choose no agent. Egress adapters (Slack, GitHub, Jira, webhook) subscribe by channel and type.

Features

• Agents are documents. Store a saige definition and it runs: no redeploy, no code, no restart. A run accepted under version 1.0.0 executes as 1.0.0 forever, even after 2.0.0 is stored.
• Subscribe by channel and type. Every adapter subscribed to a run's channel receives it, and each picks text, citations, run states, or all of them by type prefix.
• Connectors act, not just talk. Connectors contribute native verbs (transition a Jira ticket, request a review) as sandboxed agent tools, each emitting an auditable run.action.taken.v1 event.
• Backends are values. fiber.WithPostgres(pool) puts every store on one pool; fiber.WithBus takes the in-memory, Redis Streams or Postgres bus. Memory is a complete implementation, not a stub.
• Citations are first-class events, carrying the exact quoted source and resolvable provenance.
• Every run reports what it spent. Each model call is a run.usage.v1 event (tokens, cache share, model, context window, and cost in int64 micro-units for priced models), and the run's total is on its terminal event and App.Status.
• A run can wait for a person. A tool listed in a definition's approval.ask, or the built-in ask_person tool, parks the run on a question. The run frees its worker, survives restarts, and resumes with the first answer (App.Answer) or takes its default when the question expires.

Early development: v0 alpha, pre-1.0, not recommended for production. Interfaces may change between minor versions. The in-memory composition, durable replay and the Slack, GitHub, Jira, cron and webhook connectors are tested; the Postgres backends, knowledge layer, built-in author agent and history compaction pass CI's conformance suites but have not seen production load. See Limitations.

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 stores an agent, stores a second version, answers a simulated GitHub event on a channel a Slack sink and a webhook subscribe to, shows a third party listening to the channels, and demonstrates that a replayed run does not pay for the model twice.

Compose it

An agent is a saige definition: YAML frontmatter, then the system prompt.

---
apiVersion: saige/v1
name: docs-bot
version: 1.0.0
model:
  use: anthropic/claude-haiku-5-5   # offline/scripted runs with no key
---
You answer questions about the deploy pipeline. Cite your sources.

Store it, make it a member of a channel, and publish a message there:

app, err := fiber.New(ctx, fiber.WithTenant("acme")) // memory for everything
if err != nil {
    return err
}
defer app.Close()
go app.Run(ctx) // delivery and execution, in this process

// 1. Store the agent. Put is insert only: a version never changes.
if _, err := app.Definitions().Put(ctx, definition.Record{
    Tenant: "acme", CreatedBy: "ada", Content: docsBot,
}); err != nil {
    return err
}
// 2. Make it a member of a channel. It wakes for messages that name it.
if err := app.Memberships().Join(ctx, "acme", "slack://T01/C02", "docs-bot",
    membership.Rule{Types: []string{"message"}}); err != nil {
    return err
}
// 3. Listen to the channel's run events.
stream, err := app.Bus().Subscribe(ctx, bus.Subscription{
    Tenant: "acme", Channels: []string{"slack://T01/C02"}, Types: []string{"run"},
})
if err != nil {
    return err
}
defer stream.Close()
// 4. Publish a message. The channel's members decide who answers.
if err := app.Publish(ctx, event.Envelope{
    ID: "msg-1", Tenant: "acme", Time: time.Now(),
    Source: "slack://T01", Channel: "slack://T01/C02", Type: event.TypeMessagePosted,
    Data: event.MessagePosted{Request: event.Request{
        Text: "how does a release get out the door?", Mentions: []string{"docs-bot"},
    }},
}); err != nil {
    return err
}
for {
    e, err := stream.Recv(ctx)
    if err != nil {
        return err
    }
    switch b := e.Data.(type) {
    case event.Text: // run.delta.text.v1
        fmt.Print(b.Text)
    case event.Completed, event.Failed, event.Stopped:
        fmt.Println("\n" + e.Type)
        return nil
    }
}

For production, pass your backends: fiber.WithPostgres(pool) for every store and fiber.WithBus(bus.NewPostgres(pool)) or fiber.WithBus(bus.NewRedis(client)) for the bus. examples/downstream is a complete program.

Examples

Example	Description
local	Offline definitions and versions, selective delivery, citations, and replay, composed by fiber.New
downstream	A program composing fiber with its own model provider and tool

See examples/ for all runnable examples.

Usage

Use a model

A definition names its model as provider/model in model.use. Set ANTHROPIC_API_KEY for anthropic/.... To use OpenAI, or any server that speaks its Chat Completions API (vLLM, LM Studio, llama.cpp, OpenRouter, Groq), use openai/<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

offline/scripted calls no model and needs no key. Models the price catalog does not know report their token usage and no cost.

Fall back to other models

A definition 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):

model:
  use: anthropic/claude-haiku-5-5
  fallback: [openai/gpt-5, ollama/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. The Policy's 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 definition, so reordering fallbacks is a new version.

Grant MCP tools

A definition is granted an MCP provider's tools by naming the server:

tools:
  mcp:
    - server: urmzd
approval:
  ask: [urmzd__create_issue]

That grants the tools an admin accepted on the provider urmzd as read. Its write tools are granted only when the Policy names the server in mcp_write ("mcp_write": ["urmzd"]). Read only is the default, so leaving it out 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. 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 version; 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/turn/native alone supplies groups with native.WithToolGroups.

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 the event that started it, 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 event.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
    }))

Bring your own Runtime

A turn runs through one interface, turn.Runtime: a pinned definition and an input event in, catalog event bodies out.

type Runtime interface {
    Turn(ctx context.Context, def turn.Pinned, in turn.Input,
        emit func(context.Context, event.Body) error) (turn.Result, error)
}

The default wraps saige (pkg/turn/native) inside a playscape sandbox. Pass your own with fiber.WithTurnRuntime; turn.Func adapts a function, and sandbox.Sandboxed(inner, plane) puts any Runtime inside playscape under the pinned policy. Durability stays anchor's whatever runs the turn: the runtime pins the definition, journals what you emit, and replays from the ledger. Hold an implementation to the contract with runtimetest.Run.

app, err := fiber.New(ctx, fiber.WithTurnRuntime(turn.Func(
    func(ctx context.Context, def turn.Pinned, in turn.Input, emit func(context.Context, event.Body) error) (turn.Result, error) {
        text := "echo: " + in.Trigger.Type
        if err := emit(ctx, event.Text{Text: text}); err != nil {
            return turn.Result{}, err
        }
        return turn.Result{Text: text}, nil
    })))

Deliver to a webhook

The outbound webhook is an egress adapter. Each endpoint names the channels and type prefixes it receives, and POSTs one event per request:

fiber.WithWebhooks(map[string]webhook.Endpoint{
    "acme": { // webhook://acme/... addresses
        BaseURL:  "https://hooks.example.com/fiber",
        Secret:   deploySecret,
        Channels: []string{"github://acme/"},
        Types:    []string{"run.completed", "run.failed"},
    },
})

The body is version 2, the envelope itself in CloudEvents shape, with the header Fiber-Event-Version: 2:

{"specversion": "1.0", "id": "...", "source": "fiber://run/r1", "type": "run.completed.v1",
 "channel": "github://acme/handbook/issues/42", "time": "2026-10-28T14:03:11Z", "hops": 1,
 "tenant": "acme", "run": "r1", "seq": 200, "agent": "docs-bot", "version": "1.0.0",
 "digest": "sha256:...", "data": {"text": "..."}}

Every request is signed with the endpoint's secret and carries X-Fiber-Delivery: <run>/<seq>, the same on every retry, for the receiver to dedupe on. A body the catalog marks sensitive (human-authored text, private host data) is sent with its data left out and "redacted": true unless the endpoint sets Sensitive. Old Kinds selectors are still read; see Upgrading.

Validate a definition locally

fiber run runs one agent definition once, in process, and prints what the run does. It composes fiber.New with the in-memory backends and a temporary workspace, stores the definition, publishes one message.posted.v1 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.

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.agent.md "how does a release get out the door?"

[run.accepted.v1] docs-bot version 1.0.0, run r05babec9a8ca80435267133937f7a93c
[run.started.v1] attempt 1
docs-bot is running with the offline provider, so no model was called. ...
[run.usage.v1] scripted: 0 tokens in (0 cached), 0 out
[run.completed.v1] attempt 1, 1 model calls, 0 tokens in, 0 out

Each line is one event under its type, 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 definition's fallbacks are dropped with its provider.
-provider NAME, -model NAME	Replace the definition'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).
-policy FILE	Reads the agent's Policy sidecar from JSON. Without it the policy is the zero value, which grants nothing. A field that does not exist is an error rather than ignored.
-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 envelope as the bus carries it. With -tools, the listing as one object.
-timeout D	Gives up after D, such as 30s (default 5m; 0 waits forever).

Flags go before DEFINITION. DEFINITION is an .agent.md file, or - for standard input.

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, the definition or the policy 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 sub-agent the definition names (only the one definition is stored, so a delegation finds no agent). 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 the Policy's knowledge retrieves nothing. To list what a run would be granted in your own composition, call App.ToolsFor(ctx, event, agent) there.

Extend fiber

Add your own model providers, tools, connectors and MCP tool providers as options to fiber.New:

app, err := fiber.New(ctx,
    fiber.WithModelProvider("house", houseProvider),
    fiber.WithOpenAICompatible(native.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))).

To add a backend, implement the package's interface and pass its conformance suite: bustest, definitiontest, membershiptest, knowledgetest, threadtest, toolprovidertest, blobtest, and runtimetest for a turn.Runtime.

Upgrade

Drain runs before upgrading across a breaking release: a run recorded under an older shape fails terminally on replay. Each release's steps (the event union, the move from the server binary to a library, the rename from mandatum) are in docs/guides/upgrading.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_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 native.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 the 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 events or stop and answer requests from callers sets the tenant and actor from the caller's credential.
• Citations are derived from RAG context blocks. Citations a model or tool reports itself are not translated into events yet.
• The bus is at-least-once with no consumer groups.
• fiber refuses the saige definition fields it cannot honor (dials, skills, memory, guardrails, compaction, harness tools, MCP allow lists) rather than ignoring them.
• A Policy's subscriptions are recorded and hashed but do not join channels yet. Memberships (App.Memberships().Join, or a cron schedule, which joins its own agent) decide who an event wakes.
• 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 sink that subscribes only to run.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. native.WithStrictTools fails the build with native.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; native.WithStrictTools restores the old refusal.
• Tool events carry the call's arguments, capped at 8 KiB; scrub sensitive ones with native.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 Policy sets self_approval, and approvers/answerers restrict who may answer. tagged_only is a hint for hosts: an agent that sets it runs only when a message names it, and is never asked to take an unnamed one on its own; fiber itself carries and hashes it, nothing more. A tool implementing native.Approving decides per call whether it needs approval and may name that call's approvers, which then replace the Policy'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 the 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 sinks that subscribe to run.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.

Configuration

Pass every choice as an option to fiber.New. Read files or the environment in your own program and turn them into these options; fiber reads neither.

Option	Chooses	Default
WithPostgres(pool)	Definitions, memberships, run ledger and queue, knowledge, threads, tool providers on one pool	memory
WithBus(b)	bus.NewMemory, bus.NewRedis(client), bus.NewPostgres(pool)	memory
WithDefinitions, WithMemberships, 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
WithTurnRuntime(rt)	The turn.Runtime a turn runs on	saige (pkg/turn/native) in the playscape sandbox
WithRuntimeOptions, WithWorkerOptions, WithEgressOptions	Options for the durable runtime, the worker and egress	none
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. Set up connectors with the GitHub and Slack guides.

API

Method	Does
App.Publish	Puts an ingress event on the bus; the channel's members each start a run
App.Definitions().Put	Stores an agent definition (insert only)
App.Memberships().Join	Makes an agent a member of a channel prefix, with the types it wakes for
App.Runtime().Accept	Starts one agent's run for an event, without membership
App.Bus().Subscribe	Streams a tenant's events by channel and type prefix
App.Status, App.Stop, App.Compact, App.Answer	Act on a run
App.ToolsFor	Lists what an agent's run of an event would be granted, without running it
App.WebhookHandler()	An http.Handler for the connectors that receive webhooks, for your program to mount

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

Development

Action	Command
Set up	make init
Build	make build
Test	go test ./... (-race for the race detector)
Lint	golangci-lint run
Quality gate	make check
Start Postgres and Redis	make up (stop with make down)
Run the suites on real backends	make integration

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.

Documentation

Document	Covers
docs/README.md	Index of every document below
docs/architecture/events.md	The event union, catalog, channels, definitions and pins, the Runtime seam, adapters
docs/architecture/overview.md	Channels, membership, delivery, durability, stopping, usage, parking, credentials, limitations
docs/reference/github.md	GitHub connector capabilities and what it publishes
docs/reference/slack.md	Slack connector capabilities and what it publishes
docs/guides/upgrading.md	Upgrading across breaking releases: event union, server binary, rename
examples/README.md	Runnable example index
AGENTS.md	AI-facing conventions, commands, rules, extension guide
CONTRIBUTING.md	Development workflow and commit convention
SECURITY.md	Security boundaries and reporting

License

Apache 2.0. See LICENSE.

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.