ovr

package module
v0.5.5 Latest Latest
Warning

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

Go to latest
Published: Jun 14, 2026 License: Apache-2.0 Imports: 41 Imported by: 0

README

Ouvrier

Workers for your APIs.

Ouvrier is a Go framework for building small agentic middleware services. You declare a trigger, one or more LLM-powered pipes, the tools they can use, and the outcome. Ouvrier turns that declaration into a service with typed results, tool governance, traces, durable state, and deployment-oriented defaults.

package main

import (
	"context"
	"log"

	ovr "github.com/ArnaudGuiovanna/ouvrier"
)

type Ticket struct {
	ID      string `json:"id"`
	Subject string `json:"subject"`
	Body    string `json:"body"`
}

type Triage struct {
	Priority string `json:"priority"`
	Summary  string `json:"summary"`
}

func LoadTicket(ctx context.Context, args struct {
	ID string `json:"id"`
}) (Ticket, error) {
	return Ticket{ID: args.ID, Subject: "Login issue", Body: "User cannot sign in."}, nil
}

func main() {
	if err := ovr.Run(":8080",
		ovr.From("POST /tickets/{id}"),
		ovr.Pipe("Triage the support ticket.",
			ovr.Model("anthropic/claude-sonnet-4-6"),
			ovr.Tool("load_ticket", LoadTicket,
				ovr.ReadOnly(),
				ovr.Describe("Load one support ticket by ID."),
				ovr.Param("id", "Ticket identifier from the request path."),
			),
			ovr.Output[Triage](),
		),
		ovr.Reply(ovr.JSON[Triage]()),
	); err != nil {
		log.Fatal(err)
	}
}

Status

Ouvrier main includes the completed v0.1, v0.2, v0.3, and v0.4 milestone backlog plus the v0.5 Ouvrier Agent Cockpit. The latest tagged release is v0.5.5 — the Pi/Claude-Code-class agentic cockpit plus a new Ouvrier IDE (ouvrier ide): a terminal IDE to review/edit/save/rebuild a worker with live gopls diagnostics, hover, go-to-definition, completion, an Ouvrier API snippet palette, and a Catppuccin Macchiato theme. The cockpit keeps its streaming TUI, approval gate (prod double-confirm), and Codex subscription auth. The public Go module path is:

module github.com/ArnaudGuiovanna/ouvrier

Generated projects import the framework under that path. During local development before the repository is published, scaffolded projects add a replace github.com/ArnaudGuiovanna/ouvrier => <local-checkout> directive so they keep building against the working tree.

The CI gate (.github/workflows/ci.yml) runs gofmt, go vet, staticcheck, go test ./..., race tests on concurrency-sensitive packages, and builds the ouvrier CLI on every push and pull request.

What ships in the current codebase:

  • HTTP, Cron, Webhook, and Stream (NATS / Redis / Kafka) triggers with worker pools, signed inbound HMAC verification, and idempotency keys.
  • Sequential, Parallel, Map, Concurrency, and PartialOK composition.
  • The full SOTA harness: Session, EventStream, HookBus, StateStore (memory + embedded SQLite), PermissionPolicy, Sandbox, ToolExecutor, ResultSchema, governed SubAgent/Task.
  • Go tools, MCP through the official Go SDK, Skill loading with frontmatter validation, and a sandboxed Bash capability with workspace, env allowlist, timeout, output bounds, and process cleanup.
  • Provider adapters for anthropic/*, openai/*, ollama/*, mistral/*, gemini/*, vllm/*, groq/*, deepseek/*, azure/*, and bedrock/* with tool use, typed final results, and classified errors.
  • Reply (JSON[T], SSE, Accepted), Push (webhook, queue), and Sink (log, file) outputs.
  • Native provider token streaming through EventStream and Reply(SSE()), human-in-the-loop suspend/resume approvals, scoped persistent memory, model fallback chains, provider budgets, and cost accounting from pricing tables.
  • Stream DLQ routing, replay tooling, broker backpressure/ack policy controls, and queue push terminals for Kafka, Redis, NATS, SQS, and HTTP(S).
  • Protected admin endpoints (/admin/health, /admin/status, /admin/plans, /admin/capabilities, /admin/events, /admin/traces, /admin/traces/<id>, /admin/trigger, /admin/approvals, /admin/streams/replay), a Prometheus /metrics endpoint, plus a dev-mode trace viewer at /dev.
  • OpenTelemetry-compatible Tracer hook (and a built-in native OTLP/HTTP span exporter via WithOTLPExporter) for pipeline, pipe, session, LLM, tool, schema, and subagent spans.
  • The ouvrier CLI: version, new, show, status, logs, trace, add agent|trigger|tool|skill, dev, build (static cross-compile), and deploy <env>|ssh|docker, plus operate for local agentic worker building and ide for the worker review/edit IDE (gopls diagnostics + Ouvrier snippets).

Operational notes: trace persistence uses the configured StateStore (SQLite by default), cost accounting depends on configured pricing and provider usage metadata, and DLQ replay drains the runtime-retained copy for the configured stream plan rather than consuming arbitrary broker DLQ topics.

Mental Model

An Ouvrier worker has four parts:

From -> Pipe -> Pipe -> Reply / Push / Sink
  • From receives work from HTTP, cron, webhooks, or streams.
  • Pipe runs one LLM agent step with a goal, model, tools, skills, MCP, and typed output.
  • Tool exposes Go functions to the agent through a single governed executor.
  • Reply, Push, and Sink decide what happens to the final outcome.

The public model stays small: trigger, goal, tools, outcome. Internally, every Pipe runs through the Ouvrier harness: session, tool executor, sandbox, permission policy, hook bus, event stream, state store, result schema, and subagent task execution.

Requirements

  • Go 1.25 or newer.
  • A provider credential for hosted models, for example ANTHROPIC_API_KEY.
  • Optional local model runtime for ollama/*.

The repository is currently verified with:

go version go1.25.6 linux/amd64

Install

Download a prebuilt binary (Linux/macOS, amd64/arm64) with the install script; it fetches the latest release, verifies the checksum, and installs ouvrier:

curl -fsSL https://raw.githubusercontent.com/ArnaudGuiovanna/ouvrier/main/install.sh | sh

Pin a version with OUVRIER_VERSION=v0.5.5 and the target directory with OUVRIER_BIN_DIR=~/.local/bin. Re-running the script updates an existing install in place.

Or install with go install (Go 1.25+); it places ouvrier in $(go env GOPATH)/bin:

go install github.com/ArnaudGuiovanna/ouvrier/cmd/ouvrier@latest
ouvrier version

Use @v0.5.5 instead of @latest to pin a specific release.

To build from a checkout instead (for contributing or running the tests):

git clone https://github.com/ArnaudGuiovanna/ouvrier.git
cd ouvrier
go test ./...
go install ./cmd/ouvrier
ouvrier version

During local development, generated projects use a replace directive pointing back to this checkout. They also include ouvrier.worker.json, a small machine-readable worker manifest for editor/agent integrations. A prototype Pi extension lives in integrations/pi-ouvrier/; it discovers these manifests, streams /admin/events, and exposes an Ouvrier Inbox inside Pi.

Create A Worker

The non-interactive scaffold is the fastest way to create a development worker. It generates compiling HTTP, cron, webhook, and stream workers:

ouvrier new \
  --yes \
  --name ticket-triage \
  --trigger "POST /tickets/{id}" \
  --model "anthropic/claude-sonnet-4-6" \
  --dir /tmp

The --trigger flag accepts every supported trigger category. HTTP workers reply with JSON; cron, webhook, and stream workers terminate with Sink(Log()):

--trigger "POST /tickets"             # http  -> Reply(JSON)
--trigger "0 6 * * *"                 # cron  -> Sink(Log())  (or "cron 0 6 * * *")
--trigger "webhook github"            # webhook -> Sink(Log())
--trigger "stream kafka://tickets"    # stream  -> Sink(Log())

Then run it:

cd /tmp/ticket-triage
export ANTHROPIC_API_KEY=...
go run .

Call the worker:

curl -X POST http://localhost:8080/tickets/T-123 \
  -H 'Content-Type: application/json' \
  -d '{"message":"user cannot sign in"}'

Core API

Triggers
ovr.From("POST /tickets/{id}")
ovr.From("GET /health")
ovr.From(ovr.Cron("0 6 * * *"))
ovr.From(ovr.Webhook("github"))
ovr.From(ovr.Stream("kafka://tickets"))

HTTP triggers are the primary server runtime. Cron plans with Push or Sink terminals now execute through the same harness path when run as a cron-only runtime. Webhook trigger plans now expose POST /webhooks/<provider> routes, wrap the request as {trigger, provider, body}, and reuse the same signature, idempotency, harness, state, event, and policy guarantees as HTTP triggers. Mixed HTTP, Webhook, Cron, and Stream runtimes can run together. Stream receivers support NATS, Redis Streams, and Kafka boundaries, reserve message IDs in the state store when available, and emit stream dead-letter events for failed delivery handling.

Stream production hardening:

  • StreamDLQ(target, maxAttempts) retries a poisoned message up to maxAttempts deliveries, then routes it to the configured dead-letter target. The target is published to the real broker transport (kafka://, nats://, redis://) through the same producer machinery as the Queue push terminal.
  • StreamMaxInFlight(limit) bounds concurrently processed deliveries so a slow handler applies backpressure to the broker.
  • StreamAckPolicy(StreamAckAuto | StreamAckManual) selects the per-broker acknowledgement mode. StreamAckManual leaves acking to the handler; brokers whose receiver exposes no ack closure treat it as a no-op.
  • Replay a dead-letter queue with ReplayStreamDLQ in-process, or via the admin endpoint POST /admin/streams/replay (same admin auth as other /admin/* routes), which drains and reprocesses the runtime-retained copy for the configured stream plan and returns the replayed count. Broker DLQ targets are still published over the real transport; replaying an arbitrary broker topic or stream requires a dedicated consumer/custom DLQ integration. Dead-letter targets are credential-stripped in events/logs.
Pipes
ovr.Pipe("Summarize the request.",
	ovr.Model("openai/gpt-5.1"),
	ovr.Output[Summary](),
)

Every Pipe must declare an explicit model. Model IDs use a provider prefix:

  • anthropic/*
  • openai/*
  • ollama/*
  • mistral/*
  • gemini/*
  • vllm/*
  • groq/*
  • deepseek/*
  • azure/* (Azure OpenAI deployments)
  • bedrock/* (AWS Bedrock Converse models)
Tools

Tools are normal Go functions. The first parameter must be context.Context, and the return shape must be error or (value, error).

func LookupCustomer(ctx context.Context, args struct {
	Email string `json:"email"`
}) (Customer, error) {
	// Query your own system here.
}

ovr.Tool("lookup_customer", LookupCustomer,
	ovr.ReadOnly(),
	ovr.Describe("Find a customer by email address."),
	ovr.Param("email", "Customer email address."),
)

Classify tools deliberately:

  • ReadOnly() for side-effect-free work.
  • Idempotent("key.expression") for replay-safe side effects.
  • SideEffecting("label") for mutating work that needs explicit allowance.
  • RequiresApproval() for actions blocked unless policy allows them.

Ouvrier validates tool arguments against generated JSON Schema before calling the function. Tool panics become structured tool errors.

Skills

Reusable Markdown skills live under skills/<name>/SKILL.md and can be attached to a Pipe:

ovr.Pipe("Triage the ticket.",
	ovr.Model("anthropic/claude-sonnet-4-6"),
	ovr.Skill("ticket-triage"),
	ovr.Output[Triage](),
)

SKILL.md must include name and description frontmatter. Ouvrier loads the file through the workspace sandbox, injects the Markdown body into the system prompt in declaration order, and emits skill_loaded events without logging the skill body.

Typed Results

Use Output[T]() to force the final answer of a Pipe to match a Go type:

type Summary struct {
	Title string `json:"title"`
	Body  string `json:"body"`
}

ovr.Pipe("Summarize the ticket.",
	ovr.Model("anthropic/claude-sonnet-4-6"),
	ovr.NoCache(),
	ovr.Output[Summary](),
)

Ouvrier generates JSON Schema from T, injects it into the harness prompt, and validates final JSON strictly. Common model wrappers such as fenced json blocks or short prose around a JSON object/array are stripped before validation; the schema itself remains strict. Violations are recorded in EventStream and StateStore; bounded repair is supported by the harness where enabled. Static prompt-cache hints are enabled by default; NoCache() disables them for one Pipe.

Outputs
ovr.Reply(ovr.JSON[Summary]())
ovr.Reply(ovr.SSE())
ovr.Reply(ovr.Accepted())
ovr.Push(ovr.Webhook("https://example.com/result"))
ovr.Push(ovr.Queue("nats://127.0.0.1:4222/results"))
ovr.Push(ovr.Queue("kafka://broker:9092/results"))
ovr.Push(ovr.Queue("redis://127.0.0.1:6379/results"))
ovr.Push(ovr.Queue("sqs://sqs.us-east-1.amazonaws.com/123456789012/results"))
ovr.Sink(ovr.Log())
ovr.Sink(ovr.File("./out/result.json"))

Reply(JSON[T]()), Reply(SSE()), Reply(Accepted()), webhook push, HTTP/NATS/Kafka/Redis/SQS queue push, log sink, and file sink have runtime coverage. Push and file sink terminals run as governed output tools and require matching permission policy when they perform side effects.

Queue push schemes: http(s):// POSTs the body; nats:// publishes to the subject; kafka://host:9092/topic produces to the topic (via segmentio/kafka-go); redis://host:6379/stream appends to the stream with XADD (body stored under the body field); sqs://sqs.<region>.amazonaws.com/<account>/<queue> calls SendMessage as a SigV4-signed HTTPS request using the standard AWS_* environment credentials (no aws-sdk dependency). An idempotency_key query parameter is propagated where the protocol supports it: the Kafka message key, the SQS MessageDeduplicationId, and a Redis idempotency_key stream field.

SubAgents

A SubAgent exposes a child pipeline as a governed tool to a parent Pipe.

translator := ovr.Pipeline(
	ovr.Pipe("Translate the input to French.",
		ovr.Model("anthropic/claude-haiku-4-5"),
		ovr.Output[Translation](),
	),
)

ovr.Pipe("Draft a multilingual reply.",
	ovr.Model("anthropic/claude-sonnet-4-6"),
	ovr.SubAgent("translate", translator, ovr.MaxParallel(2)),
)

SubAgents run through ToolExecutor, inherit budgets, create child sessions, propagate cancellation, and attach child events to the parent trace.

Composition

Parallel fans out the same input to several Pipe branches and returns ordered outcomes. By default, one failed branch cancels sibling work and fails the composition:

ovr.Parallel(
	ovr.Pipe("Check quality", ovr.Model("anthropic/claude-haiku-4-5")),
	ovr.Pipe("Check compliance", ovr.Model("anthropic/claude-haiku-4-5")),
)

Map expects the previous step to return a JSON array and runs a Pipe sub-pipeline for each item with bounded concurrency:

ovr.Map(
	ovr.Concurrency(10),
	ovr.Pipe("Score one item", ovr.Model("anthropic/claude-haiku-4-5")),
)

PartialOK() changes Parallel, Map, and SubAgent failure behavior to return ordered partial outcomes instead of failing immediately.

Configuration

Provider configuration is read from the environment:

ANTHROPIC_API_KEY=
ANTHROPIC_BASE_URL=
OPENAI_API_KEY=
OPENAI_BASE_URL=
MISTRAL_API_KEY=
MISTRAL_BASE_URL=
GEMINI_API_KEY=
GEMINI_BASE_URL=
OLLAMA_BASE_URL=
VLLM_API_KEY=
VLLM_BASE_URL=
GROQ_API_KEY=
GROQ_BASE_URL=
DEEPSEEK_API_KEY=
DEEPSEEK_BASE_URL=
AZURE_OPENAI_API_KEY=
AZURE_OPENAI_BASE_URL=
AZURE_OPENAI_API_VERSION=
AWS_ACCESS_KEY_ID=
AWS_SECRET_ACCESS_KEY=
AWS_SESSION_TOKEN=
AWS_REGION=
AWS_BEDROCK_BASE_URL=

AWS Bedrock requests are authenticated with hand-rolled SigV4 signing (no aws-sdk dependency); set AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and AWS_REGION to enable the bedrock/* prefix. Real-AWS integration testing is tracked as follow-up; unit coverage exercises the signing helper and the Converse request/response mapping against a fake HTTP doer.

Runtime/admin configuration:

OUVRIER_ADMIN_TOKEN=
OUVRIER_STATE_BACKEND=sqlite
OUVRIER_STATE_PATH=.ouvrier/state.db

Admin endpoints are mounted under /admin/* for HTTP, webhook, cron-only, stream-only, and mixed runtimes. /admin/plans and /admin/capabilities return machine-readable compiled worker capabilities, and /admin/events streams redacted events as JSONL or SSE for live IDE integrations. Set OUVRIER_ADMIN_TOKEN outside local development when exposing a worker.

During local development, ouvrier dev auto-loads a .env file from the project directory into the worker's environment. The real process environment always takes precedence, so any variable you set explicitly is never overridden, and .env values are never printed. This is dev-only and does not affect deployed/prod binaries; pass --no-dotenv to opt out.

Observability

Ouvrier records structured events for pipeline, pipe, session, LLM, tool, permission, schema, budget, sink, and subagent task activity.

Runtime endpoints:

GET  /admin/health
GET  /admin/status
GET  /admin/plans          # compiled trigger/step/terminal capabilities
GET  /admin/capabilities   # alias shape for integrations
GET  /admin/events         # JSONL/SSE event stream (?format=sse, ?follow=false)
GET  /admin/traces?last=N
GET  /admin/traces/{exec-id}
POST /admin/trigger        # returns exec_id/trace_id/session_id when scheduled
GET  /metrics              # Prometheus text exposition (admin token required)

Events are redacted before persistence or admin output. The dev trace viewer is available at /dev behind the same admin authorization behavior.

GET /metrics renders a hand-rolled Prometheus exposition (counters and latency summaries derived from the EventStream/StateStore) behind the same admin auth. For external tracing, ovr.WithOTLPExporter("https://collector:4318") ships spans to an OTLP/HTTP collector with no OpenTelemetry SDK dependency.

CLI

Current commands:

ouvrier version
ouvrier new --yes --name NAME --trigger "POST /path" --model "provider/model"
ouvrier new
ouvrier add agent --name NAME --model "provider/model" [--goal TEXT]
ouvrier add trigger --trigger "cron @every 1h" [--model "provider/model"] [--goal TEXT]
ouvrier add tool --name LoadTicket [--describe TEXT] [--readonly|--side-effecting|--idempotent KEY]
ouvrier add skill --name ticket-triage [--description TEXT]
ouvrier show [--dir .] [--json]
ouvrier dev [--dir .] [--addr :8080]
ouvrier status [--url http://127.0.0.1:8080] [--token TOKEN]
ouvrier logs   [--url URL] [--token TOKEN] [--last N]
ouvrier trace  <exec-id> [--url URL] [--token TOKEN]
ouvrier build  [--static] [--target os/arch] [--output PATH] [--dir .]
ouvrier operate [--dir .] [--agent codex|manual]
ouvrier operate --print "create a worker that receives POST /tickets"
ouvrier operate --mode json --prompt "review this worker"
ouvrier operate --mode rpc
ouvrier operate create-worker --yes --name NAME --trigger "POST /path" --model provider/model [--dir .]
ouvrier operate patch --goal TEXT [--agent codex|manual] [--dir .]
ouvrier operate review-worker [--scope whole_worker] [--agent codex|manual] [--dir .]
ouvrier operate fix-worker [--session ID] [--agent codex|manual]
ouvrier operate audit [--session ID] [--dir .]
ouvrier operate build [--session ID] [--target os/arch] [--allow-failed]
ouvrier operate transfer --env ENV [--session ID] [--target os/arch] [--allow-failed]
ouvrier deploy ENV [--env-file FILE] [--identity FILE] [--target os/arch] [--keep 5] [--yes]
ouvrier deploy ssh --host HOST [--dir .] (same flags; bypasses the registry)
ouvrier deploy docker [--dir .] [--image IMAGE] [--tag TAG] [--push]
ouvrier fleet ls
ouvrier fleet rm <name> [--host HOST]
ouvrier console [--addr 127.0.0.1:7333] [--fleet PATH] [--token TOKEN] [--no-open]

ouvrier new opens the Bubble Tea v2 project wizard. The wizard and ouvrier new --yes support HTTP trigger strings such as "POST /tickets".

ouvrier operate is the prompt-first local agent cockpit for manufacturing workers. Run it from a worker or from a parent factory directory, then type the worker you want: the cockpit can infer a plan, scaffold a normal Go worker, load Ouvrier API context, ask the configured Codex/manual driver to patch code, review findings, run audit gates, build the binary, and transfer through the existing deploy engine. The same runtime powers the Bubble Tea UI, one-shot --print and --mode json, and JSONL --mode rpc.

Codex auth follows the Pi-style ownership boundary: /login codex probes or delegates to the local Codex CLI flow, while Ouvrier stores only profile metadata, never Codex subscription tokens. Sessions and artifacts live under .ouvrier/operate/sessions/<id>/ (transcript.jsonl, events.jsonl, tool-calls.jsonl, auth_profile.json, goal.md, patch.json, diff.patch, review.json, audit.json, build.json, and transfer.json) so a developer-operator can audit every step. The web console remains the remote observation and approval surface for workers and fleets; operate is the local construction harness.

The introspection commands (show, status, logs, trace) read from the project filesystem (pip.yaml) or talk to a running worker through /admin/* using OUVRIER_ADMIN_TOKEN for bearer auth. ouvrier build compiles the worker; --static implies CGO_ENABLED=0 with -ldflags="-s -w" and supports --target os/arch for cross-compilation (modernc.org/sqlite is pure Go, so static cross-builds work without a C toolchain).

ouvrier deploy <env> (and its deploy ssh --host alias) is the agentless SSH deploy: it cross-compiles a static binary, ships it into an immutable releases/<ts>-<sha>/ directory on hosts pinned in the committed ouvrier.known_hosts, installs the env file and a hardened systemd unit, atomically swaps the current symlink, health-gates the loopback admin endpoint, and rolls back to the previous release on failure. Environments (hosts, port, path, service, identity) come from the committed pip.yaml deploy: block; see the handbook for the sudoers snippet and a CI example. ouvrier deploy docker renders and builds a distroless container image that includes skills/ when present. ouvrier fleet lists or prunes the recorded deployments inventory at ~/.config/ouvrier/deployments.json (override with OUVRIER_FLEET_PATH or OUVRIER_CONFIG_DIR); it is a secret-free cache for tooling — live /admin/health is truth.

ouvrier console starts a loopback-only web console (default 127.0.0.1:7333, override with --addr / OUVRIER_CONSOLE_ADDR) that layers over the exact same SSH tunnels and /admin/* APIs as the headless commands: fleet overview with live tunnel state (fleet ls + tunnel status), a worker detail view (status/plans/traces, an SSE event tail, approvals, manual trigger — proxied to /admin/*), a fan-out overview (status --all), an event fan-in (logs), and a streamed deploy (deploy <env>). Every console capability has a headless CLI equivalent, enforced by a parity test. Security: the bind is refused on a non-loopback address unless OUVRIER_CONSOLE_INSECURE=1; a random 256-bit per-session token is printed in the URL fragment, held only in browser memory, sent as Authorization: Bearer, and constant-time compared; there are zero cookies, a Host-header allowlist and Origin rejection (DNS-rebinding defense), Cache-Control: no-store on the API, and a strict CSP plus X-Frame-Options: DENY on the SPA. The worker admin token is fetched into memory by the tunnel manager and injected server-side — it never reaches the browser. The SPA is vendored Preact + HTM ESM served from //go:embed: go build alone produces the whole console, with no npm/bundler in the repo or CI. --fleet overrides the inventory path and --no-open suppresses the browser auto-open.

Reference Examples

Two reference workers live under examples/:

  • examples/ticket-triage — the canonical README example with a typed Triage{Priority, Summary, Tags} reply, an in-memory load_ticket Go tool, and a ticket-triage skill.
  • examples/moodle-fsrs — a POST /reviews worker with a stub compute_fsrs tool and a typed Decision{NextDue, Stability, Difficulty, Lapses} reply.

Each example has its own go.mod with a local replace directive back to this checkout, and is exercised by the repo-level TestExamplesBuild golden build test to keep them from drifting away from the public API.

OpenTelemetry / Tracing

Ouvrier emits one span per pipeline, pipe, session, LLM call, tool call, schema validation, and subagent task through an OTel-compatible Tracer hook:

runner := ovr.NewRunner(ovr.WithTracer(myTracer))

myTracer implements ovr.Tracer (a StartSpan(ctx, name, attrs) factory that returns a ovr.Span with End, SetAttribute, and RecordError). Use ovr.NopTracer() to disable tracing or in tests. The harness pairs *_started events with their *_completed / *_failed counterparts internally, so callers see one span per logical operation.

Development

Run the default checks:

go test ./...
go vet ./...

Run race tests when touching concurrency, state, events, tools, or subagents:

GOCACHE=/tmp/go-build-cache go test -race ./...

Keep public behavior covered close to the owning package. Use small vertical slices that compile and can be demonstrated.

Security Model

Ouvrier assumes LLM output is untrusted:

  • Tool calls pass through ToolExecutor.
  • Privileged work passes through PermissionPolicy.
  • Typed outputs pass through ResultSchema.
  • Secrets are redacted before logs, events, state, and admin output.
  • Side effects must be classified and authorized.
  • Sandbox-backed capabilities must fail fast if requested guarantees cannot be enforced.

Documentation

User-facing documentation:

License

Ouvrier is licensed under the Apache License, Version 2.0. See LICENSE.

Created by Arnaud Guiovanna: https://aguiovanna.fr

Documentation

Overview

Package ovr exposes the public runtime primitives for Ouvrier.

Index

Constants

This section is empty.

Variables

View Source
var (
	// ErrPipelineEmpty means no pipeline nodes were provided.
	ErrPipelineEmpty = errors.New("pipeline is empty")
	// ErrFirstNodeNotFrom means the pipeline does not start with a From node.
	ErrFirstNodeNotFrom = errors.New("first node must be From")
	// ErrTerminalMissing means the pipeline has no Reply, Push, or Sink node.
	ErrTerminalMissing = errors.New("pipeline must include Reply, Push, or Sink")
	// ErrTerminalNotLast means a pipeline has nodes after its terminal before the next From.
	ErrTerminalNotLast = errors.New("terminal node must be last in its pipeline")
	// ErrMultipleTerminals means a pipeline has more than one terminal node.
	ErrMultipleTerminals = errors.New("pipeline must include exactly one terminal")
	// ErrIncompatibleTerminal means the terminal cannot be used with the trigger kind.
	ErrIncompatibleTerminal = errors.New("terminal is incompatible with trigger")
	// ErrPipeMissingModel means a Pipe node was declared without Model.
	ErrPipeMissingModel = errors.New("pipe must include Model")
	// ErrInvalidModelForm means a Model id was not in provider/model form.
	ErrInvalidModelForm = errors.New("model must use provider/model form")
	// ErrInvalidNode means a node was nil, unsupported, or otherwise invalid.
	ErrInvalidNode = errors.New("invalid node")
	// ErrIncompatibleTriggerOption means a From option was set on a trigger kind
	// that does not consume it (e.g. StreamDLQ on an HTTP trigger), which would
	// otherwise be silently ignored at runtime.
	ErrIncompatibleTriggerOption = errors.New("From option is incompatible with trigger")
)
View Source
var ErrMissingEnv = errors.New("required environment variable missing")
View Source
var ErrRunNotImplemented = errors.New("run runtime not implemented")

ErrRunNotImplemented means the requested runtime transport is not implemented yet.

Functions

func Handler added in v0.2.0

func Handler(nodes ...Node) (http.Handler, error)

Handler compiles nodes into an http.Handler for HTTP and Webhook triggers, ready to drive with net/http/httptest. It mirrors what Run serves, minus the listener, so a worker can be exercised end to end in a Go test. Cron and stream triggers are not served by the returned handler.

The handler uses an in-memory state store unless the Runner was built WithStateStore. Configure a Provider with WithProvider to avoid live model calls.

Handler always returns the combined surface — trigger routes, /admin/*, and /metrics on a single handler — regardless of OUVRIER_ADMIN_ADDR. The split admin listener is a property of Run's network layout, not of this in-process test seam: tests keep driving both trigger and admin routes through the one handler, exactly as before the split existed. Only Run starts a second listener (and removes admin routes from the public mux) when OUVRIER_ADMIN_ADDR is set.

func PartialOK

func PartialOK() partialOKOption

PartialOK lets a SubAgent return ordered error results for failed child tasks or lets Parallel/Map return ordered error outcomes instead of failing the whole parent pipeline immediately.

func RequireEnv

func RequireEnv(names ...string) error

func Run

func Run(addr string, nodes ...Node) error

Run validates a pipeline declaration before starting the default runtime.

func Validate

func Validate(nodes ...Node) error

Validate checks a pipeline declaration without starting any runtime work.

Types

type AcceptedReply

type AcceptedReply struct{}

func Accepted

func Accepted() AcceptedReply

Accepted declares an HTTP 202 reply while the pipeline continues asynchronously.

type ApprovalStatus added in v0.2.0

type ApprovalStatus string

ApprovalStatus is the public lifecycle state of a human-in-the-loop approval.

const (
	ApprovalPending  ApprovalStatus = "pending"
	ApprovalApproved ApprovalStatus = "approved"
	ApprovalDenied   ApprovalStatus = "denied"
)

type BackoffPolicy

type BackoffPolicy interface {
	// contains filtered or unexported methods
}

BackoffPolicy configures the delay between retry attempts.

func ExponentialBackoff

func ExponentialBackoff() BackoffPolicy

ExponentialBackoff configures retry attempts with an exponential delay.

type BashOption

type BashOption interface {
	// contains filtered or unexported methods
}

BashOption configures the sandboxed Bash capability.

func BashMaxOutputBytes

func BashMaxOutputBytes(max int) BashOption

BashMaxOutputBytes bounds captured stdout and stderr for one bash command.

func BashTimeout

func BashTimeout(value string) BashOption

BashTimeout configures the maximum wall-clock duration for one bash command.

func UnsafeBashHostExecution

func UnsafeBashHostExecution() BashOption

UnsafeBashHostExecution allows the host-shell Bash fallback.

Use this only for local/dev or trusted workloads. Ouvrier still enforces ToolExecutor permission decisions, environment allowlisting, working directory resolution, timeout, and bounded output, but this option does not provide a real OS sandbox.

type Budget

type Budget struct {
	MaxIterations int
	MaxTokens     int
	MaxCostUSD    float64
	MaxWallClock  time.Duration
}

Budget captures execution limits recorded with public sessions.

type CronTrigger

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

CronTrigger is a cron expression trigger source.

func Cron

func Cron(expr string) CronTrigger

Cron declares a cron trigger source for From.

type Effect

type Effect string

Effect describes the execution safety class declared on a Tool.

const (
	EffectReadOnly      Effect = "read_only"
	EffectIdempotent    Effect = "idempotent"
	EffectSideEffecting Effect = "side_effecting"
)

type Event

type Event struct {
	ID        uint64
	At        time.Time
	Kind      EventKind
	ExecID    string
	SessionID string
	TraceID   string
	Payload   map[string]any
}

Event is the public, redacted event shape passed through advanced hooks.

type EventKind

type EventKind string

EventKind identifies a runner lifecycle event that hooks can observe.

const (
	EventPipelineStarted   EventKind = "pipeline_started"
	EventPipelineCompleted EventKind = "pipeline_completed"
	EventPipelineFailed    EventKind = "pipeline_failed"

	EventPipeStarted   EventKind = "pipe_started"
	EventPipeCompleted EventKind = "pipe_completed"
	EventPipeFailed    EventKind = "pipe_failed"

	EventSessionStarted   EventKind = "session_started"
	EventSessionSaved     EventKind = "session_saved"
	EventSessionCancelled EventKind = "session_cancelled"

	EventLLMCallStarted   EventKind = "llm_call_started"
	EventLLMCallCompleted EventKind = "llm_call_completed"
	EventLLMCallFailed    EventKind = "llm_call_failed"
	EventLLMTokenDelta    EventKind = "llm_token_delta"
	EventModelFallback    EventKind = "model_fallback"

	EventToolCallStarted   EventKind = "tool_call_started"
	EventToolCallCompleted EventKind = "tool_call_completed"
	EventToolCallFailed    EventKind = "tool_call_failed"

	EventPermissionDecision  EventKind = "permission_decision"
	EventIdempotencyDecision EventKind = "idempotency_decision"
	EventSignatureDecision   EventKind = "signature_decision"

	EventApprovalRequested  EventKind = "approval_requested"
	EventApprovalApproved   EventKind = "approval_approved"
	EventApprovalDenied     EventKind = "approval_denied"
	EventExecutionSuspended EventKind = "execution_suspended"
	EventExecutionResumed   EventKind = "execution_resumed"

	EventHookFailed EventKind = "hook_failed"

	EventSchemaValidationPassed EventKind = "schema_validation_passed"
	EventSchemaValidationFailed EventKind = "schema_validation_failed"
	EventSchemaRepairStarted    EventKind = "schema_repair_started"
	EventSchemaRepairCompleted  EventKind = "schema_repair_completed"
	EventSchemaRepairFailed     EventKind = "schema_repair_failed"

	EventBudgetExceeded EventKind = "budget_exceeded"

	EventTaskStarted   EventKind = "task_started"
	EventTaskCompleted EventKind = "task_completed"
	EventTaskFailed    EventKind = "task_failed"

	EventSkillLoaded EventKind = "skill_loaded"

	EventStreamDeadLettered EventKind = "stream_dead_lettered"
	EventStreamRedelivered  EventKind = "stream_redelivered"

	EventCronLeaseAcquired EventKind = "cron_lease_acquired"
	EventCronLeaseLost     EventKind = "cron_lease_lost"
	EventCronTickSkipped   EventKind = "cron_tick_skipped"

	EventDurableRunPruneFailed EventKind = "durable_run_prune_failed"

	EventRunRecovered            EventKind = "run_recovered"
	EventRunAbandoned            EventKind = "run_abandoned"
	EventReplayIndeterminateTool EventKind = "replay_indeterminate_tool"

	EventSinkLogged EventKind = "sink_logged"
)

type Execution

type Execution struct {
	ExecID      string
	TraceID     string
	Status      ExecutionStatus
	StartedAt   time.Time
	CompletedAt time.Time
}

Execution is the public state-store representation of one pipeline execution.

type ExecutionStatus

type ExecutionStatus string

ExecutionStatus is the public state-store status for one execution.

const (
	ExecutionRunning   ExecutionStatus = "running"
	ExecutionCompleted ExecutionStatus = "completed"
	ExecutionFailed    ExecutionStatus = "failed"
	ExecutionTruncated ExecutionStatus = "truncated"
)

type FileSink

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

FileSink is a Sink target that writes the final outcome to a file.

func File

func File(path string) FileSink

File declares a file sink target.

type FromOption

type FromOption interface {
	// contains filtered or unexported methods
}

FromOption configures a From node.

func IdempotencyKey

func IdempotencyKey(header string) FromOption

IdempotencyKey prevents duplicate trigger deliveries based on a request header.

func StreamAckPolicy added in v0.2.0

func StreamAckPolicy(policy StreamAckMode) FromOption

StreamAckPolicy configures the per-broker acknowledgement mode for a stream trigger. See StreamAckAuto and StreamAckManual.

func StreamDLQ added in v0.2.0

func StreamDLQ(target string, maxAttempts int) FromOption

StreamDLQ routes a poisoned stream message to a dead-letter target once it has been redelivered maxAttempts times without success. maxAttempts counts total processing attempts before the message is dead-lettered.

func StreamMaxInFlight added in v0.2.0

func StreamMaxInFlight(limit int) FromOption

StreamMaxInFlight bounds the number of stream messages processed concurrently so a slow handler applies backpressure to the broker instead of buffering deliveries without bound.

func VerifySignature

func VerifySignature(envVar, header string) FromOption

VerifySignature verifies an HMAC-SHA256 request signature before executing a trigger.

func WorkerPool

func WorkerPool(limit int) FromOption

WorkerPool bounds concurrent trigger executions for a pipeline.

type Hook

type Hook func(context.Context, Event) (Event, error)

Hook observes or enriches a lifecycle event. Returning an error blocks that event.

type Hooks

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

Hooks is a public hook registration set consumed by NewRunner.

func NewHooks

func NewHooks() *Hooks

NewHooks creates an empty hook registration set.

func (*Hooks) Register

func (h *Hooks) Register(kind EventKind, hook Hook) error

Register adds a hook for one lifecycle event kind.

type JSONReply

type JSONReply[T any] struct {
	// contains filtered or unexported fields
}

JSONReply is a typed JSON reply format.

func JSON

func JSON[T any]() JSONReply[T]

JSON declares a typed JSON reply format.

type LogSink

type LogSink struct{}

LogSink is a Sink target that writes the outcome to logs and metrics.

func Log

func Log() LogSink

Log declares a log sink target.

type MapOption

type MapOption interface {
	// contains filtered or unexported methods
}

MapOption configures a Map composition step.

func Concurrency

func Concurrency(limit int) MapOption

Concurrency bounds concurrent Map item executions.

type MemoryRecord added in v0.2.0

type MemoryRecord struct {
	Scope     string
	Key       string
	Value     string
	UpdatedAt time.Time
}

MemoryRecord is the public state-store representation of one persistent, scoped agent-memory entry that survives across sessions.

type ModelRate added in v0.2.0

type ModelRate = provider.ModelRate

ModelRate captures USD-per-token rates for a single provider/model. Cache rates are optional and default to zero.

func PerMillion added in v0.2.0

func PerMillion(input, output, cacheRead, cacheWrite float64) ModelRate

PerMillion builds a ModelRate from USD-per-million-token rates, the common way vendors publish pricing.

type Node

type Node interface {
	// contains filtered or unexported methods
}

Node is a single step in an Ouvrier pipeline.

func From

func From(source any, options ...FromOption) Node

From declares the event source that starts a pipeline.

func Map

func Map(items ...any) Node

Map applies a Pipe sub-pipeline to each element of the previous JSON array.

func Parallel

func Parallel(items ...any) Node

Parallel fans out the current input to multiple Pipe branches.

func Pipe

func Pipe(goal string, options ...PipeOption) Node

Pipe declares one agent step in a pipeline.

func Push

func Push(target PushTarget) Node

Push terminates an asynchronous pipeline by sending the outcome elsewhere.

func Reply

func Reply(format ReplyFormat) Node

Reply terminates a synchronous pipeline by answering the caller.

func Sink

func Sink(target SinkTarget) Node

Sink terminates a pipeline without replying to the trigger source.

type OTLPExporterOption added in v0.2.0

type OTLPExporterOption = events.OTLPOption

OTLPExporterOption configures the OTLP exporter installed by WithOTLPExporter.

func OTLPHeaders added in v0.2.0

func OTLPHeaders(headers map[string]string) OTLPExporterOption

OTLPHeaders attaches additional HTTP headers (e.g. authorization for hosted collectors) to every OTLP export request.

func OTLPServiceName added in v0.2.0

func OTLPServiceName(name string) OTLPExporterOption

OTLPServiceName sets the service.name resource attribute on exported spans.

type ParallelOption

type ParallelOption interface {
	// contains filtered or unexported methods
}

ParallelOption configures a Parallel composition step.

type PendingApproval added in v0.2.0

type PendingApproval struct {
	ID         string
	ExecID     string
	SessionID  string
	TraceID    string
	ToolName   string
	ToolCallID string
	ToolKind   string
	Effect     string
	Reason     string
	Status     ApprovalStatus
	CreatedAt  time.Time
	DecidedAt  time.Time
	DecidedBy  string
}

PendingApproval is the public state-store representation of one human-in-the-loop approval request that suspends an execution until an operator approves or denies the gated tool call. The Reason is redaction-safe and never carries tool arguments or skill bodies.

type PermissionAction

type PermissionAction struct {
	Kind             PermissionActionKind
	ToolName         string
	ToolCallID       string
	ToolKind         string
	Target           string
	Effect           Effect
	IdempotencyKey   string
	SideEffects      []string
	RequiresApproval bool
}

PermissionAction is the public, auditable action passed to a PermissionPolicy.

type PermissionActionKind

type PermissionActionKind string

PermissionActionKind identifies an action that a runner policy can authorize.

const (
	PermissionActionToolCall    PermissionActionKind = "tool_call"
	PermissionActionPushWebhook PermissionActionKind = "push_webhook"
	PermissionActionPushQueue   PermissionActionKind = "push_queue"
	PermissionActionSinkFile    PermissionActionKind = "sink_file"
	PermissionActionSinkLog     PermissionActionKind = "sink_log"
)

type PermissionDecision

type PermissionDecision struct {
	Allowed    bool
	Reason     string
	Suspended  bool
	ApprovalID string
}

PermissionDecision is the policy verdict for one action.

type PermissionPolicy

type PermissionPolicy interface {
	Authorize(context.Context, PermissionAction) (PermissionDecision, error)
}

PermissionPolicy authorizes privileged runner actions.

func AllowSideEffectTargets

func AllowSideEffectTargets(label string, targets ...string) PermissionPolicy

AllowSideEffectTargets returns the default production policy plus an explicit target-scoped allowance for non-idempotent output actions such as Push and file Sink. Use "*" only when intentionally allowing every target for a label.

func AllowSideEffects

func AllowSideEffects(labels ...string) PermissionPolicy

AllowSideEffects returns the default production policy plus explicit non-idempotent side-effect allowances for non-targeted tool calls. Targeted actions such as Push, file Sink, MCP, and Bash require AllowSideEffectTargets.

type PipeOption

type PipeOption interface {
	// contains filtered or unexported methods
}

PipeOption configures a Pipe node.

func Bash

func Bash(sandbox SandboxConfig, options ...BashOption) PipeOption

Bash declares a bash command tool named "bash" for a Pipe.

By default Bash requires a platform sandbox that can enforce workspace, process, environment, and network boundaries. The runtime fails at startup when those guarantees are unavailable.

func Fallback added in v0.2.0

func Fallback(models ...string) PipeOption

Fallback declares an ordered list of fallback models for a Pipe. On a classified provider failure (transient, rate-limit, or auth) of the primary model, the harness tries each fallback model in order before failing. Permanent and validation failures never fall through.

func MCP

func MCP(name string) PipeOption

MCP connects an external Model Context Protocol server to a Pipe.

func MaxCostUSD

func MaxCostUSD(max float64) PipeOption

MaxCostUSD configures a Pipe cost budget in US dollars.

func MaxTokens

func MaxTokens(max int) PipeOption

MaxTokens configures a Pipe token budget.

func Model

func Model(id string) PipeOption

Model selects the explicit LLM model used by a Pipe.

func NoCache

func NoCache() PipeOption

NoCache disables provider prompt-cache hints for one Pipe.

func Output

func Output[T any]() PipeOption

Output declares the typed result schema expected from a Pipe.

func Retry

func Retry(max int, policies ...BackoffPolicy) PipeOption

Retry configures transient provider retries and retry-safe tool retries. Tool retries only apply to ReadOnly or Idempotent tools.

func SequentialTools

func SequentialTools() PipeOption

SequentialTools forces tool calls from one provider turn to execute one at a time.

func Skill

func Skill(dirName string) PipeOption

Skill registers a Markdown skill directory for a Pipe.

func SubAgent

func SubAgent(name string, pipeline PipelineSpec, options ...SubAgentOption) PipeOption

SubAgent exposes a child pipeline as a governed tool for a Pipe.

func Timeout

func Timeout(value string) PipeOption

Timeout configures a Pipe wall-clock budget using a Go duration string.

func Tool

func Tool(name string, goFunc any, options ...ToolOption) PipeOption

Tool registers a Go function as an agent tool for a Pipe.

type PipelineSpec

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

PipelineSpec is a child pipeline declaration used by SubAgent.

func Pipeline

func Pipeline(nodes ...Node) PipelineSpec

Pipeline declares a child pipeline that can be exposed to a Pipe as a SubAgent.

type PricingTable added in v0.2.0

type PricingTable = provider.PricingTable

PricingTable maps a "provider/model" identifier to its per-token rate. It is used to compute Usage.CostUSD per call and to aggregate total cost per execution. A nil or empty table leaves cost best-effort (zero).

type Provider added in v0.2.0

type Provider = provider.Provider

Provider is the LLM boundary the runtime calls for completions. Implement it — or use the ovrtest package's scripted provider — to drive a worker in tests without contacting a real model and without spending tokens. In production, providers are configured from environment credentials and this need not be set.

type PushTarget

type PushTarget interface {
	// contains filtered or unexported methods
}

PushTarget configures where Push sends the final outcome.

type QueueTarget

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

QueueTarget is a Push target that publishes the final outcome to a queue.

func Queue

func Queue(uri string) QueueTarget

Queue declares a queue push target.

type ReplyFormat

type ReplyFormat interface {
	// contains filtered or unexported methods
}

ReplyFormat configures how Reply serializes the final outcome.

type Runner

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

Runner owns advanced runtime configuration for Ouvrier pipelines.

func NewRunner

func NewRunner(options ...RunnerOption) *Runner

NewRunner creates an advanced runner. Configuration errors are returned by Run.

func (*Runner) Handler added in v0.2.0

func (r *Runner) Handler(nodes ...Node) (http.Handler, error)

Handler builds an http.Handler from this Runner's configuration. See the package-level Handler for details.

func (*Runner) Run

func (r *Runner) Run(addr string, nodes ...Node) error

Run validates a pipeline declaration before starting the runtime.

type RunnerOption

type RunnerOption func(*runnerConfig)

RunnerOption configures a Runner.

func WithHooks

func WithHooks(hooks *Hooks) RunnerOption

WithHooks installs advanced lifecycle hooks for the runner.

func WithOTLPExporter added in v0.2.0

func WithOTLPExporter(endpoint string, options ...OTLPExporterOption) RunnerOption

WithOTLPExporter installs a Tracer that ships spans to an OTLP/HTTP collector at endpoint (e.g. "https://collector:4318"); the exporter appends "/v1/traces". Spans are encoded as OTLP/HTTP JSON with no heavy otel SDK dependency, and attributes are redacted before export. It is a convenience wrapper over WithTracer; passing both means the last option wins. An empty endpoint is rejected. Default off: when this option is unset, behavior is unchanged.

func WithPermissionPolicy

func WithPermissionPolicy(permissionPolicy PermissionPolicy) RunnerOption

WithPermissionPolicy installs a production permission policy for privileged actions.

func WithPricing added in v0.2.0

func WithPricing(table PricingTable) RunnerOption

WithPricing installs a pricing table used to compute per-call and per-execution cost. When unset, cost stays best-effort (zero) with no behavior change.

func WithProvider added in v0.2.0

func WithProvider(p Provider) RunnerOption

WithProvider injects a single Provider used for every model id, bypassing the environment-derived provider registry. It is the seam that makes a worker unit-testable: pair it with Handler and httptest, or with a scripted provider from the ovrtest package.

func WithProviderBudget added in v0.2.0

func WithProviderBudget(provider string, maxInFlight int) RunnerOption

WithProviderBudget bounds the number of concurrent in-flight LLM calls for a single provider (the part before "/" in a model id). It prevents one provider's rate limit from stalling calls routed to other providers. The budget is shared across every Pipe in the runner. A limit less than or equal to zero is rejected.

func WithSandbox

func WithSandbox(sandbox SandboxConfig) RunnerOption

WithSandbox installs a filesystem workspace boundary for the runner.

func WithSchemaRepairAttempts

func WithSchemaRepairAttempts(max int) RunnerOption

WithSchemaRepairAttempts enables bounded ResultSchema repair attempts for runtime Pipe outputs.

func WithStateStore

func WithStateStore(store StateStore) RunnerOption

WithStateStore installs a custom durable execution store.

func WithTracer

func WithTracer(tracer Tracer) RunnerOption

WithTracer installs a Tracer that receives span lifecycle events for pipeline, pipe, session, LLM, tool, schema, and subagent activity. The harness pairs *_started events with their *_completed or *_failed counterparts internally; callers see one span per logical operation.

type SSEReply

type SSEReply struct{}

func SSE

func SSE() SSEReply

SSE declares a Server-Sent Events reply format.

type SandboxConfig

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

SandboxConfig describes an explicit filesystem workspace boundary.

func Sandbox

func Sandbox(root string, options ...SandboxOption) SandboxConfig

Sandbox configures a workspace root used for filesystem-bound capabilities.

type SandboxOption

type SandboxOption func(*SandboxConfig)

SandboxOption configures a sandbox boundary.

func AllowEnv

func AllowEnv(keys ...string) SandboxOption

AllowEnv allows selected environment variables inside sandboxed capabilities.

type SchemaViolation

type SchemaViolation struct {
	ID         uint64
	At         time.Time
	ExecID     string
	SessionID  string
	SchemaName string
	Error      string
}

SchemaViolation is the public state-store representation of one schema failure.

type Session

type Session struct {
	ExecID          string
	SessionID       string
	ParentSessionID string
	TraceID         string
	Model           string
	StartedAt       time.Time
	Budget          Budget
}

Session is the public state-store representation of one runtime session.

type SinkTarget

type SinkTarget interface {
	// contains filtered or unexported methods
}

SinkTarget configures where Sink records or discards the final outcome.

type Span

type Span = events.Span

Span is a single tracing span minted by a Tracer.

type StateStore

type StateStore interface {
	SaveExecution(context.Context, Execution) error
	Execution(context.Context, string) (Execution, bool, error)
	Executions(context.Context) ([]Execution, error)
	SaveSession(context.Context, Session) error
	Session(context.Context, string) (Session, bool, error)
	Sessions(context.Context) ([]Session, error)
	ReserveIdempotency(context.Context, string, string) (string, bool, error)
	AddEvent(context.Context, Event) (Event, error)
	Events(context.Context, string) ([]Event, error)
	EventsSince(context.Context, string, uint64) ([]Event, error)
	AddSchemaViolation(context.Context, SchemaViolation) (SchemaViolation, error)
	SchemaViolations(context.Context, string) ([]SchemaViolation, error)
	SaveMemory(context.Context, string, string, string) error
	Memory(context.Context, string, string) (string, bool, error)
	ListMemory(context.Context, string) ([]MemoryRecord, error)
	SaveApproval(context.Context, PendingApproval) error
	Approval(context.Context, string) (PendingApproval, bool, error)
	PendingApprovals(context.Context) ([]PendingApproval, error)
	ResolveApproval(context.Context, string, ApprovalStatus, string) (PendingApproval, error)
}

StateStore is the public durable execution store contract accepted by NewRunner.

type StreamAckMode added in v0.2.0

type StreamAckMode string

StreamAckMode selects when a stream delivery is acknowledged to the source broker.

const (
	// StreamAckAuto acknowledges a delivery automatically once the pipeline has
	// processed it successfully (or it has been routed to the DLQ). This is the
	// default and matches at-least-once delivery with runtime-managed acks.
	StreamAckAuto StreamAckMode = "auto"
	// StreamAckManual leaves acknowledgement to the message handler: the runtime
	// never auto-acks a successfully processed delivery, so the broker keeps
	// redelivering until the source's own ack closure is invoked. Brokers whose
	// receiver does not expose an ack closure treat this as a no-op (the
	// transport's own delivery semantics apply).
	StreamAckManual StreamAckMode = "manual"
)

type StreamTrigger

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

StreamTrigger is a stream trigger source.

func Stream

func Stream(uri string) StreamTrigger

Stream declares a stream trigger source for From.

type SubAgentOption

type SubAgentOption interface {
	// contains filtered or unexported methods
}

SubAgentOption configures a SubAgent registered on a Pipe.

func MaxParallel

func MaxParallel(limit int) SubAgentOption

MaxParallel bounds concurrent invocations of a SubAgent.

type ToolOption

type ToolOption interface {
	// contains filtered or unexported methods
}

ToolOption configures a Go tool registered on a Pipe.

func Describe

func Describe(text string) ToolOption

Describe sets the LLM-facing description for a Tool.

func Idempotent

func Idempotent(keyExpression string) ToolOption

Idempotent marks a side-effecting Tool as replay-safe for a stable key expression.

func Param

func Param(name, description string) ToolOption

Param sets the LLM-facing description for a Tool parameter.

func ReadOnly

func ReadOnly() ToolOption

ReadOnly marks a Tool as side-effect free and eligible for safe retry/parallel policies.

func RequiresApproval

func RequiresApproval() ToolOption

RequiresApproval marks a Tool as blocked unless an explicit policy allows it.

func SideEffecting

func SideEffecting(labels ...string) ToolOption

SideEffecting marks a Tool as mutating external state.

func ToolTimeout

func ToolTimeout(value string) ToolOption

ToolTimeout configures the maximum wall-clock duration for one Tool invocation.

type Tracer

type Tracer = events.Tracer

Tracer is the public surface for OTel-compatible span emission. Implementations are free to use OpenTelemetry, Jaeger, Honeycomb, or any other observability stack. The harness emits one span per pipeline, pipe, session, LLM call, tool call, schema validation, and subagent task.

func NopTracer

func NopTracer() Tracer

NopTracer returns a Tracer whose spans do nothing. Useful for tests or for disabling tracing without removing the WithTracer option.

type WebhookEndpoint

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

WebhookEndpoint can be used as a signed webhook trigger or a push target.

func Webhook

func Webhook(value string) WebhookEndpoint

Webhook declares a webhook trigger source or push target.

Directories

Path Synopsis
cmd
ouvrier command
internal
adminapi
Package adminapi is the shared client for a worker's authenticated /admin/* endpoints.
Package adminapi is the shared client for a worker's authenticated /admin/* endpoints.
auth
Package auth bridges the official Codex CLI for ChatGPT-subscription sign-in.
Package auth bridges the official Codex CLI for ChatGPT-subscription sign-in.
cli
console
Package console is the loopback-only operator web console for an Ouvrier fleet.
Package console is the loopback-only operator web console for an Ouvrier fleet.
deploy
Package deploy implements Ouvrier's deploy engine: building static binaries, resolving pip.yaml deploy environments and their env files, pre-flight validation, shipping over SSH, and the user-level deployments inventory.
Package deploy implements Ouvrier's deploy engine: building static binaries, resolving pip.yaml deploy environments and their env files, pre-flight validation, shipping over SSH, and the user-level deployments inventory.
envnames
Package envnames is the single source of truth for the runtime environment variable names read by Ouvrier.
Package envnames is the single source of truth for the runtime environment variable names read by Ouvrier.
lsp
operate/snippets
Package snippets is the embedded Ouvrier API + snippet pack: the single source of truth shared by the agent (read_ouvrier_api / search_ouvrier_docs) and the IDE (snippet palette + API panel).
Package snippets is the embedded Ouvrier API + snippet pack: the single source of truth shared by the agent (read_ouvrier_api / search_ouvrier_docs) and the IDE (snippet palette + API panel).
provider/codex
Package codex implements provider.Provider over the official `codex exec` transport, billed to the user's ChatGPT subscription.
Package codex implements provider.Provider over the official `codex exec` transport, billed to the user's ChatGPT subscription.
tui
tunnel
Package tunnel federates the loopback-only /admin APIs of deployed workers over SSH tunnels.
Package tunnel federates the loopback-only /admin APIs of deployed workers over SSH tunnels.
Package ovreval is a lightweight evaluation and regression harness for Ouvrier workers.
Package ovreval is a lightweight evaluation and regression harness for Ouvrier workers.
Package ovrtest provides test doubles for exercising Ouvrier workers without contacting a real model provider or spending tokens.
Package ovrtest provides test doubles for exercising Ouvrier workers without contacting a real model provider or spending tokens.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL