otel

func AppLogs ¶ added in v0.17.0

func AppLogs(on bool) DestOption

AppLogs says whether this destination takes the slog bridge's app log records (Slog, SlogHandler): on by default for Local and Studio with content on, off for Datadog, Langfuse, OTLP and Exporters — and for a Local or Studio destination under NoContent — because an app's log lines may carry anything it logged, prompts included. AppLogs(true) is the opt-in: it admits the lines, never bypasses the destination's content policy — an app log line is content like a transcript record, so a content-on destination applies ContentConfig's Redact (ContentLog) and MaxBytes to its message and string attributes, and a content-off one receives it stripped (message and string values emptied, weft.content=stripped; severity, span and the other attributes kept). AppLogs(false) keeps them from a Local or Studio destination. Only a destination that takes the lines carries the bridge mark (weft.logs.bridged on its heartbeats and root spans). Records the app emits through the OTel Logs API itself are not the bridge's and are not governed by it.

func BatchDelay ¶

func BatchDelay(d time.Duration) DestOption

BatchDelay sets the destination's batch flush interval (Studio defaults logs 200 ms / spans 1 s; the others the SDK's defaults).

func Blocking ¶ added in v0.16.0

func Blocking() DestOption

Blocking makes a full queue wait for room instead of refusing a durable record — up to 5 s and never past the emitting run's context (a span's End, which has none, waits under its own 5 s deadline); deltas are still shed. The bound is per stall: once a wait times out (the exporter is stuck), every later record is refused at once, counted, until an export begins again — a hung exporter costs the app one 5 s wait, not one per record. Development only, like RawBodies: it takes effect only when the pipeline starts under the dev gate (WEFT_ENV=dev read unless NoEnv, or DevMode — which an explicit WEFT_ENV other than dev overrules); otherwise Start logs one WARN naming the cause ("otel: Blocking ignored: WEFT_ENV is not dev", or "… otel.NoEnv reads no WEFT_ENV") and the destination refuses as without it — observability never slows a production app. Local ignores it (no queue).

package main

import (
	"context"
	"log"
	"log/slog"
	"os"
	"sync"

	"go.opentelemetry.io/otel/attribute"
	sdklog "go.opentelemetry.io/otel/sdk/log"

	"github.com/weftgo/weft/otel"
)

func main() {
	prev := slog.Default()
	defer slog.SetDefault(prev)
	slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
		ReplaceAttr: func(_ []string, a slog.Attr) slog.Attr {
			if a.Key == slog.TimeKey || a.Key == "hint" {
				return slog.Attr{}
			}
			return a
		},
	})))

	p, err := otel.Start(context.Background(),
		otel.NoEnv(), otel.NoGlobal(),
		otel.Exporters(nil, &eventNames{}, otel.Blocking()),
	)
	if err != nil {
		log.Fatal(err)
	}
	_ = p.Shutdown(context.Background())
}

// eventNames is a log exporter that remembers the weft.event.type of
// every record it exported.
type eventNames struct {
	mu    sync.Mutex
	types map[string]bool
}

func (e *eventNames) Export(_ context.Context, recs []sdklog.Record) error {
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.types == nil {
		e.types = map[string]bool{}
	}
	for _, r := range recs {
		r.WalkAttributes(func(kv attribute.KeyValue) bool {
			if kv.Key == "weft.event.type" {
				e.types[kv.Value.AsString()] = true
			}
			return true
		})
	}
	return nil
}

func (e *eventNames) saw(typ string) bool {
	e.mu.Lock()
	defer e.mu.Unlock()
	return e.types[typ]
}

func (*eventNames) Shutdown(context.Context) error   { return nil }
func (*eventNames) ForceFlush(context.Context) error { return nil }

Output:
level=WARN msg="otel: Blocking ignored: otel.NoEnv reads no WEFT_ENV"

func DatadogEndpoint ¶

func DatadogEndpoint(url string) DestOption

DatadogEndpoint overrides the Datadog destination's OTLP intake URL (default http://localhost:4318).

func Headers ¶

func Headers(h map[string]string) DestOption

Headers adds HTTP headers to the destination's requests (OTLP-family destinations).

func Insecure ¶

func Insecure() DestOption

Insecure allows http:// to a non-loopback host (OTLP-family destinations). Loopback http is always allowed, and so is the environment's OTEL_EXPORTER_OTLP_ENDPOINT written with http:// (the operator's opt-in; see envDestinations) — never WEFT_STUDIO_URL.

func NoContent ¶

func NoContent() DestOption

NoContent turns content off: the chain strips event and delta bodies (core.StripContent) and drops messages records before its exporter.

func NoDeltas ¶

func NoDeltas() DestOption

NoDeltas drops text/reasoning/args deltas before export, to save bandwidth when live streaming is not wanted. Deltas are never stored by the sinks either way (Q4); the core's durable sequence is untouched (they number on their own counter, D3).

func QueueSize ¶ added in v0.16.0

func QueueSize(n int) DestOption

QueueSize sets how many records (and, separately, spans) the destination holds between emission and export — its batch queue (default 16,384 for Studio, the SDK's 2,048 for the rest; n <= 0 keeps the default). A full queue never blocks the app: it refuses the record, deltas first (at half full), then ordinary records (at seven eighths), a run's run_start and run_finish last. Every refusal is counted: one WARN per destination ("weft/otel: destination queue full"), the count on the heartbeat and every run's run_start and run_finish (weft.records.dropped, which Studio's ingest health shows), and the run's gap badge, which counts the events lost against weft.run.events. Local has no queue (it writes synchronously) and ignores it. The OTEL_BLRP_MAX_QUEUE_SIZE and OTEL_BSP_MAX_QUEUE_SIZE variables do not apply to weft's destinations.

package main

import (
	"context"
	"fmt"
	"log"
	"sync"

	"go.opentelemetry.io/otel/attribute"
	sdklog "go.opentelemetry.io/otel/sdk/log"

	"github.com/weftgo/weft"
	"github.com/weftgo/weft/otel"
	"github.com/weftgo/weft/wefttest"
)

func main() {
	exp := &eventNames{}
	p, err := otel.Start(context.Background(),
		otel.NoEnv(), otel.NoGlobal(),
		otel.Exporters(nil, exp, otel.QueueSize(1<<15)),
	)
	if err != nil {
		log.Fatal(err)
	}
	agt := weft.New(wefttest.Script(wefttest.Say("done")),
		weft.LoggerProvider(p.LoggerProvider()))
	if _, err := agt.Generate(context.Background(), weft.Prompt("go")); err != nil {
		log.Fatal(err)
	}
	if err := p.Shutdown(context.Background()); err != nil {
		log.Fatal(err)
	}
	fmt.Println(exp.saw("run_start"), exp.saw("run_finish"))
}

// eventNames is a log exporter that remembers the weft.event.type of
// every record it exported.
type eventNames struct {
	mu    sync.Mutex
	types map[string]bool
}

func (e *eventNames) Export(_ context.Context, recs []sdklog.Record) error {
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.types == nil {
		e.types = map[string]bool{}
	}
	for _, r := range recs {
		r.WalkAttributes(func(kv attribute.KeyValue) bool {
			if kv.Key == "weft.event.type" {
				e.types[kv.Value.AsString()] = true
			}
			return true
		})
	}
	return nil
}

func (e *eventNames) saw(typ string) bool {
	e.mu.Lock()
	defer e.mu.Unlock()
	return e.types[typ]
}

func (*eventNames) Shutdown(context.Context) error   { return nil }
func (*eventNames) ForceFlush(context.Context) error { return nil }

Output:
true true

func Signals ¶

func Signals(traces, logs bool) DestOption

Signals selects which signals this destination takes. The defaults are per destination (Langfuse is traces only; the rest both).

func Timeout ¶

func Timeout(d time.Duration) DestOption

Timeout sets the destination's export request timeout.

func WithContent ¶

func WithContent(cfg ...ContentConfig) DestOption

WithContent turns content on for this destination, optionally overriding the pipeline's ContentConfig.

func Content ¶

func Content(cfg ContentConfig) Option

Content sets the pipeline-wide default for destinations with content on; a destination's WithContent overrides it.

func Datadog ¶

func Datadog(opts ...DestOption) Option

Datadog exports to the local Datadog Agent's OTLP intake (http://localhost:4318; the Agent binds the standard OTLP ports when its OTLP receiver is enabled). Content off, traces + logs, SDK default batching. DatadogEndpoint overrides the URL.

func DevDetail ¶ added in v0.20.0

func DevDetail() Option

DevDetail turns on the development detail level (ADR 0016's 2026-10-26 amendment): beside RawBodies, the record shows the whole middleware onion. Each model call and tool call gets one "layer <name>" span per WrapModel / WrapTools layer (weft.layer.name, weft.layer.kind, weft.layer.index; error.type where the layer returned an error), nested outer to inner under its chat or execute_tool span, a weft.repair event carries the original arguments mw.RepairJSON replaced (capped at 4 KiB), and a contained tool or middleware panic's stack rides its execute_tool span as a weft.panic event (16 KiB) — neither under content off. Off by default: a span per layer is 2–6 more spans per call for a signal most calls do not need. Like RawBodies it is for development only — it takes effect only when the pipeline starts under the dev gate (WEFT_ENV=dev read unless NoEnv, or DevMode); otherwise Start logs one WARN ("otel: DevDetail ignored: …") and the pipeline behaves as without it — and only the Local and Studio destinations with content on receive the detail: an OTLP, Datadog, Langfuse or Exporters destination gets the run without its layer spans and without the original arguments. The gate is the pipeline's LoggerProvider marker (LoggerProvider.DevDetail), which the core asks: pass Pipeline.LoggerProvider() to weft.LoggerProvider, or rely on Install's global registration. With the option the agent knows the level at New and builds the layer boundaries into its one model chain. Through the global provider it cannot (the pipeline may register after New): the first run builds a second chain, so the WrapModel constructors run twice — a stateful middleware holds two states — and a caller streaming Agent.Model() itself (weft/thread's compaction summary) gets the plain chain, without layer spans.

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"

	"github.com/weftgo/weft"
	"github.com/weftgo/weft/mw"
	"github.com/weftgo/weft/otel"
	"github.com/weftgo/weft/wefttest"
)

func main() {
	dir, err := os.MkdirTemp("", "weft-otel-devdetail-")
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = os.RemoveAll(dir) }()
	p, err := otel.Start(context.Background(),
		otel.NoEnv(), otel.DevMode(), otel.NoGlobal(),
		otel.Local(filepath.Join(dir, "weft.db")),
		otel.DevDetail(),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = p.Shutdown(context.Background()) }()

	lookup := weft.Tool("lookup", "", func(context.Context, struct{}) (string, error) { return "found", nil })
	agt := weft.New(wefttest.Script(
		wefttest.ToolCalls(wefttest.Call{Name: "lookup"}),
		wefttest.Say("done")),
		lookup, weft.WrapTools(mw.Audit(nil), mw.Allow(nil)),
		weft.TracerProvider(p.TracerProvider()), weft.LoggerProvider(p.LoggerProvider()))
	res, err := agt.Generate(context.Background(), weft.Prompt("look it up"))
	if err != nil {
		log.Fatal(err)
	}
	_ = p.ForceFlush(context.Background())
	spans, err := p.LocalDB().RunSpans(context.Background(), res.ID)
	if err != nil {
		log.Fatal(err)
	}
	for _, s := range spans {
		if strings.HasPrefix(s.Name, "layer ") {
			fmt.Println(s.Name, s.Attrs["weft.layer.index"])
		}
	}
}

Output:
layer mw.Audit 0
layer mw.Allow 1

func DevMode ¶ added in v0.19.0

func DevMode() Option

DevMode attests in code that this pipeline runs in development: it satisfies the dev gate (ADR 0028 §12) that RawBodies, DevDetail and a destination's Blocking need, without reading the environment. It is the code-side twin of WEFT_ENV=dev, for tests and harnesses that control their own process — typically beside NoEnv, which reads no WEFT_ENV and so refuses all three on its own. It unlocks those three and nothing else: no destination, discovery read or runtime link follows from it (runtime.Install keeps its own WEFT_ENV / Enabled rule). The gate still applies — a production binary never sets DevMode; the attestation has only moved from the process environment into code. The operator's environment wins: when the pipeline reads it (no NoEnv) and WEFT_ENV is set to anything but "dev", DevMode is refused with one WARN ("otel: DevMode ignored: WEFT_ENV=<value>") and the switches stay off; it opens the gate under NoEnv or with WEFT_ENV unset. Whenever DevMode is what opened it, Start logs one INFO line per switch it unlocked ("otel: raw bodies on by otel.DevMode()", "otel: dev detail on by otel.DevDetail() under otel.DevMode()", "otel: Blocking on by otel.DevMode()"), so a binary that ships it is visible.

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"path/filepath"

	"github.com/weftgo/weft/otel"
)

func main() {
	dir, err := os.MkdirTemp("", "weft-otel-devmode-")
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = os.RemoveAll(dir) }()

	p, err := otel.Start(context.Background(),
		otel.NoEnv(), otel.DevMode(), otel.NoGlobal(),
		otel.Local(filepath.Join(dir, "weft.db")),
		otel.RawBodies(true),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = p.Shutdown(context.Background()) }()
	fmt.Println(p.LoggerProvider().RawBodies())
}

Output:
true

func Exporters ¶

func Exporters(spans sdktrace.SpanExporter, logs sdklog.Exporter, opts ...DestOption) Option

Exporters wires your own exporters (vendor-specific, stdout, tests): whichever is non-nil gets a chain. Content off, SDK default batching.

func Heartbeat ¶

func Heartbeat(every time.Duration) Option

Heartbeat sets how often the run tracker emits one weft.heartbeat record per open run (the default 10s; 0 or less disables). Three missed intervals are what obsdb's InterruptedAfter reads as interrupted.

func Langfuse ¶

func Langfuse(host, publicKey, secretKey string, opts ...DestOption) Option

Langfuse exports traces to Langfuse's OTLP endpoint (<host>/api/public/otel/v1/traces, Basic auth base64(pk:sk); OTLP/HTTP only — gRPC is not accepted). Traces only: weft's spans carry no content (ADR 0016 O7) and Langfuse ingests traces, so it gets timing, usage, tool names and the identity chain; span-level content is the post-v1 follow-up F1.

func Local ¶

func Local(path string, opts ...DestOption) Option

Local is the local sink: obsdb/sqlite at path ("" = $WEFT_DB or ./.weft/weft.db), written synchronously — nothing emitted is lost, and the DB's in-process hub is the live lane setup A shares with Studio. Content on, traces + logs. Naming a Local switches the discovery file off, as Studio does: the pipeline writes to the stores it names (and the environment's), never into a running Studio it found on disk — add Studio(url, token) to write to one as well.

func NoDiscovery ¶ added in v0.19.0

func NoDiscovery() Option

NoDiscovery skips the discovery file a running `weft studio` / `weft dev` wrote, and nothing else: WEFT_STUDIO_URL, the OTEL_* variables and the WEFT_ENV development gate are read as usual (NoEnv drops them all). It is the per-pipeline form of WEFT_DISCOVERY=off — for a harness or test that keeps the operator's destinations but must not write into a Studio found on disk. A pipeline that names a Studio or a Local destination in code needs no NoDiscovery: either turns the read off on its own. Like NoEnv it is otel's alone: runtime.Install's discovery read (under WEFT_ENV=dev, when the pipeline has no Studio) is governed by runtime's options and WEFT_DISCOVERY=off.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/weftgo/weft"
	"github.com/weftgo/weft/otel"
	"github.com/weftgo/weft/wefttest"
	"go.opentelemetry.io/otel/sdk/trace/tracetest"
)

func main() {
	spans := tracetest.NewInMemoryExporter()
	p, err := otel.Start(context.Background(), otel.NoGlobal(), otel.NoDiscovery(),
		otel.Exporters(spans, nil), otel.Heartbeat(0))
	if err != nil {
		log.Fatal(err)
	}
	agt := weft.New(wefttest.Script(wefttest.Say("ok")), weft.Name("harness"),
		weft.TracerProvider(p.TracerProvider()), weft.LoggerProvider(p.LoggerProvider()))
	if _, err := agt.Generate(context.Background(), weft.Prompt("go")); err != nil {
		log.Fatal(err)
	}
	if err := p.ForceFlush(context.Background()); err != nil {
		log.Fatal(err)
	}
	fmt.Println("exported to the harness:", len(spans.GetSpans()) > 0)
	_ = p.Shutdown(context.Background())
}

Output:
exported to the harness: true

func NoEnv ¶

func NoEnv() Option

NoEnv ignores the WEFT_* / OTEL_* destination variables, and the discovery file a running `weft studio` / `weft dev` wrote (which otherwise stands in for WEFT_STUDIO_URL when that is unset). Explicit options are unaffected. It reads nothing, WEFT_ENV included: under NoEnv the development gate RawBodies and Blocking need holds only with DevMode (Start's WARN says "otel.NoEnv reads no WEFT_ENV" otherwise); NoDiscovery is the isolation that keeps WEFT_ENV. It is otel's alone: the runtime link's own discovery read (runtime.Install under WEFT_ENV=dev, when the pipeline has no Studio) is governed by runtime's options and WEFT_DISCOVERY=off.

func NoGlobal ¶

func NoGlobal() Option

NoGlobal leaves the OTel globals alone (Start only; Install always registers). For tests and dependency-injected programs. It isolates the globals only: it does not stop the discovery file's read — a pipeline without an explicit Studio or Local destination still joins a running Studio; NoDiscovery stops that read alone, NoEnv every environment destination.

func OTLP ¶

func OTLP(endpoint string, opts ...DestOption) Option

OTLP exports to any OTLP/HTTP endpoint (http(s)://host:port). Content off, traces + logs, SDK default batching; Headers adds request headers.

func RawBodies ¶ added in v0.14.0

func RawBodies(on bool) Option

RawBodies records each model call attempt's wire exchange — the provider request and response the first-party adapters sent and received, headers included minus the credential ones, each body capped at 256 KiB — as one weft.raw record per attempt (ADR 0028 §12), for the Local and Studio destinations only (never Datadog, Langfuse, OTLP or Exporters, whatever their content setting, and not a Local or Studio destination under NoContent). Off by default, and refused outside development: it takes effect only when the pipeline starts under the dev gate — WEFT_ENV is "dev" (read unless NoEnv), or DevMode attests it in code (under NoEnv, or with WEFT_ENV unset; an explicit WEFT_ENV other than dev wins, "otel: DevMode ignored: WEFT_ENV=<value>"). Otherwise Start logs one WARN line naming the cause ("otel: RawBodies ignored: WEFT_ENV is not dev", or under NoEnv without DevMode "otel: RawBodies ignored: otel.NoEnv reads no WEFT_ENV") and the pipeline behaves as if the option were absent: no destination asks for weft.raw, so the adapters buffer nothing (core.Reporter.WantsRaw is false). The gate is the pipeline's LoggerProvider marker (LoggerProvider.RawBodies): pass Pipeline.LoggerProvider() to weft.LoggerProvider, or rely on Install's global registration — a stock SDK provider never records raw bodies. Under it a Studio destination's log batch is capped at 16 records and each export is split at 12 MiB, keeping every post under Studio's 16 MiB ingest limit: a raw record is ≈ 650 KiB with two full ordinary bodies and at worst ≈ 3 MiB (bodies of control bytes, each six bytes in JSON; it is encoded without HTML escaping).

func Resource ¶

func Resource(r *sdkresource.Resource) Option

Resource merges r over the detected resource.

func Sampler ¶

func Sampler(s sdktrace.Sampler) Option

Sampler sets the trace sampler (default ParentBased(AlwaysOn)). Records are log records; sampling never applies to them.

func Service ¶

func Service(name string) Option

Service sets resource service.name. Else OTEL_SERVICE_NAME, else the binary's name.

func Slog ¶ added in v0.17.0

func Slog() Option

Slog bridges the program's default logger: once the pipeline is registered, slog.SetDefault wraps the default's handler with SlogHandler (bound to this pipeline), so every slog.Default line — weft.Logger's run lines (their default is slog.Default, read at log time), mw.Log(nil), mw.Audit(slog.Default()), the app's own — that is logged under a span reaches the pipeline's app-log destinations (AppLogs) as well as where it went before. Shutdown restores the previous default and the log package's writer and flags (unless the program replaced the default since).

A program still on slog's built-in default handler (it never called slog.SetDefault) keeps its lines on the log package's writer, written by a slog.TextHandler at the level slog.SetLogLoggerLevel set (INFO unless changed; read once, at Install), each line after log.Prefix() — wrapping the built-in handler itself would deadlock the log package. The format changes while the bridge is installed: the TextHandler's key=value lines (time=… level=… msg=…) replace the built-in "2009/11/10 23:00:00 INFO msg k=v" form, and the log package's flags no longer shape them; both come back at shutdown.

One Slog pipeline at a time: a second Install(…, Slog()) wraps the first's bridge. Shut them down in reverse order; out of order, the last shutdown still skips a closed pipeline's wrap and restores what the first found. Under NoGlobal (Start only) the default is left alone, like the OTel globals; bind a handler with Pipeline.SlogHandler instead.

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"path/filepath"

	"github.com/weftgo/weft"
	"github.com/weftgo/weft/mw"
	"github.com/weftgo/weft/obsdb"
	"github.com/weftgo/weft/obsdb/sqlite"
	"github.com/weftgo/weft/otel"
	"github.com/weftgo/weft/wefttest"
)

func main() {
	dir, err := os.MkdirTemp("", "weft-otel-slog-")
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = os.RemoveAll(dir) }()
	path := filepath.Join(dir, "weft.db")
	shutdown := otel.Install(otel.NoEnv(), otel.Local(path), otel.Slog())

	agt := weft.New(wefttest.Script(wefttest.Say("all done")),
		weft.Name("example"), weft.WrapModel(mw.Log(nil)))
	res, err := agt.Generate(context.Background(), weft.Prompt("go"))
	if err != nil {
		log.Fatal(err)
	}
	shutdown() // flushes, and puts slog.Default back

	db, err := sqlite.Open(path)
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = db.Close() }()
	page, err := db.OtherLogs(context.Background(), res.ID, obsdb.LogQuery{})
	if err != nil {
		log.Fatal(err)
	}
	for _, l := range page.Logs {
		fmt.Println(obsdb.SeverityText(l.Severity), l.Body)
	}
}

Output:
DEBUG run start
DEBUG model request
DEBUG model finish
DEBUG model call
DEBUG run finish

func Studio ¶

func Studio(url, token string, opts ...DestOption) Option

Studio exports OTLP/HTTP protobuf to a Weft Studio with a bearer token. Content on, traces + logs, short batches (logs 200 ms, spans 1 s). Naming a Studio here switches the discovery file off (as naming a Local does): the explicit destination is the pipeline's one Studio from the file's point of view (WEFT_STUDIO_URL still adds its own, de-duplicated).

func Start ¶

func Start(ctx context.Context, opts ...Option) (*Pipeline, error)

Start builds and starts the pipeline: one provider pair, the run tracker, one processor chain per destination (explicit options and environment combined, de-duplicated with the explicit destination winning). It fails if any destination fails to build — what was already built is shut down again, exporters handed to Exporters included; NoGlobal leaves the OTel globals alone (tests).