Skip to main content

API Reference

A comprehensive guide to TruvaG3's APIs with practical examples and best practices.

Quick Navigation​

Changed framework contracts:

Getting Started:

  • NewFramework - Bootstrap your component with the framework
  • NewBaseAgent - Create agents that discover and orchestrate
  • NewTool - Build tools that provide specific capabilities
  • NewAIAgent - Create AI-powered orchestration agents

Key Features:

By Module:

  • Core - Foundation types and component lifecycle
  • AI - AI provider integration and intelligent agents
  • Resilience - Circuit breakers and retry mechanisms
  • Telemetry - Metrics, tracing, and observability
  • Orchestration - Multi-agent coordination

Core Module​

The foundation of TruvaG3 - components, discovery, and lifecycle management.

Component Interface​

Every TruvaG3 component (tools and agents) implements this interface, providing a consistent API for initialization, identification, and capability discovery.

type Component interface {
    Initialize(ctx context.Context) error
    GetID() string
    GetName() string
    GetCapabilities() []Capability
    GetType() ComponentType
}

Why this matters: This unified interface allows the framework to manage any component type consistently, whether it's a simple tool or complex orchestration agent.

Example:

// Any component can be queried uniformly
func inspectComponent(comp core.Component) {
    fmt.Printf("Component: %s (ID: %s)\n", comp.GetName(), comp.GetID())
    fmt.Printf("Type: %s\n", comp.GetType()) // "agent" or "tool"

    caps := comp.GetCapabilities()
    fmt.Printf("Capabilities (%d):\n", len(caps))
    for _, cap := range caps {
        fmt.Printf("  - %s: %s\n", cap.Name, cap.Description)
    }
}

NewBaseAgent​

Creates an agent - an active component that can discover and orchestrate other services. Agents are the "brains" of your system, capable of finding tools and other agents to accomplish complex tasks.

func NewBaseAgent(name string) *BaseAgent
func NewBaseAgentWithConfig(config *Config) *BaseAgent

When to use agents vs tools:

  • Use an Agent when: You need to discover services, coordinate multiple components, or build orchestrators
  • Use a Tool when: You're providing a specific capability that others will use

Example - Simple Agent:

// Quick start with minimal config
agent := core.NewBaseAgent("data-processor")

// Agent can discover other services (tools can't!)
services, _ := agent.Discover(ctx, core.DiscoveryFilter{
    Type: core.ComponentTypeTool,
    Capabilities: []string{"database"},
})

agent.Initialize(ctx)
agent.Start(ctx, 8080)

Example - Production Agent:

// Full configuration for production
config, err := core.NewConfig(
    core.WithName("orchestrator"),
    core.WithPort(8080),
    core.WithRedisURL("redis://redis:6379"),      // Enable discovery
    core.WithLogLevel("info"),
    core.WithEnableMetrics(true),
)
if err != nil {
    log.Fatal("Failed to configure:", err)
}

agent := core.NewBaseAgentWithConfig(config)

// Register what this agent can do
agent.RegisterCapability(core.Capability{
    Name:        "orchestrate",
    Description: "Coordinate multiple services to complete tasks",
    Handler: func(w http.ResponseWriter, r *http.Request) {
        // Handler implementation
    },
})

// Initialize connects to Redis, sets up telemetry, etc.
if err := agent.Initialize(ctx); err != nil {
    log.Fatal("Failed to initialize:", err)
}

// Start begins serving HTTP requests
if err := agent.Start(ctx, 8080); err != nil {
    log.Fatal("Failed to start:", err)
}

NewTool​

Creates a tool - a passive component that provides specific capabilities. Tools are discovered and used by agents but cannot discover other components themselves.

func NewTool(name string) *BaseTool
func NewToolWithConfig(config *Config) *BaseTool

Tools are perfect for:

  • Microservices that perform specific tasks
  • API wrappers (weather, database, external services)
  • Stateless functions exposed as services
  • Any capability that doesn't need orchestration

Example - Weather Tool:

// Create a specialized tool
weatherTool := core.NewTool("weather-service")

// Define what it can do
weatherTool.RegisterCapability(core.Capability{
    Name:        "get_current_weather",
    Description: "Get current weather for any city",
    Endpoint:    "/api/weather/current",
    Handler: func(w http.ResponseWriter, r *http.Request) {
        var req struct {
            City string `json:"city"`
        }
        json.NewDecoder(r.Body).Decode(&req)

        weather := fetchWeatherData(req.City)

        w.Header().Set("Content-Type", "application/json")
        json.NewEncoder(w).Encode(weather)
    },
})

// Tools still need initialization and startup
weatherTool.Initialize(ctx)
weatherTool.Start(ctx, 8081) // Different port from agents

Key differences between Agents and Tools:

FeatureAgentTool
Can discover services✅ Yes❌ No
Can be discovered✅ Yes✅ Yes
Has HTTP server✅ Yes✅ Yes
Typical use caseOrchestrator, RouterMicroservice, API
ComplexityHigherLower

Discover​

Agent-only capability for finding components in your system. This is how agents build a dynamic picture of available services.

func (a *BaseAgent) Discover(ctx context.Context, filter DiscoveryFilter) ([]*ServiceInfo, error)

DiscoveryFilter fields:

  • Type - Filter by ComponentType (agent/tool)
  • Capabilities - Required capability names
  • Name - Exact service name match
  • Metadata - Custom key-value filters

Example - Smart Service Discovery:

// Find all database tools
dbTools, err := agent.Discover(ctx, core.DiscoveryFilter{
    Type: core.ComponentTypeTool,
    Capabilities: []string{"database"},
})

// Find services in specific region (using metadata)
regionalServices, err := agent.Discover(ctx, core.DiscoveryFilter{
    Metadata: map[string]interface{}{
        "region": "us-west",
        "environment": "production",
    },
})

// Find any service that can translate
translators, err := agent.Discover(ctx, core.DiscoveryFilter{
    Capabilities: []string{"translate"},
})

// Load balance across discovered services
service := translators[rand.Intn(len(translators))]
response, err := http.Post(service.Address, "application/json", payload)

Pro tip: Discovery results are cached for 5 minutes by default to reduce Redis load. Use WithDiscoveryCacheEnabled(false) for real-time discovery in development.

RegisterCapability​

Add capabilities to any component. This is how you define what your component can do.

func (c *BaseAgent) RegisterCapability(cap Capability)
func (c *BaseTool) RegisterCapability(cap Capability)

Capability structure:

// CapabilityType indicates the nature of a capability for executor behavior decisions.
type CapabilityType string

const (
    CapabilityTool         CapabilityType = "tool"         // Calls external APIs, stateless, safe to retry
    CapabilityReasoning    CapabilityType = "reasoning"    // LLM analysis only, no side effects
    CapabilityOrchestrator CapabilityType = "orchestrator"  // Nested DAG with own planning, execution, HITL
)

type Capability struct {
    Name           string           // Unique identifier
    Description    string           // Human-readable description for AI (Phase 1)
    Endpoint       string           // HTTP endpoint path
    Type           CapabilityType   // "tool", "reasoning", or "orchestrator" (default: "tool")
    InputTypes     []string         // Accepted content types
    OutputTypes    []string         // Response content types
    Handler        http.HandlerFunc // HTTP handler function

    // Phase 2: Schema discovery fields for AI-powered payload generation
    InputSummary   *SchemaSummary   // Compact field hints for AI (Phase 2)
    OutputSummary  *SchemaSummary   // Output schema hints (optional)
    SchemaEndpoint string           // Full JSON Schema endpoint (Phase 3)

    // Internal capabilities are excluded from LLM planning catalogs
    // but remain HTTP-callable. Use for orchestration endpoints,
    // admin endpoints, or deprecated capabilities.
    Internal       bool             // Exclude from LLM catalog (default: false)
}

// SchemaSummary provides compact field hints for AI payload generation (Phase 2)
type SchemaSummary struct {
    RequiredFields []FieldHint  // Required input fields
    OptionalFields []FieldHint  // Optional input fields
}

// FieldHint describes a single field for AI understanding
type FieldHint struct {
    Name        string  // Field name (exact, used in JSON)
    Type        string  // Field type (string, number, boolean, object, array)
    Example     string  // Example value for AI
    Description string  // Human-readable description
}

Capability Type controls executor behavior during step execution:

TypeTimeoutRetryUse When
"" or "tool"Shared HTTP-client timeout and enclosing context deadlinesPolicy-eligible retriesStandard tool/API calls
"reasoning"Shared HTTP-client timeout and enclosing context deadlinesPolicy-eligible retriesLLM analysis only, no side effects
"orchestrator"Additional per-call context deadline: HITL.DefaultTimeout + ExecutionOptions.StepTimeoutSingle dispatchNested delegation to another orchestrator agent

The "orchestrator" type accommodates child-agent HITL waits. With resolved positive settings, the orchestrator sets its shared HTTP-client timeout to max(ExecutionOptions.TotalTimeout, HITL.DefaultTimeout + ExecutionOptions.StepTimeout). That shared timeout also applies to ordinary tools. Only orchestrator capabilities get the additional per-call context deadline above. The phase loop similarly raises its configured phase timeout to at least that delegation allowance; this adjustment does not depend on the current plan containing a delegation. Earlier parent deadlines and the target server's timeouts can still end a call. StepTimeout is not a separate deadline for ordinary tools, and TotalTimeout does not bound the whole multi-phase request. See the execution limits.

Most capabilities should leave Type empty — the default "tool" behavior is correct. Set Type: core.CapabilityOrchestrator on capabilities that delegate to another orchestrator agent (e.g., devops-chat-agent's /query endpoint exposed to travel-chat-agent).

Example - Phase 1 (Basic - Description Only):

// Minimal capability - AI generates payloads from description alone (~85-90% accuracy)
tool.RegisterCapability(core.Capability{
    Name:        "current_weather",
    Description: "Gets current weather for a location. Required: location (city name). Optional: units (metric/imperial).",
    Handler:     handleWeather,
})

Example - Phase 2 (Recommended - With Field Hints):

// Enhanced capability with field hints for better AI accuracy (~95%)
tool.RegisterCapability(core.Capability{
    Name:        "current_weather",
    Description: "Gets current weather conditions for a location",
    InputTypes:  []string{"json"},
    OutputTypes: []string{"json"},
    Handler:     handleWeather,

    // Phase 2: Add structured field hints for AI payload generation
    InputSummary: &core.SchemaSummary{
        RequiredFields: []core.FieldHint{
            {
                Name:        "location",
                Type:        "string",
                Example:     "London",
                Description: "City name or coordinates (lat,lon)",
            },
        },
        OptionalFields: []core.FieldHint{
            {
                Name:        "units",
                Type:        "string",
                Example:     "metric",
                Description: "Temperature unit: metric or imperial",
            },
        },
    },
})

Example - Phase 3 (Mission-Critical - With Validation):

// Full capability with schema validation endpoint for maximum reliability (~99%)
tool.RegisterCapability(core.Capability{
    Name:        "process_payment",
    Description: "Process a payment transaction",
    Handler:     handlePayment,

    // Phase 2: Field hints for AI generation
    InputSummary: &core.SchemaSummary{
        RequiredFields: []core.FieldHint{
            {Name: "amount", Type: "number", Example: "99.99", Description: "Payment amount"},
            {Name: "currency", Type: "string", Example: "USD", Description: "Currency code"},
            {Name: "card_number", Type: "string", Example: "4111111111111111", Description: "Credit card number"},
        },
    },

    // Phase 3: Schema endpoint for validation (auto-generated at /api/capabilities/process_payment/schema)
    // Framework automatically serves JSON Schema at this endpoint
})

Redis Client Compatibility Defaults​

TruvaG3 uses github.com/redis/go-redis/v9. Framework-owned clients apply a stable compatibility profile instead of inheriting changing library defaults. Use the same helper for an application-owned client that you inject into core, memory, telemetry, or orchestration:

func ApplyRedisClientDefaults(options *redis.Options) *redis.Options

redisOptions, err := redis.ParseURL(os.Getenv("REDIS_URL"))
if err != nil {
    log.Fatal(err)
}
redisClient := redis.NewClient(core.ApplyRedisClientDefaults(redisOptions))

The helper returns a shallow copy and changes only unset fields, so the supplied options are not mutated. It selects RESP2 and retains the framework's established timeouts, retry backoff, and idle-connection lifetime. The dialer implementation, TCP keepalive, and buffer sizing use go-redis/v9 defaults unless the application sets them explicitly. Negative duration values that disable a go-redis timeout are also preserved. To evaluate RESP3, set Protocol: 3 on an application-owned client and validate every command family and backend before injecting it. Framework-owned clients remain on RESP2 in this release.

The v9 migration changes exported Redis client types. Applications that inject Redis clients must update imports from github.com/go-redis/redis/v8 to github.com/redis/go-redis/v9. Code that calls core.RedisClient.ZAdd must pass redis.Z values rather than *redis.Z pointers. This is a source-level change; stored Redis keys and values are unchanged.

Schema Cache (Phase 3 Validation)​

Redis-backed caching for JSON Schemas used in Phase 3 validation. Schemas are fetched once and cached forever, providing zero-overhead validation after initial fetch.

NewSchemaCache​

Create a Redis-backed schema cache for agents that perform Phase 3 validation.

func NewSchemaCache(redisClient redis.Cmdable, opts ...SchemaCacheOption) SchemaCache

SchemaCache interface:

type SchemaCache interface {
    // Get retrieves a cached schema by tool and capability name
    Get(ctx context.Context, toolName, capabilityName string) (map[string]interface{}, bool)

    // Set stores a schema in the cache
    Set(ctx context.Context, toolName, capabilityName string, schema map[string]interface{}) error

    // Stats returns cache statistics for monitoring
    Stats() map[string]interface{}
}

Example - Basic Schema Cache:

// In your agent initialization
redisOpt, _ := redis.ParseURL(os.Getenv("REDIS_URL"))
redisClient := redis.NewClient(core.ApplyRedisClientDefaults(redisOpt))

// Create schema cache with defaults (24-hour TTL)
agent.SchemaCache = core.NewSchemaCache(redisClient)

Example - Custom Configuration:

// Production configuration with custom TTL and prefix
agent.SchemaCache = core.NewSchemaCache(redisClient,
    core.WithTTL(1 * time.Hour),          // Shorter TTL for frequently changing schemas
    core.WithPrefix("myapp:schemas:"),    // Custom prefix for multi-tenant deployments
)

// Enable validation via environment variable
// export TRUVAG3_VALIDATE_PAYLOADS=true

// Agent will now:
// 1. Generate payloads using AI + field hints (Phase 1/2)
// 2. Fetch schema from tool's /schema endpoint (once)
// 3. Cache schema in Redis (shared across all agent replicas)
// 4. Validate all future payloads against cached schema

Cache configuration options:

// WithTTL sets cache expiration time
WithTTL(ttl time.Duration) SchemaCacheOption

// WithPrefix sets Redis key prefix (for namespacing)
WithPrefix(prefix string) SchemaCacheOption

Monitoring cache performance:

// Get cache statistics
stats := agent.SchemaCache.Stats()
// Returns: {"hits": 150, "misses": 3, "total_lookups": 153, "hit_rate": 0.98}

// Log statistics periodically
ticker := time.NewTicker(1 * time.Minute)
go func() {
    for range ticker.C {
        stats := agent.SchemaCache.Stats()
        logger.Info("Schema cache stats", stats)
    }
}()

When to use Schema Cache:

ScenarioUse Schema Cache?Reason
DevelopmentNoSchemas change frequently, overhead not worth it
Production agentsYesShared cache across replicas, validates critical payloads
Mission-critical APIsYes~99% accuracy with validation
Simple toolsNoPhase 1+2 sufficient for most cases

Best practices:

  • Enable only when TRUVAG3_VALIDATE_PAYLOADS=true
  • Use shared Redis instance for all agent replicas
  • Monitor cache hit rate (should be >95% after warmup)
  • Set reasonable TTL (24 hours default, schemas rarely change)
  • Use custom prefix for multi-tenant deployments

Component Lifecycle Management​

func (b *BaseAgent) GetDiscovery() Discovery returns a synchronized snapshot of the current discovery reference. It can be nil while startup recovery is pending. Use it for runtime reads when TRUVAG3_DISCOVERY_RETRY is enabled; direct Discovery assignment is supported before initialization. The returned implementation retains its own concurrency and lifecycle contract. Delayed application initialization belongs in a registered core.Runnable.

Every component follows a three-phase lifecycle: Initialize → Start → Stop/Shutdown. Understanding this lifecycle is crucial for building robust services.

// Initialize is the only lifecycle method on the Component interface itself
// (see "Component Interface" above) — every component implements it.
//
// Start is a concrete method on BaseAgent and BaseTool, not on the interface:
func (b *BaseAgent) Start(ctx context.Context, port int) error
func (t *BaseTool)  Start(ctx context.Context, port int) error

// Graceful shutdown is asymmetric — agents use Stop, tools use Shutdown:
func (b *BaseAgent) Stop(ctx context.Context) error
func (t *BaseTool)  Shutdown(ctx context.Context) error

Lifecycle best practices:

Built-in BaseAgent.Start and BaseTool.Start handle cancellation of their context: they unregister and join HTTP draining before returning. Framework.Run also cancels/drains registered runnables when serving exits, including startup failure. Close application-owned clients and telemetry only after it returns; no shutdown-wait goroutine or arbitrary sleep is required. Explicit Stop/Shutdown remains available and joins the same operation. The HTTP drain uses HTTP.ShutdownTimeout (10s by default; zero typed value means unset). Caller cancellation and deadlines limit only that caller's wait; the shared drain retains its configured budget and initiating context values. Start still waits for that drain to finish. The environment variable TRUVAG3_HTTP_SHUTDOWN_TIMEOUT requires a positive Go duration. A drain timeout returns an error, not a guarantee that active handlers or hijacked connections were terminated. Custom HTTPComponent.Start methods remain responsible for honoring cancellation.

  1. Initialize Phase - One-time setup

    • Connect to databases
    • Load configuration
    • Set up telemetry
    • Register with discovery

  2. Start Phase - Begin operations

    • Start HTTP server
    • Begin health checks
    • Accept incoming requests

  3. Stop Phase - Clean shutdown

    • Stop accepting new requests
    • Wait for in-flight requests
    • Unregister from discovery
    • Close connections

Example - Production Lifecycle:

func main() {
    ctx := context.Background()

    // Create and configure
    agent := core.NewBaseAgent("my-service")
    agent.RegisterCapability(/* ... */)

    // Initialize (connects to Redis, etc.)
    if err := agent.Initialize(ctx); err != nil {
        log.Fatal("Initialization failed:", err)
    }

    // Start in goroutine
    errChan := make(chan error, 1)
    go func() {
        if err := agent.Start(ctx, 8080); err != nil {
            errChan <- err
        }
    }()

    // Wait for shutdown signal
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)

    select {
    case <-sigChan:
        log.Info("Received shutdown signal")
    case err := <-errChan:
        log.Error("Server error:", err)
    }

    // Graceful shutdown with timeout
    shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    if err := agent.Stop(shutdownCtx); err != nil {
        log.Error("Graceful shutdown failed:", err)
        os.Exit(1)
    }

    log.Info("Shutdown complete")
}

NewFramework​

The main entry point for running TruvaG3 components. The framework handles all the complex setup - discovery, telemetry, configuration - so you can focus on your business logic.

func NewFramework(component HTTPComponent, opts ...Option) (*Framework, error)

Why use the framework:

  • Automatic configuration from environment variables
  • Built-in health checks and metrics
  • Graceful shutdown handling
  • Service discovery registration
  • Standardized logging

Example - Simple:

// Minimal setup - just works!
agent := core.NewBaseAgent("my-agent")
framework, _ := core.NewFramework(agent)
framework.Run(ctx)

Example - Production:

// Full production setup
agent := createMyAgent()

framework, err := core.NewFramework(agent,
    // Discovery
    core.WithRedisURL("redis://redis:6379"),

    // Observability
    core.WithTelemetry(true, "http://otel:4317"),
    core.WithLogLevel("info"),

    // Networking
    core.WithPort(8080),
    core.WithCORSDefaults(),

    // Features
    core.WithEnableMetrics(true),
    core.WithEnableTracing(true),
)

if err != nil {
    log.Fatal("Framework creation failed:", err)
}

// Run blocks until shutdown
if err := framework.Run(ctx); err != nil {
    log.Fatal("Framework run failed:", err)
}

Configuration​

TruvaG3's configuration system is designed for flexibility - use code, environment variables, or config files.

NewConfig​

Create configuration programmatically with validation and defaults.

func NewConfig(options ...Option) (*Config, error)
func DefaultConfig() *Config

Configuration application order: defaults, environment, then code options in the order supplied. WithConfigFile is itself an option: its file overlays the values resolved so far, and later options can override valid file values. DefaultConfig() installs defaults and detects the execution environment; it does not load the general deployment-variable overrides. Use NewConfig() for validated environment loading. Configuration files currently support JSON; .yaml and .yml files are rejected by LoadFromFile.

Core-owned numeric shared-memory/activity values fail validation when explicitly empty, malformed, out of range or overflowing. This also applies when memory is disabled. Missing values retain defaults; compaction detail 0 remains valid. Invalid environment input fails before options are applied. LoadFromFile rejects invalid numeric values and explicit nulls in these sections, and final Validate catches invalid typed options, including non-finite retrieval weights. Start direct configuration from DefaultConfig; Validate does not install defaults into an unresolved zero-value Config. Hook-only options and memory backend constructors have their own validation contracts.

Example - Different Configuration Styles:

// 1. Code-based (explicit, testable)
config, err := core.NewConfig(
    core.WithName("weather-service"),
    core.WithPort(8080),
    core.WithRedisDiscovery("redis://localhost:6379"),
)
if err != nil {
    log.Fatal(err)
}
// 2. Environment-based (12-factor app)
// Set: TRUVAG3_AGENT_NAME=weather-service
// Set: TRUVAG3_PORT=8080
// Set: REDIS_URL=redis://localhost:6379
config, err := core.NewConfig() // Loads and validates core environment settings
if err != nil {
    log.Fatal(err)
}
// 3. File-based (complex configs)
config, err := core.NewConfig(
    core.WithConfigFile("/etc/truvag3/config.json"),
)
if err != nil {
    log.Fatal(err)
}

Key Configuration Options​

Service Discovery:

// Enable Redis discovery
WithRedisURL(url string)           // Redis connection
WithRedisDiscovery(url string)     // Shorthand for Redis setup
WithDiscoveryCacheEnabled(bool)    // Cache discovery results
WithMockDiscovery(bool)            // Use mock for testing

// Registration TTL and heartbeat tuning
WithDiscoveryTTL(ttl)              // Registration TTL (default: 30s, min 5s)
WithHeartbeatInterval(d)           // Heartbeat interval (default: 0 = TTL/2, min 2s)

// Background retry for Redis connection failures: opt in via the
// TRUVAG3_DISCOVERY_RETRY=true / TRUVAG3_DISCOVERY_RETRY_INTERVAL env vars
// (no Go option function exposed today).

AI Integration:

WithAI(enabled bool, provider, apiKey string)  // Enable AI with provider + API key
WithOpenAIAPIKey(key string)        // Set OpenAI key
WithAIModel(model string)           // Choose a catalog-backed alias or concrete model
WithMockAI(bool)                    // Use mock for testing

Observability:

WithTelemetry(enabled bool, endpoint string)  // Enable telemetry
WithEnableMetrics(bool)                       // Metrics only
WithEnableTracing(bool)                       // Tracing only
WithLogLevel(level string)                    // error, warn, info, debug
WithLogFormat(format string)                  // json or text

Networking:

WithPort(port int)                  // HTTP port
WithAddress(addr string)            // Bind address
WithCORS(config CORSConfig)         // CORS settings
WithCORSDefaults()                  // Permissive CORS for dev

Advanced:

WithCircuitBreaker(threshold int, timeout time.Duration) // Resilience settings
WithRetry(maxAttempts int, initialInterval time.Duration) // Retry configuration
WithKubernetes(serviceDiscovery, leaderElection bool)    // K8s integration
WithDevelopmentMode(enabled bool)                        // Debug logging, mock services

Logging​

TruvaG3 provides structured logging with automatic context propagation. The framework automatically injects loggers into all components.

Log Levels (hierarchical):

  • error - Critical errors only
  • warn - Warnings and errors
  • info - Normal operations (default)
  • debug - Detailed debugging information

Using the Logger:

// Every component gets a logger automatically
func (t *MyTool) ProcessRequest(ctx context.Context, req Request) error {
    // Structured logging with context
    t.Logger.Info("Processing request", map[string]interface{}{
        "request_id": req.ID,
        "user_id":    req.UserID,
        "action":     req.Action,
    })

    result, err := t.performAction(req)
    if err != nil {
        t.Logger.Error("Action failed", map[string]interface{}{
            "error":      err.Error(),
            "request_id": req.ID,
            "duration":   time.Since(start).Milliseconds(),
        })
        return err
    }

    t.Logger.Debug("Action completed", map[string]interface{}{
        "request_id": req.ID,
        "result":     result,
    })

    return nil
}

Configuration via Environment:

# Production
export TRUVAG3_LOG_LEVEL=info
export TRUVAG3_LOG_FORMAT=json

# Development
export TRUVAG3_LOG_LEVEL=debug
export TRUVAG3_LOG_FORMAT=text

# Or use dev mode (sets debug + text automatically)
export TRUVAG3_DEV_MODE=true

Interfaces​

Discovery Interface​

Service discovery interface for agents. Combines registration with powerful query capabilities.

type Discovery interface {
    Registry  // Embed Registry (Register, UpdateHealth, Unregister)

    // Query methods
    Discover(ctx context.Context, filter DiscoveryFilter) ([]*ServiceInfo, error)
    FindService(ctx context.Context, serviceName string) ([]*ServiceInfo, error)
    FindByCapability(ctx context.Context, capability string) ([]*ServiceInfo, error)
}

Method usage guide:

MethodUse WhenExample
Discover()Complex multi-criteria searchFind healthy services in region with capability
FindService()Know exact service nameFind all "user-service" instances
FindByCapability()Need any service with capabilityFind any translator

Example - Building a Load Balancer:

type LoadBalancer struct {
    discovery core.Discovery
    current   uint32
}

func (lb *LoadBalancer) GetService(ctx context.Context, capability string) (*core.ServiceInfo, error) {
    // Find services advertising this capability. The Redis-backed discovery
    // implementation only returns services that are passing their heartbeat,
    // so there is no separate "healthy only" filter on DiscoveryFilter.
    services, err := lb.discovery.Discover(ctx, core.DiscoveryFilter{
        Capabilities: []string{capability},
    })

    if len(services) == 0 {
        return nil, fmt.Errorf("no services available for %s", capability)
    }

    // Round-robin selection
    index := atomic.AddUint32(&lb.current, 1) % uint32(len(services))
    return services[index], nil
}

Registry Interface​

Service registration interface for tools. Handles registration, health updates, and cleanup.

type Registry interface {
    Register(ctx context.Context, info *ServiceInfo) error
    UpdateHealth(ctx context.Context, id string, status HealthStatus) error
    Unregister(ctx context.Context, id string) error
}

Registration lifecycle:

// 1. Register on startup
info := &core.ServiceInfo{
    ID:       "weather-1",
    Name:     "weather-service",
    Type:     "tool",
    Address:  "http://localhost:8081",
    Health:   core.HealthHealthy,
}
registry.Register(ctx, info)

// 2. Periodic health updates
ticker := time.NewTicker(30 * time.Second)
go func() {
    for range ticker.C {
        health := checkHealth()
        registry.UpdateHealth(ctx, info.ID, health)
    }
}()

// 3. Unregister on shutdown
defer registry.Unregister(ctx, info.ID)

Pipeline short-circuit provenance​

The original BeforePlanningHook remains source compatible and its non-nil short circuits are treated as authoritative. New response-cache hooks should implement the additive provenance-aware contract:

type PipelineShortCircuitKind string

const (
    PipelineShortCircuitAuthoritative PipelineShortCircuitKind = "authoritative"
    PipelineShortCircuitCache         PipelineShortCircuitKind = "cache"
)

type PipelineGate interface {
    CacheVary() map[string]string
    ResponseCacheReadDisabled() bool
}

type PipelineShortCircuitDecision struct {
    ShortCircuit  *PipelineShortCircuit
    Kind          PipelineShortCircuitKind
    CachedAgainst map[string]string
}

type BeforePlanningDecisionHook interface {
    PipelineHook
    BeforePlanningDecision(
        context.Context,
        *PipelineContext,
        PipelineGate,
    ) (*PipelineShortCircuitDecision, error)
}

authoritative is for policy denials, rate limits, and other answers whose validity does not depend on response-cache freshness. cache must carry the variation dimensions persisted with the cached entry; it must not copy the gate's current dimensions during lookup. Orchestration accepts a cache hit only when every consumer-designated reserved dimension has the same presence and value in the stored and current maps; a value present on only one side is a mismatch. Other opaque dimensions are not enforced until their owning feature designates them. Cache reads are bypassed when ResponseCacheReadDisabled is true. Dimension semantics belong to the feature that adds them; core treats their names and values as opaque strings.

An opt-in decision with no payload returns an error matching orchestration.ErrInvalidPipelineShortCircuitDecision; an empty or unsupported kind matches orchestration.ErrUnknownPipelineShortCircuitKind. Ordinary hook callback errors retain the legacy fail-open behavior.

See Pipeline Hooks Guide for the full lifecycle, compatibility behavior, and cache-hook example.

Required after-planning governance​

Ordinary AfterPlanningHook implementations remain fail-open: orchestration rejects a callback error, wrong return type, clone failure, or invalid mutation and continues with the last valid plan. A mandatory plan-governance hook also implements this marker:

type RequiredAfterPlanningHook interface {
    PipelineHook
    RequireAfterPlanningSuccess()
}

The marker intentionally does not embed AfterPlanningHook. During factory construction, orchestration verifies that the same registered value implements both contracts. It also requires at most one required hook and requires it to be the final registered hook that participates in AfterPlanning. Invalid composition returns *orchestration.InvalidRequiredAfterPlanningHookError, matches orchestration.ErrInvalidOrchestratorConfig, and carries one stable reason:

type InvalidRequiredAfterPlanningHookReason string

const (
    InvalidRequiredAfterPlanningStageDrift InvalidRequiredAfterPlanningHookReason = "stage_drift"
    InvalidRequiredAfterPlanningMultiple   InvalidRequiredAfterPlanningHookReason = "multiple_required"
    InvalidRequiredAfterPlanningNotLast    InvalidRequiredAfterPlanningHookReason = "required_not_terminal"
)

At runtime, a rejected required mutation returns *orchestration.RequiredAfterPlanningError with HookName and the underlying Cause. orchestration.IsRequiredAfterPlanningError(err) and orchestration.IsInvalidRequiredAfterPlanningHook(err) provide stable checks. The request stops before phase-plan persistence, HITL evaluation, tool execution, or synthesis. Required hooks do not run again for HITL resume plans; those plans are restored approved checkpoint state.

Background Redis Retry​

TruvaG3 provides an intelligent background retry mechanism for handling Redis connection failures during service startup. This is particularly useful in Kubernetes environments where Redis may not be immediately available.

Key features:

  • Opt-in by default - Backward compatible, must be explicitly enabled
  • Exponential backoff - Retry intervals double on each failure (30s → 60s → 120s → 240s → 300s cap)
  • Automatic re-registration - When Redis becomes available, service is automatically registered
  • Thread-safe - Registry references are updated atomically

Configuration:

// Enable via environment variables
// export TRUVAG3_DISCOVERY_RETRY=true
// export TRUVAG3_DISCOVERY_RETRY_INTERVAL=30s
// export TRUVAG3_DISCOVERY_TTL=60s
// export TRUVAG3_DISCOVERY_HEARTBEAT=20s

// Or via code configuration (background retry stays env-var-driven —
// set TRUVAG3_DISCOVERY_RETRY=true alongside this code).
config, err := core.NewConfig(
    core.WithRedisURL("redis://redis:6379"),
    core.WithDiscoveryTTL(60 * time.Second),
    core.WithHeartbeatInterval(20 * time.Second),
)
if err != nil {
    log.Fatal(err)
}

How it works:

  1. Service attempts initial Redis connection during startup
  2. If connection fails and retry is enabled, service starts normally (without discovery)
  3. Background goroutine attempts reconnection at configured intervals
  4. On successful reconnection, service is registered and heartbeat begins
  5. Parent component's registry reference is updated via callback

Example - Kubernetes Deployment:

env:
  - name: REDIS_URL
    value: "redis://redis:6379"
  - name: TRUVAG3_DISCOVERY_RETRY
    value: "true"
  - name: TRUVAG3_DISCOVERY_RETRY_INTERVAL
    value: "30s"
  - name: TRUVAG3_DISCOVERY_TTL
    value: "60s"
  - name: TRUVAG3_DISCOVERY_HEARTBEAT
    value: "20s"

When to use:

  • Kubernetes environments where Redis may start after your service
  • Systems with intermittent Redis connectivity
  • Services that should remain functional even without discovery

When NOT to use:

  • Development environments (use mock discovery instead)
  • When Redis availability is a hard requirement

Memory Interface​

Abstract storage for state, sessions, and caching.

type Memory interface {
    Get(ctx context.Context, key string) (string, error)
    Set(ctx context.Context, key string, value string, ttl time.Duration) error
    Delete(ctx context.Context, key string) error
    Exists(ctx context.Context, key string) (bool, error)
}

Example - Session Management:

type SessionManager struct {
    store core.Memory
}

func (sm *SessionManager) CreateSession(userID string) (string, error) {
    sessionID := generateSessionID()
    data := map[string]string{
        "user_id": userID,
        "created": time.Now().Format(time.RFC3339),
    }

    jsonData, _ := json.Marshal(data)

    // Session expires in 24 hours
    err := sm.store.Set(ctx, "session:"+sessionID, string(jsonData), 24*time.Hour)
    return sessionID, err
}

func (sm *SessionManager) ValidateSession(sessionID string) (string, error) {
    data, err := sm.store.Get(ctx, "session:"+sessionID)
    if err != nil {
        return "", errors.New("invalid or expired session")
    }

    var session map[string]string
    json.Unmarshal([]byte(data), &session)
    return session["user_id"], nil
}

Shared Memory Interfaces​

Cross-agent shared memory for domain-scoped episodic events, investigation coordination, activity compaction, and real-time agent signals. Interfaces are in core/interfaces.go; implementations in the memory module.

// EpisodicMemory stores and queries structured agent events.
type EpisodicMemory interface {
    RecordEvent(ctx context.Context, event AgentEvent) error
    QueryEvents(ctx context.Context, callerDomain string, filter EventFilter) ([]AgentEvent, error)
    QueryEntityHistory(ctx context.Context, callerDomain, entityType, entityID string, since time.Time) ([]AgentEvent, error)
    QueryRecentEvents(ctx context.Context, callerDomain string, since time.Time, limit int) ([]AgentEvent, error)
    DeleteEvents(ctx context.Context, eventIDs []string) error
}

// InvestigationCoordinator prevents duplicate investigations across agents.
type InvestigationCoordinator interface {
    ClaimInvestigation(ctx context.Context, agentName, entityID string, ttl time.Duration) (claimed bool, holder string, err error)
    ReleaseInvestigation(ctx context.Context, agentName, entityID string) error
    GetActiveInvestigations(ctx context.Context) (map[string]string, error)
}

// Optional extension: report read health instead of hiding backend failure.
type ObservedEpisodicReader interface {
    QueryEventsObserved(ctx context.Context, callerDomain string, filter EventFilter) ([]AgentEvent, error)
}

// ActivityCompactor compresses raw events into a fixed-size LLM digest.
type ActivityCompactor interface {
    CompactEvents(ctx context.Context, events []AgentEvent, maxTokens int) (string, error)
    UpdateDigest(ctx context.Context, previousDigest string, newEvents []AgentEvent, maxTokens int) (string, error)
}

// DigestCache stores/retrieves compacted domain activity digests.
type DigestCache interface {
    GetDigest(ctx context.Context, domain string) ([]byte, error)
    SetDigest(ctx context.Context, domain string, data []byte, ttl time.Duration) error
}

// ActivityCoordinator manages real-time agent coordination signals.
type ActivityCoordinator interface {
    AnnounceActivity(ctx context.Context, signal ActivitySignal) error
    UpdateStatus(ctx context.Context, requestID, status string) error
    GetDomainActivities(ctx context.Context, domain string) ([]ActivitySignal, error)
    CompleteActivity(ctx context.Context, requestID string) error
}

// StepSummary is the per-step output produced by an EventSummarizer.
// Summary is a one-sentence factual description of what the step did.
// Entities is the LLM's identification of domain entities the step operated on.
// Implementations that do not extract entities should return Entities as nil.
type StepSummary struct {
    Summary  string      `json:"summary"`
    Entities []EntityRef `json:"entities,omitempty"`
}

// Exact input contract for built-in and custom summarizers.
type StepSummaryInput struct {
    StepID      string                 `json:"step_id"`
    AgentName   string                 `json:"agent_name"`
    Capability  string                 `json:"capability"`
    Instruction string                 `json:"instruction"`
    Parameters  map[string]interface{} `json:"parameters,omitempty"`
    Response    string                 `json:"response"`
    Success     bool                   `json:"success"`
    Error       string                 `json:"error,omitempty"`
    ToolError   json.RawMessage        `json:"tool_error,omitempty"`
}

// EventSummarizer generates actionable summaries for execution steps.
// Returns a map of stepID -> StepSummary. Implementations must be fail-open.
// StepSummaryInput also carries optional Error string and ToolError
// json.RawMessage. These are untrusted recorded facts, not inferred diagnoses.
// The recording hook copies ToolError bytes before calling custom summarizers.
type EventSummarizer interface {
    SummarizeSteps(ctx context.Context, steps []StepSummaryInput) (map[string]StepSummary, error)
}

// SharedKnowledge stores and searches reusable knowledge fragments.
type SharedKnowledge interface {
    StoreKnowledge(ctx context.Context, fragment KnowledgeFragment) error
    SearchKnowledge(ctx context.Context, callerDomain, namespace, query string, topK int, weights RetrievalWeights) ([]ScoredKnowledge, error)
}

// MemoryReflector extracts patterns from events and generates knowledge.
type MemoryReflector interface {
    Reflect(ctx context.Context, entityType, entityID string, since time.Time) ([]KnowledgeFragment, error)
}

Implementations:

InterfaceRedis/Vector (shared)In-Memory (testing)NoOp (disabled)
EpisodicMemorymemory.StreamEpisodicMemorymemory.InMemoryEpisodicMemory—
InvestigationCoordinatormemory.AtomicLockCoordinatormemory.InMemoryInvestigationCoordinator—
SharedKnowledgememory.VectorSharedKnowledge (Qdrant)——
MemoryReflectormemory.LLMMemoryReflector——
ActivityCompactor——core.NoOpActivityCompactor
DigestCachememory.RedisDigestCachememory.InMemoryDigestCachecore.NoOpDigestCache
ActivityCoordinatormemory.RedisActivityCoordinatormemory.InMemoryActivityCoordinatorcore.NoOpActivityCoordinator
EventSummarizer——core.NoOpEventSummarizer

ActivityCompactor and EventSummarizer are LLM-powered — implemented in the orchestration module as LLMActivityCompactor and LLMEventSummarizer. MemoryReflector is also LLM-powered via LLMMemoryReflector in the memory module.

For the built-in LLMActivityCompactor, a non-empty full or incremental call requires a positive target no larger than math.MaxInt / 5, so its five-times generation allowance fits a Go int. Invalid targets return an error matching core.ErrInvalidConfiguration before AI invocation; incremental rejection also returns the previous digest. Empty-event calls retain their no-op behavior. An enrichment hook selecting this compactor validates the target at construction. This implementation-specific arithmetic bound does not constrain custom ActivityCompactor implementations or guarantee a provider will accept the budget.

Convenience Constructors​

memory.NewSharedBackends — creates all memory backends from a Redis client:

backends, err := memory.NewSharedBackends(redisClient, logger,
    memory.WithAgentName("my-agent"),
    memory.WithDomain("infrastructure"),
    memory.WithEmbeddingClient(embedder),  // optional — enables Phase 2
)
if err != nil {
    return fmt.Errorf("configure memory backends: %w", err)
}
defer backends.Close()

Options: WithDomain(string), WithAgentName(string), WithKnowledgeDisabled(), WithEmbeddingClient(core.EmbeddingClient).

Investigation lease contracts — ownership used by the memory hooks:

Built-in investigation providers also implement core.InvestigationLeaseCoordinator: AcquireInvestigation(ctx, core.InvestigationLeaseRequest) returns an exact handle (or contention holder); ReleaseInvestigationLease(ctx, handle) reports the typed core.InvestigationReleaseOutcome: released, absent, not_owned, or unknown (the last requires an error). Each request segment uses a fresh private acquisition ID. Managed ProcessRequest* cleans these handles on every exit, bounded by WithExecutionCleanupTimeout / the matching environment variable (5s default). Direct hooks/ExecutePlan* retain caller-owned lifecycle; legacy custom coordinators retain name-only semantics. Claims expire without renewal and remain advisory. Ownership handles are excluded from JSON.

// core: acquisition identity is private request-segment state.
type InvestigationLeaseRequest struct {
    AgentName     string
    EntityID      string
    AcquisitionID string `json:"-"`
    TTL           time.Duration
}

type InvestigationLease struct {
    AgentName     string
    EntityID      string
    AcquisitionID string `json:"-"`
    ExpiresAt     time.Time
}

type InvestigationReleaseOutcome string

const (
    InvestigationReleased InvestigationReleaseOutcome = "released"
    InvestigationAbsent   InvestigationReleaseOutcome = "absent"
    InvestigationNotOwned InvestigationReleaseOutcome = "not_owned"
    InvestigationUnknown  InvestigationReleaseOutcome = "unknown"
)

type InvestigationLeaseCoordinator interface {
    AcquireInvestigation(context.Context, InvestigationLeaseRequest) (*InvestigationLease, string, error)
    ReleaseInvestigationLease(context.Context, InvestigationLease) (InvestigationReleaseOutcome, error)
}

TTL == 0 selects the provider default; negative TTL or empty identities are invalid. Repeating the same live acquisition is idempotent without extending expiry. Contention returns a nil lease and the current holder. Backend state, not ExpiresAt or agent-name equality, decides whether release is owned. ObservedEpisodicReader likewise exposes read health without changing legacy fail-open methods: a nil error is not an atomic snapshot or unlimited-history guarantee. These optional contracts are discovered by interface assertion.

orchestration.BuildMemoryHooks — creates pipeline hooks from backends:

This builder and memory.NewSharedBackends do not receive core.Config. Typed/JSON SharedMemory and ActivityCoordination values validated by core are not automatically applied to these bundles. Use their documented environment variables/options, or direct hook/backend constructors for explicit numeric or behavioral overrides. Direct constructors do not inherit bundle environment tuning.

hooks, activityCoord, err := orchestration.BuildMemoryHooks(
    backends.ToDeps(), aiClient, logger,
    orchestration.WithMemoryEntityExtractor(e),    // optional: override BOTH hooks
    orchestration.WithMemoryImportanceFunc(fn),    // optional behavioural override
)
if err != nil {
    return fmt.Errorf("configure memory hooks: %w", err)
}

Returns ([]core.PipelineHook, core.ActivityCoordinator, error). Check the error before serving; pass hooks and coordinator to OrchestratorDependencies. Generic numeric parsing/range checks and behavioral options are validated even without AI or episodic dependencies. Implementation-specific checks, including the built-in compactor's multiplication bound, run when that implementation is selected. With valid configuration, absent episodic memory returns (nil, nil, nil). Runtime memory fail-open behavior is unchanged.

Default extractor selection: when aiClient != nil, the record hook defaults to LLMEntityExtractor (zero-cost LLM extraction piggybacking on the existing EventSummarizer call). The enrichment hook always defaults to NoOpEntityExtractor (pre-planning, the summarizer hasn't run yet, so LLM extraction would silently return nothing). When aiClient == nil, both default to NoOpEntityExtractor.

Behavioural options:

  • WithMemoryEntityExtractor(EntityExtractor) — overrides BOTH hooks (shorthand)
  • WithBuilderRecordEntityExtractor(EntityExtractor) — overrides record hook only
  • WithBuilderRecordRequestEntityExtractor(EntityExtractor) — opt-in additional request-context entities for recording; request entities precede distinct step entities. Direct hooks use WithRecordRequestEntityExtractor. Enrichment and claim ownership are unchanged.
  • WithBuilderEnrichmentEntityExtractor(EntityExtractor) — overrides enrichment hook only
  • WithMemoryImportanceFunc(func), WithMemoryRetrievalWeights(RetrievalWeights), WithMemoryLookback(time.Duration), WithMemoryActivityFilter(ActivityFilter)

Numeric tuning uses TRUVAG3_SHARED_MEMORY_* env vars.

TRUVAG3_SHARED_MEMORY_RECORD_FAILURE_CAUSE_MAX_BYTES limits the cause in heuristic failure summaries (default 256 bytes; positive integer). Direct hooks use WithRecordFailureCauseMaxBytes(n). UTF-8-safe shortening adds [shortened] outside the cause allowance and summary_cause_shortened=true metadata. Operation/entity framing, model/custom summaries and source execution errors are unchanged. Invalid explicit builder values fail even without optional dependencies. Built-in batch exclusions record summary_fallback_reason=batch_limit and a separate batch_excluded count. Record hook effects use independent backend_assigns_event_id/backend_assigns_timestamp flags; the former combined diagnostic field is no longer emitted (historical records stay unchanged).

WithSummarizerSummaryMaxBytes(n) limits each stored derived summary to 1024 UTF-8 bytes by default (minimum 32). The builder reads TRUVAG3_SHARED_MEMORY_EVENT_SUMMARIZER_SUMMARY_MAX_BYTES; invalid explicit settings fail construction even without AI. Overlong prose gets a rune-safe [shortened] suffix within the limit, with summaries_shortened in coverage, logs and traces. Structured entities and raw LLM debug output are unchanged. This is separate from the 10000-token batch generation allowance.

TRUVAG3_SHARED_MEMORY_EVENT_SUMMARIZER_MAX_RESPONSE_BYTES sets the positive per-step encoded failure/response payload allowance (default 4000 bytes). Direct callers use WithSummarizerMaxResponseChars (legacy name, byte units). Failure facts have priority; omissions/shortening are labelled in the derived prompt, leaving raw execution evidence unchanged. Output-token budgets are independent. Event metadata records summary_source, optional fallback reason, recorded_at and event_time_basis; event timestamps prefer valid end/start times over the unknown-occurrence recording fallback. Provider retention remains ingestion-based. Historical records retain unknown provenance in derived views.

The built-in summarizer keeps complete, individually valid entries before a truncated array suffix, marking partial coverage; excluded or missing entries receive recording-hook fallback once. This does not make the incomplete array complete evidence or change custom summarizers' contract. Fallback preserves bounded failure causes; request-entity extraction is opt-in and independent of LLM summary success. Digest reuse requires an observed matching bounded event window and configuration/AI identity. Legacy episodic implementations without ObservedEpisodicReader remain usable but do not provide safe digest reuse. Independently implemented external compactors also bypass reuse: the stable behavior-fingerprint method is package-private, not a custom-provider extension. Observed read errors preserve the cache instead of treating failure as empty history. Event queries use newest-first occurrence timestamps with domain/ID tie breaks and filtering before limit; entity-history order is unchanged.

Available EntityExtractor implementations (in orchestration package):

  • NoOpEntityExtractor — honors explicit metadata["entity_type"]/metadata["entity_id"] or multi-entity metadata["entities"]. Performs no extraction of its own — the framework refuses to guess. Domain-agnostic.
  • LLMEntityExtractor — reads entities the EventSummarizer LLM produced as a side effect of summarization (via the metadata["llm_entities"] key plumbed by MemoryRecordHook). Zero additional LLM round-trips. Falls through to explicit metadata if no LLM entities are present.

See Agent Memory User Guide for full usage.

User Memory Interfaces​

Per-user private memory for personal assistant agents. Stores learned facts (preferences, identity, constraints) across sessions and injects them into the planning prompt via a <user_profile> enrichment tag. Separate from shared memory — user facts are never visible to other users or agents outside the user's session. Interfaces are in core/interfaces.go; implementations in the memory module.

// FactSource describes how a user fact was acquired.
type FactSource string

const (
    SourceExplicit   FactSource = "explicit"   // User directly stated it
    SourceCorrection FactSource = "correction" // User corrected the agent
    SourceInferred   FactSource = "inferred"   // Extracted from conversation patterns
    SourceDerived    FactSource = "derived"     // Synthesized from multiple facts/sessions
)

// UserFact represents a learned piece of information about a specific user.
type UserFact struct {
    FactID     string            `json:"fact_id"`
    UserID     string            `json:"user_id"`
    Namespace  string            `json:"namespace"`  // Agent type scope: "travel", "devops", "universal"
    Category   string            `json:"category"`   // "preference", "identity", "constraint", "context", "summary", "relationship"
    Content    string            `json:"content"`    // Natural language: "Prefers window seats on flights over 4 hours"
    Source     FactSource        `json:"source"`
    Confidence float64           `json:"confidence"` // 0.0-1.0
    CreatedAt  time.Time         `json:"created_at"`
    UpdatedAt  time.Time         `json:"updated_at"`
    Metadata   map[string]string `json:"metadata,omitempty"`
}

// UserMemory stores and retrieves per-user private facts.
// Implementations must enforce strict user isolation at the storage level.
type UserMemory interface {
    Remember(ctx context.Context, userID string, fact UserFact) error
    Recall(ctx context.Context, userID string, namespace string, queryContext string, limit int) ([]UserFact, error)
    RecallByCategory(ctx context.Context, userID string, namespace string, category string, limit int) ([]UserFact, error)
    Forget(ctx context.Context, userID string) error  // GDPR Article 17
}

// UserMemoryAdmin extends UserMemory with administrative operations.
type UserMemoryAdmin interface {
    UserMemory
    ListFacts(ctx context.Context, userID string, namespace string, offset int, limit int) ([]UserFact, int, error)
    ForgetNamespace(ctx context.Context, userID string, namespace string) error
    ForgetFact(ctx context.Context, userID string, factID string) error
}

// UserMemoryDeps holds dependencies for BuildUserMemoryHooks.
type UserMemoryDeps struct {
    UserMemory UserMemory      // Required
    Embedder   EmbeddingClient // Required for vector-backed backends; nil for InMemoryUserMemory
    Namespace  string          // Agent type scope (e.g., "travel", "devops")
}

Implementations:

InterfaceVector DB (production)In-Memory (testing)NoOp (disabled)
UserMemorymemory.VectorUserMemory (Qdrant)memory.InMemoryUserMemorycore.NoOpUserMemory
UserMemoryAdminmemory.VectorUserMemorymemory.InMemoryUserMemorycore.NoOpUserMemory

The VectorUserMemory backend is the reference implementation. Custom backends (pgvector, Redis Stack) implement the UserMemory interface — same swap pattern as EpisodicMemory.

Convenience constructors:

memory.NewUserMemoryBackend — creates a user memory backend with auto-detection:

backend, err := memory.NewUserMemoryBackend(logger,
    memory.WithUserMemoryNamespace("travel"),
    memory.WithUserMemoryEmbeddingClient(embedder),
)
defer backend.Close()

If TRUVAG3_VECTOR_DB_URL is set and an embedding client is provided, creates VectorUserMemory. Otherwise falls back to InMemoryUserMemory. Options: WithUserMemoryNamespace(string), WithUserMemoryEmbeddingClient(core.EmbeddingClient), WithUserMemoryVectorOption(Option) (wraps vector config options like WithCollectionName).

orchestration.BuildUserMemoryHooks — creates pipeline hooks from a deps struct:

hooks, closer := orchestration.BuildUserMemoryHooks(
    backend.ToDeps(), aiClient, logger,
    orchestration.WithUserFactExtractor(e),     // optional behavioural override
    orchestration.WithUserFactReconciler(r),     // optional behavioural override
)
defer closer.Close()

Returns ([]core.PipelineHook, io.Closer) — append the hooks to OrchestratorDependencies.PipelineHooks alongside shared memory hooks, and call closer.Close() during shutdown to drain any in-flight async extraction work. Separate from BuildMemoryHooks because user memory always requires an embedding client and not all agents need user memory.

Behavioural options: WithUserFactExtractor(UserFactExtractor), WithUserFactReconciler(UserFactReconciler), WithUserFactPersistencePolicy(UserFactPersistencePolicy), WithUserMemoryRetrievalWeights(RetrievalWeights), WithSynchronousExtraction() (opt out of the async Layer 1 default). Numeric tuning uses TRUVAG3_USER_MEMORY_* env vars — see Environment Variables Guide.

orchestration.BatchUserFactReconciler — optional extension interface. Reconcilers that implement it can classify all candidates from a single turn in one LLM call instead of one-per-candidate, collapsing N sequential reconciliation calls into 1:

type BatchUserFactReconciler interface {
    UserFactReconciler
    ReconcileBatch(ctx context.Context, userID, namespace string,
        candidates []core.UserFact, neighbors [][]core.UserFact) ([]ReconcileResult, error)
}

The extraction hook detects this interface via type assertion and uses the batched path when available, falling back to per-candidate Reconcile on error, parse failure, or length mismatch. The default LLMUserFactReconciler implements it. Custom reconcilers continue to work unchanged via the per-candidate path.

Well-known enrichment key:

const EnrichmentUserProfile = "user_profile"

Injected by UserMemoryEnrichmentHook into PipelineContext.Enrichments. Consumed by DefaultPromptBuilder and the orchestrator's synthesis prompt builder to include <user_profile> in the LLM prompt.

Scheduling Interfaces​

TaskConsumer Interface​

Consumer-side counterpart of TaskDispatcher. Used by the scheduled-executor to drain tasks from the transport.

// core/async_task.go
type TaskConsumer interface {
    Consume(ctx context.Context, queueName string) (TaskHandle, error)
}

Returns a TaskHandle that the worker must settle via exactly one Ack or Nack call. Returning (nil, nil) is permitted on graceful shutdown.

Implementations: orchestration.RedisTaskConsumer (BRPOP, default), orchestration.RedisStreamsTaskConsumer (Streams, at-least-once), orchestration.InMemoryTaskConsumer (dev/test)

TaskHandle Interface​

Leased reference returned by TaskConsumer.Consume. Follows the borrow-then-settle pattern (same as database/sql.Rows, net.Conn, AMQP basic.deliver/basic.ack).

// core/async_task.go
type TaskHandle interface {
    Task() *Task                           // Accessor
    Ack(ctx context.Context) error         // Success settlement
    Nack(ctx context.Context, reason string) error  // Terminal failure + DLQ persistence
}

Dead-letter persistence is folded into Nack -- there is no separate DeadLetterSink interface.

TaskDispatcher Interface​

// core/async_task.go
type TaskDispatcher interface {
    Dispatch(ctx context.Context, queueName string, task *Task) error
}

Returns core.ErrTaskAlreadyExists for duplicate task IDs (idempotency contract).

ScheduleStore Interface​

// core/async_task.go
type ScheduleStore interface {
    Create(ctx context.Context, schedule *Schedule) error
    Get(ctx context.Context, id string) (*Schedule, error)
    List(ctx context.Context) ([]*Schedule, error)
    Update(ctx context.Context, schedule *Schedule) error
    Delete(ctx context.Context, id string) error
    GetDue(ctx context.Context, now time.Time) ([]*Schedule, error)
}

RegisterScheduledEndpoint​

Mounts /api/v1/scheduled on an agent to receive scheduled tasks from the executor.

// orchestration/scheduled_endpoint.go
type OrchestratorFunc func() Orchestrator

func RegisterScheduledEndpoint(agent *core.BaseAgent, orchestratorFn OrchestratorFunc, opts ...ScheduledEndpointOption) error

Must be called before framework.Run() — HandleFunc rejects registrations after the HTTP server starts. The OrchestratorFunc is called on each request, supporting async orchestrator initialization (returns 503 until ready).

Options:

OptionPurpose
WithScheduledQueryBuilder(fn)Override how the user-query string is extracted from the request
WithScheduledMetadataBuilder(fn)Override what metadata is passed to the orchestrator
WithScheduledFilter(fn)Predicate that skips requests (returns 200 with status: filtered)
WithScheduledEndpointLogger(logger)Override the logger (default reads agent.Logger dynamically)

Example — Agent wiring (before framework.Run):

if err := orchestration.RegisterScheduledEndpoint(agent.BaseAgent, func() orchestration.Orchestrator {
    if o := agent.GetOrchestrator(); o != nil {
        return o
    }
    return nil
}); err != nil {
    agent.Logger.Warn("Failed to register scheduled endpoint", map[string]interface{}{"error": err.Error()})
}

ExecutorDeps (Scheduled-Executor Configuration)​

Code-level configuration for the scheduled-executor worker. Fields set in code take priority over environment variables; zero-valued fields fall back to env vars, then defaults.

// examples/scheduled-executor/worker.go
type ExecutorDeps struct {
    Consumer        core.TaskConsumer  // Required
    HTTPClient      *http.Client       // Required (use telemetry.NewTracedHTTPClient)
    Catalog         AgentResolver      // Required
    Logger          core.Logger        // Optional (defaults to NoOpLogger)
    QueueName       string             // Default: "scheduled-executor"
    WorkerCount     int                // Default: 5    | Env: TRUVAG3_EXECUTOR_WORKER_COUNT
    MaxRetries      int                // Default: 3    | Env: TRUVAG3_EXECUTOR_MAX_RETRIES
    RetryBaseDelay  time.Duration      // Default: 5s   | Env: TRUVAG3_EXECUTOR_RETRY_BASE_DELAY
    RetryMaxDelay   time.Duration      // Default: 60s  | Env: TRUVAG3_EXECUTOR_RETRY_MAX_DELAY
    DispatchTimeout time.Duration      // Default: 15m  | Env: TRUVAG3_EXECUTOR_DISPATCH_TIMEOUT
}

Configuration priority: code-level field > environment variable > hardcoded default.

Example — Code-level override:

worker, _ := NewWorker(ExecutorDeps{
    Consumer:        backends.TaskConsumer,
    HTTPClient:      tracedClient,
    Catalog:         catalog,
    WorkerCount:     10,               // overrides TRUVAG3_EXECUTOR_WORKER_COUNT
    MaxRetries:      5,                // overrides TRUVAG3_EXECUTOR_MAX_RETRIES
    DispatchTimeout: 15 * time.Minute, // Size for the target agent's entire synchronous request
})

See Environment Variables Guide for the full env var reference.

SchedulerBackends​

Bundle of producer + consumer backends returned by factory functions.

// orchestration/scheduler_backends.go
type SchedulerBackends struct {
    ScheduleStore  core.ScheduleStore   // Producer-side
    TaskDispatcher core.TaskDispatcher  // Producer-side
    TaskConsumer   core.TaskConsumer    // Consumer-side
}

Factory functions:

FactoryDeliveryReturns
NewRedisSchedulerBackends(client)At-most-once (BRPOP)(*SchedulerBackends, error)
NewRedisStreamsSchedulerBackends(client)At-least-once (Streams)(*SchedulerBackends, core.Runnable, error)
NewInMemorySchedulerBackends()In-memory (dev/test)*SchedulerBackends

NewRedisStreamsSchedulerBackends returns a reaper Runnable that must be registered alongside the worker — without it, crashed replicas leak pending entries.

These factories are retained convenience/compatibility layers. New application bootstrap code should expose the same interfaces through OrchestrationBackends, validate the effective scheduler feature requirements, and register the bundle's runnables.

core/conformance Sub-Package​

Reusable contract test suites for interfaces owned by Core. This package is test support: provider tests import it, while runtime packages do not.

// core/conformance/task_consumer_conformance.go
func RunTaskConsumerConformance(t *testing.T, factory TaskConsumerFactory)
func RunTaskDeliveryProfileConformance(
    t *testing.T,
    profile TaskDeliveryProfile,
    factory TaskDeliveryFactory,
)
func RunTaskStoreConformance(t *testing.T, factory TaskStoreFactory)
func RunScheduleStoreConformance(t *testing.T, factory ScheduleStoreFactory)
func RunTaskQueueConformance(t *testing.T, factory TaskQueueFactory)

const (
    TaskDeliveryAtMostOnce  TaskDeliveryProfile = "at_most_once"
    TaskDeliveryAtLeastOnce TaskDeliveryProfile = "at_least_once"
)

The task-consumer base suite runs 10 transport-neutral subtests covering roundtrip, settlement, idempotency, cancellation, concurrency, field preservation, and double settlement. The profile suite also checks terminal dead-letter behavior and the recovery semantics advertised by an at-most-once or at-least-once adapter. The storage suites verify lifecycle, namespace isolation, cross-instance visibility, cancellation, ordering, state transitions, and settlement behavior appropriate to TaskStore, ScheduleStore, and the legacy TaskQueue.

The test-only orchestration/backendconformance package supplies corresponding provider-neutral suites for execution-debug and LLM-debug stores, checkpoint persistence and leased expiry claims, command delivery, workflow state, and distributed locks. Provider packages invoke these suites from their own tests; runtime code must not import the conformance package. See Scheduled Tasks Guide for delivery details.

func RunExecutionStoreConformance(t *testing.T, factory func(*testing.T) ExecutionFixture)
func RunLLMDebugStoreConformance(t *testing.T, factory func(*testing.T) LLMDebugFixture)
func RunCheckpointConformance(t *testing.T, factory CheckpointFactory)
func RunCommandStoreConformance(t *testing.T, factory func(*testing.T) CommandFixture)
func RunWorkflowStateConformance(t *testing.T, factory func(*testing.T) WorkflowFixture)
func RunDistributedLockConformance(t *testing.T, factory LockFactory)

Upstream Error Classification​

Utilities for tools that call upstream APIs. Extracts status codes from error messages and maps them to (HTTPStatus, ErrorCategory, Retryable). Recognized input 400/404/409/422 errors become HTTP 400 with Retryable:false, stopping executor retries and correction. Tools can explicitly opt a known safely correctable condition into correction; the generic classifier does not grant that permission.

UpstreamErrorInfo​

type UpstreamErrorInfo struct {
    HTTPStatus int           // Tool HTTP response status (maps to orchestrator routing)
    Category   ErrorCategory // For ToolError.Category field
    Retryable  bool          // For ToolError.Retryable field
    Code       string        // For ToolError.Code field (e.g., "API_ERROR", "RATE_LIMIT")
}

ClassifyUpstreamError​

Extracts an HTTP status code from an error message using the regex (?:status|error|code)[:\s]+(\d{3}) and classifies it for orchestrator routing.

func ClassifyUpstreamError(err error) UpstreamErrorInfo

Classification mapping:

Upstream StatusTool HTTPCategoryRetryableCode
400, 404, 409, 422400CategoryInputErrorfalseAPI_ERROR
401401CategoryAuthErrorfalseAUTH_ERROR
403403CategoryAuthErrorfalseAUTH_ERROR
429429CategoryRateLimittrueRATE_LIMIT
500+502CategoryServiceErrortrueSERVICE_ERROR
No match / nil502CategoryServiceErrortrueSERVICE_ERROR

Example — Tool handler with upstream API:

func (h *HotelHandler) handleSearch(rw http.ResponseWriter, r *http.Request) {
    result, err := h.amadeusClient.SearchHotels(ctx, params)
    if err != nil {
        info := core.ClassifyUpstreamError(err)
        safeError := core.RedactSensitiveText(err.Error())
        h.sendUpstreamError(rw, "Hotel search failed: "+safeError, info)
        return
    }
    h.sendSuccess(rw, result)
}

RedactSensitiveText​

Explicit helpers that an application, adapter, or framework subsystem may call when it has deliberately chosen to transform diagnostic text:

func RedactSensitiveText(value string) string
func RedactSensitiveError(err error) error

logger.Warn("Upstream request failed", map[string]interface{}{
    "error": core.RedactSensitiveText(err.Error()),
})

return fmt.Errorf("upstream request failed: %w", core.RedactSensitiveError(err))

The helper recognizes common credential assignments and JSON fields, bearer or basic authorization values, credential-bearing URL user information, and common credential query parameters. It retains surrounding diagnostic structure and replaces values with [REDACTED].

This is a lossy, pattern-based transformation, not a general secret detector and not a framework-wide persistence guarantee. Built-in debug stores and recorders preserve the values supplied to them. Applications that require redaction must choose and install that policy explicitly through their logger, telemetry pipeline, recorder, storage implementation, middleware, or processing hooks.

Some older framework paths still invoke these helpers before storage or observation. In particular, the orchestration executor currently transforms downstream failure text used by retry/error-analysis and error surfaces, and selected skills and provider diagnostics have similar behavior. These are legacy implementation exceptions pending the audit required by FRAMEWORK_DESIGN_PRINCIPLES.md; adopters must not rely on them as complete or stable sanitization. Successful application payloads are not automatically rewritten.

Use RedactSensitiveText when only an observation field needs sanitization. Use RedactSensitiveError when the sanitized error is returned: its Error message is redacted while Unwrap preserves errors.Is and errors.As classification of the original cause.

BackoffConfig​

Pure calculation utility for exponential backoff with deterministic jitter. Used by the orchestration module (step retry) and resilience module (retry executor). Unlike RetryConfig (a serializable config struct), BackoffConfig is a runtime calculation type.

type BackoffConfig struct {
    InitialDelay  time.Duration // Base delay for the first attempt (e.g., 500ms)
    MaxDelay      time.Duration // Upper bound — delay never exceeds this (e.g., 10s)
    BackoffFactor float64       // Multiplier per attempt (e.g., 2.0 for doubling)
    JitterEnabled bool          // Add deterministic ±10% jitter (math.Sin-based)
}

func DefaultBackoffConfig() BackoffConfig  // 500ms initial, 10s max, 2x factor, jitter enabled
func (c BackoffConfig) Delay(attempt int) time.Duration  // Calculate delay for attempt (1-indexed)

Backoff progression (defaults):

AttemptBase DelayWith Jitter (±10%)
1500ms450–550ms
21s900ms–1.1s
32s1.8–2.2s
44s3.6–4.4s
5+capped at 10scapped at 10s

Request Context Propagation​

Functions for propagating orchestration context (request IDs, step IDs) through the call chain. Used by agents to correlate LLM debug recordings back to the originating orchestration request.

Context Helpers​

// Store and retrieve request ID in context
func WithRequestID(ctx context.Context, id string) context.Context
func GetRequestID(ctx context.Context) string

// Store and retrieve step ID in context
func WithStepID(ctx context.Context, id string) context.Context
func GetStepID(ctx context.Context) string

// Validate and carry opaque multi-turn correlation.
const MetadataConversationID = "conversation_id"

func ValidateConversationID(id string) ConversationIDValidationReason
func WithConversationID(ctx context.Context, id string) context.Context
func GetConversationID(ctx context.Context) string
func GetConversationIDCandidate(ctx context.Context) ConversationIDCandidate
func WithoutConversationID(ctx context.Context) context.Context

ValidateConversationID accepts 1–512 bytes from the documented W3C-baggage-safe visible-ASCII subset and returns a bounded reason enum. MetadataConversationID is the shared framework-owned metadata and baggage key. orchestration.MetadataConversationID remains an alias so applications can continue to set orchestration request metadata without importing a second package. WithConversationID records a programmatic candidate. An explicit X-TruvaG3-Conversation-ID candidate has precedence regardless of call order; invalid raw values are not retained. WithoutConversationID shadows inherited conversation identity while preserving unrelated context values.

ExtractRequestContext​

Extract orchestration context from HTTP request headers. Call this in agent/tool HTTP handlers to enable LLM debug recording correlation.

func ExtractRequestContext(ctx context.Context, r *http.Request) context.Context

Headers extracted:

HeaderContext KeyPurpose
X-TruvaG3-Request-IDtruvag3_request_idCorrelates agent LLM calls to orchestration request
X-TruvaG3-Step-IDtruvag3_step_idIdentifies which plan step triggered the call
X-TruvaG3-Conversation-IDvalidated conversation candidateCorrelates multiple top-level turns without assigning application session semantics

Example — Agent HTTP Handler:

func handleCapabilityRequest(w http.ResponseWriter, r *http.Request) {
    // Extract orchestration context from headers set by executor
    ctx := core.ExtractRequestContext(r.Context(), r)

    // Now any LLM call made with this ctx will be recorded
    // under the orchestration request's debug trace
    response, err := aiClient.GenerateResponse(ctx, prompt, options)
}

Memory Implementations​

NewInMemoryStore​

Fast, local memory storage with automatic expiration. Perfect for development and single-instance deployments.

func NewInMemoryStore() *InMemoryStore

Features:

  • Automatic TTL expiration - Items expire based on TTL
  • Background cleanup - Removes expired items every 10 minutes
  • Thread-safe - Concurrent access with RWMutex
  • Capacity limited - Max 1000 items (prevents memory leaks)

When to use:

  • Development and testing
  • Single-instance deployments
  • Temporary caching
  • Session storage for small apps

Example - Rate Limiting:

func RateLimitMiddleware(store core.Memory, limit int) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            clientIP := r.RemoteAddr
            key := "rate:" + clientIP

            // Check current count
            val, err := store.Get(ctx, key)
            if err != nil {
                // First request - allow and start counting
                store.Set(ctx, key, "1", 1*time.Minute)
                next.ServeHTTP(w, r)
                return
            }

            count, _ := strconv.Atoi(val)
            if count >= limit {
                http.Error(w, "Rate limit exceeded", 429)
                return
            }

            // Increment counter
            store.Set(ctx, key, strconv.Itoa(count+1), 1*time.Minute)
            next.ServeHTTP(w, r)
        })
    }
}

AI Module​

Connect to AI providers and build intelligent agents that leverage LLMs for natural language understanding and generation.

Provider packages are opt-in and register their factories from init. Link every provider an application may select or auto-detect, normally with blank imports:

import (
    _ "github.com/truvaagents/truva-g3/ai/providers/anthropic"
    _ "github.com/truvaagents/truva-g3/ai/providers/azureopenai"
    _ "github.com/truvaagents/truva-g3/ai/providers/gemini"
    _ "github.com/truvaagents/truva-g3/ai/providers/openai"
)

Bedrock additionally requires a blank import of github.com/truvaagents/truva-g3/ai/providers/bedrock from code compiled with -tags bedrock. Construction and auto-detection can use only linked provider factories.

Request-Aware AI API​

The request-aware API extends the legacy AIClient contract without breaking existing clients. Its provider-neutral types and dispatch helpers live in core; provider construction and integration options live in ai.

Construction and dispatch​

func NewRequestClient(options ...ClientOption) (core.AIRequestClient, error)

type ClientOption interface {
    // sealed framework option contract
}

func GenerateAI(
    ctx context.Context,
    client core.AIClient,
    request *core.AIRequest,
) (*core.AIResult, error)

func StreamAI(
    ctx context.Context,
    client core.AIClient,
    request *core.AIRequest,
    callback core.StreamCallback,
) (*core.AIResult, error)

type AIRequestClient interface {
    AIClient
    Generate(context.Context, *AIRequest) (*AIResult, error)
}

type StreamingAIRequestClient interface {
    AIRequestClient
    Stream(context.Context, *AIRequest, StreamCallback) (*AIResult, error)
}

var ErrAIRequestFeatureUnsupported error

type AIRequestFeatureError struct {
    ClientType string
    Feature    string
}

ClientOption is sealed; applications compose the exported option helpers rather than implementing it. Existing AIOption values satisfy ClientOption, so legacy configuration can be mixed with the advanced options. GenerateAI and StreamAI prefer the request-aware capability and fall back to legacy clients only when the request is losslessly representable. An unsupported feature returns *core.AIRequestFeatureError and matches core.ErrAIRequestFeatureUnsupported.

Built-in request-aware factories currently include Anthropic, OpenAI, Gemini, Azure OpenAI, Vertex-hosted Claude, and Bedrock with the bedrock build tag.

AIRequest and presence-aware generation​

type AIRequest struct {
    Prompt     string
    Purpose    string
    Generation AIGenerationOptions
    Patches    []AIProviderPatch
}

type AIGenerationOptions struct {
    Model           string
    Temperature     AIParameter[float32]
    TopP            AIParameter[float32]
    TopK            AIParameter[int]
    MaxTokens       AIParameter[int]
    SystemPrompt    AIParameter[string]
    ReasoningEffort AIParameter[string]
    ResponseFormat  AIParameter[string]
}

type AIParameterMode uint8

const (
    AIParameterInherit AIParameterMode = iota
    AIParameterSet
    AIParameterOmit
)

type AIParameter[T any] struct {
    Mode  AIParameterMode
    Value T
}

func NewAIRequest(prompt, purpose string) *AIRequest
func NewAIRequestFromLegacy(prompt, purpose string, options *AIOptions) *AIRequest
func CloneAIRequest(request *AIRequest) (*AIRequest, error)

func (request *AIRequest) LegacyOptions() *AIOptions
func (request *AIRequest) LegacyRepresentable() bool

func InheritAIParameter[T any]() AIParameter[T]
func SetAIParameter[T any](value T) AIParameter[T]
func OmitAIParameter[T any]() AIParameter[T]

The zero value of AIParameter is Inherit. Set explicitly supplies a value, including its zero value. Omit requires the provider field to be absent. Model is structural: an empty string inherits and the model cannot be explicitly omitted.

NewAIRequestFromLegacy and LegacyOptions recursively copy legacy map, slice, and array containers. Opaque legacy Extra leaves retain their backward-compatible shared-reference behavior and are never mutated by the framework. LegacyOptions is not an application mutation surface. LegacyRepresentable reports whether dispatch to a legacy-only AIClient would preserve every request-aware semantic.

Purpose must be a stable, non-secret operation label because it may appear in policy selectors, sanitized reports, and traces.

AIResult and request reports​

type AIResult struct {
    Response      *AIResponse
    RequestReport *AIRequestReport
    UsageDetails  *AIUsageDetails
}

type AIRequestReport struct {
    Provider             string
    ProviderAlias        string
    Surface              string
    Operation            string
    Purpose              string
    RequestedModel       string
    ResolvedModel        string
    EffectiveTemperature AIParameter[float32]
    EffectiveMaxTokens   AIParameter[int]
    Adjustments          []AIRequestAdjustment
    Fingerprint          string
    Stable               bool
}

type AIRequestAdjustment struct {
    Source string
    Rule   string
    Path   string
    Action string
    Reason string
}

type AIUsageDetails struct {
    CachedInputTokens int64
    ReasoningTokens   int64
    AudioInputTokens  int64
    AudioOutputTokens int64
    Counters          map[string]int64
}

type AIRequestFingerprinter interface {
    RequestFingerprint(
        context.Context,
        *AIRequest,
    ) (fingerprint string, stable bool)
}

Reports contain only sanitized preparation facts. They must not contain prompt text, credentials, raw bodies, complete secret-bearing endpoints, or secret field values. AIRequestFingerprinter optionally exposes the same stable, secret-free semantic identity before execution; AI-output caches must bypass reads and writes when it reports stable=false.

EffectiveTemperature and EffectiveMaxTokens describe the final provider body after built-ins, application rules, middleware, per-request patches, and compatibility handling. Set(value) means the exact value was sent, Omit means the field was absent, and Inherit means that provider surface could not report the effective field. These fields are evidence, not a second request configuration surface.

Provider patches​

type AIProviderPatch struct {
    Name          string
    Version       string
    Selector      AIProviderSelector
    Set           map[string]interface{}
    Remove        []string
    SetHeaders    map[string]string
    RemoveHeaders []string
}

type AIProviderSelector struct {
    Provider      string
    ProviderAlias string
    Surface       string
    Model         string
    Operation     string
    Purpose       string
    AllProviders  bool
}

Body paths use RFC 6901 JSON Pointer syntax. Set values must be JSON-native: string-keyed maps, slices, arrays, finite scalars, or nil. Pointers, structs, non-finite floats, non-string map keys, and cyclic values are rejected.

Request-aware construction accepts:

func WithRequestRules(rules ...core.AIProviderPatch) ClientOption
func WithRequestMiddleware(m ...requestpolicy.RequestMiddleware) ClientOption
func WithCompatibilityMode(mode requestpolicy.CompatibilityMode) ClientOption

The middleware and compatibility contracts are:

type RequestMiddleware interface {
    Name() string
    Version() string
    Apply(context.Context, RequestEditor) error
}

type StableRequestMiddleware interface {
    RequestMiddleware
    StablePolicyFingerprint() bool
}

type RequestInfo struct {
    Provider       string
    ProviderAlias  string
    Surface        string
    Operation      string
    Purpose        string
    RequestedModel string
    ResolvedModel  string
}

type RequestEditor interface {
    Info() RequestInfo
    Get(path string) (interface{}, bool)
    Set(path string, value interface{}) error
    Remove(path string) error
    SetHeader(name, value string) error
    RemoveHeader(name string) error
}

type CompatibilityMode uint8

const (
    CompatibilityCompatible CompatibilityMode = iota
    CompatibilityStrict
)

type PolicyError struct {
    Stage string
    Rule  string
    Path  string
    Err   error
}

The order is provider built-ins, application rules, middleware, per-request patches, then compatibility and provider-draft validation. Compatible mode reports built-in adjustments. Strict mode rejects a built-in adjustment to explicit intent unless application policy acknowledges the affected path.

PolicyError preserves its cause via Unwrap. Deterministic local stages request-patch-validation, mutation, compatibility, draft-validation and output-capacity also match core.ErrInvalidConfiguration through errors.Is. The default AI circuit-breaker classifier excludes these caller-side failures. Middleware and unknown extension stages retain their cause's classification; they are not blanket-exempted from provider-health accounting.

Middleware must be safe for concurrent use and must not retain its call-local editor. It is fingerprint-unstable unless it implements requestpolicy.StableRequestMiddleware and explicitly declares deterministic, versioned behavior.

Provider authors can use the complete shared policy engine and logical-document surface:

type Draft interface {
    RequestEditor
    Validate() error
}

type Config struct {
    BuiltIns   []core.AIProviderPatch
    AppRules   []core.AIProviderPatch
    Middleware []RequestMiddleware
    Mode       CompatibilityMode
}

type Engine struct { /* unexported fields */ }

func NewEngine(config Config) (*Engine, error)
func (engine *Engine) WithBuiltInRules(
    patches ...core.AIProviderPatch,
) (*Engine, error)
func (engine *Engine) Apply(
    ctx context.Context,
    draft Draft,
    perRequest []core.AIProviderPatch,
) (*core.AIRequestReport, error)

type DocumentConfig struct {
    Info                 RequestInfo
    Body                 map[string]interface{}
    Headers              map[string]string
    ProtectedPaths       []string
    ProtectedHeaders     []string
    CaseInsensitivePaths []string
}

type Document struct { /* unexported fields */ }

func NewDocument(config DocumentConfig) (*Document, error)
func (document *Document) Info() RequestInfo
func (document *Document) Get(path string) (interface{}, bool)
func (document *Document) Set(path string, value interface{}) error
func (document *Document) Remove(path string) error
func (document *Document) SetHeader(name, value string) error
func (document *Document) RemoveHeader(name string) error
func (document *Document) Header(name string) (string, bool)
func (document *Document) Body() map[string]interface{}
func (document *Document) Headers() map[string]string
func (document *Document) WouldSetChange(path string, value interface{}) bool
func (document *Document) Validate() error

func CloneJSONValue(value interface{}) (interface{}, error)
func ClonePatches(
    patches []core.AIProviderPatch,
) ([]core.AIProviderPatch, error)

NewEngine snapshots its configuration and returns an immutable, concurrency-safe engine. WithBuiltInRules derives a separate immutable engine with additional provider-owned rules, validating and cloning the supplied patches. It preserves application rules, middleware and Compatible/Strict precedence without mutating the original engine. A nil receiver or invalid patch returns an error. Providers can use it for a call-local capability snapshot; applications should use application rules instead. Matching built-ins contribute to the existing request report and policy fingerprint.

NewDocument adopts Body, so that map must already be call-local; it snapshots the other configuration. Body exposes the document-owned map for provider validation and encoding, while Headers returns an isolated copy. CloneJSONValue and ClonePatches reject unsupported or cyclic values instead of retaining caller-owned mutable containers.

Enterprise integration options​

func WithCredentialSource(source CredentialSource) ClientOption
func WithEndpointResolver(resolver EndpointResolver) ClientOption
func WithAuthHeader(name string, value AuthHeaderFunc) ClientOption
func WithHTTPClient(client *http.Client) ClientOption

The corresponding public contracts are:

type EndpointRequest struct {
    Provider      string
    ProviderAlias string
    Surface       string
    ResolvedModel string
    Operation     string
    Purpose       string
}

type EndpointResolver interface {
    ResolveEndpoint(context.Context, EndpointRequest) (ResolvedEndpoint, error)
}

type AIRequestFailureReason string

const (
    AIRequestFailureReasonRoute AIRequestFailureReason = "route"
)

type AIRequestFailureReasoner interface {
    AIRequestFailureReason() AIRequestFailureReason
}

type ResolvedEndpoint struct {
    URL             *url.URL
    Deployment      string
    RouteIdentity   string
    CredentialScope string
    Query           url.Values
}

type CredentialRequest struct {
    Provider        string
    ProviderAlias   string
    Surface         string
    Operation       string
    ResolvedModel   string
    RouteIdentity   string
    Deployment      string
    CredentialScope string
}

type HeaderCredential struct {
    Name  string
    Value string
}

func NewHeaderCredential(name, value string) HeaderCredential

type CredentialSource interface {
    Credential(context.Context, CredentialRequest) (HeaderCredential, error)
}

type CredentialRejectionObserver interface {
    CredentialRejected(context.Context, CredentialRequest, statusCode int) error
}

type AuthHeaderFunc func(context.Context) (string, error)

CredentialSource is called once per transport attempt after policy and route resolution. Implementations must be concurrency-safe; credential values are never included in reports, fingerprints, or logs. A source may implement CredentialRejectionObserver to observe HTTP 401/403 responses without replacing the original provider error.

For HTTP providers, EndpointResolver returns a complete URL and a stable, non-secret RouteIdentity. An SDK-native provider may define a narrower destination contract. Bedrock requires URL == nil, no query or credential scope, an opaque Converse modelId in Deployment, and a stable route identity. A resolver may run during cache fingerprint preflight and again on a cache miss, so it must be concurrency-safe, stable, and side-effect-free. WithHTTPClient is supported by the request-aware HTTP providers, which shallow-copy and do not mutate the supplied client. SDK-native Bedrock rejects HTTP and credential integration options while accepting its constrained destination resolver.

AIRequestFailureReasoner is an optional error contract for provider-owned failures whose chain meaning cannot be expressed accurately by core.ProviderError. Route-resolution and local route-viability errors return the typed AIRequestFailureReasonRoute constant. Generate and streaming chains use it for bounded route failure metadata on attempts, recovery, abort, and exhaustion. Unknown values degrade to unknown; callers still receive the original wrapped error. Caller cancellation and deadlines retain precedence.

Heterogeneous chains​

func NewChain(entries ...ChainEntry) (*ChainClient, error)
func ProviderEntry(name, providerAlias string, options ...ClientOption) ChainEntry
func ClientEntry(name string, client core.AIClient) ChainEntry

Entry names must be unique, stable, non-secret labels of 1–256 bytes with no surrounding whitespace or control characters. ProviderEntry aliases have the same length and character constraints. ProviderEntry constructs an independently configured request-aware provider. ClientEntry invokes a caller-owned client without mutating it through optional setters. Streaming failover is allowed only before the first chunk is delivered.

NewChainClient remains the legacy homogeneous-chain constructor. Use NewChain when entries need different policy, routing, credentials, or client implementations.

Retry defaults differ deliberately. A legacy NewChainClient gives each provider zero in-provider retries by default so provider retries do not multiply ordered failover attempts. Each ProviderEntry in NewChain is independently materialized through NewRequestClient, so it uses the single-client default of three retries unless that entry supplies WithMaxRetries(0) or another explicit value. A ClientEntry keeps whatever retry behavior its caller-owned client implements. Unless an explicit option wins, a positive TRUVAG3_AI_RETRY_ATTEMPTS value overrides the constructor-specific default for single clients, ProviderEntry, and legacy NewChainClient; it never reconfigures an injected ClientEntry.

Entries using ai/providers/openai and its registered aliases use providers.BaseClient.ExecuteWithRetryDecoded to share this same budget across HTTP/transport and eligible decoded API failures. The provider-owned ResponseAttemptDecoder returns a DecodedAttempt containing result/error, same-provider Retryable, semantic status and optional sanitized response identity. The helper closes decoded bodies and retains terminal partial results. Stream output prevents replay. ExecuteWithRetry / ExecuteWithRetryPrepared remain HTTP-only; direct Azure classic/v1, other providers and injected clients retain their existing policy.

// ai/providers: adapter-owned decision for one decoded HTTP response.
type DecodedAttempt struct {
    Result            *core.AIResult
    Err               error
    Retryable         bool
    SemanticStatus    int
    ProviderRequestID string
}

type ResponseAttemptDecoder func(context.Context, *http.Response) DecodedAttempt

func (b *BaseClient) ExecuteWithRetryDecoded(
    ctx context.Context, req *http.Request,
    prepare RequestAttemptPreparer, decode ResponseAttemptDecoder,
) (*core.AIResult, error)

decode is required; prepare is optional and runs once per wire attempt. The decoder owns retry permission for real HTTP errors (including 429/5xx) as well as in-body errors under HTTP 200. There is no second status-based retry fallback when it returns false. Transport failures and the HTML-error guard remain base-owned. The returned result belongs only to the terminal attempt; earlier partial results are not merged. SemanticStatus is not the real HTTP status or proof the provider supplied it; only 429/503 enable Retry-After handling on an otherwise eligible retry. ProviderRequestID must be bounded and sanitized by the adapter (base observation allows at most 128 bytes). Retryable here controls same-provider recovery, unlike core.ProviderError.IsRetryable, which permits chain failover.

For OpenRouter, recognized transient types can retry; unknown nonempty types are conservative, and untyped status-only retries allow 408/429/500/502/503/504, not 501/505. Actual 401/402/403 stop same-provider recovery even if an error body claims otherwise. Native OpenAI and the other aliases served by ai/providers/openai also stop same-provider retries for classified quota/billing exhaustion (including insufficient_quota reported as HTTP 429). Their ordinary transient 429/5xx status policy is otherwise unchanged. Quota exhaustion still permits chain failover when another entry is available; it does not imply an automatic wait for quota reset. Azure uses its separate adapter.

OpenRouter's credible HTTP/in-body 408 normalizes to timeout with status 408, including a typed 408 timeout; it uses the existing bounded retry and core.ProviderError.IsRetryable chain signals. This is OpenRouter-local and does not expand the shared status allowlist or other aliases' 408 behavior. Zero local budget permits no same-provider retry, and failover needs another entry. Caller cancellation, caller/provider-call deadlines and delivered streaming output stop recovery.

Buffered openaiwire.Codec.Decode failures preserve the reader/JSON cause through errors.Is/errors.As while keeping decode OpenAI wire response as the outward message. Its error exposes the optional fact method ResponseBodyReadError() error: a non-nil observed non-EOF reader failure maps to transport; nil maps to decode for JSON syntax/type errors, including incomplete JSON at clean EOF. Cancellation, deadline and partial-stream classifications retain precedence. Compatible execution adds expired attempt-context identity only when the returned error does not already match it. The chain stops on returned cancellation/deadline identity, including an expired provider-call deadline with an active outer caller. It does not select recovery based only on the outer context. Successful decoded responses retain their existing behavior; these observation changes add no retries.

Model capability lookup and adaptation​

// ai/providers: static adapter contract, not a live model catalog.
type ModelCapabilities struct {
    ProviderAlias             string
    ModelPrefix               string
    ExactModel                bool
    MatchFamilyBoundary       bool
    ReasoningStyle            string
    SupportsJSONMode          bool
    SupportsJSONSchema        bool
    SupportsHeaderPassthrough bool
}

func LookupModelCapabilities(providerAlias, model string) ModelCapabilities
func LogInheritedEffortOmitted(context.Context, core.Logger, string, string)

Lookup chooses the most specific matching row for the provider alias. Model matching is case-insensitive: ExactModel takes precedence; a family-boundary row matches the name or a hyphen suffix (not gpt-60 for gpt-6); other rows retain prefix matching. Unknown capabilities are conservative. Applications customize requests through the existing request-policy APIs, not by mutating this built-in table. LogInheritedEffortOmitted(ctx, logger, providerAlias, model) is an adapter-authoring Debug diagnostic; the adapter separately records the adjustment. It emits no degraded-translation event or warning metric.

Native OpenAI's GPT-6 family, including gpt-6.1-sol and its hyphen-suffixed variants/snapshots, uses low, medium, high, xhigh or max. Only original GPT-6 Sol and GPT-6 Luna also allow none; GPT-6 Astra and GPT-6.1 Sol reject it. Leaving effort unset inherits the client/chain default, if any. Explicit OmitAIParameter[string]() suppresses that inheritance and the wire effort field; the provider chooses its default. Inherited active effort activates reasoning-token scaling on profiles that enable it; explicit omission does not. Active/default reasoning rejects sampling/logprob controls and native Chat Completions tools after application policy; original GPT-6 Sol/GPT-6 Luna with none can use the non-reasoning Chat Completions tool path. GPT-6.1 Sol native tools require a Responses API adapter, which this inclusion does not add. These restrictions are native-route contracts, not rules automatically applied to every compatible endpoint.

Exact OpenRouter deepseek/deepseek-v4.1-flash and z-ai/glm-5.3-flash rows declare JSON object support. Exact openai/gpt-5.6-luna, openai/gpt-6-luna and openai/gpt-6.1-sol add JSON schema and OpenAI-style reasoning. GPT-6 Luna and GPT-6.1 Sol each have a scoped built-in temperature-removal rule: Compatible mode reports adaptation; Strict mode requires application acknowledgement for a change to an explicit value. Protected privacy/required-parameter routing remains in effect. These capability rules do not change unrelated models. OpenRouter's fast-alias migration is documented under model alias resolution below; capability metadata is not a live availability guarantee.

Routed openai/gpt-6.1-sol has a final validator separate from native OpenAI: it accepts low, medium, high, xhigh and max at reasoning.effort, or omitted effort, and rejects none, noncanonical keys, top-level reasoning_effort, and sampling/logprob controls left after policy. Native reasoning extras accept non-null JSON objects, including typed/named string-keyed Go maps and named string effort values, without rewriting them. enabled and exclude require boolean values; enabled:false is rejected, while exclude:true only hides reasoning text and remains allowed. Standalone max_tokens requires a positive integer JSON number; OpenRouter owns its budget-to-effort mapping. Supplying both effort and max_tokens in the final object is rejected, including an effort inherited from the client or inserted by policy. Application patches can remove the conflict before final validation. Its openrouter-sol-6.1-v3 identity versions that validation contract for caches. The shared codec still validates before policy: an explicit sampling setting with active effort can fail before a removal rule runs. Use explicit temperature omission on the presence-aware API when appropriate. Native tool restrictions are not copied to this routed guard; OpenRouter owns its upstream translation. Supported OpenRouter portable effort now activates the configured reasoning multiplier, as native scaling profiles do. Reviewed exact native GPT-5.6 Sol, GPT-6 Astra/Sol/Luna/GPT-6.1 Sol and routed GPT-5.6 Luna/GPT-6 Luna/GPT-6.1 Sol have a 128000-token output ceiling. Added headroom for a valid base is bounded before policy; over-cap base budgets retain Compatible/Strict handling. The GPT-5.6 capacity entries add no default-reasoning inference. Unknown model/route capacity is not inferred.

Provider factory extension contracts​

type ProviderFactory interface {
    Create(*AIConfig) core.AIClient
    DetectEnvironment() (priority int, available bool)
    Name() string
    Description() string
}

type ValidatedProviderFactory interface {
    ProviderFactory
    CreateValidated(*AIConfig) (core.AIClient, error)
}

type RequestProviderFactory interface {
    ProviderFactory
    CreateRequestClient(*AIConfig, ProviderIntegrationConfig) (core.AIRequestClient, error)
}

type ProviderRequestTimeoutFactory interface {
    ProviderFactory
    DefaultRequestTimeout() time.Duration
}

type ProviderIntegrationConfig struct {
    RequestRules      []core.AIProviderPatch
    RequestMiddleware []requestpolicy.RequestMiddleware
    CompatibilityMode requestpolicy.CompatibilityMode
    CredentialSource  CredentialSource
    EndpointResolver  EndpointResolver
    HTTPClient        *http.Client
}

NewClient prefers CreateValidated. NewRequestClient uses CreateRequestClient and passes an isolated integration snapshot. The reusable ProviderRequestTimeoutFactory is optional: without it, a provider receives the framework's 300-second default. When implemented, its positive duration is used only for an explicitly selected standalone provider if the application did not supply a positive WithTimeout. Bedrock uses this contract to declare a 60-minute standalone default. Auto-detected clients and framework-managed chain entries retain 300 seconds; caller-owned ClientEntry values remain untouched. The reusable ai/providerkit/openaiwire package supplies the OpenAI chat-completions draft, encode, decode, and stream-decode contract without owning credentials, routing, retries, provider identity, or telemetry.

Compatible response and codec contracts​

Responses expose ReasoningContent, exact FinishReason and ContentSource (final, reasoning_fallback, mixed; empty means unknown). StreamChunk.ReasoningContent is separate from answer Content. The default compatible decoder keeps final text intact, including an empty final answer. It can deliver reasoning-only callbacks with empty Content and nonempty ReasoningContent and ContentSource=final; these count as delivered output for no-replay rules. Here final labels the delivery policy, not the presence of nonempty answer text in that chunk. ai.WithReasoningContentFallback(true) / ai.WithChainReasoningContentFallback(true) or direct openaiwire.Config.ReasoningContentFallback explicitly restores the legacy substitution/concatenation; both policies retain truthful provenance and have distinct cache fingerprints. Token/retry/time limits are unchanged.

The core response/provenance types and provider-authoring codec surface are:

type AIResponse struct {
    Content          string
    ReasoningContent string          `json:"reasoning_content,omitempty"`
    FinishReason     string          `json:"finish_reason,omitempty"`
    ContentSource    AIContentSource `json:"content_source,omitempty"`
    Model            string
    Provider         string
    Usage            TokenUsage
}

type AIContentSource string

const (
    AIContentSourceFinal             AIContentSource = "final"
    AIContentSourceReasoningFallback AIContentSource = "reasoning_fallback"
    AIContentSourceMixed             AIContentSource = "mixed"
)

These facts describe provider completion, not application success. A length finish reason is preserved rather than silently relabelled; empty final text is not automatically replaced or retried. Opt-in LLM debug records preserve exact available facts, including orchestration-owned planning and synthesis records; normal traces carry bounded classifications/presence flags, not reasoning text.

type Codec interface {
    BuildDraft(*core.AIRequest, string, bool) (*Draft, error)
    Encode(*Draft) ([]byte, error)
    Decode(io.Reader) (*core.AIResult, error)
    DecodeStream(io.Reader, core.StreamCallback) (*core.AIResult, error)
    SurfaceVersion() string
}

type ProfiledCodec interface {
    Codec
    BuildDraftWithProfile(
        *core.AIRequest,
        RequestProfile,
        bool,
    ) (*Draft, error)
}

type RequestProfile struct {
    SemanticModel   string
    WireModel       string
    ModelField      ModelFieldMode
    TokenLimit      TokenLimitField
    TokenBudget     TokenBudgetPolicy
    ReasoningEffort ReasoningEffortStyle
    Sampling        SamplingPolicy
    ReasoningActiveByDefault bool // reviewed provider fact; budget only, no wire effort
    ReasoningHeadroomLimit int // reviewed capacity for added headroom; 0 unknown, negative invalid
}

type Config struct {
    SurfaceVersion           string
    ReasoningTokenMultiplier int
    DefaultReasoningEffort   string
    ForceReasoningObject     bool
    MaxSSEEventBytes          int
    ReasoningContentFallback bool
}

const DefaultSurfaceVersion = "openai-chat-completions-v2"
const DefaultReasoningTokenMultiplier = 3

type ModelFieldMode uint8

const (
    ModelFieldRequired ModelFieldMode = iota + 1
    ModelFieldOmitted
)

type TokenLimitField uint8

type TokenBudgetPolicy uint8

const (
    TokenBudgetExact TokenBudgetPolicy = iota + 1
    TokenBudgetScaleForReasoning
)

const (
    TokenLimitMaxTokens TokenLimitField = iota + 1
    TokenLimitMaxCompletionTokens
)

type ReasoningEffortStyle uint8

const (
    ReasoningEffortOmitted ReasoningEffortStyle = iota + 1
    ReasoningEffortTopLevel
    ReasoningEffortNestedObject
)

type SamplingPolicy uint8

const (
    SamplingOrdinary SamplingPolicy = iota + 1
    SamplingReasoningRestricted
)

func NewCodec(surfaceVersion string) (Codec, error)
func NewConfiguredCodec(config Config) (Codec, error)
func NewProfiledCodec(config Config) (ProfiledCodec, error)

func (profile RequestProfile) Validate() error

type Draft struct {
    *requestpolicy.Document
    // unexported wire-profile state
}

func (draft *Draft) BindIdentity(provider, providerAlias string) error
func (draft *Draft) HasExplicitIntent(path string) bool
func (draft *Draft) PolicyFingerprintIdentity() string
func (draft *Draft) Adjustments() []core.AIRequestAdjustment
func (draft *Draft) ProtectedHeaderConflicts() []string
func (draft *Draft) Validate() error

func IsReasoningModel(model string) bool
func BuildRequestBody(
    model string,
    messages []map[string]string,
    maxTokens int,
    temperature float32,
    stream bool,
    reasoningTokenMultiplier int,
    reasoningEffort string,
) map[string]interface{}

RequestProfile is the explicit boundary between semantic model identity and route-owned wire identity. Use its exported mode constants rather than numeric values, and call Validate before relying on a profile constructed outside the codec. BuildRequestBody is a compatibility bridge for providers that adopted the earlier OpenAI reasoning helper; new provider integrations should use a Codec. The map-only helper saturates overflowing token multiplication at math.MaxInt; the error-returning draft API rejects it unless a reviewed limit safely bounds added headroom. Saturation is an integer-safety fallback, not a model-capacity guarantee.

For scaling profiles, ReasoningActiveByDefault reserves headroom when effective effort is absent. The provider owns evidence for that fact; the codec does not infer model defaults. It does not change sampling or emit effort. Explicit reasoning omission, none, and native-extra-only reasoning/reasoning_effort controls suppress inference. The legacy BuildRequestBody helper has no such reviewed profile and retains effort-only scaling.

ReasoningHeadroomLimit bounds only framework-added headroom for a base budget that fits the reviewed capacity. It does not clamp an over-cap base or validate application patches; the adapter must retain those capacity guards. Its zero value means unknown and negative values fail profile validation. On active scaling profiles, the limit participates in cache identity (including token-free preflight). Bounding records wire-profile / reasoning-headroom-ceiling-v1, action cap. This is a provider capability fact, not a new operator budget setting.

OpenAI-wire token fields accept positive integral numeric values that fit a Go int, including json.Number("1e3") and json.Number("1000.0") supplied by request policy. Validation and EffectiveMaxTokens agree on the value; the encoded numeric representation remains unchanged. Strings, fractions, non-finite values and out-of-range integers fail before dispatch.

IsReasoningModel, BuildDraft and BuildRequestBody recognize gpt-6 and gpt-6-*, plus gpt-6.1-sol and its hyphen-suffixed variants/snapshots. Recognition is case-insensitive and does not include unrelated gpt-6.1-*, gpt-6.10-*, gpt-60 or vendor-prefixed names. The compatibility helpers select reasoning token/budget and sampling wire handling; they do not independently provide every native adapter model-contract check. Azure v1 selects its existing reasoning profile for matching resolved semantic models; Azure classic rejects them. This does not certify availability or effort support for a route-owned deployment. The matcher accepts any hyphen suffix, including -pro; it does not validate snapshot dates. Such variants inherit the matching family's wire rules without establishing a separately reviewed native endpoint contract. OpenRouter's exact model rows remain independent and do not extend to suffixed variants.

The side-effect-free OpenAI-compatible semantic catalog is also public:

// Package ai/providers/openai/modelcatalog
func DefaultAliases() map[string]map[string]string
func Resolve(providerAlias, model string) string
func ResolveWithAliases(
    aliases map[string]map[string]string,
    providerAlias string,
    model string,
) string

// Package ai/providers/openai
var ModelAliases map[string]map[string]string
func ResolveModel(providerAlias, model string) string

modelcatalog.Resolve uses the immutable built-in catalog plus non-empty TRUVAG3_{PROVIDER}_MODEL_{ALIAS} overrides. DefaultAliases returns a deep copy. openai.ResolveModel deliberately uses the mutable compatibility map openai.ModelAliases; Azure uses modelcatalog.Resolve, so runtime mutations of that compatibility map do not affect Azure semantic resolution.

The shared ai/providers package also supplies the provider-observation and retry boundary:

type RequestObservation struct {
    Provider      string
    ProviderAlias string
    SemanticModel string
    PromptLength  int
}

type ResponseObservation struct {
    Provider          string
    ProviderAlias     string
    SemanticModel     string
    ResponseModel     string
    ProviderRequestID string
    Usage             core.TokenUsage
    Duration          time.Duration
}

type ErrorObservation struct {
    Operation     string
    Provider      string
    ProviderAlias string
    ErrorType     string
    Duration      time.Duration
}

type RequestAttemptPreparer func(context.Context, *http.Request) error

func (client *BaseClient) ExecuteWithRetry(
    ctx context.Context,
    request *http.Request,
) (*http.Response, error)

func (client *BaseClient) ExecuteWithRetryPrepared(
    ctx context.Context,
    request *http.Request,
    prepare RequestAttemptPreparer,
) (*http.Response, error)

func (client *BaseClient) LogRequestMetadata(ctx context.Context, observation RequestObservation)
func (client *BaseClient) LogResponseMetadata(ctx context.Context, observation ResponseObservation)
func (client *BaseClient) LogErrorMetadata(ctx context.Context, observation ErrorObservation)

func NormalizeObservationErrorType(value string) string
func SanitizedObservationError(err error, fallback string) (string, error)
func RecordObservationError(span core.Span, err error, fallback string) string
func AddObservationRequestID(
    ctx context.Context,
    fields map[string]interface{},
)
func LogTranslationDegraded(
    ctx context.Context,
    logger core.Logger,
    providerAlias string,
    model string,
    warningType string,
    capability string,
)

ResponseModel and ProviderRequestID are optional, bounded provider-response identifiers. The built-in OpenRouter client populates them only after validation; LogResponseMetadata may emit them as the response_model and provider_request_id log fields. They are never metric labels. OpenRouter's provider-local span uses the corresponding ai.response.model and ai.provider_request_id attributes; other provider aliases leave them empty.

Request bodies must be replayable before ExecuteWithRetry performs network I/O. ExecuteWithRetryPrepared invokes its preparer once per attempt on a fresh request, which is the supported point for rotating credentials. Observation error types are closed, bounded classifiers; raw provider error text must not be used as a log, metric, or span label. AddObservationRequestID uses telemetry baggage first and the core request context as a fallback.

See the Custom AI Providers and Enterprise Integration Guide for implementation patterns and security requirements.

NewClient​

Create an AI client with automatic provider detection. The client intelligently selects the best available AI provider based on environment variables.

func NewClient(opts ...AIOption) (core.AIClient, error)
func MustNewClient(opts ...AIOption) core.AIClient  // Panics on error

Provider auto-detection order (by priority):

  1. OpenAI (OPENAI_API_KEY) - Priority 1000
  2. Anthropic (ANTHROPIC_API_KEY) - Priority 900
  3. OpenRouter (OPENROUTER_API_KEY) - Priority 850
  4. Google Gemini (GOOGLE_API_KEY then GEMINI_API_KEY) - Priority 800
  5. Groq (GROQ_API_KEY) - Priority 700
  6. DeepSeek (DEEPSEEK_API_KEY) - Priority 600
  7. xAI Grok (XAI_API_KEY) - Priority 500
  8. Mistral (MISTRAL_API_KEY) - Priority 450
  9. Qwen (QWEN_API_KEY) - Priority 400
  10. Together AI (TOGETHER_API_KEY) - Priority 300
  11. AWS Bedrock (AWS credentials, -tags bedrock) - Priority 200; priority 250 when availability is detected from a Lambda/ECS-style managed-runtime indicator (static credential and profile indicators are checked first)
  12. Ollama (OLLAMA_BASE_URL must be set) - Priority 100

Azure OpenAI and Vertex-hosted Claude deliberately do not participate in environment auto-detection. Their route and short-lived credential contracts must be supplied explicitly through NewRequestClient.

Example - Auto-detect Provider:

// Automatically uses first available provider
client, err := ai.NewClient()
if err != nil {
    log.Fatal("No AI provider available. Set OPENAI_API_KEY, ANTHROPIC_API_KEY, or OPENROUTER_API_KEY")
}

response, _ := client.GenerateResponse(ctx, "Explain quantum computing", nil)
fmt.Println(response.Content)

Example - Specific Provider:

// Use specific provider with custom settings
client, err := ai.NewClient(
    ai.WithProvider("anthropic"),
    ai.WithModel("smart"),
    ai.WithTemperature(0.7),
    ai.WithMaxTokens(2000),
)

// Use for complex reasoning tasks
response, _ := client.GenerateResponse(ctx,
    "Analyze this code for potential issues and suggest improvements",
    &core.AIOptions{
        Temperature: 0.3,  // Lower temperature for technical analysis
    },
)

WithProviderAlias​

Select a registered provider profile. Some aliases select an automatically configured OpenAI-compatible service; hosted-enterprise aliases select a request-aware wire profile whose route and credentials must be supplied by the application.

func WithProviderAlias(alias string) AIOption

Automatically configured OpenAI-compatible aliases:

  • "openai" - Standard OpenAI (default)
  • "openai.openrouter" - OpenRouter with protected privacy/routing defaults
  • "openai.deepseek" - DeepSeek with reasoning models
  • "openai.groq" - Groq for ultra-fast inference
  • "openai.together" - Together AI for open models
  • "openai.xai" - xAI Grok models
  • "openai.mistral" - Mistral AI
  • "openai.qwen" - Alibaba Qwen models
  • "openai.ollama" - Local Ollama

Request-aware hosted profiles:

  • "azureopenai.v1" - Azure OpenAI v1 chat completions
  • "azureopenai.classic" - Azure deployment-scoped classic chat completions
  • "anthropic.vertex" - Anthropic Claude on Vertex AI

The hosted profiles require NewRequestClient and WithEndpointResolver. Azure accepts either WithAPIKey or an application-owned credential source; Vertex requires the credential source. Google Cloud's OpenAI-compatible endpoint uses the ordinary "openai" profile with a Google route and credential source; it does not have a separate provider alias.

Features:

  • Automatic endpoint configuration for the openai.* service aliases
  • Explicit, route-owned endpoints for Azure and Google-hosted profiles
  • Model aliases such as "smart", "fast", and "code"
  • Configuration precedence of explicit option → environment → built-in default where that provider supports environment configuration

Example - Using Alternative Providers:

// Use DeepSeek's reasoning model
client, _ := ai.NewClient(
    ai.WithProviderAlias("openai.deepseek"),
    ai.WithModel("smart"),  // Resolves to "deepseek-reasoner"
)

// Use Groq for fast inference
client, _ := ai.NewClient(
    ai.WithProviderAlias("openai.groq"),
    ai.WithModel("fast"),   // Resolves to "llama-3.1-8b-instant"
)

// Use Together AI with explicit model
client, _ := ai.NewClient(
    ai.WithProviderAlias("openai.together"),
    ai.WithModel("deepseek-ai/DeepSeek-V4-Flash-0731"), // Explicit model name
)

// Use OpenRouter with a concrete default model
client, _ := ai.NewClient(
    ai.WithProviderAlias("openai.openrouter"),
    ai.WithModel("openai/gpt-5.6-sol"),
)

// Retain the environment-overridable fast alias. This resolves
// TRUVAG3_OPENROUTER_MODEL_FAST first, then the built-in fast model.
fastClient, _ := ai.NewClient(
    ai.WithProviderAlias("openai.openrouter"),
    ai.WithModel("fast"),
)

// Override the constructor default for one call with an exact OpenRouter slug.
response, err := client.GenerateResponse(ctx, prompt, &core.AIOptions{
    Model: "moonshotai/kimi-k3",
})

For OpenRouter, constructor and per-call model selections accept default, smart, fast, and code, or an exact OpenRouter slug. Alias environment variables use TRUVAG3_OPENROUTER_MODEL_{ALIAS}. Per-call selection wins over WithModel; a selected alias then resolves through its non-empty environment override before the built-in catalog.

Configuration priority:

// 1. Explicit configuration (highest priority)
client, _ := ai.NewClient(
    ai.WithProviderAlias("openai.groq"),
    ai.WithAPIKey("explicit-key"),        // Overrides environment
    ai.WithBaseURL("https://custom.url"), // Overrides defaults
)

// 2. Environment variables (medium priority)
// Set: GROQ_API_KEY=your-key
// Set: GROQ_BASE_URL=https://custom.groq.com  // Optional override
client, _ := ai.NewClient(
    ai.WithProviderAlias("openai.groq"),
)

// 3. Hardcoded defaults (lowest priority)
// Uses built-in endpoints like https://api.groq.com/openai/v1

Model Aliases​

Use portable model names across providers that define an alias catalog. OpenAI-compatible, Anthropic, and Gemini providers resolve the aliases below; Bedrock and custom providers without a catalog pass model strings through unchanged. For a chain that includes such a provider, leave a per-request model empty to use each entry's configured default, or supply only a concrete ID accepted by every entry.

Standard aliases:

  • "smart" - Most capable model for complex tasks
  • "fast" - Quick responses for simple queries
  • "code" - Optimized for code generation
  • "vision" - Multimodal/vision capabilities (if available)
  • "default" - Provider's recommended default

Model alias resolution examples (snapshot — see ai/providers/openai/models.go for the canonical mapping):

OpenAI's general default resolves to gpt-6.1-sol. Planning and synthesis inherit it when no phase/client override selects another model. The fast aliases select gpt-6-luna (OpenAI) and openai/gpt-6-luna (OpenRouter); other alias mappings are unchanged. Override with a concrete model or a non-empty TRUVAG3_OPENAI_MODEL_DEFAULT (for example, gpt-5.6-terra to retain the prior default). Azure/shared-catalog endpoint resolvers also receive the new semantic default; route-owned deployment names are not automatically changed. GPT-6.1 Sol rejects inherited none effort, so use a supported effort or explicit omission rather than expecting the alias change to rewrite request settings. Fast-tier overrides use TRUVAG3_OPENAI_MODEL_FAST and TRUVAG3_OPENROUTER_MODEL_FAST; existing pins retain precedence. Fast selection does not change effort or token budgets. Azure/shared-catalog resolvers also receive Luna for the fast semantic model.

Anthropic fast selects claude-haiku-5-5; smart, code, and vision select claude-opus-5-5. default and premium remain Sonnet 5 and Fable 5. Nonempty TRUVAG3_ANTHROPIC_MODEL_<ALIAS> overrides still win. anthropic.vertex resolves the same semantic catalog before application-owned publisher mapping. Budgets are unchanged; adaptive thinking consumes part of max_tokens. See the Anthropic migration note for native-extra changes, sampling policy, rollback and verification scope.

// "smart" resolves differently per provider
ai.WithProviderAlias("openai")          // → gpt-5.6-sol
ai.WithProviderAlias("openai.deepseek") // → deepseek-reasoner
ai.WithProviderAlias("openai.groq")     // → openai/gpt-oss-120b
ai.WithProviderAlias("openai.openrouter") // → openrouter/auto

// "fast" for quick responses
ai.WithProviderAlias("openai")          // → gpt-6-luna
ai.WithProviderAlias("openai.openrouter") // → openai/gpt-6-luna
ai.WithProviderAlias("openai.deepseek") // → deepseek-chat
ai.WithProviderAlias("openai.groq")     // → llama-3.1-8b-instant

// Direct model names still work (pass-through)
client, _ := ai.NewClient(
    ai.WithProviderAlias("openai.deepseek"),
    ai.WithModel("deepseek-chat"), // Exact model, not an alias
)

NewChainClient​

Create a chain client that automatically fails over between multiple AI providers. Perfect for production systems requiring high availability and resilience.

func NewChainClient(opts ...ChainOption) (*ChainClient, error)

Features:

  • Automatic failover - Seamlessly switches to backup providers
  • Auto-detection - Discovers available providers from API keys when no explicit chain is specified
  • Graceful degradation - Works with single provider if needed
  • Smart error handling - Only fails over on infrastructure errors
  • Configurable chain - Define your provider priority order

Example - Auto-Detect (recommended for development):

// Auto-detects available providers from environment API keys,
// ordered by priority: OpenAI(1000) > Anthropic(900) > OpenRouter(850) >
// Gemini(800) > Groq(700) > ...
chain, err := ai.NewChainClient(
    ai.WithChainLogger(logger),
    ai.WithChainTimeout(120*time.Second),
)

// Automatically tries providers in priority order until one succeeds
response, err := chain.GenerateResponse(ctx, "Explain quantum computing", nil)

Example - Explicit Chain (recommended for production):

// Define exact failover order — providers are tried left to right
chain, err := ai.NewChainClient(
    ai.WithProviderChain("openai", "anthropic", "openai.groq"),
    ai.WithChainTimeout(120*time.Second),
)

// Tries: OpenAI → Anthropic → Groq
response, err := chain.GenerateResponse(ctx, "Explain quantum computing", nil)

Example - Using Provider Aliases:

// Chain with OpenAI-compatible providers using dot-notation aliases
chain, _ := ai.NewChainClient(
    ai.WithProviderChain("openai.deepseek", "openai.groq", "openai"),
)

WithProviderChain​

Configure the provider chain for automatic failover. Providers are tried in the order specified.

func WithProviderChain(aliases ...string) ChainOption
  • Each alias is a provider name ("openai", "anthropic") or a dot-notation sub-provider alias ("openai.groq", "openai.deepseek")
  • When omitted, NewChainClient() auto-detects available providers from environment API keys and orders them by priority

Example - Production Failover Strategy:

chain, _ := ai.NewChainClient(
    ai.WithProviderChain(
        "openai",           // Primary: Best quality
        "openai.deepseek",  // Backup 1: Good reasoning, lower cost
        "anthropic",        // Backup 2: Alternative provider
        "openai.groq",      // Emergency: Fast inference
    ),
    ai.WithChainTimeout(120*time.Second),
)

AI Circuit-Breaker Composition​

The ai module accepts the provider-neutral core.CircuitBreaker interface; applications may supply an implementation from resilience without coupling the two optional framework modules.

type CircuitBreakerFactory func(
    entryName string,
    providerAlias string,
    shouldCountFailure func(error) bool,
) (core.CircuitBreaker, error)

func WithCircuitBreaker(breaker core.CircuitBreaker) AIOption
func WithChainCircuitBreakerFactory(factory CircuitBreakerFactory) ChainOption
func NewCircuitBreakerClient(
    client core.AIClient,
    breaker core.CircuitBreaker,
) (core.AIClient, error)
func ShouldCountAICircuitBreakerFailure(err error) bool

Use WithCircuitBreaker for clients created by NewClient, NewRequestClient, or a heterogeneous ProviderEntry. Use NewCircuitBreakerClient to decorate a directly constructed provider client. For NewChainClient, the factory must create a fresh breaker for every entry and configure it with the supplied classifier; shared breaker state across providers is unsupported. The classifier counts provider-health and transport failures while excluding caller/configuration errors, ordinary non-transient 4xx responses, and post-output streaming failures. Breaker wrappers preserve the wrapped client's request-fingerprint behavior.

See resilience/README.md for a complete application-composition example.

Client Configuration Options​

AIOption values configure NewClient, NewRequestClient, and individual heterogeneous ProviderEntry values. The legacy NewChainClient accepts the separate ChainOption helpers documented below; similarly named options are not interchangeable Go types.

WithTimeout​

Set the provider request timeout. The framework default is 300 seconds (5 minutes); an explicitly selected standalone Bedrock provider declares 60 minutes. Auto-detected clients and framework-managed chain entries use 300 seconds. Positive code overrides TRUVAG3_AI_TIMEOUT, then the applicable default. A present environment value must be a positive Go duration; invalid values fail construction even with a code override. Zero and negative code values mean unset, not unbounded. Environment resolution belongs to root AI constructors and managed entries, not direct provider constructors/factory calls or caller-owned ClientEntry values; core.Config.LoadFromEnv does not read it. The total provider deadline covers retries and streaming. Shorter caller, phase, auxiliary-stage or injected HTTP-client deadlines still win. Token caps are not latency guarantees: deadlines are not derived from guessed token throughput, and deadline errors do not trigger chain failover.

func WithTimeout(timeout time.Duration) AIOption
func WithChainTimeout(timeout time.Duration) ChainOption  // For chain clients

Example:

import "time"

// Single client with an explicit five-minute timeout
client, _ := ai.NewClient(
    ai.WithTimeout(300 * time.Second),  // 5 minutes
)

// Chain client with custom timeout
chainClient, _ := ai.NewChainClient(
    ai.WithProviderChain("openai", "anthropic"),
    ai.WithChainTimeout(240 * time.Second),  // 4 minutes
)

WithReasoningTokenMultiplier​

Configure the token multiplier for OpenAI reasoning profiles (including GPT-5, GPT-6, o1, o3 and o4). Reasoning consumes the completion budget even when it is not returned as visible answer text. The multiplier applies when a scaling wire profile has an effective nonempty effort other than none. This includes effort inherited from WithReasoningEffort or WithChainReasoningEffort, even when the call supplies no effort. Reviewed exact native GPT-6 Astra/Sol/Luna/ GPT-6.1 Sol and routed GPT-6 Luna/GPT-6.1 Sol also scale when effort is absent, without sending a wire effort. Unknown defaults (including Azure), explicit reasoning omission, none and native-extra-only reasoning remain exact.

func WithReasoningTokenMultiplier(multiplier int) AIOption
func WithChainReasoningTokenMultiplier(multiplier int) ChainOption  // For chain clients
func ResolveReasoningTokenMultiplier(value int) (int, error)         // Provider authors

Default: 3x on scaling profiles, including supported OpenRouter effort. Explicit 1x keeps the configured cap exact. Code zero is unset; otherwise code overrides TRUVAG3_AI_REASONING_TOKEN_MULTIPLIER, then the default. Invalid explicit environment values (including empty/zero) and negative code values fail construction. An explicit 5x turns 2000 requested tokens into 10000. ResolveReasoningTokenMultiplier exposes the same resolution for provider factories: positive code value > environment > 3; zero code means unset. It validates a present environment value even when code overrides it, returning an error wrapping core.ErrInvalidConfiguration for invalid settings. It resolves configuration only; model-specific scaling and capacity checks happen later. Active scaling includes the resolved multiplier in the policy/cache identity, including preflight requests without a token count. Inferred defaults also carry model-default-reasoning-v1 and a model-default-reasoning-headroom-v1 report adjustment when scaling occurs. Unknown-default effort, none, explicit reasoning omission and non-scaling profiles do not gain that identity. The OpenAI-compatible adapter bounds reviewed exact native GPT-5.6 Sol, GPT-6 Astra/Sol/Luna/GPT-6.1 Sol and routed GPT-5.6 Luna/GPT-6 Luna/GPT-6.1 Sol IDs to 128000. GPT-5.6 entries add capacity only. For a valid base, headroom is bounded in both modes (50000 × 3 → 128000), with a wire-profile cap adjustment. For an over-cap caller base, Compatible reports a built-in clamp; Strict rejects an unacknowledged explicit change. Final validation still checks patches. Unknown models/variants have no inferred ceiling. Remaining input-context capacity is not calculated, and request-specific routing can still reject a budget. Reasoning visibility does not change the allowance or create retries.

Example:

// Opt into additional reasoning/output headroom
client, _ := ai.NewClient(
    ai.WithReasoningTokenMultiplier(3),  // 3x multiplier
)

// Higher multiplier for complex reasoning tasks
client, _ := ai.NewClient(
    ai.WithReasoningTokenMultiplier(8),  // 8x multiplier
)

// Chain client with custom multiplier
chainClient, _ := ai.NewChainClient(
    ai.WithProviderChain("openai", "anthropic"),
    ai.WithChainReasoningTokenMultiplier(6),
)

Note: The multiplier affects requests classified into an OpenAI reasoning-family wire profile, including stock OpenAI and Azure v1. Ordinary-model profiles are unaffected, and Azure classic rejects reasoning families because this release does not claim a verified classic reasoning contract.

Other Configuration Options​

OptionTypeDefaultDescription
WithProvider(name)stringauto-detectAI provider name
WithProviderAlias(alias)string-Provider alias (e.g., "openai.groq")
WithModel(model)stringprovider defaultConcrete model ID, or an alias when the selected provider defines a model catalog
WithAPIKey(key)stringfrom envOverride API key
WithBaseURL(url)stringprovider defaultCustom API endpoint
WithRegion(region)stringus-east-1 for BedrockBedrock AWS region
WithAWSCredentials(access, secret, session)three string valuesAWS SDK credential chainExplicit Bedrock credentials
WithTemperature(t)float320.7Sampling temperature; permitted range is provider/model-specific
WithMaxTokens(n)int1000Default max tokens (per ai/providers/base.go)
WithTimeout(d)time.Duration300s; explicit standalone Bedrock 60mProvider request timeout
WithMaxRetries(n)int3In-provider retries; explicit 0 disables
WithReasoningContentFallback(enabled)boolfalseCompatible-codec legacy reasoning-as-answer opt-in; separate reasoning and provenance retained
WithChainReasoningContentFallback(enabled)boolfalseSame opt-in for compatible entries constructed by legacy NewChainClient; a ChainOption, not an AIOption
WithCircuitBreaker(breaker)core.CircuitBreakernoneProtect each logical provider call; configure it with ShouldCountAICircuitBreakerFailure semantics
WithReasoningTokenMultiplier(n)int3 (environment override supported)Active scaling-profile multiplier; explicit 1 keeps the cap exact
WithReasoningEffort(effort)stringmodel defaultOpenAI-compatible reasoning effort, model dependent
WithHeaders(headers)map[string]stringnoneSnapshot custom eligible request headers
WithExtra(key, value)string, interface{}noneProvider-specific legacy configuration
WithLogger(l)core.LoggernilLogger for AI operations
WithTelemetry(t)core.TelemetrynilTelemetry for distributed tracing

Azure requires an endpoint resolver, rejects WithBaseURL, and accepts either WithAPIKey or a credential source. Vertex requires endpoint and credential sources and rejects both direct WithAPIKey and WithBaseURL. Request-aware Bedrock accepts the region and AWS-credential options but rejects HTTP headers, HTTP-client injection, and HTTP credential sources.

GenerateResponse​

Generate AI responses with optional parameters for fine-tuning behavior.

// On the core.AIClient interface:
GenerateResponse(ctx context.Context, prompt string, options *AIOptions) (*AIResponse, error)

// Same shape on the concrete *ChainClient (failover wrapper):
func (c *ChainClient) GenerateResponse(ctx context.Context, prompt string, options *core.AIOptions) (*core.AIResponse, error)

StreamResponse​

Generate AI responses with real-time streaming. Tokens are delivered as they're generated by the AI provider, enabling lower time-to-first-token and real-time UX.

// Streaming lives on the core.StreamingAIClient interface (which embeds AIClient).
// All built-in providers implement it; type-assert with `client.(core.StreamingAIClient)`
// when you need to call StreamResponse on a value typed as core.AIClient.
StreamResponse(ctx context.Context, prompt string, options *AIOptions, callback StreamCallback) (*AIResponse, error)

// Same shape on the concrete *ChainClient:
func (c *ChainClient) StreamResponse(ctx context.Context, prompt string, options *core.AIOptions, callback core.StreamCallback) (*core.AIResponse, error)

StreamCallback type:

// StreamCallback is called for each chunk of the streaming response
type StreamCallback func(chunk StreamChunk) error

// StreamChunk represents a single chunk of streaming output
type StreamChunk struct {
    Content          string                 `json:"content,omitempty"`
    ReasoningContent string                 `json:"reasoning_content,omitempty"`
    ContentSource    core.AIContentSource    `json:"content_source,omitempty"`
    Delta            bool                   `json:"delta"`
    Index            int                    `json:"index"`
    FinishReason     string                 `json:"finish_reason,omitempty"`
    Model            string                 `json:"model,omitempty"`
    Usage            *core.TokenUsage       `json:"usage,omitempty"`
    Metadata         map[string]interface{} `json:"metadata,omitempty"`
}

Streaming errors are surfaced via the error returned from StreamResponse, not through a field on StreamChunk. The end of the stream is signalled by a final chunk with Delta: false; FinishReason and Usage are populated when reported by the provider. Empty completion facts remain unknown. Delivery of either answer or reasoning content prevents transparent stream replay.

Example - Basic Streaming:

client, _ := ai.NewClient()

// NewClient returns core.AIClient; type-assert to StreamingAIClient for streaming.
streaming, ok := client.(core.StreamingAIClient)
if !ok {
    log.Fatal("provider does not support streaming")
}

resp, err := streaming.StreamResponse(ctx, "Explain quantum computing", nil,
    func(chunk core.StreamChunk) error {
        // Print each token as it arrives
        fmt.Print(chunk.Content)

        // Final chunk has Delta=false and a populated FinishReason
        if !chunk.Delta && chunk.FinishReason != "" {
            fmt.Println("\n--- Complete ---")
            if chunk.Usage != nil {
                fmt.Printf("Tokens: %d\n", chunk.Usage.TotalTokens)
            }
        }

        return nil
    },
)
if err != nil {
    log.Printf("stream failed: %v", err)
}
_ = resp // resp.Content holds the accumulated text; resp.Usage holds totals.

Example - Streaming with Chain Client (Failover):

chain, _ := ai.NewChainClient(
    ai.WithProviderChain("openai", "anthropic", "openai.groq"),
)

// ChainClient exposes StreamResponse directly — no type assertion needed.
_, err := chain.StreamResponse(ctx, prompt, nil, func(chunk core.StreamChunk) error {
    sendToUI(chunk.Content)
    return nil
})
// If OpenAI fails, automatically tries Anthropic, then Groq

Example - Canceling a Stream:

ctx, cancel := context.WithCancel(context.Background())

go func() {
    time.Sleep(5 * time.Second)
    cancel() // Stop streaming after 5 seconds
}()

_, err := streaming.StreamResponse(ctx, prompt, nil, callback)
if errors.Is(err, context.Canceled) {
    fmt.Println("Stream was canceled")
}

Provider streaming support:

ProviderStreamingNotes
OpenAI✅ FullNative streaming
Anthropic✅ FullNative streaming
Gemini✅ FullNative streaming
Azure OpenAI v1✅ FullRequest-aware OpenAI-compatible SSE
Azure OpenAI classic✅ FullRequest-aware deployment-scoped SSE
Vertex-hosted Claude✅ FullRequest-aware Vertex streamRawPredict SSE
Google Cloud OpenAI compatibility✅ FullOpenAI-compatible SSE through the configured Google route
Bedrock✅ FullNative streaming
Groq✅ FullOpenAI-compatible
DeepSeek✅ FullOpenAI-compatible
xAI✅ FullOpenAI-compatible
Mistral✅ FullOpenAI-compatible
Qwen✅ FullOpenAI-compatible
Together✅ FullOpenAI-compatible
Ollama✅ FullOpenAI-compatible

For a complete production example with SSE streaming, session management, and conversation history, see the Chat Agent Implementation Guide.

AIOptions parameters:

  • Temperature - Sampling temperature; accepted range and whether the field is allowed are provider/model-specific
  • MaxTokens - Maximum response length
  • TopP - Nucleus sampling (alternative to temperature)
  • Model - Override default model

Example - Different Use Cases:

// Technical analysis (low temperature for accuracy)
codeReview, _ := client.GenerateResponse(ctx,
    "Review this Go code for bugs",
    &core.AIOptions{
        Temperature: 0.2,
        MaxTokens:   1500,
    },
)

// Creative writing (high temperature for variety)
story, _ := client.GenerateResponse(ctx,
    "Write a short story about AI",
    &core.AIOptions{
        Temperature: 0.9,
        MaxTokens:   2000,
    },
)

// Structured data extraction (deterministic)
data, _ := client.GenerateResponse(ctx,
    "Extract JSON data from this text",
    &core.AIOptions{
        Temperature: 0,  // Deterministic
        MaxTokens:   500,
    },
)

NewAIAgent​

Create an intelligent agent with both AI capabilities and service discovery powers. Perfect for building orchestrators and assistants.

func NewAIAgent(name string, apiKey string) (*AIAgent, error)

AIAgent capabilities:

  • Full agent powers (discovery, orchestration)
  • Built-in AI for natural language processing
  • Conversation memory management
  • Tool use and function calling
  • Autonomous decision making

Example - Intelligent Orchestrator:

// Create an AI-powered orchestrator
agent, err := ai.NewAIAgent("ai-orchestrator", os.Getenv("OPENAI_API_KEY"))
if err != nil {
    log.Fatal(err)
}

// It can process natural language
response, _ := agent.GenerateResponse(ctx,
    "Find the weather service and get weather for NYC",
    nil,
)

// It remembers conversations
agent.ProcessWithMemory(ctx, "My name is John")
agent.ProcessWithMemory(ctx, "What's my name?") // "Your name is John"

// It can discover and use other services
tools, _ := agent.Discover(ctx, core.DiscoveryFilter{
    Capabilities: []string{"weather", "news"},
})

// Orchestrate multiple tools based on user request
result := agent.ProcessRequest(ctx,
    "Get weather and news for NYC and summarize both",
    tools,
)

NewAITool​

Create an AI-powered tool that exposes AI capabilities as a service. Unlike AIAgent, tools are passive and cannot discover other services.

func NewAITool(name string, apiKey string, opts ...AIToolOption) (*AITool, error)

// Options
func WithAIToolLogger(logger core.Logger) AIToolOption

// Capabilities are registered after construction. Each call mounts an HTTP
// endpoint at /ai/<name> that runs `prompt` against the tool's AI client and
// returns the response body.
func (t *AITool) RegisterAICapability(name, description, prompt string)

When to use AITool vs AIAgent:

FeatureAIToolAIAgent
PurposeProvide AI serviceOrchestrate services
DiscoveryCan be discoveredCan discover & be discovered
MemoryStatelessStateful conversations
Use caseTranslation, summarizationOrchestrators, assistants

Example - AI Microservices:

// Create a specialized AI tool, then attach one or more AI-backed capabilities
translator, _ := ai.NewAITool("translator-service", apiKey)
translator.RegisterAICapability(
    "translate",
    "Translate text between languages",
    "You are a translation assistant. Translate the user's input faithfully.",
)

summarizer, _ := ai.NewAITool("summarizer-service", apiKey)
summarizer.RegisterAICapability(
    "summarize",
    "Summarize long-form text",
    "Summarize the user's input in 3 sentences or fewer.",
)

// Each runs as an independent microservice
go translator.Start(ctx, 8081)
go summarizer.Start(ctx, 8082)

// Agents can discover and use these tools
// POST http://localhost:8081/ai/translate
// (request body is forwarded to the LLM as input)

AI Provider Support​

TruvaG3 supports both environment-detected providers and explicitly configured hosted-enterprise profiles behind the unified core.AIClient API. Concrete model lists evolve faster than this document; the canonical portable mappings live in ai/providers/openai/models.go, ai/providers/anthropic/models.go, and ai/providers/gemini/models.go. Every row assumes its provider package has been linked as described at the start of this module.

Provider or surfaceAlias / selectionSupported constructionAuto-detected
OpenAIopenaiNewClient, NewRequestClientYes
Anthropic directanthropicNewClient, NewRequestClientYes
Google GeminigeminiNewClient, NewRequestClientYes
Azure OpenAI v1azureopenai.v1NewRequestClient with route and credentialsNo
Azure OpenAI classicazureopenai.classicNewRequestClient with route and credentialsNo
Vertex-hosted Claudeanthropic.vertexNewRequestClient with route and credentialsNo
Google Cloud OpenAI compatibilityopenai plus Google route/credentialsNewRequestClientNo distinct profile
AWS Bedrockai.WithProvider("bedrock")NewClient, NewRequestClient; requires -tags bedrockYes, when compiled in
Groqopenai.groqNewClient, NewRequestClientYes
DeepSeekopenai.deepseekNewClient, NewRequestClientYes
xAI Grokopenai.xaiNewClient, NewRequestClientYes
Mistralopenai.mistralNewClient, NewRequestClientYes
Qwenopenai.qwenNewClient, NewRequestClientYes
Together AIopenai.togetherNewClient, NewRequestClientYes
OpenRouteropenai.openrouterNewClient, NewRequestClientYes
Ollamaopenai.ollamaNewClient, NewRequestClientOnly when OLLAMA_BASE_URL is set

Azure and Vertex model-selection environment overrides affect the semantic model only. Azure uses TRUVAG3_OPENAI_MODEL_*; Vertex-hosted Claude uses TRUVAG3_ANTHROPIC_MODEL_*. Resolver maps are keyed by the post-alias semantic model, while deployment and publisher-model identifiers remain route-owned and are never rewritten by those variables.

For Bedrock, an invocation outside us-east-1 that would use the implicit bare Sonnet 5 default fails during request preparation, after per-request model selection. Explicit client/request models and SDK-destination resolvers remain application-owned. A Bedrock resolver must preserve semantic equivalence between EndpointRequest.ResolvedModel and its opaque deployment because sampling policy follows the semantic model.

Build-Tagged Bedrock Helpers​

The following package API is present only in binaries built with -tags bedrock:

// Package github.com/truvaagents/truva-g3/ai/providers/bedrock
const ModelClaudeSonnet5 = "anthropic.claude-sonnet-5"
const ModelTitanEmbedV2 = "amazon.titan-embed-text-v2:0"
const ModelTitanEmbedV1 = "amazon.titan-embed-text-v1"

func CreateAWSConfig(
    ctx context.Context,
    region string,
    credentials ...aws.CredentialsProvider,
) (aws.Config, error)

func NewClient(cfg aws.Config, region string, logger core.Logger) *Client
func (client *Client) SetDefaultModel(model string) error

func NewDraft(
    resolvedModel string,
    request *core.AIRequest,
) (*Draft, error)
func NewStreamDraft(
    resolvedModel string,
    request *core.AIRequest,
) (*Draft, error)

type Draft struct { /* unexported Converse state */ }

func (draft *Draft) SetHeader(name, value string) error
func (draft *Draft) RemoveHeader(name string) error
func (draft *Draft) Header(name string) (string, bool)
func (draft *Draft) HasExplicitIntent(path string) bool
func (draft *Draft) PolicyFingerprintIdentity() string
func (draft *Draft) Adjustments() []core.AIRequestAdjustment
func (draft *Draft) Validate() error
func (draft *Draft) SDKInput() (*bedrockruntime.ConverseInput, error)
func (draft *Draft) SDKStreamInput() (*bedrockruntime.ConverseStreamInput, error)

func (client *Client) InvokeModel(
    ctx context.Context,
    modelID string,
    body []byte,
) ([]byte, error)

func (client *Client) GetEmbeddings(
    ctx context.Context,
    text string,
    options ...EmbeddingOption,
) ([]float32, error)

func WithEmbeddingModel(model string) EmbeddingOption
func WithEmbeddingDimensions(dimensions int) EmbeddingOption
func WithEmbeddingNormalization(normalize bool) EmbeddingOption
func WithoutEmbeddingNormalization() EmbeddingOption

GetEmbeddings is a Titan-shaped, single-text helper with a V2 default, not an implementation of the batch core.EmbeddingClient. EmbeddingOption has an unexported configuration parameter and is intentionally constructed only through the four helpers above. Dimensions are 256, 512, or 1024; zero omits the field and uses AWS's V2 default. ModelTitanEmbedV1 is an explicit migration pin for existing 1536-dimensional V1 stores. A per-call V1 pin discards V2-only dimensions and normalization inherited from client configuration; those controls remain invalid when a non-zero dimension or normalization is explicitly supplied on the V1 call. Zero dimensions means omission. WithoutEmbeddingNormalization omits an inherited V2 normalization field for one call. Recognizable V1 variants and foundation-model ARNs use the same V1 payload validation. Embedding observations expose only the bounded Titan V1/V2 semantic family, never the selected SDK model ID or ARN.

SetDefaultModel is the error-returning construction-time API for a direct package client to declare explicit model routing. Call it before concurrent use; assigning the embedded DefaultModel field does not record explicit intent for the implicit Sonnet 5 region guard.

Generation uses Converse/ConverseStream; an optional request-aware resolver selects only the opaque SDK modelId in Deployment. Current-Claude sampling validation covers both inferenceConfig and case-insensitive sampling keys in additionalModelRequestFields. For Fable, unique common sampling fields from legacy Extra remain policy-editable in the additional container; policy may remove them or set the canonical common field and remove the legacy copy. Duplicates and unremediated post-policy placement fail locally. JSON UseNumber, named, and unsigned numeric values remain numeric in common and nested additional fields. See AWS Bedrock SDK-Native Routing.

Example - Multi-Provider Strategy:

// Use different providers for different tasks
type AIService struct {
    reasoning  core.AIClient  // Claude for complex reasoning
    creative   core.AIClient  // OpenAI for creative tasks
    fast       core.AIClient  // Groq for quick responses
}

func NewAIService() *AIService {
    return &AIService{
        reasoning: ai.MustNewClient(
            ai.WithProvider("anthropic"),
            ai.WithModel("smart"),
        ),
        creative: ai.MustNewClient(
            ai.WithProvider("openai"),
            ai.WithModel("smart"),
        ),
        fast: ai.MustNewClient(
            ai.WithProviderAlias("openai.groq"),
            ai.WithModel("fast"), // resolves to llama-3.1-8b-instant; see ai/providers/openai/models.go
        ),
    }
}

func (s *AIService) AnalyzeCode(ctx context.Context, code string) (string, error) {
    // Use Claude for code analysis
    resp, err := s.reasoning.GenerateResponse(ctx,
        fmt.Sprintf("Analyze this code:\n%s", code),
        &core.AIOptions{Temperature: 0.3},
    )
    if err != nil {
        return "", err
    }
    return resp.Content, nil
}

func (s *AIService) GenerateDocumentation(ctx context.Context, code string) (string, error) {
    // Use GPT-4 for documentation
    resp, err := s.creative.GenerateResponse(ctx,
        fmt.Sprintf("Generate comprehensive docs for:\n%s", code),
        &core.AIOptions{Temperature: 0.7},
    )
    if err != nil {
        return "", err
    }
    return resp.Content, nil
}

InstrumentedAIClient​

Decorator that wraps any core.AIClient, emits the common logical ai.generate / ai.stream spans, and optionally records every LLM call to a telemetry.LLMCallRecorder for debugging. NewClient and NewRequestClient always install this wrapper so logical telemetry is provider-independent. Consequently, code must depend on capability interfaces rather than asserting a constructor result to a concrete provider-client type.

func NewInstrumentedClient(
    client core.AIClient,
    recorder telemetry.LLMCallRecorder,
    opts ...InstrumentedOption,
) *InstrumentedAIClient

Options:

OptionDescription
WithComponentName(name)Source component name for recordings (e.g., "research-assistant")
WithDefaultCallType(type)Override default call type (default: "agent_llm_call")
WithInstrumentedLogger(logger)Logger for recording failure warnings
WithInstrumentedTelemetry(provider)Telemetry provider for logical ai.generate / ai.stream spans

Key behaviors:

  • Async recording — LLM calls are never blocked by debug recording
  • Nil-safe — Defaults to NoOpLLMCallRecorder if recorder is nil
  • Request ID resolution — Tries OTel baggage first, then core.GetRequestID(ctx)
  • Graceful shutdown — Shutdown(ctx) drains in-flight recordings via sync.WaitGroup
  • Factory wrapper collapse — wrapping a NewClient or NewRequestClient result for debug recording replaces its internal no-op-recorder layer, avoiding duplicate logical spans

Example — Agent Setup:

// Create base AI client
baseClient, _ := ai.NewClient(ai.WithProvider("anthropic"))

// Create LLM call recorder
recorder, _ := telemetry.NewRedisLLMCallRecorder()

// Wrap with instrumentation
aiClient := ai.NewInstrumentedClient(baseClient, recorder,
    ai.WithComponentName("research-assistant"),
)
defer aiClient.Shutdown(context.Background())

// Use as normal — all LLM calls are recorded automatically
response, err := aiClient.GenerateResponse(ctx, prompt, options)

Resilience Module​

Build fault-tolerant systems with circuit breakers and intelligent retry mechanisms.

Circuit Breakers​

Protect your application from cascading failures by automatically stopping calls to failing services.

NewCircuitBreaker​

Create a production-ready circuit breaker with comprehensive configuration.

func NewCircuitBreaker(config *CircuitBreakerConfig) (*CircuitBreaker, error)
func NewCircuitBreakerLegacy(failureThreshold int, recoveryTimeout time.Duration) *CircuitBreaker

How it works:

  • Closed State: Normal operation, requests pass through
  • Open State: Service is down, requests fail immediately
  • Half-Open State: Testing recovery with limited requests

Configuration parameters:

  • ErrorThreshold - Percentage of failures to open (0.0-1.0)
  • VolumeThreshold - Minimum requests before evaluation
  • SleepWindow - How long to wait before testing recovery
  • HalfOpenRequests - Test requests in half-open state
  • SuccessThreshold - Success rate to close again

Example - Production Configuration:

// Sophisticated circuit breaker for production
breaker, err := resilience.NewCircuitBreaker(&resilience.CircuitBreakerConfig{
    Name:             "payment-api",
    ErrorThreshold:   0.5,              // Open at 50% error rate
    VolumeThreshold:  10,                // Need 10+ requests to evaluate
    SleepWindow:      30 * time.Second, // Wait 30s before recovery test
    HalfOpenRequests: 5,                 // Test with 5 requests
    SuccessThreshold: 0.6,               // Need 60% success to recover

    // Smart error classification
    ErrorClassifier: func(err error) bool {
        // Only infrastructure errors trip the breaker
        if errors.Is(err, context.Canceled) {
            return false  // User cancelled, don't count
        }
        if httpErr, ok := err.(*HTTPError); ok {
            return httpErr.StatusCode >= 500  // Only server errors
        }
        return true  // Network/timeout errors count
    },
})

Execute Methods​

Execute functions with circuit breaker protection.

func (cb *CircuitBreaker) Execute(ctx context.Context, fn func() error) error
func (cb *CircuitBreaker) ExecuteWithTimeout(ctx context.Context, timeout time.Duration, fn func() error) error

What Execute does:

  1. Checks circuit state (fails fast if open)
  2. Executes your function
  3. Records success/failure
  4. Updates circuit state
  5. Handles panics gracefully

Example - API Call Protection:

func CallPaymentAPI(order Order) error {
    return paymentBreaker.Execute(ctx, func() error {
        // This is protected by the circuit breaker
        resp, err := http.Post(paymentURL, "application/json", order)
        if err != nil {
            return err
        }

        if resp.StatusCode >= 500 {
            return fmt.Errorf("server error: %d", resp.StatusCode)
        }

        return nil
    })
}

// Handle circuit breaker states
err := CallPaymentAPI(order)
if errors.Is(err, core.ErrCircuitBreakerOpen) {
    // Service is down, use fallback
    log.Warn("Payment service unavailable, using queue")
    return queuePaymentForLater(order)
}

Monitoring and Control​

Monitor circuit breaker state and metrics for operational visibility.

// Get current state
state := breaker.GetState()  // "closed", "open", or "half-open"

// Get detailed metrics
metrics := breaker.GetMetrics()
// Returns: state, success, failure, error_rate, total_executions, rejected

// Manual control for maintenance
breaker.ForceOpen()    // Block all requests
breaker.ForceClosed()  // Allow all requests
breaker.ClearForce()   // Resume automatic operation

// State change notifications
breaker.AddStateChangeListener(func(name string, from, to CircuitState) {
    if to == resilience.StateOpen {
        alert.Send("Circuit breaker %s opened!", name)
    }
})

Retry Mechanisms​

Automatically retry failed operations with configurable backoff strategies.

Retry Function​

Simple retry with exponential backoff.

func Retry(ctx context.Context, config *RetryConfig, fn func() error) error

RetryConfig parameters:

  • MaxAttempts - Maximum retry attempts
  • InitialDelay - First retry delay
  • BackoffFactor - Delay multiplier (2.0 = double each time)
  • MaxDelay - Maximum delay between retries
  • JitterEnabled - Add randomness to prevent thundering herd

Example:

config := &resilience.RetryConfig{
    MaxAttempts:   5,
    InitialDelay:  100 * time.Millisecond,
    BackoffFactor: 2.0,      // 100ms, 200ms, 400ms, 800ms, 1600ms
    MaxDelay:      5 * time.Second,
    JitterEnabled: true,      // Prevent synchronized retries
}

err := resilience.Retry(ctx, config, func() error {
    return callFlakyService()
})

RetryExecutor​

Production-ready retry with logging and telemetry.

func NewRetryExecutor(config *RetryConfig) *RetryExecutor
func (r *RetryExecutor) Execute(ctx context.Context, operation string, fn func() error) error

Why use RetryExecutor:

  • Named operations in logs
  • Detailed retry logging
  • Telemetry integration
  • Success/failure metrics

Example - Database Operations:

executor := resilience.NewRetryExecutor(&resilience.RetryConfig{
    MaxAttempts:   3,
    InitialDelay:  50 * time.Millisecond,
    BackoffFactor: 2.0,
    JitterEnabled: true,
})
executor.SetLogger(logger)

// Named operation appears in logs
err := executor.Execute(ctx, "fetch-user-profile", func() error {
    return db.QueryRow("SELECT * FROM users WHERE id = ?", userID).Scan(&user)
})

// Logs show:
// INFO: Starting retry operation [fetch-user-profile]
// DEBUG: Attempt 1 failed, retrying...
// INFO: Operation succeeded on attempt 2

RetryWithCircuitBreaker​

Combine retry and circuit breaker for maximum resilience.

func RetryWithCircuitBreaker(ctx context.Context, config *RetryConfig, cb *CircuitBreaker, fn func() error) error

Best practice pattern:

type ResilientClient struct {
    breaker *resilience.CircuitBreaker
    retry   *resilience.RetryExecutor
}

func (c *ResilientClient) Call(ctx context.Context, request Request) (Response, error) {
    var response Response

    // Retry handles transient failures
    // Circuit breaker prevents cascading failures
    err := c.breaker.Execute(ctx, func() error {
        return c.retry.Execute(ctx, "api-call", func() error {
            return c.makeHTTPCall(request, &response)
        })
    })

    return response, err
}

Telemetry Module​

Comprehensive observability with metrics, distributed tracing, and context propagation.

Basic Metrics​

Simple functions for common metrics without boilerplate.

func Counter(name string, labels ...string)
func Histogram(name string, value float64, labels ...string)
func Gauge(name string, value float64, labels ...string)
func Duration(name string, startTime time.Time, labels ...string)

Example - Request Metrics:

// Count requests
telemetry.Counter("api.requests",
    "method", r.Method,
    "endpoint", r.URL.Path,
    "status", strconv.Itoa(status),
)

// Track latency
start := time.Now()
processRequest()
telemetry.Duration("api.latency", start,
    "endpoint", r.URL.Path,
)

// Monitor concurrent requests
telemetry.Gauge("api.concurrent_requests", float64(active))

// Track response sizes
telemetry.Histogram("api.response_bytes", float64(len(response)),
    "endpoint", r.URL.Path,
)

Context-Aware Telemetry​

Advanced telemetry that automatically correlates metrics with distributed traces.

func EmitWithContext(ctx context.Context, name string, value float64, labels ...string)

// Baggage is `type Baggage map[string]string`.
func GetBaggage(ctx context.Context) Baggage

// WithBaggage attaches W3C baggage labels to the context. Pass flat
// key-value pairs (variadic), not a map.
func WithBaggage(ctx context.Context, labels ...string) context.Context

// WithBaggageExact adds/replaces one untruncated member or returns a bounded
// typed rejection while leaving ctx unchanged.
func WithBaggageExact(
    ctx context.Context,
    key string,
    rawValue string,
    options ...BaggageMemberOption,
) (context.Context, error)

func WithMetricLabelEligibility(eligible bool) BaggageMemberOption
func BaggageExactErrorReasonOf(err error) (BaggageExactErrorReason, bool)
func WithoutBaggageMember(ctx context.Context, key string) context.Context
func CopyBaggage(dst, src context.Context) context.Context

Both baggage construction paths enforce 64 members, 128-byte keys, 512-byte values, and an 8192-byte complete serialized W3C baggage value (including separators, encoding, and member properties). WithBaggage keeps its truncation/silent-drop convenience behavior. WithBaggageExact never truncates. WithMetricLabelEligibility(false) preserves the member for propagation, traces, and structured logs while excluding it from automatic metric labels.

Span creation is not a top-level function. Spans are started through a core.Telemetry instance — typically the one stored on your component (agent.Telemetry.StartSpan(ctx, name)). The framework wires this for you when you enable telemetry; bypassing it is rarely needed.

Why use context-aware telemetry:

  • Automatic trace correlation
  • Request metadata propagation
  • Cross-service tracking
  • Debugging complex flows

Example - Distributed Request Tracking:

// API Gateway
func (gw *Gateway) Handle(ctx context.Context, req Request) error {
    // Start a span using the component's Telemetry instance, then attach
    // baggage that downstream services and metrics will see.
    ctx, span := gw.Telemetry.StartSpan(ctx, "gateway.handle")
    defer span.End()

    ctx = telemetry.WithBaggage(ctx,
        "request_id", req.ID,
        "user_id", req.UserID,
    )

    // Metrics include trace context
    telemetry.EmitWithContext(ctx, "gateway.requests", 1,
        "endpoint", req.Path,
    )

    // Call downstream service (context propagates)
    result, err := gw.authService.Authenticate(ctx, req.UserID)

    return err
}

// Auth Service (automatically gets context)
func (auth *AuthService) Authenticate(ctx context.Context, userID string) error {
    // Access propagated metadata
    baggage := telemetry.GetBaggage(ctx)
    requestID := baggage["request_id"]  // From gateway!

    // Metrics correlated to same trace
    telemetry.EmitWithContext(ctx, "auth.attempts", 1,
        "user_id", userID,
        "request_id", requestID,
    )

    return nil
}

Type-Specific Helpers​

Semantic helper functions for common metric types.

func RecordError(name string, errorType string, labels ...string)
func RecordSuccess(name string, labels ...string)
func RecordLatency(name string, milliseconds float64, labels ...string)
func RecordBytes(name string, bytes int64, labels ...string)

Example:

start := time.Now()
err := processOrder(order)

if err != nil {
    telemetry.RecordError("order.processing", err.Error(),
        "order_id", order.ID,
    )
} else {
    telemetry.RecordSuccess("order.processing",
        "order_id", order.ID,
    )
}

telemetry.RecordLatency("order.processing",
    float64(time.Since(start).Milliseconds()),
    "order_id", order.ID,
)

Unified Metrics API​

Cross-module metrics that enable consistent observability across agents and orchestration. These helpers emit standardized metrics with a module label, enabling unified Grafana dashboards regardless of which TruvaG3 module you use.

// Module constants
const (
    ModuleAgent         = "agent"
    ModuleOrchestration = "orchestration"
    ModuleCore          = "core"
)

// Request metrics
func RecordRequest(module, operation string, durationMs float64, status string)
func RecordRequestError(module, operation, errorType string)

// Tool/capability call metrics
func RecordToolCall(module, toolName string, durationMs float64, status string)
func RecordToolCallError(module, toolName, errorType string)
func RecordToolCallRetry(module, toolName string)

// AI provider metrics
func RecordAIRequest(module, provider string, durationMs float64, status string)
func RecordAITokens(module, provider, tokenType string, count int64)

Why use unified metrics:

  • Single Grafana dashboard works for both agent and orchestration examples
  • Consistent metric names across all TruvaG3 modules
  • Easy to compare performance between different module implementations
  • Prometheus queries work regardless of which module emits the data

Example - Agent Request Handler:

func handleResearchRequest(w http.ResponseWriter, r *http.Request) {
    startTime := time.Now()
    var requestStatus = "success"

    defer func() {
        durationMs := float64(time.Since(startTime).Milliseconds())
        // Unified metric - works in cross-module dashboards
        telemetry.RecordRequest(telemetry.ModuleAgent, "research", durationMs, requestStatus)
    }()

    // Parse request
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        requestStatus = "error"
        telemetry.RecordRequestError(telemetry.ModuleAgent, "research", "validation")
        http.Error(w, "Invalid request", http.StatusBadRequest)
        return
    }

    // Process...
}

Example - Orchestration with AI:

func executeWithAI(ctx context.Context, prompt string) (string, error) {
    startTime := time.Now()

    response, err := aiClient.GenerateResponse(ctx, prompt, nil)

    // Record AI metrics
    status := "success"
    if err != nil {
        status = "error"
    }
    telemetry.RecordAIRequest(telemetry.ModuleOrchestration, "openai",
        float64(time.Since(startTime).Milliseconds()), status)

    if err != nil {
        return "", err
    }
    return response.Content, nil
}

Prometheus Queries for Unified Dashboards:

# Request rate across all modules
sum(rate(request_total[5m])) by (module, operation)

# P95 latency by module
histogram_quantile(0.95, sum(rate(request_duration_ms_bucket[5m])) by (le, module))

# Error rate comparison: agent vs orchestration
sum(rate(request_errors[5m])) by (module, error_type)

# AI request latency by provider
histogram_quantile(0.95, sum(rate(ai_request_duration_ms_bucket[5m])) by (le, provider))

LLM Call Recording​

Interface and implementation for recording LLM calls to Redis for debugging. Used by ai.InstrumentedAIClient to capture every LLM call an agent makes.

LLMCallRecorder Interface​

type LLMCallRecorder interface {
    // RecordLLMCall appends an LLM call record to the debug trace for a request.
    // If requestID is empty, implementations silently skip recording.
    RecordLLMCall(ctx context.Context, requestID string, record LLMCallRecord) error
}

LLMCallRecord​

type LLMCallRecord struct {
    CallType         string               `json:"call_type"`
    SourceComponent  string               `json:"source_component,omitempty"`
    Description      string               `json:"description,omitempty"`
    StepID           string               `json:"step_id,omitempty"`
    PhaseNumber      int                  `json:"phase_number,omitempty"`
    Timestamp        time.Time            `json:"timestamp"`
    DurationMs       int64                `json:"duration_ms"`
    Prompt           string               `json:"prompt"`
    SystemPrompt     string               `json:"system_prompt,omitempty"`
    Temperature      float64              `json:"temperature"`
    MaxTokens        int                  `json:"max_tokens"`
    Model            string               `json:"model,omitempty"`
    Provider         string               `json:"provider,omitempty"`
    Response         string               `json:"response"`
    ReasoningContent string               `json:"reasoning_content,omitempty"`
    FinishReason     string               `json:"finish_reason,omitempty"`
    ContentSource    core.AIContentSource `json:"content_source,omitempty"`
    PromptTokens     int                  `json:"prompt_tokens"`
    CompletionTokens int                  `json:"completion_tokens"`
    TotalTokens      int                  `json:"total_tokens"`
    Success          bool                 `json:"success"`
    Error            string               `json:"error,omitempty"`
}

PhaseNumber identifies the producing phase when supplied; zero is omitted. The three completion facts describe that call's provider response, not domain success. Empty finish/provenance fields mean unreported. Error is an observation string sanitized by the producer; the recorder stores it verbatim.

NoOpLLMCallRecorder​

Safe default that discards all recordings. Used when debug recording is not enabled.

recorder := &telemetry.NoOpLLMCallRecorder{}

NewRedisLLMCallRecorder​

Write-only Redis/Valkey-backed recorder that writes to the versioned LLM-debug subspace of shared DB 0 in the same format as the orchestration module's RedisLLMDebugStore. Agents use this instead of importing the orchestration module directly.

func NewRedisLLMCallRecorder(opts ...RecorderOption) (*RedisLLMCallRecorder, error)

Options:

OptionDescription
WithRecorderRedisURL(url)Redis connection URL
WithRecorderKeyspace(keyspace)Validated deployment namespace in shared DB 0
WithRecorderLogger(logger)Logger for recorder operations
WithRecorderTTL(ttl)Base TTL for successful records (default: 72h / 3 days); an existing longer/persistent lifetime or retention floor wins
WithRecorderErrorTTL(ttl)Base TTL for error records (default: 72h / 3 days); an existing longer/persistent lifetime or retention floor wins

Built-in resilience: Layer 1 retry with exponential backoff (3 attempts, 100ms→2s) and failure cooldown (5 failures within 30s triggers cooldown).

The request-local Redis write atomically appends the interaction and updates metadata and minimum TTL. The cross-request recent index is a separate advisory write so the portable Redis/Valkey Cluster baseline never requires cross-slot atomicity. The authoritative write does not shorten a longer lifetime or make a persistent record expiring, so it can safely share a request record with orchestration.RedisLLMDebugStore after execution or lineage retention has been promoted. The built-in recorder preserves the prompt, system prompt, response, description, and error text it receives; it does not infer sensitive application fields or silently redact their values. Applications that require sanitization must supply it explicitly through their logger, telemetry pipeline, recorder, storage implementation, or middleware and must govern access, encryption, and retention for this opt-in debug data.

Example:

recorder, err := telemetry.NewRedisLLMCallRecorder(
    telemetry.WithRecorderRedisURL("redis://redis:6379"),
    telemetry.WithRecorderLogger(logger),
)
if err != nil {
    log.Printf("LLM recording disabled: %v", err)
    recorder = nil // InstrumentedClient falls back to NoOp
} else {
    defer func() {
        if closeErr := recorder.Close(); closeErr != nil {
            log.Printf("LLM recorder close failed: %v", closeErr)
        }
    }()
}

Distributed Tracing​

HTTP instrumentation for automatic trace context propagation across service boundaries.

Provider instance identity​

func NewOTelProvider(serviceName, serviceType, endpoint string, options ...OTelProviderOption) (*OTelProvider, error)
func WithServiceInstanceID(id string) OTelProviderOption

telemetry.Config.ServiceInstanceID string provides the same override for Initialize / InitializeForComponent; Config.WithOverrides copies a non-empty value. Existing three-argument NewOTelProvider calls use the defaults (the constructor's function type is now variadic).

Non-empty code configuration takes precedence over TRUVAG3_TELEMETRY_SERVICE_INSTANCE_ID, then a generated UUID v4. Empty code values mean unset. A selected empty environment value, invalid UTF-8, surrounding whitespace or control characters fails construction. Nil options are rejected. Identity is fixed for one provider lifetime and shared by trace and metric resources as service.instance.id. Explicit identities must be unique among concurrent emitters of the same service; collectors must retain them in metric-series identity. Initialize returns its first result on every call, including failure; it does not support reinitialization.

Initialize(Config) uses the supplied endpoint, defaulting an empty value to localhost:4318. It does not automatically consume core's endpoint/name configuration; application bootstrap passes those resolved values. The separate EnableTelemetry(agent, "") helper reads OTEL_EXPORTER_OTLP_ENDPOINT before its localhost fallback. The built-in provider explicitly uses OTLP/HTTP, AlwaysSample, and a constructor-owned resource. Profiles and generic OTEL_TRACES_SAMPLER / OTEL_RESOURCE_ATTRIBUTES variables do not override those choices. Config.Verbose and core.TelemetryConfig.SamplingRate do not exist; legacy PIIRedaction/PIIPatterns fields do not enable payload filtering.

TracingMiddleware​

Wrap HTTP handlers to automatically extract and propagate W3C TraceContext headers.

func TracingMiddleware(serviceName string) func(http.Handler) http.Handler
func TracingMiddlewareWithConfig(serviceName string, config *TracingMiddlewareConfig) func(http.Handler) http.Handler

What it does:

  • Extracts traceparent and tracestate headers from incoming requests
  • Creates a span for each HTTP request
  • Records HTTP metrics (status codes, latency)
  • Propagates trace context to handler code via context.Context

TracingMiddlewareConfig options:

  • ExcludedPaths - Paths to skip tracing (e.g., /health, /metrics)
  • SpanNameFormatter - Custom function to generate span names

Example - Basic Usage:

// Initialize telemetry FIRST
telemetry.Initialize(telemetry.Config{
    ServiceName: "my-service",
    Endpoint:    "http://otel-collector:4318",
})
defer telemetry.Shutdown(context.Background())

// Create handlers
mux := http.NewServeMux()
mux.HandleFunc("/api/users", handleUsers)

// Wrap with tracing middleware
tracedHandler := telemetry.TracingMiddleware("my-service")(mux)
http.ListenAndServe(":8080", tracedHandler)

Example - With Configuration:

config := &telemetry.TracingMiddlewareConfig{
    // Don't trace health checks
    ExcludedPaths: []string{"/health", "/metrics", "/ready"},

    // Custom span names
    SpanNameFormatter: func(op string, r *http.Request) string {
        return r.Method + " " + r.URL.Path
    },
}

tracedHandler := telemetry.TracingMiddlewareWithConfig("my-service", config)(mux)

NewTracedHTTPClient​

Create an HTTP client that automatically propagates trace context to downstream services.

func NewTracedHTTPClient(baseTransport http.RoundTripper) *http.Client
func NewTracedHTTPClientWithTransport(transport *http.Transport) *http.Client

What it does:

  • Injects traceparent and tracestate headers into outgoing requests
  • Creates child spans for each HTTP call
  • Enables distributed tracing across service boundaries

Example - Basic Usage:

// Create once, reuse for all requests (connection pooling)
client := telemetry.NewTracedHTTPClient(nil)

func callDownstreamService(ctx context.Context) error {
    // Context carries trace information
    req, _ := http.NewRequestWithContext(ctx, "GET", "http://other-service/api/data", nil)

    // Trace headers automatically injected
    resp, err := client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    // Process response...
    return nil
}

Example - With Custom Transport:

// Production configuration with connection pooling
transport := &http.Transport{
    MaxIdleConns:        100,
    MaxIdleConnsPerHost: 10,
    IdleConnTimeout:     90 * time.Second,
}

client := telemetry.NewTracedHTTPClientWithTransport(transport)

Best practices:

  • Create TracedHTTPClient once and reuse (connection pooling)
  • Always pass context.Context from incoming request to outgoing calls
  • Initialize telemetry before creating traced clients

End-to-End Tracing Example​

Complete example showing trace propagation across services:

// === Service A (API Gateway) ===
func main() {
    telemetry.Initialize(telemetry.Config{ServiceName: "api-gateway"})
    defer telemetry.Shutdown(context.Background())

    // Create traced client for calling Service B
    serviceB := telemetry.NewTracedHTTPClient(nil)

    mux := http.NewServeMux()
    mux.HandleFunc("/api/request", func(w http.ResponseWriter, r *http.Request) {
        ctx := r.Context()  // Contains trace from middleware

        // Call Service B - trace context propagates automatically
        req, _ := http.NewRequestWithContext(ctx, "GET", "http://service-b/process", nil)
        resp, _ := serviceB.Do(req)
        defer resp.Body.Close()

        // Respond...
    })

    // Wrap with tracing middleware
    traced := telemetry.TracingMiddleware("api-gateway")(mux)
    http.ListenAndServe(":8080", traced)
}

// === Service B ===
func main() {
    telemetry.Initialize(telemetry.Config{ServiceName: "service-b"})
    defer telemetry.Shutdown(context.Background())

    mux := http.NewServeMux()
    mux.HandleFunc("/process", func(w http.ResponseWriter, r *http.Request) {
        ctx := r.Context()  // Trace context from Service A!

        // Metrics are correlated with the trace
        telemetry.EmitWithContext(ctx, "service-b.processed", 1)

        w.WriteHeader(http.StatusOK)
    })

    traced := telemetry.TracingMiddleware("service-b")(mux)
    http.ListenAndServe(":8081", traced)
}

// Result in Jaeger/Tempo:
// Trace abc123:
// ├── api-gateway: /api/request (100ms)
// │   └── service-b: /process (50ms)

Orchestration Module​

Intelligently coordinate multiple agents and tools to accomplish complex tasks.

Read-only execution inspection​

type StepResultFilter struct {
    AgentName  string
    Capability string
    Namespace  string
    Parameters map[string]any
}

func LastSuccessfulStep([]StepResult, StepResultFilter) (index int, found bool)
func SuccessfulToolData(StepResult) (json.RawMessage, error)
func ExecutionRequestID(*OrchestratorResponse, *ExecutionResult) (string, error)

LastSuccessfulStep searches in reverse slice order, excluding failed/skipped steps. Empty identity filters are wildcards; parameter filters use exact, type-sensitive subset matching. Not found is (-1, false). Do not mutate the execution concurrently with inspection.

SuccessfulToolData is an opt-in decoder for core.ToolResponse, not a universal capability format. It rejects unknown envelope fields, duplicate JSON keys, multiple values, unsuccessful envelopes, non-null errors, and missing/null data. It uses the final dispatch's raw response when dispatches exist and rejects a failed final dispatch. For legacy/custom results without dispatches, it uses StepResult.Response; that representation may already have been normalized by its producer. The returned detached data value preserves the selected response's bytes. Original wire fidelity and duplicate-key detection cannot be recovered from a previously normalized legacy response. Typed decoding and domain receipt validation remain with the application. Report a selected result's decoding error rather than silently trying an older result or dispatch.

ExecutionRequestID returns the response ID or recorded execution ID; missing identity is ("", nil), conflicting/mistyped identity is an error. It does not read parent context, create identity, or modify logging/tracing context. All three helpers are read-only and do not alter execution success or continuation.

Canonical Orchestrator Construction​

Use the canonical path when configuration errors must fail before the agent starts and environment ownership must be explicit:

type ConfigResolution struct {
    Base        *OrchestratorConfig
    Environment EnvironmentMode
    LookupEnv   func(string) (string, bool)
    Options     []OrchestratorOption
}

func NewDefaultOrchestratorConfig() *OrchestratorConfig
func ResolveOrchestratorConfig(ConfigResolution) (*ConfigResolutionResult, error)
func DefaultConfigWithDiagnostics() ConfigResolutionResult
func ValidateOrchestratorConfig(*OrchestratorConfig) error
func ResolveOrchestratorConfigIdentity(*OrchestratorConfig) (OrchestratorConfigIdentity, error)
func CreateResolvedOrchestrator(*OrchestratorConfig, OrchestratorDependencies) (*AIOrchestrator, error)
func CreateOrchestratorFromEnvironment(OrchestratorDependencies, ...OrchestratorOption) (*AIOrchestrator, error)

type ConfigResolutionResult struct {
    Config      *OrchestratorConfig
    Diagnostics []ConfigDiagnostic
    Err         error // Fatal error from legacy DefaultConfigWithDiagnostics; Config is nil.
}

NewDefaultOrchestratorConfig is deterministic and never reads process environment. ResolveOrchestratorConfig applies layers in this order:

framework defaults/base -> selected environment mode -> code options -> normalization -> validation

The environment modes are:

ModeBehavior
EnvironmentDisabledPerforms no environment reads
EnvironmentCompatiblePreserves accepted legacy fallbacks and returns bounded ConfigDiagnostic values for ignored/coerced/defaulted settings
EnvironmentStrictRejects malformed, empty, out-of-range, or inconsistent documented values without echoing their raw contents

CreateResolvedOrchestrator performs no environment reads and never initializes a telemetry provider. A nil telemetry dependency safely becomes NoOp and emits the documented bounded diagnostic when telemetry is enabled. CreateOrchestratorFromEnvironment selects strict environment resolution and then calls the canonical constructor. CreateOrchestrator, CreateOrchestratorWithOptions, and CreateSimpleOrchestrator remain supported compatibility entry points; they preserve their documented environment/bootstrap behavior.

DefaultConfigWithDiagnostics preserves legacy environment loading and bounded per-setting fallback diagnostics. Check its Err before reading Config: fatal resolution failures return a nil config rather than pure defaults that could disable valid HITL or execution-store settings. The canonical resolver continues to return errors through its second return value.

DefaultConfig cannot return an error through its legacy signature; it retains fatal failure on a non-executable config. Factories reject that config. The errorless NewAIOrchestrator instead returns an inert instance whose Start and execution methods reject it. Prefer error-returning constructors for startup validation. Legacy partial configs default absent cleanup to 5s and reject explicit non-positive cleanup or negative failed-operation limits before component construction; unrelated legacy zero-value semantics are unchanged. Cleanup-timeout environment errors use bounded reasons empty_value, invalid_duration or out_of_range; their diagnostics do not include the raw rejected value. ResolveOrchestratorConfigIdentity returns a sanitized behavior summary, fingerprint, and explicit CacheEligible bit. Prompt bodies, credentials, endpoints, and raw environment values are excluded. Cache consumers must bypass reuse unless CacheEligible is true.

OrchestratorConfig.RequestIDPrefix is a code-owned compatibility-visible setting. Every buffered, native-streaming, simulated-streaming, and direct-plan execution now generates <prefix>-<unix-nanoseconds>; the default prefix is orch. Keep the prefix stable and low-cardinality because it appears in logs, traces, and execution records.

resolved, err := orchestration.ResolveOrchestratorConfig(orchestration.ConfigResolution{
    Environment: orchestration.EnvironmentStrict,
    Options: []orchestration.OrchestratorOption{
        orchestration.WithMaxConcurrency(12), // code wins over environment
    },
})
if err != nil {
    log.Fatal(err)
}
orchestrator, err := orchestration.CreateResolvedOrchestrator(resolved.Config, deps)

Agent Skills​

Skills are orchestration-owned, provider-neutral instruction packages. An agent developer explicitly binds eligibility; runtime candidate resolution pins exact versions before hooks and progressively loads only active content.

For a complete developer and operator walkthrough, see the Agent Skills Guide.

type SkillBinding struct {
    Namespace  string
    Name       string
    Version    string          // "published" or a positive immutable revision
    Activation SkillActivation // always, auto, or explicit
    Required   bool            // availability requirement, not forced activation
}

type SkillRegistry interface {
    ListMetadata(context.Context, SkillMetadataFilter) ([]SkillMetadata, error)
    ResolveCandidates(context.Context, []SkillCandidateRequest) ([]SkillCandidate, error)
    GetManifest(context.Context, SkillVersionRef) (SkillManifest, error)
    GetResource(context.Context, SkillResourceRef) (SkillResource, error)
}

Binding fields are eligibility policy, not discovery:

SettingEffect
Version: "published"Resolve the current publication once at request start and pin its exact immutable identity for the execution.
Version: "N"Resolve positive immutable revision N; later publications do not move the binding.
Activation: alwaysActivate every admitted candidate without an LLM decision.
Activation: autoPermit bounded selection at initial planning and continuation boundaries.
Activation: explicitActivate only when trusted host code supplies the request-local carrier.
Required: trueMake applicable availability/content failures fatal; it does not force activation.

DomainCompatibilityMode is off, warn (default), or enforce. It only compares an explicitly bound skill's authored domains with PromptConfig.Domain; it never searches a domain catalog or substitutes another skill. A missing agent domain or an empty skill-domain list is not a mismatch. off skips comparison; warn records a diagnostic and honors the binding; enforce fails a mismatched required binding and omits a mismatched optional binding.

Configure the complete binding list and inject the registry through code:

skillConfig := orchestration.SkillConfig{
    Enabled: true,
    Bindings: []orchestration.SkillBinding{{
        Namespace:  "travel",
        Name:       "weather-assessment",
        Version:    "published",
        Activation: orchestration.SkillActivationAuto,
    }},
}

resolved, err := orchestration.ResolveOrchestratorConfig(orchestration.ConfigResolution{
    Environment: orchestration.EnvironmentStrict,
    Options: []orchestration.OrchestratorOption{
        orchestration.WithSkills(skillConfig),
        orchestration.WithSkillRegistry(skillRegistry),
    },
})
if err != nil {
    log.Fatal(err)
}
orch, err := orchestration.CreateResolvedOrchestrator(resolved.Config, deps)

When set, TRUVAG3_SKILL_BINDINGS_JSON replaces the complete code binding list; it never merges. WithSkillActivationAIOptions and WithSkillResourceAIOptions accept only model and reasoning-effort intent. WithSkillPromptGuidance is bounded additive guidance. Custom deterministic selection implementations use WithSkillActivationPolicy, WithSkillResolver, or WithSkillResourceResolver; set a stable SkillConfig.RuntimePolicyID when custom behavior must remain response-cache eligible.

Response-cache reuse is eligible only when RuntimePolicyID is present, or when no custom activation policy, activation/resource resolver, or token counter is installed and both selector models are explicitly pinned. An ineligible configuration continues to execute skills normally and may still cache verified immutable content; only cached-answer short-circuits are bypassed.

WithSkillTokenCounter(core.TokenCounter) installs the advisory counter used for runtime skill prompt admission. A typed-nil counter is rejected at construction. An error, negative count, or zero count for non-empty input falls back to the framework heuristic and records a bounded diagnostic. Because a custom counter can change admission/projection behavior, set SkillConfig.RuntimePolicyID when response-cache short-circuits must remain eligible.

The built-in selectors deliberately leave the provider-native response-format option unset. A fixed JSON prompt, strict decoding and identity validation, and two attempts form the portable structured-output contract. Activation retries malformed output only. Resource selector v2 additionally accepts a complete outer JSON/unlabelled fence and retries a child-attempt timeout. Configure SkillConfig.Limits.ResourceSelectionAttemptTimeout through TRUVAG3_SKILL_RESOURCE_SELECTION_ATTEMPT_TIMEOUT (default 0s, inherit parent). Zero in code is unset; nonzero code wins. Invalid consulted env values fail in strict and compatible modes. Parent cancellation is terminal and optional selection omission is visible in diagnostics and operation outcome. Applications that require a provider-specific protocol use a custom resolver.

The additive runtime field is ResourceSelectionAttemptTimeout time.Duration with JSON tag resource_selection_attempt_timeout,omitempty on SkillRuntimeLimits. Omitting zero preserves the previous budget identity. The two-attempt allowance is shared by malformed-output and child-timeout correction, not two retries per failure category. A successfully returned answer is parsed even if the child deadline expires at the same boundary. Each timed-out attempt is diagnosed; the final optional omission is counted once. This setting does not add an activation-selector deadline or change activation's correction rule.

Trusted host code can request an explicit activation or named resource using WithTrustedSkillActivations and WithTrustedSkillResourceRequests. Ordinary request metadata and user text cannot populate these carriers. Each carrier is bounded to 32 entries.

Trusted host code can also attach selection-only capability hints:

ctx, err := orchestration.WithSkillExpectedCapabilities(
    ctx,
    "weather-forecasting",
    "route-planning",
)

The helper trims, deduplicates, sorts, and pins at most 32 values of at most 128 bytes each for the execution. The hints are included in selector context and cache projection but never add a discovered capability, tool, or permission. V1 does not infer them from user text or model-generated plans; no carrier means an empty set.

For management hosts, NewSkillAdminHandler composes narrow capabilities:

handler, err := orchestration.NewSkillAdminHandler(
    orchestration.SkillAdminHandlerDependencies{
        Registry:             registry,
        RevisionReader:       revisions,
        Administration:       administration,
        Deletions:            deletions,
        AuthoringLimits:      customAuthoringLimits,  // optional; zero uses defaults
        AdministrationLimits: customAdminLimits,      // optional; zero uses defaults
        ValidationRules:      customValidationRules,  // optional deterministic rules
        Audit:                auditSink,
        AuditAttribution:     actorProvider,           // optional
        Advisor:              optionalAdvisor,
        Logger:               logger,
        Telemetry:            telemetryProvider,
    },
)

The package body for first publication, validation, and later publication is the same complete representation (the namespace and name come from the path):

{
  "display_name": "Travel Weather Assessment",
  "description": "Use when a travel request asks about forecasts or weather-related risk.",
  "domains": ["travel"],
  "tags": ["weather", "forecast", "risk"],
  "planning_instructions": [
    "Resolve the destination to coordinates before requesting forecast data."
  ],
  "response_instructions": [
    "Separate observed conditions from forecast uncertainty."
  ],
  "tool_hints": ["geocode", "get_weather"],
  "resources": [
    {
      "name": "weather-risk-checklist",
      "description": "Checklist for weather-sensitive recommendations.",
      "load_when": "Load when disruption risk affects the recommendation.",
      "applies_to": ["planning", "continuation", "synthesis"],
      "required_when_selected": false,
      "content_type": "text/markdown",
      "content": "Check forecast horizon, confidence, precipitation, wind, and official advisories."
    }
  ],
  "activation_examples": {
    "should_activate": ["Will storms disrupt my trip next week?"],
    "should_not_activate": ["Convert 100 USD to JPY"]
  },
  "change_reason": "Initial publication"
}

For the first publication through the bundled Registry Viewer host:

curl -X PUT \
  -H 'Content-Type: application/json' \
  -H 'If-None-Match: *' \
  -H 'Idempotency-Key: weather-assessment-initial' \
  --data @weather-assessment.json \
  http://registry.localhost/api/v1/skills/travel/weather-assessment

Validate the same file without mutation by sending it to POST /api/v1/skills/travel/weather-assessment/validate. For an update, read the current representation and ETag, then replace If-None-Match with If-Match: <current-etag>.

Resources accept text/plain or text/markdown. An empty applies_to applies at every eligible boundary; otherwise the values are planning, continuation, and/or synthesis. required_when_selected does not force selection, but once selected it turns load, integrity, and admission failures for that resource into fatal errors. Activation examples are retained for evaluation and authoring analysis and do not enter ordinary runtime prompts.

The handler owns /api/v1/skills: schema and deterministic validation, optional non-mutating analysis, catalog/published/history reads, conditional publication, and guarded single/range deletion. Creation requires If-None-Match: *; updates and deletions require the current If-Match ETag. Deletion also requires X-Audit-Reason. The host—not orchestration—owns route exposure, authentication, and authorization.

PUT may also carry a bounded Idempotency-Key header. Every mutation requires an audit sink. The store mutation commits before audit delivery; if the audit write then fails, the handler does not roll back the mutation. Instead it returns audit_recorded: false and a skill_audit_not_recorded warning so the host can repair that control-plane record. Deletion protects the currently published revision and the immediately preceding revision; older retained versions are removed only through an explicit guarded delete, preserving the roll-forward operational model.

The permanent V1 routes are:

Method and pathQuery parametersCapability required
GET /api/v1/skills/schemanonenone
POST /api/v1/skills/{namespace}/{name}/validatenonenone
POST /api/v1/skills/{namespace}/{name}/analyzenoneadvisor
GET /api/v1/skillsoptional namespace, domain, tag, limitregistry
GET /api/v1/skills/{namespace}/{name}nonerevision reader
GET /api/v1/skills/{namespace}/{name}/versionsoptional before, limitrevision reader
GET /api/v1/skills/{namespace}/{name}/versions/{version}nonerevision reader
PUT /api/v1/skills/{namespace}/{name}noneadministration and audit
DELETE /api/v1/skills/{namespace}/{name}/versions/{version}nonedeletion and audit
DELETE /api/v1/skills/{namespace}/{name}/versionsrequired from, to inclusivedeletion and audit

Query parsing is strict: malformed encoding, unknown keys, and repeated keys are rejected. Numeric versions and limits must be positive, and deletion ranges must fit the configured maximum. The complete PUT package carries the required change_reason, whose authoritative ceiling is 1,024 UTF-8 bytes; the served JSON Schema can express only the corresponding character-count hint. Use the validate endpoint for the authoritative normalized and semantic result.

The optional default authoring advisor analyzes normalized metadata, main guidance, tool hints, activation examples, validation results, and the body-free resource index; it receives neither resource bodies nor change_reason and cannot publish or mutate a package. It returns bounded suggestions after deterministic validation. Runtime selector and advisor model calls appear in LLM Debug as skill_activation_selection, skill_resource_selection, and skill_authoring_analysis; ordinary traces and execution skill records remain body-free.

The included Redis implementation is redisprovider.SkillStore. Custom stores can use backendconformance.RunSkillConformance to verify publication, versioning, deletion, integrity, and cross-instance visibility semantics.

CreateSimpleOrchestrator​

Zero-configuration orchestrator for getting started quickly.

This is a compatibility convenience. Use the canonical construction path above for production startup validation and explicit environment ownership.

func CreateSimpleOrchestrator(discovery core.Discovery, aiClient core.AIClient) *AIOrchestrator
func CreateSimpleOrchestratorWithError(discovery core.Discovery, aiClient core.AIClient) (*AIOrchestrator, error)

Prefer CreateSimpleOrchestratorWithError when startup must fail immediately. The pointer-only convenience preserves source compatibility by returning a poisoned orchestrator on construction failure; its first execution returns the stored construction error.

Example:

// Just works - no configuration needed!
orchestrator := orchestration.CreateSimpleOrchestrator(discovery, aiClient)

// Process complex natural language requests
response, err := orchestrator.ProcessRequest(ctx,
    "Get weather for NYC and find related news articles, then summarize everything",
    nil,  // Auto-discovers needed services
)

fmt.Println(response.Response)
// Output: "Today in NYC, it's 72°F with partly cloudy skies. Recent news includes..."

CreateOrchestratorWithOptions​

Create an orchestrator with flexible configuration options.

func CreateOrchestratorWithOptions(deps OrchestratorDependencies, opts ...OrchestratorOption) (*AIOrchestrator, error)

Available options:

  • WithCapabilityProvider(type, url) - Configure capability discovery
  • WithTieredResolution(enabled) - Enable tiered capability resolution (token optimization)
  • WithTieredSelectionRetry(enabled, maxRetries) - Retry tiered selection on empty LLM responses and parse failures (default: true, 2)
  • WithTelemetry(enabled) - Enable metrics and tracing
  • WithFallback(enabled) - Graceful degradation
  • WithPlanParseRetry(enabled, maxRetries) - Retry plan generation on JSON parse failures
  • WithHallucinationRetry(enabled, maxRetries) - Retry on hallucinated agent names
  • WithPlanAIOptions(opts) - Per-phase overrides for planning LLM calls
  • WithSynthesisAIOptions(opts) - Per-phase overrides for synthesis LLM calls
  • WithMicroResolutionAIOptions(opts) - Per-phase overrides for micro-resolution, semantic retry, and plan refinement calls
  • WithTieredSelectionAIOptions(opts) - Per-phase overrides for tiered capability selection
  • WithErrorAnalysisAIOptions(opts) - Per-phase overrides for error analysis calls
  • WithResultDistillAIOptions(opts) - Per-phase overrides for result distillation calls
  • WithSkills(config) - Replace the complete developer-owned skill configuration
  • WithSkillRegistry(registry) - Inject the provider-neutral runtime skill registry
  • WithSkillContentCache(cache) - Replace the optional verified immutable-content cache
  • WithSkillActivationAIOptions(opts) / WithSkillResourceAIOptions(opts) - Set selector model/reasoning intent
  • WithSkillPromptGuidance(guidance) - Add bounded activation/resource selector guidance
  • WithSkillActivationPolicy(policy) - Add deterministic activation refinement before the model selector
  • WithSkillResolver(resolver) / WithSkillResourceResolver(resolver) - Replace only the corresponding model-selection task
  • WithSkillTokenCounter(counter) - Replace advisory runtime skill token counting; invalid output falls back safely
  • WithResultTrimming(enabled, maxResultBytes) - Enable/configure structural result trimming (Layer 1)
  • WithResultPreserveKeys(keys) - JSON keys the structural trimmer prioritizes keeping (a strong preference in its scoring, not a guarantee)
  • WithResultDistill(enabled, distillThreshold) - Enable/configure LLM result distillation (Layer 2)
  • WithResultDistillModel(model) - Model or portable alias for distillation calls; aliases are portable only across catalog-backed entries, so leave it empty for a chain containing Bedrock
  • WithMaxConcurrency(n) - Max parallel step executions in DAG (default: 25)
  • WithStepTimeout(d) - Execution allowance added to the HITL timeout for delegated orchestrator calls (default: 15m); not a direct deadline on ordinary tool calls
  • WithTotalTimeout(d) - Base HTTP client timeout per tool/agent call (default: 20m); delegation allowance may enlarge it. Not an end-to-end request deadline.
  • WithStepRetryBackoff(initial, max) - Code-owned retry timing; canonical validation rejects non-positive or inverted values
  • WithExecutionStoreWriteTimeout(d) - Per-write deadline for execution-debug snapshots (default: 5s)

Iterative planning is configured through OrchestratorConfig fields (start from NewDefaultOrchestratorConfig) or the documented environment layer; there is no dedicated With* option. DefaultConfig() remains the compatibility environment-loading helper. See IterativePlanConfig for details.

To build the Layer-2 distillation result processor outside CreateOrchestrator (e.g. in a custom runner), use BuildDistillationEnabledResultProcessor(cfg ResultDistillConfig, ai core.AIClient, cache core.DigestCache, logger core.Logger) ResultProcessor — a StructuralTrimmer wrapped by the LLMDistiller wrapped by a fail-open cache (nil ai → bare structural floor; nil cache → no caching).

OrchestratorDependencies fields:

FieldTypeRequiredDescription
Discoverycore.DiscoveryYesService discovery provider
AIClientcore.AIClientYesLLM provider client
CircuitBreakercore.CircuitBreakerNoSophisticated resilience patterns
Loggercore.LoggerNoStructured logging
Telemetrycore.TelemetryNoMetrics and tracing
PromptBuilderPromptBuilderNoCustom prompt building (default: DefaultPromptBuilder)
EnableErrorAnalyzerboolNoLLM-based error analysis (Layer 3)
ResultProcessorResultProcessorNoCustom synthesis-prompt result compactor (default: StructuralTrimmer, or the LLM distiller when distillation is enabled)
SourceResultProcessorResultProcessorNoDeterministic trimmer for micro-/semantic-retry source data. A custom override is always honored; the built-in StructuralTrimmer default is installed only when result trimming is enabled
AgentInputProcessorAgentInputProcessorNoTransform/redact/trim tool→tool input parameters before dispatch (default: identity; built-in byte guard when MaxAgentInputBytes > 0)
ContinuationDistillerResultProcessorNoSummarizer for non-JSON continuation escalations (default: built-in LLM distiller when distillation is enabled and an AIClient is present)
DistillCachecore.DigestCacheNoContent-addressed cache for distillation results (nil = no caching)
PipelineHooks[]core.PipelineHookNoPer-stage middleware for context engineering. A nil or empty slice is valid; a nil or typed-nil entry fails factory construction with an error wrapping ErrInvalidOrchestratorConfig. Ordinary hooks use documented fail-open fallbacks; one terminal RequiredAfterPlanningHook may provide fail-closed plan governance. Invalid required composition fails factory construction with InvalidRequiredAfterPlanningHookError. See Pipeline Hooks: Registering Hooks
ConversationHistoryPreparerConversationHistoryPreparerNoShared conversation-history preparer for the metadata and hook ingress paths. If nil, the factory auto-builds the default Tier 1 processor from config.
ActivityCoordinatorcore.ActivityCoordinatorNoReal-time agent coordination signals — typically the second return value of orchestration.BuildMemoryHooks

Orchestration Backend Composition​

OrchestrationBackends is a provider-neutral, immutable-style composition value with a private layout. Runtime features consume its narrow interfaces; they do not inspect a provider or infer one from discovery.

func NewOrchestrationBackends(...OrchestrationBackendOption) (*OrchestrationBackends, error)
func (b *OrchestrationBackends) With(...OrchestrationBackendOption) (*OrchestrationBackends, error)
func (b *OrchestrationBackends) ValidateFor(BackendRequirements) error

func RequirementsForFeatures(*OrchestratorConfig, ...BackendFeature) (BackendRequirements, error)
func WireOrchestratorBackends(*OrchestratorConfig, *OrchestrationBackends) error
func (b *OrchestrationBackends) SkillAdministrationDependencies() (SkillAdminHandlerDependencies, error)

Typed options populate or override individual capabilities:

WithExecutionBackend(ExecutionStore)
WithLLMDebugBackend(LLMDebugStore)
WithCheckpointPersistence(CheckpointPersistence)
WithCheckpointExpiry(ExpiredCheckpointSource)
WithCheckpointExpiryProcessor(core.Runnable)
WithCommandBackend(CommandStore)
WithWorkflowBackend(StateStore)
WithScheduleBackend(core.ScheduleStore)
WithTaskBackend(core.TaskStore)
WithTaskQueueBackend(core.TaskQueue)
WithTaskDispatcherBackend(core.TaskDispatcher)
WithTaskConsumerBackend(core.TaskConsumer)
WithLockBackend(core.DistributedLock)
WithSkillRegistryBackend(SkillRegistry)
WithSkillRevisionReader(SkillRevisionReader)
WithSkillAdministrationStore(SkillAdministrationStore)
WithSkillRevisionDeletionStore(SkillRevisionDeletionStore)
WithSkillAuditSink(SkillAuditSink)
WithRunnables(...core.Runnable)

Matching getters are Execution, LLMDebug, Checkpoints, CheckpointExpiry, CheckpointExpiryProcessor, Commands, Workflow, Schedules, Tasks, TaskQueue, TaskDispatcher, TaskConsumer, Lock, SkillRegistry, SkillRevisionReader, SkillAdministrationStore, SkillRevisionDeletionStore, SkillAuditSink, and Runnables. Getters are nil-safe; Runnables returns a defensive copy. WithRunnables appends rather than replacing previously composed lifecycle components. Replacing checkpoint persistence or the expired-checkpoint source invalidates the old processor so it cannot remain bound to stale dependencies. Within one option list, dependency options must precede an explicit checkpoint processor; the reverse order fails instead of silently substituting a default.

RequirementsForFeatures combines execution/LLM-debug requirements from the effective orchestrator configuration with explicit feature flags such as BackendFeatureWorkflow, BackendFeatureCrossInstanceHITL, BackendFeatureCheckpointExpiry, BackendFeatureSchedulerProducer, BackendFeatureScheduledWorker, BackendFeatureDistributedLock, BackendFeatureSkillsRuntime, and BackendFeatureSkillsAdministration. ValidateFor fails startup with the complete sorted missing-capability list. The scheduler-producer feature requires schedules, tasks, dispatcher, and lock. Checkpoint expiry requires persistence, source, and processor capabilities. BackendSkills remains a source-compatible alias for the runtime skill-registry capability; revision reading, publication, deletion, and audit are distinct control-plane capabilities.

WireOrchestratorBackends fills only enabled dependencies that are still nil; explicitly injected config dependencies win. SkillAdministrationDependencies validates and unpacks the complete management dependency set for NewSkillAdminHandler.

Included Redis preset​

The included implementation lives in github.com/truvaagents/truva-g3/orchestration/redisprovider, outside runtime orchestration imports:

ownedBackends, err := redisprovider.NewDefaultBackends(
    logger,
    redisprovider.WithDefaultBackendProviderOptions(
        redisprovider.WithNamespace("travel-chat-agent"),
    ),
)
if err != nil {
    log.Fatal(err)
}
defer ownedBackends.Close()
backends := ownedBackends.Backends()

NewDefaultBackends is the Layer-1 path. It loads documented environment configuration, applies explicit client/provider/backend options last, creates and validates either all roles or the roles selected by WithDefaultBackendRoles, and returns OwnedBackends for deterministic client cleanup. WithDefaultBackendClientConfig, WithDefaultBackendProviderOptions, and WithDefaultBackendOverrides expose the lower-layer customization seams without duplicating their implementation.

ClientSet accepts one application-owned default Redis client plus optional role-specific clients through WithRoleClient. A default client is an explicit fallback for every role; pass nil plus role clients to construct only selected capability groups. For configuration-owned connections, use DefaultClientConfig, LoadClientConfigFromEnvironment, and NewOwnedClients; WithOwnedClientRoles restricts creation to the roles the process needs. Close the returned OwnedClients after the framework stops. Redis URLs are parsed without being repeated in errors. Default role/database assignments preserve legacy behavior and are compatibility values, not deployment recommendations.

Preset options are created with NewOptions. Deployment-owned operational values can then be applied using LoadOptionsFromEnvironment, followed by ConfigureOptions when code must win. Supported options include namespace, logger, execution-store configuration, LLM-debug retention, checkpoint retention, workflow TTL, task-queue retry policy, and checkpoint expiry. LoadOptionsFromEnvironment forwards TRUVAG3_EXECUTION_DEBUG_TTL and TRUVAG3_EXECUTION_DEBUG_ERROR_TTL into the Redis execution store's configuration; both default to 72 hours. Present empty, malformed, non-positive, or overflowing durations fail loading. NewDefaultBackends invokes this loader automatically and ignores these two variables when the execution role is not selected. Later WithExecutionStoreConfig provider options override valid environment values; they do not suppress invalid-environment errors. These provider options configure the actual injected store, independently of the harness configuration. Caller-supplied root backend overrides are applied last:

backends, err := redisprovider.NewOrchestrationBackends(
    clients,
    options,
    orchestration.WithExecutionBackend(customExecutionStore),
)

The preset never starts background work and never closes injected clients. Register each returned backends.Runnables() value with core.Framework before Run. Individual NewRedis* constructors remain supported compatibility APIs, but new applications should prefer composition so switching providers does not change runtime feature wiring.

The Redis preset's explicit configuration surface is:

func NewDefaultBackends(core.Logger, ...DefaultBackendsOption) (*OwnedBackends, error)
func WithDefaultBackendRoles(...ClientRole) DefaultBackendsOption
func WithDefaultBackendClientConfig(...ClientConfigOption) DefaultBackendsOption
func WithDefaultBackendProviderOptions(...Option) DefaultBackendsOption
func WithDefaultBackendOverrides(...orchestration.OrchestrationBackendOption) DefaultBackendsOption
func (*OwnedBackends) Backends() *orchestration.OrchestrationBackends
func (*OwnedBackends) Close() error

func DefaultClientConfig() ClientConfig
func ConfigureClientConfig(ClientConfig, ...ClientConfigOption) (ClientConfig, error)
func LoadClientConfigFromEnvironment(ClientConfig, func(string) (string, bool)) (ClientConfig, error)
func WithClientURL(string) ClientConfigOption
func NewClientSet(redis.UniversalClient, ...ClientSetOption) (*ClientSet, error)
func WithRoleClient(ClientRole, redis.UniversalClient) ClientSetOption
func NewOwnedClients(ClientConfig, ...OwnedClientsOption) (*OwnedClients, error)
func WithOwnedClientRoles(...ClientRole) OwnedClientsOption
func (*OwnedClients) ClientSet() *ClientSet
func (*OwnedClients) Close() error

const (
    ClientRoleExecution  ClientRole = "execution"
    ClientRoleLLMDebug   ClientRole = "llm_debug"
    ClientRoleHITL       ClientRole = "hitl"
    ClientRoleWorkflow   ClientRole = "workflow"
    ClientRoleScheduling ClientRole = "scheduling"
    ClientRoleSkills     ClientRole = "skills"
)

func NewOptions(...Option) (Options, error)
func ConfigureOptions(Options, ...Option) (Options, error)
func LoadOptionsFromEnvironment(Options, func(string) (string, bool)) (Options, error)
func WithNamespace(string) Option
func WithExecutionStoreConfig(orchestration.ExecutionStoreConfig) Option
func WithLLMDebugRetention(time.Duration, time.Duration) Option
func WithCheckpointTTL(time.Duration) Option
func WithLogger(core.Logger) Option
func WithWorkflowStateTTL(time.Duration) Option
func WithTaskQueueRetryPolicy(int, time.Duration) Option
func WithCheckpointExpiry(
    orchestration.ExpiryProcessorConfig,
    orchestration.ExpiryCallback,
    ...orchestration.CheckpointExpiryProcessorOption,
) Option

For adapter-specific composition, the application-owned client constructors are supported direct-composition APIs:

func NewRedisExecutionDebugStoreWithClient(redis.UniversalClient, ExecutionStoreConfig, ...RedisExecutionDebugStoreOption) (*RedisExecutionDebugStore, error)
func NewRedisLLMDebugStoreWithClient(redis.UniversalClient, ...RedisLLMDebugStoreOption) (*RedisLLMDebugStore, error)
func NewRedisCheckpointStoreWithClient(redis.UniversalClient, ...RedisCheckpointStoreOption) (*RedisCheckpointStore, error)
func NewRedisCommandStoreWithClient(redis.UniversalClient, ...RedisCommandStoreOption) (*RedisCommandStore, error)
func NewRedisStateStoreWithClient(redis.UniversalClient, core.RedisKeyspace, time.Duration) (*RedisStateStore, error)
func NewRedisTaskQueue(redis.Cmdable, *RedisTaskQueueConfig) *RedisTaskQueue
func NewRedisTaskStore(redis.Cmdable, *RedisTaskStoreConfig) *RedisTaskStore
func NewRedisTaskDispatcherWithPrefix(redis.Cmdable, string) (*RedisTaskDispatcher, error)
func NewRedisTaskConsumerWithPrefix(redis.Cmdable, string, string) (*RedisTaskConsumer, error)

These constructors never close the supplied client. Workflow state requires workflowID for GetExecution and UpdateStepExecution; StateStore aliases WorkflowStateStore. The obsolete execution-ID-only constructors are removed. The task-prefixed forms remain supported routing seams: derive their prefix from keyspace.Plain("tasks") and give producer and consumer the same scope.

The skills adapter lives in redisprovider:

func NewSkillStore(redis.UniversalClient, ...SkillStoreOption) (*SkillStore, error)
func WithSkillStoreKeyspace(core.RedisKeyspace) SkillStoreOption
func WithSkillStoreLogger(core.Logger) SkillStoreOption

It borrows the client and uses only the versioned deployment-scoped skills keyspace. The default deployment is default; WithSkillStoreKeyPrefix is removed. See the skills storage contract.

Checkpoint expiry processor​

CheckpointExpiryProcessor separates provider-neutral expiry policy/lifecycle from storage. It implements core.Runnable and requires both CheckpointPersistence and ExpiredCheckpointSource:

runtimeConfig, err := orchestration.LoadCheckpointExpiryRuntimeConfigFromEnvironment(
    orchestration.DefaultCheckpointExpiryRuntimeConfig(),
    os.LookupEnv,
)
processor, err := orchestration.NewCheckpointExpiryProcessor(
    persistence,
    expiredSource,
    callback,
    expiryConfig,
    orchestration.WithCheckpointExpiryRuntimeConfig(runtimeConfig),
    orchestration.WithCheckpointExpiryLogger(logger),
    orchestration.WithCheckpointExpiryTelemetry(telemetryProvider),
)
framework.RegisterRunnable(processor)

The included Redis source uses atomic owner/lease claims so multiple replicas do not process the same expiry concurrently. The default claim lease is 30 seconds and can be overridden with TRUVAG3_HITL_EXPIRY_CLAIM_LEASE or code configuration. When the preset assembles expiry, it binds the processor after applying final checkpoint overrides. BackendFeatureCheckpointExpiry validation therefore fails if only persistence/source are present without a runnable processor.

Example - Production Orchestrator:

// Set up dependencies
deps := orchestration.OrchestratorDependencies{
    Discovery: redisDiscovery,
    AIClient:  aiClient,
    Logger:    logger,
}

// Create with options
orchestrator, err := orchestration.CreateOrchestratorWithOptions(deps,
    orchestration.WithCapabilityProvider("service", "http://capability-service:8080"),
    orchestration.WithTelemetry(true),
    orchestration.WithFallback(true),
    orchestration.WithPlanAIOptions(&orchestration.AIOptionsOverride{
        MaxTokens: orchestration.IntPtr(15000),
    }),
)

// Process requests with automatic service discovery and coordination
response, err := orchestrator.ProcessRequest(ctx,
    "Analyze sales data and generate report",
    nil,
)

Example - With Pipeline Hooks (Context Engineering):

deps := orchestration.OrchestratorDependencies{
    Discovery: redisDiscovery,
    AIClient:  aiClient,
    Logger:    logger,
    PipelineHooks: []core.PipelineHook{
        cacheHook,    // BeforePlanningHook — short-circuits on cache hit
        ragHook,      // BeforePlanningHook — injects RAG context
        analytics,    // AfterExecutionHook — logs to Kafka
        guardrails,   // AfterSynthesisHook — filters response
    },
}

For chat agents, the preferred conversation-history path is request metadata, not a hook. Pass raw turns in metadata[orchestration.MetadataConversationTurns] and the stable canonical identity in metadata[orchestration.MetadataConversationID], then let the shared ConversationHistoryPreparer build the <conversation_history> enrichment before planning. Set MetadataConversationID on the first turn even when the turns slice is empty. Application session_id remains a separate concept, although an application may deliberately map the same opaque UUID to both. ConversationHistoryHook remains available as an adapter for memory-backed integrations that cannot supply raw turns directly. For Tier 2 recursive compaction, the ergonomic Layer 2 path is BuildCompactionEnabledConversationHistoryPreparer(config, aiClient, ...options), which installs the default cache and LLM compactor while still allowing WithConversationSummaryCache(...) and WithConversationCompactor(...) overrides.

For detailed pipeline hook implementation patterns, see Adding Context to Your Agent. For the conversation-history metadata path plus Tier 2 / Layer 3 reference implementations, see Conversation History Guide.

Orchestrator Interface Methods​

The Orchestrator interface provides the following methods:

type Orchestrator interface {
    // ProcessRequest handles a natural language request by orchestrating multiple agents
    ProcessRequest(ctx context.Context, request string, metadata map[string]interface{}) (*OrchestratorResponse, error)

    // ExecutePlan executes a pre-defined routing plan (raw results, no synthesis)
    ExecutePlan(ctx context.Context, plan *RoutingPlan) (*ExecutionResult, error)

    // ExecutePlanWithSynthesis executes a pre-defined routing plan with synthesis.
    // Unlike ExecutePlan(), this method:
    // 1. Uses the orchestrator's synthesizer (which auto-records to LLM Debug Store)
    // 2. Returns a complete OrchestratorResponse (not raw ExecutionResult)
    // 3. Stores execution to ExecutionStore for DAG visualization
    // 4. Sets up context baggage for request_id propagation
    // It does not run pipeline hooks or store a post-hook FinalResponse.
    ExecutePlanWithSynthesis(ctx context.Context, plan *RoutingPlan, originalRequest string) (*OrchestratorResponse, error)

    // GetExecutionHistory returns recent execution history
    GetExecutionHistory() []ExecutionRecord

    // GetMetrics returns orchestrator metrics
    GetMetrics() OrchestratorMetrics
}

When to use each method:

MethodUse Case
ProcessRequestNatural language requests with AI-driven planning
ExecutePlanPre-defined workflows when you need raw results for custom synthesis
ExecutePlanWithSynthesisPre-defined workflow execution with synthesis, DAG visualization, and LLM debug evidence; no application pipeline hooks or post-hook FinalResponse

Example - ExecutePlanWithSynthesis:

// Build a routing plan
plan := &orchestration.RoutingPlan{
    PlanID: "travel-research-123",
    Steps: []orchestration.RoutingStep{
        {
            StepID:      "step-1",
            AgentName:   "weather-tool",
            Instruction: "Get current weather",
            Metadata: map[string]interface{}{
                "capability": "get_weather",
                "parameters": map[string]interface{}{"location": "Tokyo"},
            },
        },
        {
            StepID:      "step-2",
            AgentName:   "currency-tool",
            Instruction: "Convert USD to JPY",
            Metadata: map[string]interface{}{
                "capability": "convert_currency",
                "parameters": map[string]interface{}{"from": "USD", "to": "JPY", "amount": 1000},
            },
        },
    },
}

// Execute with synthesis and DAG storage
response, err := orchestrator.ExecutePlanWithSynthesis(ctx, plan, "Plan my trip to Tokyo")
if err != nil {
    return err
}

// Response includes synthesized result and step details
fmt.Println(response.Response)      // AI-synthesized summary
fmt.Println(response.Steps)         // Individual step results
fmt.Println(response.RequestID)     // For DAG visualization lookup

ExecutionOptions Configuration​

Configure execution behavior for the orchestrator, including retry logic and type safety features.

type ExecutionOptions struct {
    MaxConcurrency           int           // Default: 25 | Env: TRUVAG3_EXECUTION_MAX_CONCURRENCY
    StepTimeout              time.Duration // Delegation execution allowance: 15m | Env: TRUVAG3_EXECUTION_STEP_TIMEOUT
    TotalTimeout             time.Duration // Base HTTP-client timeout: 20m | Env: TRUVAG3_ORCHESTRATION_TIMEOUT
    RetryAttempts            int           // Total attempts per step (default: 3 = 1 initial + 2 retries; overridable via TRUVAG3_STEP_RETRY_MAX_ATTEMPTS)
    RetryDelay               time.Duration // Legacy field; not consumed by SmartExecutor
    CircuitBreaker           bool          // Legacy field; does not enable a SmartExecutor breaker
    FailureThreshold         int           // Legacy field; not consumed by SmartExecutor
    RecoveryTimeout          time.Duration // Legacy field; not consumed by SmartExecutor

    // Type Safety (Layer 3 - Validation Feedback)
    ValidationFeedbackEnabled bool         // Enable LLM-based parameter correction (default: true)
    MaxValidationRetries      int          // Max correction attempts (default: 2)
}

The four legacy fields above retain stored defaults of 2s, true, 5 and 30s, respectively, but changing them does not configure the built-in executor. Use WithStepRetryBackoff(initial, max) with canonical orchestrator construction, or SmartExecutor.SetStepRetryBackoff for a directly constructed executor. Effective backoff defaults are 500 ms initially and a 10 s maximum, with a 2× factor and jitter. The CapabilityService.CircuitBreaker dependency is a separate capability-discovery concern, not a breaker on tool execution.

Type Safety Configuration:

// Default configuration - maximum reliability (~99% success rate)
config := orchestration.DefaultConfig()
// ValidationFeedbackEnabled: true (default)
// MaxValidationRetries: 2 (default)

// Cost-sensitive configuration (~95% success rate)
config.ExecutionOptions.ValidationFeedbackEnabled = false

// Maximum reliability configuration
config.ExecutionOptions.ValidationFeedbackEnabled = true
config.ExecutionOptions.MaxValidationRetries = 3  // More retries for edge cases

See Intelligent Error Handling for details on how type safety layers work together.

Step Retry Backoff​

The executor uses exponential backoff (via core.BackoffConfig) for step retries instead of linear delay. Configurable via setter or environment variables.

// SetStepRetryBackoff configures the backoff strategy for step retries.
// Default is core.DefaultBackoffConfig() (500ms initial, 10s max, 2x factor, jitter enabled).
// Also overridable via TRUVAG3_STEP_RETRY_INITIAL_DELAY and TRUVAG3_STEP_RETRY_MAX_DELAY env vars.
func (e *SmartExecutor) SetStepRetryBackoff(config core.BackoffConfig)

Environment variables:

VariableDefaultDescription
TRUVAG3_STEP_RETRY_INITIAL_DELAY500msInitial backoff delay. Go duration format (e.g., 500ms, 2s).
TRUVAG3_STEP_RETRY_MAX_DELAY10sMaximum backoff delay cap. Go duration format.

Precedence: SetStepRetryBackoff() (explicit) > TRUVAG3_STEP_RETRY_* env vars > core.DefaultBackoffConfig() defaults.

TieredCapabilityConfig​

Configures tiered capability resolution for LLM token optimization. When enabled with 20+ tools, uses a two-phase approach: lightweight summaries for tool selection, then full schemas only for selected tools.

type TieredCapabilityConfig struct {
    // MinToolsForTiering is the minimum tool count to trigger tiered resolution.
    // Below this threshold, all tools are sent directly (simpler, one LLM call).
    // Default: 20 | Env: TRUVAG3_TIERED_MIN_TOOLS
    // Research: "Less is More" (Nov 2024) shows LLM accuracy degradation at ~20 tools
    MinToolsForTiering int `json:"min_tools_for_tiering,omitempty"`

    // SelectionMaxTokens is the maximum output tokens for tiered selection LLM calls.
    // Higher values allow complex multi-tool selections but cost more tokens.
    // Default: 2000 | Env: TRUVAG3_TIERED_SELECTION_MAX_TOKENS
    SelectionMaxTokens int `json:"selection_max_tokens,omitempty"`

    // RetryEnabled enables retry on empty LLM responses and parse failures.
    // Default: true | Env: TRUVAG3_TIERED_SELECTION_RETRY_ENABLED
    RetryEnabled bool `json:"retry_enabled,omitempty"`

    // MaxRetries is the number of additional attempts after the initial failure.
    // 2 means up to 3 total attempts (1 initial + 2 retries).
    // Default: 2 | Env: TRUVAG3_TIERED_SELECTION_RETRY_MAX
    MaxRetries int `json:"max_retries,omitempty"`
}

Example - Enable Tiered Resolution:

config := orchestration.DefaultConfig()
config.EnableTieredResolution = true
config.TieredResolution = orchestration.TieredCapabilityConfig{
    MinToolsForTiering: 25,    // Override threshold (default: 20)
    SelectionMaxTokens: 3000,  // Override selection output tokens (default: 2000)
    RetryEnabled:       true,  // Retry on empty responses / parse failures (default: true)
    MaxRetries:         2,     // Up to 2 retries after initial failure (default: 2)
}

PromptConfig​

Configures prompt building behavior including persona customization and type rules.

type PromptConfig struct {
    // SystemInstructions defines the orchestrator's core behavioral context.
    // Similar to LangChain's system_prompt, AutoGen's system_message, OpenAI's instructions.
    // When set, the developer's persona becomes the primary identity.
    SystemInstructions string `json:"system_instructions,omitempty"`

    // AdditionalTypeRules extend default type handling rules.
    AdditionalTypeRules []TypeRule `json:"additional_type_rules,omitempty"`

    // TemplateFile is the path to a Go text/template file for prompt customization.
    TemplateFile string `json:"template_file,omitempty"`

    // Template is an inline Go text/template string.
    Template string `json:"template,omitempty"`

    // CustomInstructions are additional instructions appended to the prompt.
    CustomInstructions []string `json:"custom_instructions,omitempty"`

    // Domain provides context for domain-specific prompt adjustments.
    // Examples: "healthcare", "finance", "legal", "retail"
    Domain string `json:"domain,omitempty"`

    // IncludeAntiPatterns controls whether to show "what NOT to do" examples.
    IncludeAntiPatterns *bool `json:"include_anti_patterns,omitempty"`

    // IterativePlanConfig provides budget information (MaxPhases, MaxTotalSteps)
    // to prompt builders so they can embed budget-aware iterative planning
    // instructions. Populated automatically by NewAIOrchestrator when
    // iterative planning is enabled.
    IterativePlanConfig *IterativePlanConfig `json:"iterative_plan_config,omitempty"`
}

Example - Custom Agent Persona:

config := orchestration.DefaultConfig()
config.PromptConfig = orchestration.PromptConfig{
    SystemInstructions: `You are a travel planning assistant.
Always check weather before recommending outdoor activities.
Prefer real-time data sources over cached information.`,
    Domain: "travel",
}

Planning system prompts are finalized after custom-builder output and PlanAIOptions.SystemPrompt merging. The framework guarantees exactly one canonical <runtime_context> section for initial, continuation, and regeneration planning. Application prompts must not declare that reserved tag; malformed, embedded, duplicate, or raw-override copies fail before the AI call. A legacy custom SystemPromptBuilder whose output already ends with the exact canonical block is retained byte-for-byte. Correction and synthesis prompts do not receive this planning-only finalizer.

Callers can classify these preparation failures with orchestration.ErrReservedRuntimeContext and orchestration.ErrInvalidPromptAssembly.

IterativePlanConfig​

Configures multi-phase DAG planning behavior. When enabled, the LLM planner can signal that a plan is partial (terminal: false), causing the orchestrator to execute the known phase, feed results back to the planner, and generate continuation plans until the planner produces a terminal plan.

Fresh model responses need nonempty steps, explicit terminal:true for no work, or needs_user_input with a nonblank question. Zero-step ambiguity is corrected within the calling generation/recovery path's budget; direct/typed and resumed plans retain their contracts. Generated-plan normalization rewrites numeric bracket references and derives prior-phase implicit_deps, but does not repair missing same-phase depends_on. Regeneration reuses the selected phase catalog/skill projection with bounded rejected-attempt feedback. Repeated non-terminal dependency-only phases with unchanged failed ancestors stop after an opportunity to adapt; actual dispatched work, including polling, resets that structural guard.

This enables "discovery → action" queries where later steps depend on the semantic content of earlier results — e.g., "find famous tourist destinations in Germany and get their weather."

type IterativePlanConfig struct {
    // Enabled controls whether iterative planning is active.
    // When false, the terminal field is ignored and all plans are treated as terminal.
    // Default: true | Env: TRUVAG3_ITERATIVE_PLANNING_ENABLED
    Enabled bool `json:"enabled"`

    // MaxPhases is the maximum number of planning phases allowed per request.
    // If reached without a terminal plan, the orchestrator forces termination
    // and synthesizes with available results.
    // Default: 20 | Env: TRUVAG3_ITERATIVE_MAX_PHASES
    MaxPhases int `json:"max_phases"`

    // MaxTotalSteps is the maximum total steps across all phases.
    // Prevents runaway plan generation.
    // Default: 200 | Env: TRUVAG3_ITERATIVE_MAX_TOTAL_STEPS
    MaxTotalSteps int `json:"max_total_steps"`

    // PhaseTimeout is the configured phase budget (plan generation + execution).
    // The phase loop raises it to the delegation allowance when that is larger;
    // earlier parent deadlines still apply.
    // Prevents a single continuation phase from hanging indefinitely.
    // Default: 25m | Env: TRUVAG3_ITERATIVE_PHASE_TIMEOUT
    PhaseTimeout time.Duration `json:"phase_timeout"`

    // Regeneration attempts after the first validation failure. Malformed JSON
    // consumes an attempt; only parsed candidates reach validation.
    // Default: 4 | Env: TRUVAG3_ITERATIVE_MAX_VALIDATION_ROUNDS
    MaxValidationRounds int `json:"max_validation_rounds"`

    // Encoded additional correction context, not the whole prompt/token budget.
    // The effective value must be greater than zero when validated.
    // Default: 16384 | Env: TRUVAG3_ITERATIVE_REGENERATION_FEEDBACK_MAX_BYTES
    RegenerationFeedbackMaxBytes int `json:"regeneration_feedback_max_bytes"`
}

Fresh model JSON is decoded against the plan-intent contract before validation. An empty plan supplies explicit terminal:true or a clarification with a nonempty question. Initial/continuation generation uses its configured plan-parse retry path; initial unavailable-agent replacements share the hallucination recovery budget. During validation regeneration, a parse/intent failure consumes MaxValidationRounds instead. Continuation unknown-agent recovery makes one regeneration attempt and returns the original unknown-agent error if it fails; it does not use the initial generation's replacement-parse loop. The outer phase loop may subsequently attempt fallback regeneration, subject to its existing context and preparation gates; the inner attempt is not a whole-request cap. Direct/typed plans and approved HITL checkpoints retain their existing contract.

MaxValidationRounds is a ceiling, not a guarantee of that many model calls. Continuation regeneration checks for conflicts with already executed step IDs before returning its candidate. A remaining conflict returns an error immediately and can stop before the outer validation budget is exhausted.

Example - Default (enabled, suitable for most queries):

config := orchestration.DefaultConfig()
// IterativePlanning is already configured with sensible defaults:
// Enabled: true, MaxPhases: 20, MaxTotalSteps: 200, PhaseTimeout: 25m

Example - Complex research queries:

config := orchestration.DefaultConfig()
config.IterativePlanning.MaxPhases = 8
config.IterativePlanning.MaxTotalSteps = 300
config.IterativePlanning.PhaseTimeout = 5 * time.Minute
// Keep the default MaxValidationRounds and RegenerationFeedbackMaxBytes.

Example - Cost-sensitive (disable multi-phase):

config := orchestration.DefaultConfig()
config.IterativePlanning.Enabled = false // All plans treated as single-shot

Environment-backed defaults (read by DefaultConfig):

export TRUVAG3_ITERATIVE_PLANNING_ENABLED=true
export TRUVAG3_ITERATIVE_MAX_PHASES=20
export TRUVAG3_ITERATIVE_MAX_TOTAL_STEPS=200
export TRUVAG3_ITERATIVE_PHASE_TIMEOUT=25m

DefaultConfig applies these variables while constructing the returned value. Programmatic field mutations or functional options applied afterward take precedence over that environment-backed snapshot.

ContinuationResultMaxChars​

Floor-preview size (in chars) for a non-JSON completed-step result in continuation planning prompts (Phase 14). JSON step results are rendered as a structure-complete digest (skeleton) instead — this knob governs only the fallback preview for non-JSON blobs (logs, markdown, CSV) that the planner reads when the continuation distiller does not summarize the step.

type OrchestratorConfig struct {
    // ...
    ContinuationResultMaxChars int `json:"continuation_result_max_chars,omitempty"`
    // Default: 10000 | Env: TRUVAG3_CONTINUATION_RESULT_MAX_CHARS
}

Default: 10000. Related: When an orchestrator step's response contains a steps[] array, the continuation builder extracts a structured child-sub-step summary (from the full response, before any trimming) and appends a "Do NOT duplicate" directive — so child-step visibility holds regardless of this knob.

Continuation Digest Budgeting (Phase 14)​

The continuation prompt's <completed_steps> section renders each completed step as a decision digest under an aggregate budget. These top-level OrchestratorConfig fields tune it (all env-overridable; no With* options):

type OrchestratorConfig struct {
    // ...
    ContinuationResultMaxTotalChars int `json:"continuation_result_max_total_chars,omitempty"` // Default: 196608 | Env: TRUVAG3_CONTINUATION_RESULT_MAX_TOTAL_CHARS
    ContinuationMaxEscalations      int `json:"continuation_max_escalations,omitempty"`        // Default: 8     | Env: TRUVAG3_CONTINUATION_MAX_ESCALATIONS
    ContinuationDigestArraySample   int `json:"continuation_digest_array_sample,omitempty"`    // Default: 3     | Env: TRUVAG3_CONTINUATION_DIGEST_ARRAY_SAMPLE
    ContinuationDigestScalarMax     int `json:"continuation_digest_scalar_max,omitempty"`      // Default: 200   | Env: TRUVAG3_CONTINUATION_DIGEST_SCALAR_MAX
    ContinuationDigestMaxKeys       int `json:"continuation_digest_max_keys,omitempty"`        // Default: 50    | Env: TRUVAG3_CONTINUATION_DIGEST_MAX_KEYS
}
  • ContinuationResultMaxTotalChars — a soft byte target, despite the historical Chars name, for the whole completed-steps section; defaults to 196608 bytes (192 KiB) and drives newest-first eviction. Older successful steps evict with a "showing N of M" note; the number that fits depends on their actual rendered sizes. Not a hard cap: failed steps are always kept, the newest successful step is always kept even if it alone exceeds the budget, and the N-of-M note plus any orchestrator child-summaries are appended after the budget decision — so the rendered section can exceed it.
  • ContinuationMaxEscalations — max non-JSON steps escalated to the continuation distiller (a fast-model summary) per phase, newest-first. Sequential fast-model calls on the phase-gating path; 0 disables. Fires ~never on all-JSON workloads.
  • ContinuationDigestArraySample — per-array head-sample size (arrays render as the first N elements + a length sentinel).
  • ContinuationDigestScalarMax — max length of a string value kept inline before elision; raise it to surface longer salient values (status/error strings) at plan time.
  • ContinuationDigestMaxKeys — per-object key cap; schema objects are kept whole, map-shaped objects (many dynamic-ID keys) are sampled to N sorted keys + a sentinel.

RemediationFailurePattern​

Tunes the shared-error pattern summary that is embedded into the remediation continuation prompt when multiple upstream steps fail with the same error. When one or more steps are skipped because their template-referenced dependencies failed, the orchestrator forces a remediation continuation so the planner can adapt. If the causal failures share a dominant error signature, a one-line summary — e.g. "Upstream failure pattern: 2 of 2 prior steps failed with the same error; retries exhausted (flight-tool/search_airports): upstream API error 500" — is prepended to that prompt so the planner has concrete evidence that the upstream is persistently unavailable.

type OrchestratorConfig struct {
    // ...
    // Minimum distinct failed upstream steps required before a pattern
    // summary is emitted. 1 would make every skip emit a summary (noisy).
    // Default: 2 | Env: TRUVAG3_FAILURE_PATTERN_MIN_FAILURES
    RemediationFailurePatternMinFailures int `json:"remediation_failure_pattern_min_failures,omitempty"`

    // Max chars of error text used to bucket failures into a shared
    // signature for classification. Wider than the display cap to reduce
    // false-positive collisions between distinct errors that share a
    // common prefix.
    // Default: 120 | Env: TRUVAG3_FAILURE_PATTERN_SIGNATURE_LEN
    RemediationFailurePatternSignatureLen int `json:"remediation_failure_pattern_signature_len,omitempty"`

    // Max chars of the shared error rendered into the remediation prompt
    // (trailing `…` appended on truncation). Kept short so the
    // continuation prompt stays slim per EFFECTIVE_PROMPTS_GUIDE §4.5.
    // Default: 80 | Env: TRUVAG3_FAILURE_PATTERN_DISPLAY_LEN
    RemediationFailurePatternDisplayLen int `json:"remediation_failure_pattern_display_len,omitempty"`
}

Observability: Emission is observable via the orchestrator.remediation.triggered span event (attribute has_failure_pattern) and the remediation_failure_pattern DEBUG log (fields emitted, reject_reason). See DISTRIBUTED_TRACING_GUIDE and LOGGING_IMPLEMENTATION_GUIDE §11.

Failed-operation view configuration​

type FailedOperationViewConfig struct {
    Disabled       bool                       `json:"disabled,omitempty"`
    MaxValueBytes  int                        `json:"max_value_bytes,omitempty"`
    MaxStepBytes   int                        `json:"max_step_bytes,omitempty"`
    MaxPromptBytes int                        `json:"max_prompt_bytes,omitempty"`
    InputPolicy    FailedOperationInputPolicy `json:"-"`
}

type FailedOperationIdentity struct {
    StepID     string
    AgentName  string
    Capability string
    Attempt    int
}

type FailedOperationInputPolicy interface {
    AllowFailedOperationInput(context.Context, FailedOperationIdentity, string, interface{}) (bool, error)
}

func WithFailedOperationView(FailedOperationViewConfig) OrchestratorOption
func ProjectFailedOperationEvidence(context.Context, []StepResult, FailedOperationViewConfig) (map[string]json.RawMessage, error)

WithFailedOperationView replaces the entire view configuration after the environment layer; it does not merge individual fields. Zero numeric fields then resolve to defaults, and an omitted Disabled is false. For example, a code option setting only InputPolicy resets environment-set limits and disabled state. Supply the other desired fields explicitly to retain them.

Example configuration (the types above belong to orchestration):

resolved, err := orchestration.ResolveOrchestratorConfig(orchestration.ConfigResolution{
    Base: orchestration.NewDefaultOrchestratorConfig(),
    Environment: orchestration.EnvironmentStrict,
    Options: []orchestration.OrchestratorOption{
        orchestration.WithFailedOperationView(orchestration.FailedOperationViewConfig{
            MaxValueBytes: 4096, MaxStepBytes: 16384, MaxPromptBytes: 32768,
            // Disabled: true restores legacy prompt rendering.
            // InputPolicy: applicationOwnedRestriction,
        }),
    },
})

Check err before construction. FailedOperationInputPolicy implements AllowFailedOperationInput(context.Context, FailedOperationIdentity, string, interface{}) (bool, error). It may restrict disclosure but cannot replace values. Policy errors omit inputs; implementations must be concurrency-safe and immutable. An injected policy makes semantic config cache identity ineligible. Limits apply to derived views, not raw execution retention. ProjectFailedOperationEvidence(ctx, steps, config) is an explicit application seam for authorized delegation; it returns JSON keyed by failed step ID, with a null value for a budget-omitted view. It does not authorize recipients or change result status. Custom synthesizers opt in separately.

For a directly constructed built-in synthesizer, call (*AISynthesizer).SetFailedOperationView(FailedOperationViewConfig) error and check the returned configuration error. Factory-built orchestrators wire their resolved configuration automatically. Defaults are 4,096 bytes per value, 16,384 per failed step and 32,768 per prompt; a tighter enclosing continuation allocation still wins. These are encoded derived-view budgets, not new raw record limits. See result compaction.

Exact whole-value step templates retain native JSON types. Object/array values cannot silently become string identifiers under a known scalar schema. Select the scalar path explicitly. Semantic auto-wire nested extraction remains supported.

ResultTrimConfig​

Configures automatic structural trimming of large step results before they are embedded in LLM prompts. When tool/agent responses are large (e.g., full web pages, large API payloads), they can overflow the LLM's token budget during synthesis. The StructuralTrimmer intelligently reduces result sizes — preserving JSON structure, key names, and representative samples while dropping repetitive array elements and deeply nested data.

type ResultTrimConfig struct {
    // Enabled controls whether result trimming is active.
    // Default: true | Env: TRUVAG3_RESULT_TRIM_ENABLED
    Enabled bool `json:"enabled"`

    // MaxResultBytes is the maximum bytes per individual step result.
    // Results exceeding this are structurally trimmed.
    // Note: Single-step requests use MaxTotalPromptBytes (default 192 KiB) instead,
    // since there is no multi-result allocation needed.
    // Default: 65536 (64 KiB, ~16K tokens) | Env: TRUVAG3_RESULT_TRIM_MAX_BYTES
    MaxResultBytes int `json:"max_result_bytes"`

    // MaxTotalPromptBytes is the maximum total bytes across all step results
    // in a synthesis prompt. When combined results exceed this limit,
    // BudgetAllocator distributes budget proportionally by result size,
    // then redistributes savings from results clamped by MaxResultBytes
    // to other eligible results. Allocation is distillation-aware: a result the
    // processor will distill is sized at its post-distill footprint (~TargetSize,
    // via EffectiveSizer), not its raw bytes, so a large distillable result
    // doesn't crowd out the others.
    // Default: 196608 (192 KiB, ~48K tokens) | Env: TRUVAG3_RESULT_TRIM_MAX_TOTAL_BYTES
    MaxTotalPromptBytes int `json:"max_total_prompt_bytes"`

    // MaxMicroResolutionBytes controls how much source data is included when
    // the LLM resolves missing parameters from prior step results (micro-resolution)
    // or retries failed steps (semantic retry).
    // Default: 65536 (64 KiB) | Env: TRUVAG3_RESULT_TRIM_MAX_MICRO_BYTES
    MaxMicroResolutionBytes int `json:"max_micro_resolution_bytes"`

    // MaxAgentInputBytes is the maximum bytes per parameter value for
    // agent/tool HTTP calls. Trims large data values before they are sent
    // as input parameters to downstream agents or tools.
    // Default: 0 (no cap — fidelity-first: tool→tool data flows raw so
    // downstream steps receive the full upstream output). Set > 0 (or supply
    // deps.AgentInputProcessor) to enable the byte-budget guard.
    // Env: TRUVAG3_RESULT_TRIM_MAX_AGENT_INPUT_BYTES
    MaxAgentInputBytes int `json:"max_agent_input_bytes"`

    // SchemaGuidedMappingThreshold is the result size (bytes) above which
    // schema-guided mapping is used instead of direct value extraction for
    // micro-resolution. Uses JSON schema analysis to map fields without
    // passing raw data to the LLM. Set to 0 to disable schema-guided mapping.
    // Default: 16384 (16 KB) | Env: TRUVAG3_RESULT_TRIM_SCHEMA_MAPPING_THRESHOLD
    SchemaGuidedMappingThreshold int `json:"schema_guided_mapping_threshold"`

    // PreserveKeys lists JSON keys the trimmer favors keeping (prioritized, not guaranteed).
    PreserveKeys []string `json:"preserve_keys,omitempty"`
}

Example - Default (enabled, suitable for most queries):

config := orchestration.DefaultConfig()
// ResultTrim is already configured with sensible defaults:
// Enabled: true, MaxResultBytes: 65536, MaxTotalPromptBytes: 196608,
// MaxMicroResolutionBytes: 65536, MaxAgentInputBytes: 0 (no cap, fidelity-first),
// SchemaGuidedMappingThreshold: 16384

Example - Larger budgets for data-heavy workflows:

config := orchestration.DefaultConfig()
config.ResultTrim = orchestration.ResultTrimConfig{
    Enabled:                      true,
    MaxResultBytes:               131072, // 128 KiB per result
    MaxTotalPromptBytes:          262144, // 256 KiB total
    MaxMicroResolutionBytes:      131072, // 128 KB for resolution prompts
    MaxAgentInputBytes:           131072, // 128 KB per agent parameter
    SchemaGuidedMappingThreshold: 16384,  // Schema mapping above 16 KB
}

Example - Disable schema-guided mapping:

config := orchestration.DefaultConfig()
config.ResultTrim.SchemaGuidedMappingThreshold = 0 // Always use direct value extraction

Environment variable overrides:

export TRUVAG3_RESULT_TRIM_ENABLED=true
export TRUVAG3_RESULT_TRIM_MAX_BYTES=65536
export TRUVAG3_RESULT_TRIM_MAX_TOTAL_BYTES=196608
export TRUVAG3_RESULT_TRIM_MAX_MICRO_BYTES=65536
export TRUVAG3_RESULT_TRIM_MAX_AGENT_INPUT_BYTES=0   # default 0 = no cap (fidelity-first); set > 0 to enable the guard
export TRUVAG3_RESULT_TRIM_SCHEMA_MAPPING_THRESHOLD=16384

ResultTrimMetadata​

Captures metadata about each result trimming operation. Populated by the ResultProcessor during ProcessForPrompt() and exposed via StepResult.Metadata["result_trim"]. Also emitted as attributes on the result_trim.completed span event.

type ResultTrimMetadata struct {
    OriginalBytes    int      `json:"original_bytes"`              // Pre-trim size in bytes
    TrimmedBytes     int      `json:"trimmed_bytes"`               // Post-trim size in bytes
    Method           string   `json:"method"`                      // "structural", "structural_array", "structural_text", "truncate", "distill", "distill_mapreduce"
    FieldsKept       int      `json:"fields_kept,omitempty"`       // Number of fields/items retained
    FieldsDropped    int      `json:"fields_dropped,omitempty"`    // Number of fields/items dropped
    BackfilledCount  int      `json:"backfilled_count,omitempty"`  // Fields recovered via multi-field backfill
    ThresholdSkipped int      `json:"threshold_skipped,omitempty"` // Candidates skipped below minimum relevance (0.3)
    Keywords         []string `json:"keywords,omitempty"`          // Extracted query keywords from step instruction
    MatchedPaths     []string `json:"matched_paths,omitempty"`     // Field paths selected by keyword match
    BudgetAllocated  int      `json:"budget_allocated,omitempty"`  // Per-result budget from BudgetAllocator (multi-result only)
    Degenerate       bool     `json:"degenerate,omitempty"`        // Structural trim kept a non-representative fraction (< 5%); triggers the "severely reduced … treat as UNKNOWN" disclosure
    KeptRatio        float64  `json:"kept_ratio,omitempty"`        // trimmed_bytes / original_bytes

    // Phase 16/17 — coverage accounting. Byte-based values are APPROXIMATE (JSON is parsed
    // and re-serialized, so byte counts shift); pair them with the exact FieldsKept/FieldsDropped counts.
    SourceCoverageRatio float64 `json:"source_coverage_ratio,omitempty"` // ~ source represented / original (approximate)
    LLMInputBytes       int     `json:"llm_input_bytes,omitempty"`       // total DATA bytes sent across LLM calls (chunk/combine payloads — not the full prompts/system messages); can exceed OriginalBytes once map-reduce wrappers replicate per chunk
    SegmentsAnalyzed    int     `json:"segments_analyzed,omitempty"`     // map-reduce N (single-call: 1)
    SegmentsTotal       int     `json:"segments_total,omitempty"`        // map-reduce M (single-call: 1)
    PartialCoverage     bool    `json:"partial_coverage,omitempty"`      // the LLM did not see the full source: a pre-LLM structural drop (single-call pre-filter) OR incomplete map-reduce coverage (failed/timed-out chunks)
    CombineTruncated    bool    `json:"combine_truncated,omitempty"`     // reduce output was deterministically truncated
    ChunkStrategy       string  `json:"chunk_strategy,omitempty"`        // map-reduce chunk mode: "array" | "wrapper" | "lines" | "bytes" (lines/bytes are byte-lossless but may tear records mid-JSON)

    // ContentLost is the AUTHORITATIVE loss signal, set by every lossy trim operation (field/item/
    // sentence drops, threshold skips, value truncation, byte cuts). Disclosure gating keys on it,
    // never on byte ratios (re-serialization shrinks bytes without losing content). Deliberately
    // NOT omitempty: an explicit false means "verified lossless", distinct from the key being absent
    // on legacy records — the registry viewer renders that tri-state.
    ContentLost bool `json:"content_lost"`
}

Accessing trim metadata programmatically:

// After orchestration completes, inspect per-step trim metadata:
for _, step := range response.Steps {
    if meta, ok := step.Metadata["result_trim"]; ok {
        trimMeta := meta.(*orchestration.ResultTrimMetadata)
        fmt.Printf("Step %s: %d → %d bytes, %d fields kept, %d dropped\n",
            step.StepID, trimMeta.OriginalBytes, trimMeta.TrimmedBytes,
            trimMeta.FieldsKept, trimMeta.FieldsDropped)
    }
}

CaptureResultTrimMetadata​

func CaptureResultTrimMetadata(ctx context.Context, meta ResultTrimMetadata)

The supported reporting hook a custom ResultProcessor calls to record what it trimmed. It is the producer side of the metadata contract: the framework prepares the per-step capture slot before invoking ProcessForPrompt, so a custom processor only calls this with its outcome, and the record surfaces on StepResult.Metadata["result_trim"] and the result_trim.completed span.

Honesty contract. A custom processor that drops, truncates, or samples content must call this with ContentLost: true — downstream disclosure gating keys exclusively on that signal (a lossy processor that skips it makes real loss read as full coverage). The function folds any specific loss flag (Degenerate / PartialCoverage / CombineTruncated) into ContentLost, so a report cannot claim "verified lossless" while flagging a loss. It is a safe no-op outside the framework's capture context. (The framework sets up that context with the companion WithTrimMetadataCapture(ctx) (context.Context, *ResultTrimMetadata), which callers driving ProcessForPrompt by hand use to read the metadata back.)

ResultDistillConfig​

Configures default-on LLM-based result distillation — the primary compaction path for over-budget results. Uses a two-stage pipeline: structural pre-filtering (Stage 1) followed by LLM-based summarization (Stage 2); results whose estimated token count exceeds ModelContextTokens — or, when MapReduceThresholdBytes is set, whose byte size exceeds it — are chunked and map-reduced. Active when the orchestrator has an AIClient and Result Trimming is enabled (both true by default); without an AIClient it falls back to the StructuralTrimmer floor. Opt out with TRUVAG3_RESULT_DISTILL_ENABLED=false. Every distillation is bounded by CompactionDeadline (fail-open). Distillation results are cached only when a DigestCache is supplied via deps.DistillCache — there is no cache by default — and CacheTTL applies only then.

type ResultDistillConfig struct {
    // Enabled controls whether LLM distillation is active.
    // Default: true | Env: TRUVAG3_RESULT_DISTILL_ENABLED
    Enabled bool `json:"enabled"`

    // DistillThreshold is the minimum result size (bytes) to trigger distillation.
    // Results below this threshold use structural trimming only.
    // Default: 16384 (16 KB) | Env: TRUVAG3_RESULT_DISTILL_THRESHOLD
    DistillThreshold int `json:"distill_threshold"`

    // PreFilterBudget is the StructuralTrimmer byte budget applied before
    // LLM distillation (Stage 1 pre-filter). Reduces LLM input size.
    // Default: 131072 (128 KB — fits the 64K fast-tier context floor) | Env: TRUVAG3_RESULT_DISTILL_PREFILTER
    PreFilterBudget int `json:"prefilter_budget"`

    // TargetSize is the target output size for LLM distillation (Stage 2).
    // The LLM summarizes the pre-filtered result to approximately this size.
    // Default: 4096 (4 KB) | Env: TRUVAG3_RESULT_DISTILL_TARGET
    TargetSize int `json:"target_size"`

    // Model overrides the default AI model for distillation calls. Defaults to
    // the "fast" alias, which resolves for Anthropic, OpenAI-compatible, and
    // Gemini providers. Bedrock has no portable-alias catalog; set this empty
    // programmatically for a chain containing Bedrock so each entry uses its
    // configured default. The environment loader ignores an empty value.
    // Default: "fast" | Env: TRUVAG3_RESULT_DISTILL_MODEL
    Model string `json:"model,omitempty"`

    // CacheTTL is how long a distillation result stays cached (keyed by result
    // content + instruction + query + budget) — but only when a DigestCache is
    // supplied via deps.DistillCache; there is no cache by default (a nil cache is
    // a fail-open no-op). CacheTTL=0 does NOT reliably disable caching: a
    // Redis-backed cache treats 0 as "no expiration" (cached indefinitely). To
    // disable caching, don't supply a cache.
    // Go duration; env accepts positive values only.
    // Default: 5m | Env: TRUVAG3_RESULT_DISTILL_CACHE_TTL
    CacheTTL time.Duration `json:"cache_ttl,omitempty"`

    // CompactionDeadline bounds the wall-clock time a single compaction may
    // spend in the synthesis hot path. On timeout it fails open (single-call →
    // structural floor; map-reduce → completed chunks + a "partial" disclosure).
    // The programmatic zero value disables the deadline; the env var accepts
    // positive durations only. Keep it under the HTTP gateway timeout.
    // Default: 45s | Env: TRUVAG3_RESULT_DISTILL_DEADLINE
    CompactionDeadline time.Duration `json:"compaction_deadline,omitempty"`

    // ModelContextTokens is the usable context (tokens) of the compaction model.
    // Results estimated above this are chunked and map-reduced instead of sent
    // in one call (~525 KB at the default, using the ≈3.5 bytes/token counter).
    // Default: 150000 | Env: TRUVAG3_RESULT_DISTILL_CONTEXT_TOKENS
    ModelContextTokens int `json:"model_context_tokens,omitempty"`

    // MapReduceThresholdBytes routes results larger than this (bytes) to map-reduce
    // even when they fit the model context — so the WHOLE result reaches an LLM, not
    // only the pre-filtered head (Phase 17). INDEPENDENT of ModelContextTokens.
    // 0 = disabled (context-only routing, today's default). A positive value below
    // PreFilterBudget is ignored with a factory warning — a result that small is a
    // single chunk, so routing it to map-reduce adds no fan-out over the single-call
    // path (that lone chunk still runs an LLM extract call when it fits context).
    // Parsed with >= 0 (not the repo's usual > 0) so an explicit 0 can disable a
    // future nonzero default.
    // Default: 0 (disabled) | Env: TRUVAG3_RESULT_DISTILL_MAPREDUCE_THRESHOLD
    MapReduceThresholdBytes int `json:"mapreduce_threshold_bytes,omitempty"`

    // MapConcurrency caps how many chunks are compacted concurrently in the
    // map-reduce path. <= 0 falls back to the default.
    // Default: 8 | Env: TRUVAG3_RESULT_DISTILL_MAP_CONCURRENCY
    MapConcurrency int `json:"map_concurrency,omitempty"`
}

Example - Tune distillation (default-on; adjust individual fields):

config := orchestration.DefaultConfig()
// Distillation is already enabled with sensible defaults — mutate individual
// fields rather than reassigning the whole struct, which would zero the fields
// you omit. Most self-heal at construction — PreFilterBudget, DistillThreshold,
// ModelContextTokens, and MapConcurrency all backfill to their defaults when
// non-positive (so ModelContextTokens=0 becomes 150000, NOT "disabled" — there is
// no zero-value opt-out for token-based routing). Two do NOT self-heal:
// CompactionDeadline=0 disables the hot-path deadline, and CacheTTL=0 does NOT
// cleanly disable caching (a Redis-backed cache treats 0 as no expiration — see
// the CacheTTL note above).
config.ResultDistill.DistillThreshold = 65536 // only distill results > 64 KB
config.ResultDistill.TargetSize = 2048        // tighter ~2 KB summaries
config.ResultDistill.Model = "fast"           // portable across catalog-backed providers
// To opt out entirely:
// config.ResultDistill.Enabled = false

Environment variable overrides (values shown are the defaults):

export TRUVAG3_RESULT_DISTILL_ENABLED=true            # default-on; =false to opt out
export TRUVAG3_RESULT_DISTILL_THRESHOLD=16384         # min bytes to trigger (default 16 KB)
export TRUVAG3_RESULT_DISTILL_PREFILTER=131072        # Stage-1 pre-filter budget (default 128 KB)
export TRUVAG3_RESULT_DISTILL_TARGET=4096             # Stage-2 LLM output target (default 4 KB)
export TRUVAG3_RESULT_DISTILL_MODEL=fast              # catalog-backed providers; Bedrock needs a programmatic empty override
export TRUVAG3_RESULT_DISTILL_CACHE_TTL=5m            # distillation cache TTL
export TRUVAG3_RESULT_DISTILL_DEADLINE=45s            # hot-path compaction bound (positive durations only)
export TRUVAG3_RESULT_DISTILL_CONTEXT_TOKENS=150000   # above this (tokens), chunk → map-reduce
export TRUVAG3_RESULT_DISTILL_MAPREDUCE_THRESHOLD=0   # bytes; 0=disabled; above this, route to map-reduce even if it fits context (accepts 0 as a kill switch)
export TRUVAG3_RESULT_DISTILL_MAP_CONCURRENCY=8       # concurrent chunk distillations

How the pipeline works:

  1. Stage 1 — Structural Pre-Filter: The StructuralTrimmer reduces the large result to PreFilterBudget bytes, preserving JSON structure and key data points.
  2. Stage 2 — LLM Distillation: An LLM call summarizes the pre-filtered result to approximately TargetSize bytes, preserving the most relevant information for the user's query.
  3. Very large results — map-reduce: when the result is estimated above ModelContextTokens (or exceeds MapReduceThresholdBytes bytes, when that opt-in knob is set — independent of the token estimate), it is chunked and the chunks are compacted concurrently (MapConcurrency at a time), then reduced — instead of being sent in a single call. This routes the whole result through an LLM rather than only its Stage-1 pre-filtered head.
  4. Bounded & (optionally) cached: each compaction is bounded by CompactionDeadline (fail-open). Results are cached for CacheTTL only when a DigestCache is supplied via deps.DistillCache — there is no cache by default.
  5. Fallback: if the LLM call fails (or the deadline trips), the Stage 1 pre-filtered structural result is used directly.

DeserializeStringValues​

func DeserializeStringValues(data interface{}) interface{}

Recursively walks a decoded JSON value (map[string]interface{} / []interface{} / string) and, wherever a string field itself contains JSON (a value starting with { or [ — e.g. a Loki log line that carries JSON as a string), re-parses that inner JSON so downstream code sees structured data instead of an opaque string. Exported for edge-case use by tool handlers that post-process results.

Compatibility change (Phase 17). The inner re-parse now decodes with UseNumber, so numbers inside those decoded JSON-in-string fields are returned as json.Number, not float64. This preserves large integer identifiers (snowflake IDs, 64-bit keys) that a float64 round-trip would corrupt. Callers that type-assert float64 on such nested numbers must instead handle json.Number (use .Int64() / .Float64() / .String()). Only numbers that were nested inside a JSON-in-string field are affected — numbers already present in the top-level decoded structure pass through unchanged.

ProcessRequestStreaming​

Process requests with real-time streaming responses. Tokens are delivered as they're generated, enabling SSE/WebSocket chat interfaces and real-time UX.

func (o *AIOrchestrator) ProcessRequestStreaming(
    ctx context.Context,
    query string,
    tools []*ServiceInfo,
    callback core.StreamCallback,
) (*StreamingOrchestratorResponse, error)

Native delivery requires the streaming interface and SupportsStreaming() to return true. Otherwise orchestration uses simulated delivery and emits the bounded streaming_fallback diagnostic. Simulated chunks are split on UTF-8 rune boundaries, so a callback never receives half of a multibyte character.

If native synthesis returns partial output with core.ErrStreamPartiallyCompleted, including a wrapped/joined error, orchestration returns that output without a retry: PartialContent=true, StreamCompleted=false, and a nil Go error. The response's Errors retains the exact interruption error. Delivery FinishReason is cancelled for context.Canceled, deadline_exceeded for context.DeadlineExceeded, otherwise interrupted. The execution and synthesis debug interaction remain unsuccessful; the provider's reported finish reason is preserved separately in the debug interaction. A partial sentinel without an AI response follows the ordinary error-return path.

StreamingOrchestratorResponse fields:

type StreamingOrchestratorResponse struct {
    OrchestratorResponse              // Embedded base response (see below)

    // Streaming-specific fields
    ChunksDelivered int               // Number of chunks delivered via callback
    StreamCompleted bool              // Whether streaming completed successfully
    PartialContent  bool              // True if response is partial due to error
    StepResults     []StepResult      // Detailed results from each execution step
    FinishReason    string            // Why streaming stopped ("stop", "length", "error")
}

type OrchestratorResponse struct {
    RequestID       string                    // Unique request identifier
    OriginalRequest string                    // The original user query
    Response        string                    // Complete response (accumulated from chunks)
    RoutingMode     RouterMode                // Routing mode used
    ExecutionTime   time.Duration             // Total execution time
    AgentsInvolved  []string                  // Tools/agents used during execution
    Metadata        map[string]interface{}    // Additional metadata
    Errors          []string                  // Any errors encountered
    Confidence      float64                   // Orchestration confidence score
    Steps           []StepResult              // Individual step results

    // Aggregated token usage across all LLM calls in this request
    Usage           *core.TokenUsage          // Total prompt + completion tokens
    UsageByPhase    map[string]core.TokenUsage // Breakdown by orchestration phase
}

Phase names in UsageByPhase: planning, correction, synthesis, micro_resolution, schema_mapping, distillation, semantic_retry, error_analysis, tiered_selection.

Example - SSE Chat Endpoint:

func (h *ChatHandler) HandleStream(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()
    flusher := w.(http.Flusher)

    // Set SSE headers
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")

    var req ChatRequest
    json.NewDecoder(r.Body).Decode(&req)

    // Stream orchestration results
    result, err := h.orchestrator.ProcessRequestStreaming(ctx, req.Message, nil,
        func(chunk core.StreamChunk) error {
            if chunk.Content != "" {
                fmt.Fprintf(w, "data: %s\n\n", chunk.Content)
                flusher.Flush()
            }
            return nil
        },
    )

    if err != nil {
        fmt.Fprint(w, "event: error\ndata: request processing failed\n\n")
        return
    }

    // Send completion event
    fmt.Fprintf(w, "event: done\ndata: {\"request_id\": \"%s\"}\n\n", result.RequestID)
}

Example - With Step Callbacks:

// Add per-request step callback for real-time tool progress
ctx = orchestration.WithStepCallback(ctx,
    func(stepIndex, totalSteps int, step orchestration.RoutingStep, result orchestration.StepResult) {
        // Send SSE event for each tool completion
        callback.SendStep(step.AgentName, result.Success, result.Duration.Milliseconds())
    },
)

result, err := orchestrator.ProcessRequestStreaming(ctx, query, nil,
    func(chunk core.StreamChunk) error {
        callback.SendChunk(chunk.Content)
        return nil
    },
)

For a complete implementation with conversation history, session management, and SSE events, see the Chat Agent Implementation Guide.

Orchestration Strategies​

Different strategies for different scales and use cases:

StrategyUse CaseScale
AutonomousAI decides which services to useSmall-medium
DirectedExplicit service specificationAny scale
HybridMix of AI and rulesMedium-large

Example - Scaling Orchestration:

// Small scale (< 20 tools) - Let AI figure it out
config := &orchestration.OrchestratorConfig{
    RoutingMode:       orchestration.ModeAutonomous,
    SynthesisStrategy: orchestration.StrategyLLM,
}

// Medium scale (20-100 tools) - Use tiered resolution (default)
// Reduces token usage by 50-75% via two-phase tool selection
config := &orchestration.OrchestratorConfig{
    RoutingMode:            orchestration.ModeAutonomous,
    EnableTieredResolution: true,  // Default: true
    TieredResolution: orchestration.TieredCapabilityConfig{
        MinToolsForTiering: 20,    // Trigger threshold
        SelectionMaxTokens: 2000,  // Selection output tokens
    },
    PromptConfig: orchestration.PromptConfig{
        SystemInstructions: "You are a helpful assistant specializing in travel planning.",
    },
}

// Large scale (100+ services) - Use capability service
config := &orchestration.OrchestratorConfig{
    RoutingMode:            orchestration.ModeHybrid,
    CapabilityProviderType: "service",
    CapabilityServiceURL:   "http://capability-registry:8080",
}

// Capability service indexes all available capabilities
// and provides fast, structured search without hitting discovery

OAuth Bearer Token Propagation​

The orchestrator propagates OAuth Bearer tokens on all outbound HTTP calls to tool/agent endpoints. Two mechanisms are supported:

OrchestratorConfig field (Scenario 2: machine-to-machine):

config := &orchestration.OrchestratorConfig{
    OAuthToken: "eyJhbGciOiJSUzI1NiIs...",  // or via TRUVAG3_OAUTH_TOKEN env var
}

Context helpers (Scenario 1: user token pass-through):

// Attach per-request Bearer token
ctx = orchestration.WithOAuthToken(ctx, token)

// Retrieve token from context
token := orchestration.GetOAuthToken(ctx)

Resolution order: context token > config token > none. When set, outbound requests include Authorization: Bearer <token>.

Custom Header Propagation​

The orchestrator supports injecting custom headers into all outbound HTTP calls to tool/agent endpoints. This is useful for correlation IDs, tenant routing, and audit logging across non-OTel services.

OrchestratorConfig field (instance-level defaults):

config := &orchestration.OrchestratorConfig{
    PropagatedHeaders: map[string]string{
        "X-Tenant-ID":      tenantID,
        "X-Correlation-ID": correlationID,
    },
}

Context helpers (per-request overrides):

// Attach multiple headers to context
ctx = orchestration.WithPropagatedHeaders(ctx, map[string]string{
    "X-Tenant-ID":      r.Header.Get("X-Tenant-ID"),
    "X-Correlation-ID": r.Header.Get("X-Correlation-ID"),
})

// Add a single header to context
ctx = orchestration.AddPropagatedHeader(ctx, "X-Request-Source", "chat-ui")

// Retrieve headers from context
headers := orchestration.GetPropagatedHeaders(ctx)

Runtime setter (update without restart):

orch.SetPropagatedHeaders(map[string]string{"X-Tenant-ID": newTenantID})

Resolution order: context headers > config headers > none. Context headers override config headers on key conflict; config-only keys are preserved.

Reserved header protection: Authorization, Content-Type, and X-TruvaG3-* headers cannot be overridden by propagated headers. OAuth is managed by dedicated logic, Content-Type is set by the framework, and framework headers (X-TruvaG3-Request-ID, X-TruvaG3-Step-ID) are set last.

No env var is provided for PropagatedHeaders (maps don't map to a single env var). Set programmatically via config or SetPropagatedHeaders().

LLM Debug Store​

Captures complete LLM request/response payloads for production debugging. Unlike Jaeger spans which truncate large payloads, the debug store preserves full prompts and responses.

LLMDebugStore Interface​

type LLMDebugStore interface {
    // RecordInteraction appends an LLM interaction to the debug record
    RecordInteraction(ctx context.Context, requestID string, interaction LLMInteraction) error

    // GetRecord retrieves the complete debug record for a request
    GetRecord(ctx context.Context, requestID string) (*LLMDebugRecord, error)

    // SetMetadata adds metadata to an existing record (for investigation notes)
    SetMetadata(ctx context.Context, requestID string, key, value string) error

    // ExtendTTL extends retention for important records
    ExtendTTL(ctx context.Context, requestID string, duration time.Duration) error

    // ListRecent returns recent records for UI listing (newest first)
    ListRecent(ctx context.Context, limit int) ([]LLMDebugRecordSummary, error)
}

// Optional capability used by final execution recording. It establishes a
// minimum lifetime for existing and future writes without creating an empty
// debug record.
type LLMDebugRetentionPreserver interface {
    PreserveRetention(ctx context.Context, requestID string, duration time.Duration) error
}

RedisLLMDebugStore implements LLMDebugRetentionPreserver with a small request-scoped retention-floor key. Both the orchestration writer and telemetry.RedisLLMCallRecorder consult it atomically, so late or cross-process writes inherit the final execution lifetime. Custom stores that do not implement the optional capability continue through ExtendTTL.

LLMInteraction Type​

type LLMInteraction struct {
    Type              string                     `json:"type"`
    HookPhase         string                     `json:"hook_phase,omitempty"`
    Timestamp         time.Time                  `json:"timestamp"`
    DurationMs        int64                      `json:"duration_ms"`
    Prompt            string                     `json:"prompt"`
    SystemPrompt      string                     `json:"system_prompt,omitempty"`
    Temperature       float64                    `json:"temperature"`
    MaxTokens         int                        `json:"max_tokens"`
    Model             string                     `json:"model,omitempty"`
    Provider          string                     `json:"provider,omitempty"`
    RequestedModel    string                     `json:"requested_model,omitempty"`
    EffectiveModel    string                     `json:"effective_model,omitempty"`
    Adjustments       []core.AIRequestAdjustment `json:"adjustments,omitempty"`
    PolicyFingerprint string                     `json:"policy_fingerprint,omitempty"`
    PolicyStable      bool                       `json:"policy_stable"`
    Response          string                     `json:"response"`
    ReasoningContent  string                     `json:"reasoning_content,omitempty"`
    FinishReason      string                     `json:"finish_reason,omitempty"`
    ContentSource     core.AIContentSource        `json:"content_source,omitempty"`
    PromptTokens      int                        `json:"prompt_tokens"`
    CompletionTokens  int                        `json:"completion_tokens"`
    TotalTokens       int                        `json:"total_tokens"`
    Success           bool                       `json:"success"`
    Error             string                     `json:"error,omitempty"`
    Attempt           int                        `json:"attempt"`
    StepID            string                     `json:"step_id,omitempty"`
    SourceComponent   string                     `json:"source_component,omitempty"`
    CallDescription   string                     `json:"call_description,omitempty"`
    PhaseNumber       int                        `json:"phase_number,omitempty"`
    Category          string                     `json:"category,omitempty"`
    PrecedenceAudit   *PrecedenceAudit           `json:"precedence_audit,omitempty"`
}

PhaseNumber is the producing phase when provided; zero is omitted. HookPhase classifies hook-local interactions (pre/post), and Category distinguishes model calls from embedding, vector-database, storage or logic observations. Empty category retains the legacy model-call rendering default. Completion facts describe provider output, not application success; unavailable facts remain empty. PrecedenceAudit is optional prompt-enrichment audit data, not a provider completion verdict.

Factory Options and Direct Redis Constructors​

// Environment-compatible orchestrator options
orchestration.WithLLMDebug(true)                    // Enable debug capture
orchestration.WithLLMDebugStore(customStore)        // Inject custom store
orchestration.WithLLMDebugTTL(48 * time.Hour)       // Custom TTL for success
orchestration.WithLLMDebugErrorTTL(14 * 24 * time.Hour) // Custom TTL for errors

// Direct Redis constructor options
keyspace, _ := core.NewRedisKeyspace("team-a")
orchestration.WithDebugRedisURL("redis://localhost:6379")
orchestration.WithDebugLogger(logger)
orchestration.WithDebugCircuitBreaker(cb)
orchestration.WithDebugTTL(24 * time.Hour)
orchestration.WithDebugErrorTTL(168 * time.Hour)
orchestration.WithDebugKeyspace(keyspace)

NewRedisLLMDebugStore creates and owns its Redis client. For application-owned connection routing, use:

func NewRedisLLMDebugStoreWithClient(
    client redis.UniversalClient,
    opts ...RedisLLMDebugStoreOption,
) (*RedisLLMDebugStore, error)

The supplied client remains open when the store is closed. The WithClient constructor does not read Redis connection or database environment variables; configure DB-0 routing on the supplied client and use WithDebugKeyspace for key isolation. WithDebugRedisURL does not reconfigure an already-supplied client.

New applications should normally construct the included adapter through the Redis preset and inject the provider-neutral store. Direct Redis constructors remain available for explicit adapter composition.

Example - Enable Debug Capture:

ownedBackends, err := redisprovider.NewDefaultBackends(
    logger,
    redisprovider.WithDefaultBackendRoles(redisprovider.ClientRoleLLMDebug),
)
if err != nil {
    return err
}
defer ownedBackends.Close()

config := orchestration.NewDefaultOrchestratorConfig()
config.LLMDebug.Enabled = true
if err := orchestration.WireOrchestratorBackends(config, ownedBackends.Backends()); err != nil {
    return err
}
deps := orchestration.OrchestratorDependencies{
    Discovery: discovery,
    AIClient:  aiClient,
    Logger:    logger,
}
orchestrator, err := orchestration.CreateResolvedOrchestrator(config, deps)

// Query debug records (from your application's API)
record, _ := orchestrator.GetLLMDebugStore().GetRecord(ctx, requestID)
for _, interaction := range record.Interactions {
    fmt.Printf("Site: %s, Model: %s, Provider: %s\n",
        interaction.Type, interaction.Model, interaction.Provider)
    fmt.Printf("Prompt: %s\n", interaction.Prompt[:100])  // First 100 chars
}

LLMCallRecorderAdapter​

Bridges orchestration.LLMDebugStore to telemetry.LLMCallRecorder. Use this when the orchestrator already has a debug store and you want agents to write to the same store without creating a separate Redis connection.

func NewLLMCallRecorderAdapter(store LLMDebugStore) telemetry.LLMCallRecorder

Example:

debugStore, _ := orchestration.NewRedisLLMDebugStore()
recorder := orchestration.NewLLMCallRecorderAdapter(debugStore)
aiClient := ai.NewInstrumentedClient(baseClient, recorder)

LLMDebugRecordSummary​

Lightweight summary for listing records. Includes source attribution.

type LLMDebugRecordSummary struct {
    RequestID         string    `json:"request_id"`
    OriginalRequestID string    `json:"original_request_id,omitempty"`
    TraceID           string    `json:"trace_id"`
    CreatedAt         time.Time `json:"created_at"`
    InteractionCount  int       `json:"interaction_count"`
    TotalTokens       int       `json:"total_tokens"`
    HasErrors         bool      `json:"has_errors"`
    // SourceComponents lists unique agent names that made LLM calls in this record.
    // Empty for orchestrator-only records.
    SourceComponents []string   `json:"source_components,omitempty"`
}

Environment Variables:

VariableDefaultDescription
TRUVAG3_LLM_DEBUG_ENABLEDfalseEnable debug capture
TRUVAG3_LLM_DEBUG_TTL72hBase TTL for successful records; longer lineage floors are preserved
TRUVAG3_LLM_DEBUG_ERROR_TTL72hBase TTL for error records (3 days); HITL or investigation retention may extend it

Execution Debug Store​

Execution debug records and list summaries expose optional ExecutionLifecycleObservation (state, request-local revision, observed_at) and record_revision. Managed requests transition running → completed / failed / cancelled / suspended after delivery. These fields are independent of step success and domain outcome; omitted lifecycle on legacy/direct-plan records means unknown. Late hook effects advance the record revision without reverting phases.

Step dispatch evidence and request lifecycle​

orchestration.StepResult distinguishes retry-budget attempts from physical component invocations. Relevant fields (other existing fields omitted):

type StepResult struct {
    Success       bool                   `json:"success"`
    Error         string                 `json:"error,omitempty"`
    Attempts      int                    `json:"attempts"`
    DispatchCount int                    `json:"dispatch_count"`
    Dispatches    []StepDispatch         `json:"dispatches,omitempty"`
    ToolError     json.RawMessage        `json:"tool_error,omitempty"`
    Parameters    map[string]interface{} `json:"parameters,omitempty"`
    StartTime     time.Time              `json:"start_time"`
    EndTime       time.Time              `json:"end_time"`
    Skipped       bool                   `json:"skipped,omitempty"`
    SkipReason    string                 `json:"skip_reason,omitempty"`
    Metadata      map[string]interface{} `json:"metadata,omitempty"`
}

type StepDispatch struct {
    Number     int                    `json:"number"`
    Parameters map[string]interface{} `json:"parameters"`
    Response   string                 `json:"response"`
    Error      string                 `json:"error,omitempty"`
    Success    bool                   `json:"success"`
}

type ExecutionLifecycleObservation struct {
    State      string    `json:"state"`
    Revision   uint64    `json:"revision"`
    ObservedAt time.Time `json:"observed_at"`
}

const (
    RequestLifecycleRunning   = "running"
    RequestLifecycleCompleted = "completed"
    RequestLifecycleFailed    = "failed"
    RequestLifecycleCancelled = "cancelled"
    RequestLifecycleSuspended = "suspended"
)

DispatchCount includes corrected-input dispatches that the historical Attempts counter does not count separately. Dispatches retains actual inputs and original response/error per invocation; it does not infer sanitization. ToolError retains the structured terminal component error, including whether retry permission was absent. Terminal success clears earlier terminal error fields, while dispatch history remains. Existing StepResult.Error keeps its legacy text transformation; it is not an exact copy of every raw response. Protect raw records with the configured store's access/retention policies.

Structured retryable:false stops all same-step retries on every HTTP status. Omission is distinct from false (the Go ToolError bool always serializes). Transport/408/429/5xx failures use unchanged admitted parameters; only eligible non-transient 4xx can reach correction. A non-transient 4xx is not resent unchanged, and exact-bound/HITL-approved values stay protected. Ordinary retry settings do not give delegated orchestrator calls more than one dispatch.

Dependency skips have Skipped=true, Success=false, no dispatch, and planned inputs in Metadata.planned_parameters. Refinement-satisfied skips have Success=true. Neither represents a new action for episodic recording. RoutingStep.ImplicitDeps represents referenced prior-phase IDs, participates in failure propagation, and does not schedule same-phase work. Fresh generated plans derive this field; same-phase depends_on remains validated planner input.

Managed request segments observe running revision 1 and terminal revision 2. Caller cancellation is cancelled; a deadline/error or partial-content delivery is failed; HITL interruption is suspended. A completed pipeline does not prove a positive domain outcome or that every step succeeded. StoredExecution and ExecutionSummary expose Lifecycle and RecordRevision; record revision also advances on late effects without reverting the latest snapshot. Legacy/direct-plan omissions and zero record revision mean unknown, not failure. Resume uses a new linked request ID. MetadataKeyExecutionRequestID (execution_request_id) identifies the owning segment on terminal execution results; legacy records may omit it.

Before each HTTP dispatch, including retries and corrected inputs, validateDispatchParameters rejects unresolved reserved references, absent required keys, declared enum violations and known object/array/scalar container mismatches. This is not full JSON Schema validation: explicit nulls remain application data, scalar coercion retains its existing behavior, and unknown custom types are not rejected solely by this guard. Exact bindings have an additional integrity check before the dispatch loop.

Component retry permission is read from the structured wire envelope only when top-level success is JSON false and error is an object containing boolean retryable. Omitted, null, misplaced or non-boolean declarations mean unspecified, not false. For example:

{"success": false, "error": {"code": "INVALID_INPUT", "retryable": false}}

A non-Go tool returning only {"error":{"retryable":false}}, or a top-level retryable field, does not declare retry permission to this parser. HTTP classification and immutable-input guards still apply when policy is unspecified; a true flag does not authorize a retry that those guards forbid.

ExecutionResult.Metadata["termination_reason"] records a phase-loop reason when available, such as repeated_failed_dependencies or plan_validation_exhausted. It is not a StepResult field or a domain-success verdict. Request-completion spans/logs also use termination_reason, but their values describe the request boundary (completed, clarification, forced_terminal, partial, or a failure/interruption cause) and can differ from the more specific result-metadata reason. Legacy records can omit both.

Persistence contract​

ExecutionStore remains the required request-oriented persistence contract. Conversation lookup is additive capability discovery:

type StoredExecution struct {
    // Existing correlation, request, plan, result, HITL, phase, skill, and
    // metadata fields are omitted here for brevity.

    // PipelineHooks is the ordered, request-local hook invocation record.
    // It is provider-neutral and does not require a tracing backend.
    PipelineHooks []PipelineHookExecution `json:"pipeline_hooks,omitempty"`

    // Request outcome is independent of step outcomes. Legacy records may
    // omit it; absence is unknown, not inferred success/failure.
    Lifecycle *ExecutionLifecycleObservation `json:"lifecycle,omitempty"`
    RecordRevision uint64 `json:"record_revision,omitempty"`

    // FinalResponse is the terminal application response captured after
    // AfterSynthesis hooks or an accepted BeforePlanning short-circuit.
    FinalResponse       *string `json:"final_response,omitempty"`
    FinalResponseSource string  `json:"final_response_source,omitempty"`
}

type PipelineHookExecution struct {
    HookName  string                      `json:"hook_name"`
    Phase     string                      `json:"phase"`
    Status    PipelineHookExecutionStatus `json:"status"`
    Sequence  int                         `json:"sequence"`
    PlanPhase int                         `json:"plan_phase,omitempty"`
    StartedAt time.Time                   `json:"started_at"`
    Duration  time.Duration               `json:"duration"`
    Error     string                      `json:"error,omitempty"`
    Effects   []core.PipelineHookEffect   `json:"effects,omitempty"`
    Decision  *PipelineHookDecision       `json:"decision,omitempty"`
}

// Orchestration-owned boundary semantics, distinct from invocation/effect status.
type PipelineHookFailurePolicy string

const (
    PipelineHookFailOpen   PipelineHookFailurePolicy = "fail_open"
    PipelineHookFailClosed PipelineHookFailurePolicy = "fail_closed"
)

type PipelineHookAction string

const (
    PipelineHookContinue       PipelineHookAction = "continue"
    PipelineHookTerminate      PipelineHookAction = "terminate"
    PipelineHookShortCircuit   PipelineHookAction = "short_circuit"
    PipelineHookPropagatePanic PipelineHookAction = "propagate_panic"
)

type PipelineHookDecision struct {
    FailurePolicy PipelineHookFailurePolicy `json:"failure_policy"`
    Action        PipelineHookAction        `json:"action"`
    Reason        string                    `json:"reason"`
}

type PipelineHookExecutionStatus string

const (
    PipelineHookSucceeded PipelineHookExecutionStatus = "succeeded"
    PipelineHookFailed    PipelineHookExecutionStatus = "failed"
    PipelineHookSkipped   PipelineHookExecutionStatus = "skipped"
)

const (
    PipelineHookPhaseBeforePlanning = "before_planning"
    PipelineHookPhaseAfterPlanning  = "after_planning"
    PipelineHookPhaseAfterExecution = "after_execution"
    PipelineHookPhaseAfterSynthesis = "after_synthesis"
)

const (
    FinalResponseSourceAfterSynthesisHooks = "after_synthesis_hooks"
    FinalResponseSourceBeforePlanningShortCircuit = "before_planning_short_circuit"
)

type PipelineHookEffect struct {
    EffectID      string                   `json:"effect_id"`
    SchemaVersion int                      `json:"schema_version"`
    Name          string                   `json:"name"`
    Status        PipelineHookEffectStatus `json:"status"`
    Summary       string                   `json:"summary,omitempty"`
    Data          json.RawMessage          `json:"data,omitempty"`
    Error         string                   `json:"error,omitempty"`
    StartedAt     time.Time                `json:"started_at,omitempty"`
    Duration      time.Duration            `json:"duration,omitempty"`
}

type PipelineHookEffectStatus string

const (
    PipelineHookEffectPending   PipelineHookEffectStatus = "pending"
    PipelineHookEffectSucceeded PipelineHookEffectStatus = "succeeded"
    PipelineHookEffectPartial   PipelineHookEffectStatus = "partial"
    PipelineHookEffectFailed    PipelineHookEffectStatus = "failed"
    PipelineHookEffectSkipped   PipelineHookEffectStatus = "skipped"
)

var ErrInvalidPipelineHookEffect = errors.New("invalid pipeline hook effect")

Sequence is one-based among hooks eligible for the same lifecycle phase. PlanPhase is populated for after_planning invocations so an iterative plan can place the hook without timestamp inference. Duration uses Go's time.Duration JSON representation (nanoseconds). Error preserves the exact hook or validation failure because execution-debug persistence is an explicit, fidelity-preserving opt-in. The field is carried by the existing ExecutionStore.Store contract, so custom providers receive the same record without any Redis-specific method or key.

Status is the outcome of the framework invocation. It does not assert that all hook-initiated effects completed: hooks may fail open or schedule work that finishes after the invocation returns.

Decision records what the framework did at a completed hook boundary. FailurePolicy is fail_open or fail_closed; Action is continue, terminate, short_circuit, or propagate_panic; Reason is a bounded framework classification such as hook_error, accepted, or cache_dimension_mismatch. Required after-planning success is fail_closed / continue; optional callback failure is fail_open / continue. Rejected cache provenance is fail_closed / continue because the cached response is withheld while planning may continue. A nil decision means not recorded, not inferred success. propagate_panic records a panic crossing the synchronous hook boundary: invocation status is failed, reason is panic, and the ordinary boundary policy is retained. The framework closes the hook span, preserves known prior work in the failed request record, and re-throws the original panic value. Independent goroutine panics and process-exit durability are outside this contract. Normal-return capture does not emit it. Existing hook spans carry the same three fields as pipeline.hook.* attributes, independently of debug-store enablement.

Effects separately records those concrete outcomes. Effect IDs are stable within one invocation and carry a producer-owned schema version. A hook must announce a new pending effect before its invocation returns; detached work may then update that same ID to a terminal state. Hooks call core.ReportPipelineHookEffect; orchestration automatically captures exact BeforePlanning enrichment changes. Data and Error are exact execution-debug evidence and are not redacted, truncated, or summarized by the framework. Status is producer-reported, not independent proof of durable backend persistence; this distinction matters for documented fail-open memory providers that can return a neutral value after logging an internal runtime failure.

Terminal effect rewrites apply to successful requests, failed requests, partial native streams, and HITL suspensions. Request-local revisions are admitted monotonically before the existing execution recorder serializes ExecutionStore.Store calls, preventing an older concurrent effect snapshot from overwriting a newer one without adding a provider-specific API.

The LLM debug store retains the model's synthesis output. That output is a raw LLM draft, not proof of what the application ultimately returned. When the normal buffered or native-streaming ProcessRequest path reaches its terminal response, the execution recorder stores the post-AfterSynthesis value in FinalResponse and labels its source with FinalResponseSourceAfterSynthesisHooks. Registry Viewer displays that value under Post-Execution and keeps the synthesis interaction under LLM Calls. On native streaming, tokens were already emitted before AfterSynthesis ran; the stored value is the post-hook response object and may differ from text the client already received.

An accepted BeforePlanning short-circuit stores a successful zero-step result and the exact response with FinalResponseSourceBeforePlanningShortCircuit. No model plan is required for that execution to appear as a successful terminal record.

ExecutePlanWithSynthesis is workflow mode: it executes a supplied plan and performs synthesis, but it does not run application pipeline hooks and does not currently store FinalResponse. Consumers must therefore treat its synthesis record as model output, not as post-hook application evidence. Historical records written before these fields existed also legitimately omit them.

type ConversationExecutionLister interface {
    ListByConversationID(
        ctx context.Context,
        conversationID string,
        limit int,
    ) ([]ExecutionSummary, error)
}

type IndexTTLManager interface {
    ExtendIndexTTL(
        ctx context.Context,
        indexKey string,
        minTTL time.Duration,
    ) error
}

type KeyTTLManager interface {
    // ExtendKeyTTL preserves an existing key for at least minTTL. It does not
    // create a missing key, shorten a longer TTL, or expire a persistent key.
    ExtendKeyTTL(ctx context.Context, key string, minTTL time.Duration) error

    // SetKeyWithMinimumTTL writes value atomically while preserving the larger
    // of the previous remaining TTL and minTTL. Persistent keys stay persistent.
    SetKeyWithMinimumTTL(
        ctx context.Context,
        key string,
        value string,
        minTTL time.Duration,
    ) error
}

type ExecutionStorageProvider interface {
    StorageProvider
    KeyTTLManager
}

Use lister, ok := store.(ConversationExecutionLister); the NoOp store does not advertise the capability. Provider-backed stores may implement IndexTTLManager, but its absence is supported through bounded lazy index cleanup. NewExecutionStoreWithProvider and WithExecutionStoreProvider require ExecutionStorageProvider, making atomic lineage promotion and TTL-preserving content rewrites compile-time provider requirements.

An execution result with Success=false or a request lifecycle of failed/cancelled selects error retention, even if every stored step succeeded or no result is available. The request lifecycle can fail after successful tools, for example during synthesis. An existing longer or persistent lifetime is preserved. Interrupted executions receive at least the maximum of normal retention, error retention, and the checkpoint's remaining approval lifetime. Within the execution store, storing a related execution promotes its existing root execution, trace mapping, and conversation index to the selected minimum lifetime. Final execution recording separately promotes current/root LLM-debug evidence through LLMDebugRetentionPreserver, with LLMDebugStore.ExtendTTL used as a compatibility fallback. ExecutionStore.ExtendTTL follows retained root links recursively with cycle protection. Normal extension reads a small retention-link projection instead of decoding the full execution payload, and direct Redis performs the related key updates inside its retry/circuit-breaker boundary. A missing requested execution returns ErrExecutionRecordNotFound. If the requested execution exists but a later ancestor has expired, the available retained chain is extended successfully and traversal stops without recreating the ancestor.

StoredExecution.Metadata and ExecutionSummary.Metadata carry the framework-owned MetadataConversationID. Read it with ExecutionConversationID or ExecutionSummaryConversationID. Record metadata is authoritative; the SHA-256 conversation index is an accelerator.

type ExecutionStoreConfig struct {
    Enabled                    bool
    TTL                        time.Duration
    ErrorTTL                   time.Duration
    KeyPrefix                  string
    ConversationQueryLimit     int // default 1000
    ConversationIndexScanLimit int // default 5000
}

Non-positive TTL and ErrorTTL values in programmatic configuration are normalized to the standard 24-hour and 7-day defaults. They do not disable persistence or produce an invalid Redis expiry.

The included direct Redis adapter exposes both store-owned and application-owned construction paths:

func NewRedisExecutionDebugStore(
    opts ...RedisExecutionDebugStoreOption,
) (*RedisExecutionDebugStore, error)

func NewRedisExecutionDebugStoreWithConfig(
    config ExecutionStoreConfig,
    opts ...RedisExecutionDebugStoreOption,
) (*RedisExecutionDebugStore, error)

func NewRedisExecutionDebugStoreWithClient(
    client redis.UniversalClient,
    config ExecutionStoreConfig,
    opts ...RedisExecutionDebugStoreOption,
) (*RedisExecutionDebugStore, error)

func WithExecutionDebugRedisURL(url string) RedisExecutionDebugStoreOption
func WithExecutionDebugLogger(logger core.Logger) RedisExecutionDebugStoreOption
func WithExecutionDebugCircuitBreaker(cb core.CircuitBreaker) RedisExecutionDebugStoreOption
func WithExecutionDebugKeyspace(keyspace core.RedisKeyspace) RedisExecutionDebugStoreOption
func WithExecutionDebugTTL(ttl time.Duration) RedisExecutionDebugStoreOption
func WithExecutionDebugErrorTTL(ttl time.Duration) RedisExecutionDebugStoreOption

The first two constructors create and own their Redis clients. The WithClient form leaves the application-owned client open on Close and does not read Redis connection or database environment variables. config supplies the normalized execution-store limits and retention policy; explicit adapter options apply afterward. Redis key isolation uses WithExecutionDebugKeyspace; a custom ExecutionStoreConfig.KeyPrefix is only for generic StorageProvider keys and is rejected by this adapter unless an explicit typed keyspace overrides it. WithExecutionDebugRedisURL does not reconfigure an already-supplied client, which must use DB 0.

Positive programmatic values are authoritative. Per-call result limits are clamped to ConversationQueryLimit, and stale-index work is bounded by ConversationIndexScanLimit. After explicit adapter options are applied, non-positive normal or error TTLs are normalized to the framework defaults.

VariableDefault
TRUVAG3_EXECUTION_DEBUG_CONVERSATION_QUERY_LIMIT1000
TRUVAG3_EXECUTION_DEBUG_INDEX_SCAN_LIMIT5000

Human-in-the-Loop (HITL)​

Add human oversight to AI orchestration. HITL pauses execution at critical points (checkpoints) and waits for human approval before proceeding.

For getting started and implementation patterns, see the Human-in-the-Loop User Guide.

Core Interfaces​

HITL uses small, focused interfaces following Go idioms:

// InterruptPolicy composes all approval behaviors
type InterruptPolicy interface {
    PlanApprover      // ShouldApprovePlan(ctx, plan) (*InterruptDecision, error)
    StepApprover      // ShouldApproveBeforeStep/AfterStep(ctx, step, ...) (*InterruptDecision, error)
    ErrorEscalator    // ShouldEscalateError(ctx, step, err, attempts) (*InterruptDecision, error)
}

// CheckpointPersistence is the provider-neutral request-path contract.
type CheckpointPersistence interface {
    SaveCheckpoint(ctx context.Context, checkpoint *ExecutionCheckpoint) error
    LoadCheckpoint(ctx context.Context, checkpointID string) (*ExecutionCheckpoint, error)
    UpdateCheckpointStatus(ctx context.Context, checkpointID string, expected, status CheckpointStatus) error
    ListPendingCheckpoints(ctx context.Context, filter CheckpointFilter) ([]*ExecutionCheckpoint, error)
    DeleteCheckpoint(ctx context.Context, checkpointID string) error
}

type ExpiredCheckpointClaimRequest struct {
    Before time.Time
    Limit  int
    Owner  string
    Lease  time.Duration
}

// ExpiredCheckpointSource atomically leases expiry work to one processor.
type ExpiredCheckpointSource interface {
    ClaimExpiredCheckpoints(ctx context.Context, request ExpiredCheckpointClaimRequest) ([]*ExecutionCheckpoint, error)
    ReleaseExpiredCheckpointClaim(ctx context.Context, checkpointID, owner string) error
}

// CheckpointStore is the legacy convenience contract. New runtime code uses
// CheckpointPersistence plus a separate CheckpointExpiryProcessor Runnable.
type CheckpointStore interface {
    CheckpointPersistence
    StartExpiryProcessor(ctx context.Context, config ExpiryProcessorConfig) error
    StopExpiryProcessor(ctx context.Context) error
    SetExpiryCallback(callback ExpiryCallback) error
}

// InterruptController coordinates all HITL functionality
type InterruptController interface {
    SetPolicy(policy InterruptPolicy)
    SetHandler(handler InterruptHandler)
    SetCheckpointStore(store CheckpointStore)
    CheckPlanApproval(ctx context.Context, plan *RoutingPlan) (*ExecutionCheckpoint, error)
    CheckBeforeStep(ctx context.Context, step RoutingStep, plan *RoutingPlan) (*ExecutionCheckpoint, error)
    CheckAfterStep(ctx context.Context, step RoutingStep, result *StepResult) (*ExecutionCheckpoint, error)
    CheckOnError(ctx context.Context, step RoutingStep, err error, attempts int) (*ExecutionCheckpoint, error)
    ProcessCommand(ctx context.Context, command *Command) (*ResumeResult, error)
    UpdateCheckpointProgress(ctx context.Context, checkpointID string, completedSteps []StepResult) error
}

// CommandStore provides distributed command delivery
type CommandStore interface {
    PublishCommand(ctx context.Context, command *Command) error
    SubscribeCommand(ctx context.Context, checkpointID string) (<-chan *Command, func(), error)
    Close() error
}

// InterruptHandler manages notification and response collection
type InterruptHandler interface {
    NotifyInterrupt(ctx context.Context, checkpoint *ExecutionCheckpoint) error
    WaitForCommand(ctx context.Context, checkpointID string, timeout time.Duration) (*Command, error)
    SubmitCommand(ctx context.Context, command *Command) error
}

Constructors​

InterruptController:

func NewInterruptController(
    policy InterruptPolicy,
    store CheckpointPersistence,
    handler InterruptHandler,
    opts ...InterruptControllerOption,
) *DefaultInterruptController

// Options
func WithControllerLogger(logger core.Logger) InterruptControllerOption
func WithControllerTelemetry(telemetry core.Telemetry) InterruptControllerOption
func WithControllerCommandStore(store CommandStore) InterruptControllerOption
func WithControllerCheckpointPersistence(store CheckpointPersistence) InterruptControllerOption

WithControllerCheckpointPersistence is the provider-neutral path for request processing. It avoids requiring the legacy store-owned expiry lifecycle from a custom backend.

Policy:

func NewRuleBasedPolicy(config HITLConfig, opts ...PolicyOption) *RuleBasedPolicy
func NewNoOpPolicy() *NoOpPolicy  // For testing or HITL-disabled scenarios

// Options
func WithPolicyLogger(logger core.Logger) PolicyOption

CheckpointStore:

// Canonical provider-neutral expiry lifecycle.
func NewCheckpointExpiryProcessor(
    persistence CheckpointPersistence,
    source ExpiredCheckpointSource,
    callback ExpiryCallback,
    config ExpiryProcessorConfig,
    options ...CheckpointExpiryProcessorOption,
) (*CheckpointExpiryProcessor, error)

type CheckpointExpiryRuntimeConfig struct {
    ClaimLease time.Duration // default 30s
}

func DefaultCheckpointExpiryRuntimeConfig() CheckpointExpiryRuntimeConfig
func LoadCheckpointExpiryRuntimeConfigFromEnvironment(
    base CheckpointExpiryRuntimeConfig,
    lookup func(string) (string, bool),
) (CheckpointExpiryRuntimeConfig, error)

func WithCheckpointExpiryOwner(owner string) CheckpointExpiryProcessorOption
func WithCheckpointExpiryLease(lease time.Duration) CheckpointExpiryProcessorOption
func WithCheckpointExpiryRuntimeConfig(config CheckpointExpiryRuntimeConfig) CheckpointExpiryProcessorOption
func WithCheckpointExpiryLogger(logger core.Logger) CheckpointExpiryProcessorOption
func WithCheckpointExpiryTelemetry(telemetry core.Telemetry) CheckpointExpiryProcessorOption
func WithCheckpointExpiryPolicy(policy CheckpointExpiryPolicy) CheckpointExpiryProcessorOption

// Included Redis owning constructor. Prefer redisprovider composition
// for new application bootstrap code.
func NewRedisCheckpointStore(opts ...interface{}) (*RedisCheckpointStore, error)
func NewRedisCheckpointStoreWithClient(
    client redis.UniversalClient,
    opts ...RedisCheckpointStoreOption,
) (*RedisCheckpointStore, error)

// Options
func WithCheckpointRedisURL(url string) RedisCheckpointStoreOption
func WithCheckpointKeyspace(keyspace core.RedisKeyspace, agentScope string) RedisCheckpointStoreOption
func WithCheckpointTTL(ttl time.Duration) RedisCheckpointStoreOption
func WithInstanceID(id string) RedisCheckpointStoreOption  // Legacy store-owned expiry lifecycle only
func WithCheckpointStoreLogger(logger core.Logger) RedisCheckpointStoreOption
func WithCheckpointStoreTelemetry(telemetry core.Telemetry) RedisCheckpointStoreOption

NewRedisCheckpointStoreWithClient leaves the supplied client open on Close. Both HITL stores derive identity from TRUVAG3_REDIS_NAMESPACE and TRUVAG3_AGENT_NAME (falling back to TRUVAG3_K8S_SERVICE_NAME). Explicit WithCheckpointKeyspace / WithCommandStoreKeyspace options override that identity. Connection options do not reconfigure a supplied client; the application must configure DB 0.

CheckpointExpiryProcessor implements core.Runnable. Register it before Framework.Run() and allow framework shutdown to cancel it. The processor owns polling, delivery policy, logging, tracing, and claim-owner identity; the storage provider owns persistence and atomic leased claims. The included Redis preset can assemble both with redisprovider.WithCheckpointExpiry(...) and returns the processor through OrchestrationBackends.Runnables().

CommandStore:

func NewRedisCommandStore(opts ...RedisCommandStoreOption) (*RedisCommandStore, error)
func NewRedisCommandStoreWithClient(client redis.UniversalClient, opts ...RedisCommandStoreOption) (*RedisCommandStore, error)

// Options
func WithCommandStoreRedisURL(url string) RedisCommandStoreOption
func WithCommandStoreKeyspace(keyspace core.RedisKeyspace, agentScope string) RedisCommandStoreOption
func WithCommandStoreLogger(logger core.Logger) RedisCommandStoreOption
func WithCommandStoreTelemetry(t core.Telemetry) RedisCommandStoreOption

NewRedisCommandStore is the environment-aware owning constructor. NewRedisCommandStoreWithClient uses an application-owned client, derives its default versioned key prefix from deployment and agent-identity configuration, and leaves the supplied client open on Close. Canonical composition passes the same typed keyspace and agentScope to WithCommandStoreKeyspace and WithCheckpointKeyspace.

InterruptHandler:

func NewWebhookInterruptHandler(webhookURL string, commandStore CommandStore, opts ...WebhookHandlerOption) *WebhookInterruptHandler
func NewNoOpInterruptHandler() *NoOpInterruptHandler  // For testing

// Options
func WithHandlerCircuitBreaker(cb core.CircuitBreaker) WebhookHandlerOption
func WithHandlerLogger(logger core.Logger) WebhookHandlerOption
func WithHandlerTelemetry(telemetry core.Telemetry) WebhookHandlerOption

HITLHandler (HTTP API):

func NewHITLHandler(controller InterruptController, store CheckpointPersistence, opts ...HITLHandlerOption) (*HITLHandler, error)

// Options
func WithHITLHandlerLogger(logger core.Logger) HITLHandlerOption
func WithHITLHandlerTelemetry(t core.Telemetry) HITLHandlerOption
func WithHITLResumer(resumer HITLResumer) HITLHandlerOption

// Register all routes
func (h *HITLHandler) RegisterRoutes(mux *http.ServeMux)

Framework-owned resume:

type HITLResumer interface {
    ResumeExecution(context.Context, string) (*ExecutionResult, error)
}
type ResumeExecutor interface {
    ExecuteResume(context.Context, *ExecutionCheckpoint) (*ExecutionResult, error)
}
type ResumeExecutorFunc func(context.Context, *ExecutionCheckpoint) (*ExecutionResult, error)

type CheckpointResumePersistence interface {
    LoadCheckpoint(context.Context, string) (*ExecutionCheckpoint, error)
    ClaimCheckpointForResume(context.Context, ResumeClaimRequest) (*CheckpointResumeClaim, error)
    RenewCheckpointResumeClaim(ctx context.Context, checkpointID, attemptID string, lease time.Duration) error
    ReleaseCheckpointResumeClaim(ctx context.Context, checkpointID, attemptID string) error
    FinalizeCheckpointResume(context.Context, ResumeFinalization) error
}
type ResumeClaimRequest struct {
    CheckpointID          string
    AttemptID             string
    Owner                 string
    Lease                 time.Duration
    SuccessorCheckpointID string
}
type CheckpointResumeClaim struct {
    Checkpoint     *ExecutionCheckpoint
    AttemptID      string
    PreviousStatus CheckpointStatus
    LeaseExpiresAt time.Time
}
type ResumeFinalizationOutcome string
const (
    ResumeFinalizationCompleted ResumeFinalizationOutcome = "completed"
    ResumeFinalizationContinued ResumeFinalizationOutcome = "continued"
)
type ResumeFinalization struct {
    CheckpointID          string
    AttemptID             string
    Outcome               ResumeFinalizationOutcome
    SuccessorCheckpointID string
}
type ResumeCoordinatorRuntimeConfig struct {
    ClaimLease     time.Duration
    CleanupTimeout time.Duration
}
func DefaultResumeCoordinatorRuntimeConfig() ResumeCoordinatorRuntimeConfig
func (c ResumeCoordinatorRuntimeConfig) Validate() error
func LoadResumeCoordinatorRuntimeConfigFromEnvironment(
    base ResumeCoordinatorRuntimeConfig, lookup func(string) (string, bool),
) (ResumeCoordinatorRuntimeConfig, error)

func NewResumeCoordinator(
    store CheckpointResumePersistence, executor ResumeExecutor,
    config ResumeCoordinatorRuntimeConfig, opts ...ResumeCoordinatorOption,
) (*ResumeCoordinator, error)
func WithResumeLogger(logger core.Logger) ResumeCoordinatorOption
func WithResumeOwner(owner string) ResumeCoordinatorOption
func (r *ResumeCoordinator) ResumeExecution(ctx context.Context, checkpointID string) (*ExecutionResult, error)
func (r *ResumeCoordinator) ResumeWithExecutor(ctx context.Context, checkpointID string, executor ResumeExecutor) (*ExecutionResult, error)

func WithCheckpointResume(value CheckpointResumePersistence) OrchestrationBackendOption
func (b *OrchestrationBackends) CheckpointResume() CheckpointResumePersistence

func CheckpointResponseFrom(cp *ExecutionCheckpoint) *CheckpointResponse
func HITLErrorResponse(err error) (status int, code, message string)

BackendFeatureHITLResume requires ordinary checkpoint persistence plus the BackendCheckpointResume capability. The Redis provider supplies both from the same adapter and keyspace. Runtime contracts remain provider-neutral.

Constructors reject invalid configuration and nil/typed-nil dependencies. WithHITLResumer(nil) is invalid; omitting it is valid and registers no resume route. The coordinator borrows dependencies, joins its per-call renewal goroutine, and has no Close() or permanent background worker.

The lease defaults to 30 seconds (valid: 3 seconds–24 hours). Cleanup defaults to 5 seconds (positive, ≤1 minute, ≤lease/3). The explicit loader recognizes TRUVAG3_HITL_RESUME_CLAIM_LEASE and TRUVAG3_HITL_RESUME_CLEANUP_TIMEOUT; construction itself reads no environment. These durations do not extend checkpoint retention or human approval time.

ExecutionCheckpoint persists ResumeState *CheckpointResumeState, ParentCheckpointID string, and ParentResumeAttemptID string. The ownership state retains attempt/owner, previous approval status, lease deadline, outcome, and reserved/finalized successor identities. CheckpointResponse maps the original public fields and parent/successor navigation, excluding internal ownership. Nested application data is unchanged.

A claim returns an isolated snapshot with its preceding approval status while the stored record is resuming. Only approved and expired_approved start new attempts; an expired owned attempt may be recovered. continued is terminal for the parent and points to a new checkpoint. edited is not resumable. State updates use expected-status CAS; owned records cannot be overwritten or deleted by generic persistence calls.

Resume outcomeDefault HTTP response
Persisted successful terminal result200, execution result
Persisted successor202, ResumeInterruptedResponse{Interrupted: true, Checkpoint: ...}
Another live owner409, resume_in_progress
Not approved / already terminal409, resume_not_resumable
Ownership lost after claiming409, resume_claim_lost; work may have executed
Missing target / invalid input404 checkpoint_not_found / 400 invalid_resume_request
Execution, restoration, recovery, or finalization failure500, resume_failed

ErrCheckpointResumeLifecycle preserves execution and lifecycle causes through errors.Is/As. Check it before interpreting a nested ErrInterrupted. HITLErrorResponse distinguishes pre-claim rejection from post-claim failure. No response implies automatic retry. Deletion conflicts use checkpoint_deletion_conflict; a pending-decision CAS conflict uses checkpoint_status_conflict.

Application HTTP/SSE contexts cancel active work on disconnect. Accepted queue tasks use their independent worker context. Bounded detached finalization occurs only after execution settles. An SSE adapter uses ResumeWithExecutor, sends progress normally, and holds done/checkpoint until finalization. After stream headers, failures are error events with retryable: false, not appended JSON.

Actual terminal result access:

func (o *AIOrchestrator) ProcessRequestWithExecution(
    ctx context.Context, request string, metadata map[string]interface{},
) (*OrchestratorResponse, *ExecutionResult, error)
func (o *AIOrchestrator) ProcessRequestStreamingWithExecution(
    ctx context.Context, request string, metadata map[string]interface{},
    callback core.StreamCallback,
) (*StreamingOrchestratorResponse, *ExecutionResult, error)

These share the existing processing and terminal-recording lifecycle, including synthesis, hooks, and delivery errors. Partial results and original errors are preserved; no result is invented before execution produces one. Response-only entry points continue to use that same lifecycle.

Orchestrator Integration:

// Wire HITL into orchestrator after creation
func (o *AIOrchestrator) SetInterruptController(controller InterruptController)
func (o *AIOrchestrator) GetInterruptController() InterruptController

HTTP API Endpoints​

The framework provides these endpoints via HITLHandler.RegisterRoutes:

EndpointMethodPurpose
POST /hitl/commandPOSTSubmit approve, reject, or abort; other commands return 400
POST /hitl/resume/{checkpoint_id}POSTResume through the injected resumer; route absent when none is configured
GET /hitl/checkpointsGETList pending checkpoints
GET /hitl/checkpoints/{id}GETGet checkpoint details

Submit Command:

curl -X POST http://agent:8080/hitl/command \
  -H "Content-Type: application/json" \
  -d '{"checkpoint_id": "cp-abc123", "type": "approve"}'

Response:

{
  "checkpoint_id": "cp-abc123",
  "action": "approve",
  "should_resume": true
}

Interrupt Points​

Interrupt PointTriggerUse Case
plan_generatedAfter AI creates execution planReview full plan before any execution
before_stepBefore executing each tool/agentApprove individual high-risk operations
on_errorWhen a step fails and is recoverableHuman approves another attempt, rejects, or aborts
after_stepAfter step completes (reserved)Output validation before proceeding
context_gatheringDuring context collection (reserved)Approve data access requests

Supported default commands​

CommandDefault controller behavior
approveCAS pending → approved; caller may request resume
rejectCAS pending → rejected; no resume
abortCAS pending → aborted; no resume
edit, skip, retry, respond400 unsupported_command, before state writes, publication, or success callbacks

The supported decisions apply to pending checkpoints, including on_error. Approval there permits the interrupted step to run again. Reserved interrupt points and declared enum values do not implement additional command protocols. A stale decision returns 409 checkpoint_status_conflict.

Key Types​

// ExecutionCheckpoint contains all state needed to resume execution
type ExecutionCheckpoint struct {
    CheckpointID       string                 `json:"checkpoint_id"`
    RequestID          string                 `json:"request_id"`
    OriginalRequestID  string                 `json:"original_request_id,omitempty"` // For trace correlation
    InterruptPoint     InterruptPoint         `json:"interrupt_point"`
    Decision           *InterruptDecision     `json:"decision"`
    Plan               *RoutingPlan           `json:"plan"`
    CompletedSteps     []StepResult           `json:"completed_steps"`
    CurrentStep        *RoutingStep           `json:"current_step,omitempty"`
    CurrentStepResult  *StepResult            `json:"current_step_result,omitempty"`
    StepResults        map[string]*StepResult `json:"step_results"`
    ResolvedParameters map[string]interface{} `json:"resolved_parameters,omitempty"`
    OriginalRequest    string                 `json:"original_request"`
    UserContext        map[string]interface{} `json:"user_context,omitempty"`
    RequestMode        RequestMode            `json:"request_mode,omitempty"`
    CreatedAt          time.Time              `json:"created_at"`
    ExpiresAt          time.Time              `json:"expires_at"`
    Status             CheckpointStatus       `json:"status"`
}

// InterruptPoint enum
const (
    InterruptPointPlanGenerated    InterruptPoint = "plan_generated"
    InterruptPointBeforeStep       InterruptPoint = "before_step"
    InterruptPointAfterStep        InterruptPoint = "after_step"
    InterruptPointOnError          InterruptPoint = "on_error"
    InterruptPointContextGathering InterruptPoint = "context_gathering"
)

// CheckpointStatus enum - includes human-initiated and expiry statuses
const (
    // Human-initiated
    CheckpointStatusPending   CheckpointStatus = "pending"
    CheckpointStatusApproved  CheckpointStatus = "approved"
    CheckpointStatusRejected  CheckpointStatus = "rejected"
    CheckpointStatusEdited    CheckpointStatus = "edited"
    CheckpointStatusCompleted CheckpointStatus = "completed"
    CheckpointStatusResuming  CheckpointStatus = "resuming"
    CheckpointStatusContinued CheckpointStatus = "continued"
    CheckpointStatusAborted   CheckpointStatus = "aborted"

    // Framework-owned transient state while the durable run snapshot is
    // attached. It is never resumable or listed as pending human work.
    CheckpointStatusPreparing CheckpointStatus = "preparing"

    // Streaming expiry (implicit deny)
    CheckpointStatusExpired CheckpointStatus = "expired"

    // Non-streaming expiry (policy-driven)
    CheckpointStatusExpiredApproved CheckpointStatus = "expired_approved"
    CheckpointStatusExpiredRejected CheckpointStatus = "expired_rejected"
    CheckpointStatusExpiredAborted  CheckpointStatus = "expired_aborted"
)

// CommandType enum
const (
    CommandApprove CommandType = "approve"
    CommandEdit    CommandType = "edit"
    CommandReject  CommandType = "reject"
    CommandSkip    CommandType = "skip"
    CommandAbort   CommandType = "abort"
    CommandRetry   CommandType = "retry"
    CommandRespond CommandType = "respond"
)

// RequestMode determines expiry behavior
const (
    RequestModeStreaming    RequestMode = "streaming"     // Implicit deny on expiry
    RequestModeNonStreaming RequestMode = "non_streaming" // Apply DefaultAction on expiry
)

// Command represents a human decision
type Command struct {
    CommandID    string                 `json:"command_id"`
    CheckpointID string                 `json:"checkpoint_id"`
    Type         CommandType            `json:"type"`
    EditedPlan   *RoutingPlan           `json:"edited_plan,omitempty"`
    EditedStep   *RoutingStep           `json:"edited_step,omitempty"`
    EditedParams map[string]interface{} `json:"edited_params,omitempty"`
    Feedback     string                 `json:"feedback,omitempty"`
    Response     string                 `json:"response,omitempty"`
    UserID       string                 `json:"user_id,omitempty"`
    Timestamp    time.Time              `json:"timestamp"`
}

// ResumeResult contains the outcome of processing a command
type ResumeResult struct {
    CheckpointID string       `json:"checkpoint_id"`
    Action       CommandType  `json:"action"`
    ShouldResume bool         `json:"should_resume"`
    ModifiedPlan *RoutingPlan `json:"modified_plan,omitempty"`
    SkipStep     bool         `json:"skip_step,omitempty"`
    Abort        bool         `json:"abort,omitempty"`
    Feedback     string       `json:"feedback,omitempty"`
}

Context Helpers​

Request Context:

func WithRequestID(ctx context.Context, requestID string) context.Context
func GetRequestID(ctx context.Context) string

func WithRequestMode(ctx context.Context, mode RequestMode) context.Context
func GetRequestMode(ctx context.Context) RequestMode

func WithMetadata(ctx context.Context, metadata map[string]interface{}) context.Context
func GetMetadata(ctx context.Context) map[string]interface{}

Resume Context:

// Lower-level helper called and cleaned up by ResumeCoordinator after claim.
// Calling this alone does not acquire or finalize ownership.
func BuildResumeContext(ctx context.Context, checkpoint *ExecutionCheckpoint) (context.Context, func(), error)

// Individual context helpers are not alternatives to the resume coordinator.
func WithResumeMode(ctx context.Context, checkpointID string) context.Context
func WithPlanOverride(ctx context.Context, plan *RoutingPlan) context.Context
func WithCompletedSteps(ctx context.Context, results map[string]*StepResult) context.Context
func WithPreResolvedParams(ctx context.Context, params map[string]interface{}, stepID string) context.Context

Iterative planning note: During multi-phase execution, WithCompletedSteps populates prior-phase results into context. The executor uses these for cross-phase template resolution (e.g., {{step-1.response.data}} in Phase 2 referencing Phase 1 results) while the loop termination logic only counts steps belonging to the current phase's plan.

Error Handling​

// A failed finalization may wrap an interruption. Preserve that failure first.
var lifecycle *orchestration.ErrCheckpointResumeLifecycle
if errors.As(err, &lifecycle) {
    // Handle failure; inspect effects before retrying.
} else if orchestration.IsInterrupted(err) {
    checkpoint := orchestration.GetCheckpoint(err)
    checkpointID := orchestration.GetCheckpointID(err)
    // Handle checkpoint - workflow is paused, not failed
}

// Error type checking
if orchestration.IsCheckpointNotFound(err) { /* Checkpoint doesn't exist */ }
if orchestration.IsCheckpointExpired(err) { /* Timeout reached */ }
if orchestration.IsInvalidCommand(err) { /* Invalid command for state */ }
if orchestration.IsHITLDisabled(err) { /* HITL not enabled */ }

Status Helpers​

// Check if checkpoint can be resumed
if orchestration.IsResumableStatus(checkpoint.Status) {
    // Status is: approved or expired_approved
}

// Check if checkpoint is in terminal state
if orchestration.IsTerminalStatus(checkpoint.Status) {
    // Status is: continued, completed, rejected, aborted, expired, expired_rejected, expired_aborted
}

// Check if checkpoint is awaiting response
if orchestration.IsPendingStatus(checkpoint.Status) {
    // Status is: pending
}

Configuration​

type HITLConfig struct {
    Enabled                   bool          `json:"enabled"`
    RequirePlanApproval       bool          `json:"require_plan_approval"`
    SensitiveCapabilities     []string      `json:"sensitive_capabilities"`      // Plan + Step approval
    SensitiveAgents           []string      `json:"sensitive_agents"`            // Plan + Step approval
    StepSensitiveCapabilities []string      `json:"step_sensitive_capabilities"` // Step-only approval
    StepSensitiveAgents       []string      `json:"step_sensitive_agents"`       // Step-only approval
    EscalateAfterRetries      int           `json:"escalate_after_retries"`
    DefaultTimeout            time.Duration `json:"default_timeout"`
    DefaultAction             CommandType   `json:"default_action"`
    CheckpointTTL             time.Duration `json:"checkpoint_ttl"`
    KeyPrefix                 string        `json:"key_prefix"`
    ExpiryProcessor           ExpiryProcessorConfig `json:"expiry_processor"`
}

// Create with smart defaults
func DefaultHITLConfig() HITLConfig
func NewHITLConfig(opts ...HITLOption) HITLConfig

// Options
func WithExpiryProcessor(config ExpiryProcessorConfig) HITLOption
Environment VariableDefaultDescription
TRUVAG3_HITL_ENABLEDfalseEnable HITL system
TRUVAG3_HITL_REQUIRE_PLAN_APPROVALfalsePause after every plan generation
TRUVAG3_HITL_SENSITIVE_CAPABILITIES``Capabilities requiring plan + step approval
TRUVAG3_HITL_SENSITIVE_AGENTS``Agents requiring plan + step approval
TRUVAG3_HITL_STEP_SENSITIVE_CAPABILITIES``Capabilities requiring step-only approval
TRUVAG3_HITL_STEP_SENSITIVE_AGENTS``Agents requiring step-only approval
TRUVAG3_HITL_ESCALATE_AFTER_RETRIES3Escalate to human after N retry failures
TRUVAG3_HITL_DEFAULT_TIMEOUT5mCheckpoint expiration time
TRUVAG3_HITL_DEFAULT_ACTIONrejectAction on timeout (approve, reject, abort)
TRUVAG3_REDIS_NAMESPACEdefaultDeployment namespace for canonical versioned DB 0 HITL keys

Expiry Processor Configuration​

The expiry processor handles checkpoint timeouts in the background:

type ExpiryProcessorConfig struct {
    Enabled           bool              // Run processor (default: true)
    ScanInterval      time.Duration     // Scan frequency (default: 10s)
    BatchSize         int               // Max per scan (default: 100)
    DeliverySemantics DeliverySemantics // Callback timing
}

const (
    DeliveryAtMostOnce  DeliverySemantics = "at_most_once"  // Default, safe
    DeliveryAtLeastOnce DeliverySemantics = "at_least_once" // Callback must be idempotent
)

type ExpiryCallback func(ctx context.Context, checkpoint *ExecutionCheckpoint, appliedAction CommandType)
Environment VariableDefaultDescription
TRUVAG3_HITL_EXPIRY_ENABLEDtrueEnable expiry processor
TRUVAG3_HITL_EXPIRY_INTERVAL10sScan interval
TRUVAG3_HITL_EXPIRY_BATCH_SIZE100Max checkpoints per scan
TRUVAG3_HITL_EXPIRY_DELIVERYat_least_onceDelivery semantics
TRUVAG3_HITL_STREAMING_EXPIRYimplicit_denyStreaming request expiry behavior
TRUVAG3_HITL_NON_STREAMING_EXPIRYapply_defaultNon-streaming request expiry behavior

Metrics​

HITL emits these Prometheus metrics:

MetricTypeLabelsDescription
orchestration.hitl.checkpoint_created_totalCounterinterrupt_point, reasonCheckpoints created
orchestration.hitl.checkpoint_status_totalCounterfrom_status, to_statusStatus transitions
orchestration.hitl.command_processed_totalCountercommand_type, successCommands processed
orchestration.hitl.checkpoint_expired_totalCounteraction, request_mode, interrupt_pointExpired checkpoints
orchestration.hitl.approval_latency_secondsHistogramcommand_typeHuman response time
orchestration.hitl.expiry_scan_duration_secondsHistogram-Expiry scan duration

Common Patterns​

Production Service Template​

Complete template for production-ready services:

package main

import (
    "context"
    "errors"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/truvaagents/truva-g3/core"
)

func main() {
    // signal.NotifyContext returns a context that is cancelled on SIGINT/SIGTERM.
    // Framework.Run() blocks until this context is cancelled, then drains
    // registered runnables within TRUVAG3_FRAMEWORK_RUNNABLE_DRAIN_TIMEOUT.
    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer stop()

    // Create component
    agent := core.NewBaseAgent("production-service")

    // Register capabilities
    agent.RegisterCapability(core.Capability{
        Name:        "process",
        Description: "Process data",
        Handler:     handleProcess,
    })

    // Create framework with production settings
    framework, err := core.NewFramework(agent,
        // Discovery
        core.WithRedisURL(os.Getenv("REDIS_URL")),

        // Observability
        core.WithTelemetry(true, os.Getenv("OTEL_ENDPOINT")),
        core.WithLogLevel("info"),

        // Networking
        core.WithPort(8080),
        core.WithCORSDefaults(),

        // Resilience
        core.WithCircuitBreaker(5, 30*time.Second),
    )
    if err != nil {
        log.Fatalf("Failed to create framework: %v", err)
    }

    // Run blocks until ctx is cancelled, then performs the graceful drain.
    if err := framework.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
        log.Fatalf("Framework error: %v", err)
    }
}

Multi-Provider AI Strategy​

Use different AI providers for different tasks:

type SmartAI struct {
    fast      core.AIClient  // Groq for speed
    accurate  core.AIClient  // Claude for accuracy
    creative  core.AIClient  // OpenAI for creativity
}

func NewSmartAI() *SmartAI {
    return &SmartAI{
        fast: ai.MustNewClient(
            ai.WithProviderAlias("openai.groq"),
            ai.WithModel("fast"),
        ),
        accurate: ai.MustNewClient(
            ai.WithProvider("anthropic"),
            ai.WithModel("smart"),
        ),
        creative: ai.MustNewClient(
            ai.WithProvider("openai"),
            ai.WithModel("smart"),
        ),
    }
}

func (s *SmartAI) QuickAnswer(question string) (string, error) {
    // Use fast model for simple queries
    resp, err := s.fast.GenerateResponse(ctx, question, &core.AIOptions{
        MaxTokens:   200,
        Temperature: 0.3,
    })
    return resp.Content, err
}

func (s *SmartAI) AnalyzeCode(code string) (string, error) {
    // Use accurate model for code analysis
    resp, err := s.accurate.GenerateResponse(ctx,
        fmt.Sprintf("Analyze this code for bugs and improvements:\n%s", code),
        &core.AIOptions{
            Temperature: 0.1,  // Very low for technical accuracy
            MaxTokens:   2000,
        },
    )
    return resp.Content, err
}

Service Mesh Pattern​

Build a resilient service mesh with discovery and circuit breakers:

type ServiceMesh struct {
    discovery core.Discovery
    breakers  map[string]*resilience.CircuitBreaker
    mu        sync.RWMutex
}

func (sm *ServiceMesh) Call(ctx context.Context, capability string, request interface{}) (interface{}, error) {
    // Discover service
    services, err := sm.discovery.FindByCapability(ctx, capability)
    if err != nil || len(services) == 0 {
        return nil, fmt.Errorf("no service found for %s", capability)
    }

    service := services[0]  // TODO: Add load balancing

    // Get or create circuit breaker for this service
    breaker := sm.getBreaker(service.ID)

    var response interface{}
    err = breaker.Execute(ctx, func() error {
        // Call the service
        return sm.callService(service, request, &response)
    })

    return response, err
}

func (sm *ServiceMesh) getBreaker(serviceID string) *resilience.CircuitBreaker {
    sm.mu.RLock()
    if breaker, exists := sm.breakers[serviceID]; exists {
        sm.mu.RUnlock()
        return breaker
    }
    sm.mu.RUnlock()

    // Create new breaker
    sm.mu.Lock()
    defer sm.mu.Unlock()

    breaker := resilience.NewCircuitBreakerLegacy(5, 30*time.Second)
    sm.breakers[serviceID] = breaker
    return breaker
}

Best Practices​

1. Always Use the Framework​

The framework handles complex setup, so you don't have to:

// ❌ Don't do this
agent := core.NewBaseAgent("my-agent")
agent.Initialize(ctx)
agent.Start(ctx, 8080)

// ✅ Do this instead
framework, _ := core.NewFramework(agent, options...)
framework.Run(ctx)

2. Configure via Environment​

Follow 12-factor app principles:

// ✅ Good - configuration from environment
framework, _ := core.NewFramework(agent,
    core.WithRedisURL(os.Getenv("REDIS_URL")),
    core.WithPort(getPortFromEnv()),
)

// ❌ Bad - hardcoded values
framework, _ := core.NewFramework(agent,
    core.WithRedisURL("redis://localhost:6379"),
    core.WithPort(8080),
)

3. Use Structured Logging​

Always log with context:

// ✅ Good - structured with context
logger.Info("Processing request", map[string]interface{}{
    "request_id": req.ID,
    "user_id":    req.UserID,
    "action":     req.Action,
})

// ❌ Bad - unstructured string
log.Printf("Processing request %s for user %s", req.ID, req.UserID)

4. Handle Circuit Breaker States​

Always check for circuit breaker errors:

err := breaker.Execute(ctx, func() error {
    return callService()
})

if errors.Is(err, core.ErrCircuitBreakerOpen) {
    // Service is down, use fallback
    return useFallback()
}

if err != nil {
    // Other error, handle accordingly
    return err
}

5. Use Context for Cancellation​

Always respect context cancellation:

func LongRunningOperation(ctx context.Context) error {
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()  // Respect cancellation
        default:
            // Do work
        }
    }
}

Error Handling​

TruvaG3 uses typed errors for better error handling:

// Defined in core/errors.go — match against them with errors.Is(err, core.Err...)
var (
    ErrCircuitBreakerOpen   = errors.New("circuit breaker open")
    ErrMaxRetriesExceeded   = errors.New("maximum retries exceeded")
    ErrServiceNotFound      = errors.New("service not found")
    ErrInvalidConfiguration = errors.New("invalid configuration")
)

Example - Comprehensive Error Handling:

func HandleRequest(ctx context.Context, req Request) error {
    err := processWithResilience(ctx, req)

    switch {
    case errors.Is(err, core.ErrCircuitBreakerOpen):
        // Service down, use fallback
        return handleWithFallback(ctx, req)

    case errors.Is(err, core.ErrMaxRetriesExceeded):
        // Temporary failure, queue for later
        return queueForRetry(req)

    case errors.Is(err, context.DeadlineExceeded):
        // Timeout, return error to client
        return fmt.Errorf("request timeout: %w", err)

    case err != nil:
        // Unexpected error
        logger.Error("Unexpected error", map[string]interface{}{
            "error": err.Error(),
            "request_id": req.ID,
        })
        return fmt.Errorf("internal error: %w", err)
    }

    return nil
}

Environment Variables​

TruvaG3 supports configuration through environment variables:

Core Configuration​

  • TRUVAG3_AGENT_NAME - Component name
  • TRUVAG3_PORT - HTTP port (default: 8080)
  • REDIS_URL - Redis connection URL
  • TRUVAG3_LOG_LEVEL - Logging level (error/warn/info/debug)
  • TRUVAG3_LOG_FORMAT - Log format (json/text)
  • TRUVAG3_DEV_MODE - Development mode (true/false)

Discovery Retry Configuration​

  • TRUVAG3_DISCOVERY_RETRY - Enable background Redis retry on connection failure (true/false, default: false)
  • TRUVAG3_DISCOVERY_RETRY_INTERVAL - Initial retry interval (e.g., "30s", "1m", default: 30s)

Schema Discovery & Validation​

  • TRUVAG3_VALIDATE_PAYLOADS - Enable Phase 3 schema validation (true/false, default: false)

AI Configuration​

  • OPENAI_API_KEY - OpenAI API key
  • ANTHROPIC_API_KEY - Anthropic API key
  • GOOGLE_API_KEY / GEMINI_API_KEY - Google Gemini API key (GOOGLE_API_KEY takes precedence)
  • GROQ_API_KEY - Groq API key
  • TOGETHER_API_KEY - Together AI API key
  • OPENROUTER_API_KEY - OpenRouter API key
  • DEEPSEEK_API_KEY - DeepSeek API key
  • XAI_API_KEY - xAI Grok API key
  • QWEN_API_KEY - Qwen API key

OpenAI-Compatible Provider URLs (Optional)​

Kubernetes Configuration​

  • TRUVAG3_K8S_SERVICE_NAME - Kubernetes service name
  • TRUVAG3_K8S_SERVICE_PORT - Kubernetes service port
  • TRUVAG3_K8S_POD_IP - Pod IP address
  • HOSTNAME - Pod hostname

Telemetry Configuration​

  • OTEL_EXPORTER_OTLP_ENDPOINT - OpenTelemetry endpoint
  • OTEL_SERVICE_NAME - Service name for telemetry

Shared Memory Configuration​

Episodic Memory & Coordination:

  • TRUVAG3_AGENT_DOMAIN - Groups agents for memory scoping (default: "default")
  • TRUVAG3_SHARED_MEMORY_PROVIDER - Storage provider: redis or noop (default: "noop")
  • TRUVAG3_SHARED_MEMORY_STREAM_MAXLEN - Max events in domain stream (default: 100000)
  • TRUVAG3_SHARED_MEMORY_INVESTIGATION_TTL - Investigation claim auto-expiry (default: 30m)
  • TRUVAG3_SHARED_MEMORY_ENRICHMENT_MAX_TOKENS - Memory view allowance at four encoded bytes/token (default: 2000, or 8000 bytes); not an exact tokenizer count
  • TRUVAG3_SHARED_MEMORY_EVENT_SUMMARIZER_MAX_RESPONSE_BYTES - Positive per-step summary-input payload allowance shared by encoded failure facts and response (default: 4000 bytes)
  • TRUVAG3_SHARED_MEMORY_RECORD_FAILURE_CAUSE_MAX_BYTES - Positive heuristic failure cause allowance (default: 256 bytes; framing/shortening marker additional); direct option WithRecordFailureCauseMaxBytes.
  • TRUVAG3_SHARED_MEMORY_RECENT_EVENTS_LIMIT - Recent domain events for baseline awareness (default: 20)

Activity Compaction (LLM-powered digest):

  • TRUVAG3_SHARED_MEMORY_SUMMARIZER_MODEL - Model for summarization/compaction LLM calls. Accepts aliases such as fast and smart only when the selected provider defines them, or a concrete model ID (default: agent's model)
  • TRUVAG3_SHARED_MEMORY_EVENT_SUMMARIZER_MAX_TOKENS - Positive fixed output-token allowance per event-summary call, read by BuildMemoryHooks at startup. Unset retains 10000; invalid explicit values fail construction, including without an AI client. Direct construction uses WithSummarizerMaxTokens(n), which rejects non-positive values and overrides WithSummarizerTokensPerStep(n) regardless of option order. Independent of synthesis and compaction budgets.
  • TRUVAG3_SHARED_MEMORY_ENRICHMENT_SUMMARY_MAX_TOKENS - Requested digest length (default: 500 tokens). The built-in full/incremental compactor sends AIOptions.MaxTokens = 5 × target (default: 2500); provider reasoning scaling may further alter the wire budget. Its positive target must fit math.MaxInt / 5, checked at hook construction when that compactor is selected and before direct non-empty AI calls. Custom compactors retain their own generation rules. The target is a prompt instruction, not a tokenizer-enforced output cap. The final enrichment view has its own byte budget.
  • TRUVAG3_SHARED_MEMORY_COMPACTION_RAW_LIMIT - Max events fetched before compaction (default: 200)
  • TRUVAG3_SHARED_MEMORY_COMPACTION_RECENT_DETAIL - Raw events appended after digest for detail (default: 15)

Digest Caching (reduces redundant LLM calls):

  • TRUVAG3_SHARED_MEMORY_DIGEST_CACHE_TTL - Cached digest expiry (default: 5m)
  • TRUVAG3_SHARED_MEMORY_DIGEST_INCREMENTAL_THRESHOLD - Max new events for incremental update; above this triggers full recompaction (default: 20)

Activity Coordination (real-time agent signals):

  • TRUVAG3_ACTIVITY_COORDINATION_ENABLED - Enable/disable coordination layer (default: true)
  • TRUVAG3_ACTIVITY_SIGNAL_TTL - Signal auto-expiry (default: 5m)
  • TRUVAG3_ACTIVITY_SIGNAL_MAX_IN_PROMPT - Max signals in <agent_coordination> prompt section (default: 10)
  • TRUVAG3_ACTIVITY_SIGNAL_QUERY_MAX_LEN - Max query chars in signals (default: 200)

Troubleshooting​

Common Issues​

"No AI provider available"

  • Set one of: OPENAI_API_KEY, ANTHROPIC_API_KEY, OPENROUTER_API_KEY, GEMINI_API_KEY, GROQ_API_KEY

"Failed to connect to Redis"

  • Check REDIS_URL is correct
  • Ensure Redis is running
  • Use WithMockDiscovery(true) for testing without Redis

"Circuit breaker is open"

  • Check if downstream service is healthy
  • Review circuit breaker thresholds
  • Check logs for failure patterns

"Port already in use"

  • Another service is using the port
  • Change port with WithPort(different_port)

Debug Mode​

Enable detailed logging for troubleshooting:

export TRUVAG3_LOG_LEVEL=debug
export TRUVAG3_LOG_FORMAT=text
# or
export TRUVAG3_DEV_MODE=true

Support​

Terminal correlation and event-summary timeout​

orchestration.MetadataKeyExecutionRequestID identifies the owning request segment on existing terminal execution results, returned and stored. Older results can lack it. orchestration.WithRequestID(ctx, id) binds it for post-return evidence projection.

orchestration.WithSummarizerTimeout(time.Duration) configures a positive builtin AI-invocation timeout (default 2m). Invalid explicit values fail construction. The standard BuildMemoryHooks layer instead resolves environment then default; a direct-constructor option does not override an independently built hook bundle. WithSummarizerMaxStepsPerBatch(int) already controls direct-construction batch size (default 20); remaining steps use heuristic summaries, without extra model calls.