Skip to main content

TruvaG3 Agent Development Guide

This guide provides a comprehensive, step-by-step tutorial for developing agents in the TruvaG3 framework. Agents are active components that use AI-powered orchestration to coordinate multiple tools and provide intelligent responses to natural language queries.

Table of Contents​

  1. Understanding Agents in TruvaG3
  2. Choosing Your Agent Type
  3. Project Structure
  4. Step 1: Create the Agent Struct
  5. Step 2: Configure the Orchestrator
  6. Step 3: Register Capabilities
  7. Step 4: Implement Handlers
  8. Step 5: Add SSE Streaming (Streaming Agents Only)
  9. Step 6: Add Session Management (Chat Agents Only)
  10. Step 7: Create the Main Entry Point
  1. Step 8: Add Deployment Files
  1. Logging and Observability
  2. Distributed Tracing
  3. Testing Your Agent
  1. Adding Human-in-the-Loop (HITL) Approval
  1. Best Practices
  2. Troubleshooting
  3. Quick Reference

1. Understanding Agents in TruvaG3​

What is an Agent?​

In TruvaG3, an Agent is an active component that:

  • Uses AI-powered orchestration to coordinate multiple tools
  • Processes natural language requests and generates execution plans
  • Can discover other components via service discovery (Redis)
  • Synthesizes results from multiple tool calls into coherent responses
  • Optionally streams responses in real-time via SSE

Agents vs Tools​

AspectToolAgent
Base type*core.BaseTool*core.BaseAgent
DiscoveryCan register, cannot discoverCan both register and discover
OrchestrationCannot orchestrateOrchestrates other components via AI
AI clientRarely neededRequired for orchestration
PurposeProvide specific capabilities (API wrapper)Coordinate tools, process natural language
Importscore, telemetrycore, telemetry, ai, orchestration
Exampleweather-tool, currency-tooltravel-chat-agent, devops-chat-agent

Architecture Overview​

┌──────────────────────────────────────────────────────────────────────┐
│                          Your Agent                                   │
│                                                                       │
│  ┌──────────────────────────────────────────────────────────────────┐ │
│  │                    AI Orchestrator                                │ │
│  │  ┌──────────┐  ┌──────────┐  ┌───────────┐  ┌──────────────┐   │ │
│  │  │ Planner  │  │ Executor │  │ Synthesizer│  │ Error Recovery│   │ │
│  │  │ (LLM)   │→│ (DAG)    │→│ (LLM)      │  │ (4-Layer)    │   │ │
│  │  └──────────┘  └──────────┘  └───────────┘  └──────────────┘   │ │
│  └──────────────────────────────────────────────────────────────────┘ │
│                                │                                      │
│  ┌──────────────┐   ┌──────────▼──────────┐   ┌──────────────────┐  │
│  │ Session Store│   │  Service Discovery  │   │  SSE Handler     │  │
│  │ (DB 0 keys)  │   │  (DB 0 keys)        │   │  (Streaming)     │  │
│  └──────────────┘   └─────────────────────┘   └──────────────────┘  │
│                                │                                      │
└────────────────────────────────┼──────────────────────────────────────┘
                                 │
            ┌────────────────────┼────────────────────┐
            │                    │                    │
       ┌────▼───┐          ┌────▼───┐          ┌────▼───┐
       │  Tool  │          │  Tool  │          │  Tool  │
       │weather │          │currency│          │ devops │
       └────────┘          └────────┘          └────────┘

2. Choosing Your Agent Type​

TruvaG3 supports two agent patterns. Choose based on your use case:

Non-Streaming Agent (Request/Response)​

Use when: Building API backends, batch processing, or server-to-server orchestration.

Client → POST /orchestrate/natural → Agent → Orchestrator → Tools → JSON Response
  • Uses orchestrator.ProcessRequest() — returns complete response
  • Standard HTTP JSON request/response
  • Simpler to implement
  • Reference implementation: examples/agent-with-orchestration

Streaming Agent (SSE Chat)​

Use when: Building chat UIs, real-time dashboards, or interactive experiences.

Client → POST /chat/stream → Agent → Orchestrator → Tools → SSE Events (streaming)
  • Uses orchestrator.ProcessRequestStreaming() — delivers tokens as they're generated
  • Server-Sent Events (SSE) for real-time delivery
  • Per-step progress callbacks for tool completion events
  • Session management with conversation history
  • Reference implementation: examples/travel-chat-agent

Feature Comparison​

FeatureNon-StreamingStreaming
Response deliveryComplete JSONToken-by-token SSE
Time to first byteAfter full processingImmediate (planning status)
Step progressIn final responseReal-time SSE events
Conversation historyOptionalTypically required
Session managementOptionalTypically required
Agent skillsOptionalOptional
Implementation complexityLowerHigher
Files needed3-4 Go files4-5 Go files

3. Project Structure​

Non-Streaming Agent​

examples/your-agent/
├── main.go              # Entry point, telemetry, framework setup, orchestrator init
├── your_agent.go        # Agent struct, AI client, orchestrator config, capability registration
├── handlers.go          # HTTP handlers for orchestration and utility endpoints
├── go.mod               # Go module with ai, core, orchestration, telemetry dependencies
├── .env                 # Environment variables (API keys, Redis URL, port)
├── Dockerfile           # Standalone container build
├── Dockerfile.workspace # Development container (builds from truvag3 root)
├── k8-deployment.yaml   # Kubernetes deployment manifest
├── skills/packages/    # Optional reviewable skill packages grouped by namespace
└── setup.sh             # Full lifecycle: build, deploy, test, clean

Streaming Agent (Chat)​

examples/your-chat-agent/
├── main.go              # Entry point, telemetry, framework setup, orchestrator init
├── chat_agent.go        # Agent struct, AI client, orchestrator config, streaming logic
├── sse_handler.go       # SSE handler, StreamCallback interface, SSE event formatting
├── session.go           # Redis-backed session store, conversation history management
├── handlers.go          # REST handlers for session CRUD, health, discovery
├── go.mod               # Go module with ai, core, orchestration, telemetry dependencies
├── .env                 # Environment variables (API keys, Redis URL, port)
├── Dockerfile           # Standalone container build
├── Dockerfile.workspace # Development container (builds from truvag3 root)
├── k8-deployment.yaml   # Kubernetes deployment manifest
├── skills/packages/    # Optional reviewable skill packages grouped by namespace
└── setup.sh             # Full lifecycle: build, deploy, test, clean

File Responsibilities​

FileNon-StreamingStreaming
main.goConfig validation, telemetry init, framework setup, managed deferred initialization, graceful shutdownSame
*_agent.goAgent struct, AI chain client, orchestrator config with PromptConfig, capability registrationSame + ProcessWithStreaming() method, conversation context formatting
handlers.gohandleNaturalOrchestration(), handleHealth(), handleDiscover(), helper functionsSession CRUD handlers, health, discover, helper functions
sse_handler.goN/AStreamCallback interface, SSECallback impl, SSEHandler.ServeHTTP()
session.goN/ASessionStore (Redis-backed), Session/Message types, conversation history
skills/packages/Optional skill package source grouped as <namespace>/<name>.jsonSame

4. Step 1: Create the Agent Struct​

The agent struct embeds *core.BaseAgent and holds the orchestrator, HTTP client, and any domain-specific state.

Non-Streaming Agent​

// your_agent.go
package main

import (
    "fmt"
    "net/http"
    "sync"
    "time"

    "github.com/truvaagents/truva-g3/ai"
    "github.com/truvaagents/truva-g3/core"
    "github.com/truvaagents/truva-g3/orchestration"
    "github.com/truvaagents/truva-g3/telemetry"
)

type YourAgent struct {
    *core.BaseAgent
    orchestrator *orchestration.AIOrchestrator
    httpClient   *http.Client
    mu           sync.RWMutex
}

func NewYourAgent() (*YourAgent, error) {
    agent := core.NewBaseAgent("your-agent")

    // Create AI client with provider chain for automatic failover
    // Provider chain: OpenAI (primary) → Anthropic (backup)
    chainClient, err := ai.NewChainClient(
        ai.WithProviderChain("openai", "anthropic"),
        ai.WithChainTelemetry(telemetry.GetTelemetryProvider()),
        ai.WithChainLogger(agent.Logger),
    )
    if err != nil {
        // Fallback to single provider
        singleClient, err := ai.NewClient()
        if err != nil {
            agent.Logger.Warn("AI client creation failed", map[string]interface{}{
                "error": err.Error(),
            })
        } else {
            agent.AI = singleClient
        }
    } else {
        agent.AI = chainClient
    }

    // Declare custom metrics for observability
    telemetry.DeclareMetrics("your-agent", telemetry.ModuleConfig{
        Metrics: []telemetry.MetricDefinition{
            {
                Name:    "orchestration.request.duration_ms",
                Type:    "histogram",
                Help:    "Orchestration request duration in milliseconds",
                Labels:  []string{"status"},
                Unit:    "milliseconds",
                Buckets: []float64{100, 500, 1000, 5000, 10000, 30000},
            },
            {
                Name:   "orchestration.requests",
                Type:   "counter",
                Help:   "Number of orchestration requests",
                Labels: []string{"status"},
            },
        },
    })

    // Create traced HTTP client for distributed tracing propagation
    tracedClient := telemetry.NewTracedHTTPClientWithTransport(&http.Transport{
        MaxIdleConns:        100,
        MaxIdleConnsPerHost: 10,
        IdleConnTimeout:     90 * time.Second,
        ForceAttemptHTTP2:   true,
    })
    tracedClient.Timeout = 300 * time.Second

    yourAgent := &YourAgent{
        BaseAgent:  agent,
        httpClient: tracedClient,
    }

    yourAgent.registerCapabilities()
    return yourAgent, nil
}

Streaming Chat Agent​

// chat_agent.go
package main

import (
    "context"
    "fmt"
    "net/http"
    "os"
    "strings"
    "sync"
    "time"

    "github.com/truvaagents/truva-g3/ai"
    "github.com/truvaagents/truva-g3/core"
    "github.com/truvaagents/truva-g3/orchestration"
    "github.com/truvaagents/truva-g3/telemetry"
    "go.opentelemetry.io/otel/attribute"
)

type YourChatAgent struct {
    *core.BaseAgent
    orchestrator *orchestration.AIOrchestrator
    sessionStore *SessionStore
    httpClient   *http.Client
    mu           sync.RWMutex
}

func NewYourChatAgent() (*YourChatAgent, error) {
    agent := core.NewBaseAgent("your-chat-agent")

    // AI client with auto-detect chain (discovers providers from available API keys)
    // For explicit ordering, add: ai.WithProviderChain("openai", "anthropic", "openai.groq")
    chainClient, err := ai.NewChainClient(
        ai.WithChainTelemetry(telemetry.GetTelemetryProvider()),
        ai.WithChainLogger(agent.Logger),
        ai.WithChainTimeout(240*time.Second), // Extended for reasoning models
    )
    if err != nil {
        singleClient, _ := ai.NewClient(ai.WithTimeout(240 * time.Second))
        agent.AI = singleClient
    } else {
        agent.AI = chainClient
    }

    // Declare metrics (same pattern)
    telemetry.DeclareMetrics("your-chat-agent", telemetry.ModuleConfig{
        Metrics: []telemetry.MetricDefinition{
            {
                Name:    "chat.request.duration_ms",
                Type:    "histogram",
                Help:    "Chat request duration in milliseconds",
                Labels:  []string{"session_id", "status"},
                Unit:    "milliseconds",
                Buckets: []float64{100, 500, 1000, 2000, 5000, 10000, 30000},
            },
            {
                Name:   "chat.requests",
                Type:   "counter",
                Help:   "Number of chat requests",
                Labels: []string{"status"},
            },
        },
    })

    // Traced HTTP client (same pattern)
    tracedClient := telemetry.NewTracedHTTPClientWithTransport(&http.Transport{
        MaxIdleConns:        100,
        MaxIdleConnsPerHost: 10,
        IdleConnTimeout:     90 * time.Second,
        ForceAttemptHTTP2:   true,
    })
    tracedClient.Timeout = 300 * time.Second

    // Resolve one standalone, Sentinel, or cluster topology and DB-0 keyspace.
    redisResolution, err := core.ResolveRedisConnectionConfig(nil, os.LookupEnv)
    if err != nil {
        return nil, fmt.Errorf("resolve Redis topology: %w", err)
    }
    redisKeyspace, err := core.NewRedisKeyspace(os.Getenv("TRUVAG3_REDIS_NAMESPACE"))
    if err != nil {
        return nil, fmt.Errorf("resolve Redis namespace: %w", err)
    }
    sessionStore, err := NewSessionStore(redisResolution, redisKeyspace, 48*time.Hour, 50, agent.Logger)
    if err != nil {
        return nil, fmt.Errorf("failed to create session store: %w", err)
    }

    chatAgent := &YourChatAgent{
        BaseAgent:    agent,
        sessionStore: sessionStore,
        httpClient:   tracedClient,
    }

    chatAgent.registerCapabilities()
    return chatAgent, nil
}

Thread-Safe Orchestrator Access​

The request and scheduling examples later in this guide call GetOrchestrator(). Define the matching getter on the agent type so those callers do not race with background initialization:

func (a *YourAgent) GetOrchestrator() *orchestration.AIOrchestrator {
    a.mu.RLock()
    defer a.mu.RUnlock()
    return a.orchestrator
}

func (a *YourChatAgent) GetOrchestrator() *orchestration.AIOrchestrator {
    a.mu.RLock()
    defer a.mu.RUnlock()
    return a.orchestrator
}

In a real agent, include only the method for that agent's concrete type.

Key Points​

  1. Embed *core.BaseAgent: Provides Logger, AI, Discovery, RegisterCapability(), and GetID()
  2. AI provider chain: Use ai.NewChainClient() for automatic failover between providers. Falls back to ai.NewClient() for single provider
  3. Traced HTTP client: Always use telemetry.NewTracedHTTPClientWithTransport() for trace context propagation to downstream tools
  4. Mutex protection: The orchestrator is initialized asynchronously, so protect access with sync.RWMutex
  5. Declare metrics: Use telemetry.DeclareMetrics() to register custom metrics at startup
  6. Optional memory fields: BaseAgent has two optional fields — ConversationMemory (session-scoped history) and SemanticMemory (cross-session similarity search). Set them when your agent needs conversation context or long-term knowledge retrieval. See Adding Context to Your Agent for implementation patterns

AI Client Configuration​

The code above uses ai.NewChainClient() for multi-provider failover. Key concepts:

  • Provider aliases: One-line identifiers ("openai", "anthropic", "openai.groq", "openai.ollama") that auto-resolve base URL + API key env var
  • Model aliases: Portable names ("default", "fast", "smart", "premium", "code", "vision") that resolve per provider. Override a mapping at runtime when needed: TRUVAG3_OPENAI_MODEL_SMART=gpt-5.6-luna
  • Explicit vs auto-detect chain: Use ai.WithProviderChain("openai", "anthropic") for deterministic order, or omit for auto-detection from available API keys
  • Reasoning-capable models: Reasoning and visible output share the completion cap. Scaling profiles offer a configurable multiplier (default 3x), not a reserved reasoning allowance; follow provider guidance and measure truncation rather than coupling application code to a model family name

Reference: AI_PROVIDERS_SETUP_GUIDE.md for provider/model alias tables, failover behavior, K8s secrets/ConfigMaps, and operational scenarios.

5. Step 2: Configure the Orchestrator​

The orchestrator is initialized after the framework starts (because it needs Discovery, which connects to Redis during framework.Run()). This is done in a background goroutine from main.go.

InitializeOrchestrator Method​

// your_agent.go (continued)

func (a *YourAgent) InitializeOrchestrator(discovery core.Discovery) error {
    a.mu.Lock()
    defer a.mu.Unlock()

    if discovery == nil {
        return fmt.Errorf("discovery service not available")
    }

    // 1. Create orchestrator config
    config := orchestration.DefaultConfig()
    config.RoutingMode = orchestration.ModeAutonomous     // AI-driven, no workflow
    config.SynthesisStrategy = orchestration.StrategyLLM  // LLM synthesizes results
    config.MetricsEnabled = true
    config.EnableTelemetry = true

    // 2. Phase-specific LLM configuration
    config.PlanAIOptions = &orchestration.AIOptionsOverride{
        MaxTokens: orchestration.IntPtr(15000),
    }
    config.SynthesisAIOptions = &orchestration.AIOptionsOverride{
        MaxTokens:   orchestration.IntPtr(5000),
        Temperature: orchestration.Float32Ptr(0.5),  // 0.5 for non-streaming, 0.7 for more natural streaming
    }

    // 3. Execution timeouts
    config.ExecutionOptions.TotalTimeout = 5 * time.Minute
    config.ExecutionOptions.StepTimeout = 120 * time.Second

    // 4. Configure PromptConfig for your domain
    config.PromptConfig = orchestration.PromptConfig{
        // SystemInstructions: The agent's persona and behavioral context.
        // This becomes the primary identity for the LLM planner and synthesizer.
        SystemInstructions: `You are a helpful DevOps assistant.
You help users manage Kubernetes clusters, check system status, and troubleshoot issues.
Provide clear, actionable information and warn about potentially destructive operations.`,

        // Domain helps the LLM understand the context
        Domain: "devops",

        // Domain-specific type rules for correct JSON generation
        AdditionalTypeRules: []orchestration.TypeRule{
            {
                TypeNames:   []string{"namespace"},
                JsonType:    "JSON strings",
                Example:     `"default"`,
                Description: "Kubernetes namespace names",
            },
            {
                TypeNames:   []string{"replicas", "replica_count"},
                JsonType:    "JSON integers",
                Example:     `3`,
                Description: "Number of pod replicas (0-10)",
            },
        },

        // Custom instructions guide the LLM planner
        CustomInstructions: []string{
            "Prefer parallel execution when steps are independent",
            "Use the smallest discovered capability set that fully satisfies the request",
            "Preserve dependency order when a step consumes an earlier step's output",
        },
    }

    // 5. Create dependencies with dependency injection
    deps := orchestration.OrchestratorDependencies{
        Discovery:           discovery,
        AIClient:            a.AI,
        Logger:              a.Logger,
        Telemetry:           telemetry.GetTelemetryProvider(),
        EnableErrorAnalyzer: true, // Enable LLM-based error analysis (Layer 3)
    }

    // 6. Create and start orchestrator
    orch, err := orchestration.CreateOrchestrator(config, deps)
    if err != nil {
        return fmt.Errorf("failed to create orchestrator: %w", err)
    }

    ctx := context.Background()
    if err := orch.Start(ctx); err != nil {
        return fmt.Errorf("failed to start orchestrator: %w", err)
    }

    a.orchestrator = orch

    a.Logger.Info("Orchestrator initialized", map[string]interface{}{
        "routing_mode":       config.RoutingMode,
        "synthesis_strategy": config.SynthesisStrategy,
    })

    return nil
}

For JSON-shaped plans, an agent can set config.PlanAIOptions.ResponseFormat = orchestration.StringPtr("json"). Configure synthesis separately through SynthesisAIOptions. This is a provider-portable preference inherited by model requests, not a guarantee: an adapter omits it when the selected model is not declared to support JSON mode and records that adjustment. An explicitly required JSON format instead fails with an unsupported-feature error. Inspect effective provider requests before claiming JSON enforcement, especially on a fallback route. JSON mode does not replace the orchestrator's plan validation.

PromptConfig Tips​

PromptConfig is the most important configuration — it controls how the LLM plans and synthesizes:

  • SystemInstructions: Define persona in first sentence, state what agent does, include constraints. Keep to 3-5 sentences
  • CustomInstructions: Keep short, agent-specific guidance that applies to every request here. Use one clear decision per instruction
  • AdditionalTypeRules: Teach LLM domain-specific JSON types (e.g., namespace names, replica counts)

Use an agent skill instead of CustomInstructions when procedural guidance is reusable, versioned, relevant only to some requests, or large enough to benefit from loading resources only when needed. Keep the agent's identity and universal behavior in SystemInstructions. Skills guide planning and responses; they do not register, authorize, or hardcode tool endpoints. The orchestrator continues to discover capabilities at runtime.

See also: docs/orchestration/LLM_PLANNING_PROMPT_GUIDE.md for full PromptConfig documentation.

Optional: Enable Agent Skills​

Skills are disabled by default. Enabling them is an explicit application choice with two separate parts. The same integration applies to request/response, streaming, asynchronous, and event-driven agents that use orchestration; the agent's delivery mode does not change the skill lifecycle.

  1. The application startup code (the composition root) creates a provider-backed implementation of the provider-neutral orchestration.SkillRegistry interface. Keep the registry and any provider-client owner alive for the orchestrator's lifetime, and close owned clients during application shutdown.
  2. The agent supplies the complete set of skill bindings through code or deployment environment configuration.

When the effective skill runtime is inactive—because the binding list is empty, or because skills are disabled without a required binding—no skill registry is required, no selector or registry calls run, and no skill sections are added to prompts. Disabling skills while retaining a Required: true binding is invalid configuration and fails startup.

The following replaces the configuration-resolution and construction portions of the earlier example. For this path, initialize the earlier config value with orchestration.NewDefaultOrchestratorConfig() instead of orchestration.DefaultConfig(), apply the ordinary fields shown above, and let ResolveOrchestratorConfig perform the one selected environment-resolution pass. EnvironmentStrict validates all supported orchestration environment settings, not only the skill settings.

The skillRegistry argument comes from the application startup code; it is not created by WithSkills. The included Redis/Valkey composition and a custom provider example are shown separately in the Agent Skills Guide. See the included provider and custom storage adapters.

func createSkillEnabledOrchestrator(
    config *orchestration.OrchestratorConfig,
    deps orchestration.OrchestratorDependencies,
    skillRegistry orchestration.SkillRegistry,
) (*orchestration.AIOrchestrator, error) {
    skillConfig := orchestration.SkillConfig{
        Enabled: true,
        Bindings: []orchestration.SkillBinding{
            {
                Namespace:  "travel",
                Name:       "weather-assessment",
                Version:    "published",
                Activation: orchestration.SkillActivationAuto,
                Required:   false,
            },
        },
    }

    resolved, err := orchestration.ResolveOrchestratorConfig(
        orchestration.ConfigResolution{
            Base:        config,
            Environment: orchestration.EnvironmentStrict,
            Options: []orchestration.OrchestratorOption{
                orchestration.WithSkills(skillConfig),
                orchestration.WithSkillRegistry(skillRegistry),
            },
        },
    )
    if err != nil {
        return nil, fmt.Errorf("resolve orchestrator configuration: %w", err)
    }

    orch, err := orchestration.CreateResolvedOrchestrator(resolved.Config, deps)
    if err != nil {
        return nil, fmt.Errorf("create orchestrator: %w", err)
    }
    return orch, nil
}

This helper constructs the orchestrator but does not start it. The caller must still call orch.Start(ctx) and assign the started instance as shown in the earlier InitializeOrchestrator example.

always activates on every request, auto lets the bounded activation policy choose, and explicit requires a trusted activation from application code. Required: true means that an unavailable bound package fails the applicable request boundary; it does not force an auto skill to activate. A published binding is resolved to one exact revision at request start and remains pinned for that execution.

Either environment-enabled resolution mode lets deployment configuration replace the code-owned binding list. Skill environment values are validated strictly in both EnvironmentCompatible and EnvironmentStrict. EnvironmentStrict additionally rejects malformed non-skill orchestration environment values instead of applying compatible fallbacks. For example, place the complete JSON array on one line in the agent's .env file:

TRUVAG3_SKILLS_ENABLED=true
TRUVAG3_SKILL_BINDINGS_JSON='[{"namespace":"travel","name":"weather-assessment","version":"published","activation":"auto","required":false}]'

TRUVAG3_SKILL_BINDINGS_JSON replaces the complete code list; it never appends or merges. This keeps all replicas in a Kubernetes Deployment on the same binding configuration. There is no V1 API for changing the bindings of one running replica. An empty array is an explicit deployment-owned binding list with no skills:

TRUVAG3_SKILL_BINDINGS_JSON='[]'

6. Step 3: Register Capabilities​

Agent capabilities define the HTTP endpoints your agent exposes. Critical rule: All agent capabilities must be marked Internal: true to prevent the orchestrator's LLM from including the agent's own endpoints in execution plans (which would cause recursive self-calls).

Non-Streaming Agent Capabilities​

func (a *YourAgent) registerCapabilities() {
    // Primary endpoint: Natural language orchestration
    a.RegisterCapability(core.Capability{
        Name:        "orchestrate_natural",
        Description: "Process natural language requests using AI-powered orchestration. " +
                     "Required: request (natural language query). " +
                     "Optional: ai_synthesis (boolean, default: true), metadata (object).",
        Endpoint:    "/orchestrate/natural",
        InputTypes:  []string{"json", "text"},
        OutputTypes: []string{"json"},
        Handler:     a.handleNaturalOrchestration,
        Internal:    true, // CRITICAL: Prevents recursive self-calls
        InputSummary: &core.SchemaSummary{
            RequiredFields: []core.FieldHint{
                {
                    Name:        "request",
                    Type:        "string",
                    Example:     "What is the cluster status?",
                    Description: "Natural language request to process",
                },
            },
            OptionalFields: []core.FieldHint{
                {
                    Name:        "ai_synthesis",
                    Type:        "boolean",
                    Example:     "true",
                    Description: "Enable AI synthesis of results (default: true)",
                },
                {
                    Name:        "metadata",
                    Type:        "object",
                    Example:     `{"namespace": "production"}`,
                    Description: "Additional context and preferences",
                },
            },
        },
    })

    // Utility: Health check with orchestrator status
    a.RegisterCapability(core.Capability{
        Name:        "health",
        Description: "Health check with orchestrator status and metrics",
        Endpoint:    "/health",
        InputTypes:  []string{},
        OutputTypes: []string{"json"},
        Handler:     a.handleHealth,
        Internal:    true,
    })

    // Utility: Discover available tools
    a.RegisterCapability(core.Capability{
        Name:        "discover_tools",
        Description: "Discovers available tools and their capabilities",
        Endpoint:    "/discover",
        InputTypes:  []string{},
        OutputTypes: []string{"json"},
        Handler:     a.handleDiscover,
        Internal:    true,
    })
}

Streaming Chat Agent Capabilities​

func (a *YourChatAgent) registerCapabilities() {
    // SSE streaming chat endpoint
    a.RegisterCapability(core.Capability{
        Name:        "chat_stream",
        Description: "SSE streaming chat endpoint",
        Endpoint:    "/chat/stream",
        Handler:     NewSSEHandler(a).ServeHTTP,
        Internal:    true,
    })

    // Session management + health + discover — all Internal: true
    a.RegisterCapability(core.Capability{Name: "create_session", Endpoint: "/chat/session", Handler: a.handleCreateSession, Internal: true})
    a.RegisterCapability(core.Capability{Name: "get_session", Endpoint: "/chat/session/{id}", Handler: a.handleGetSession, Internal: true})
    a.RegisterCapability(core.Capability{Name: "get_history", Endpoint: "/chat/session/{id}/history", Handler: a.handleGetHistory, Internal: true})
    a.RegisterCapability(core.Capability{Name: "list_sessions", Endpoint: "/chat/sessions", Handler: a.handleListSessions, Internal: true})
    a.RegisterCapability(core.Capability{Name: "update_title", Endpoint: "/chat/session/{id}/title", Handler: a.handleUpdateTitle, Internal: true})
    a.RegisterCapability(core.Capability{Name: "delete_session", Endpoint: "/chat/session/delete", Handler: a.handleDeleteSession, Internal: true})
    a.RegisterCapability(core.Capability{Name: "health", Endpoint: "/health", Handler: a.handleHealth, Internal: true})
    a.RegisterCapability(core.Capability{Name: "discover", Endpoint: "/discover", Handler: a.handleDiscover, Internal: true})
}

Why Internal: true is Critical​

Without Internal: true:
  User: "Check cluster status"
  LLM Plan: step-1: your-agent.orchestrate_natural("Check cluster status")  ← RECURSIVE!

With Internal: true:
  User: "Check cluster status"
  LLM Plan: step-1: devops-tool.get_cluster_status()  ← Correct!

The Internal flag excludes capabilities from the LLM's tool catalog while keeping them callable via HTTP. This prevents the orchestrator from planning steps that call itself.

When to Use Internal: false (Agent-as-Tool)​

For multi-agent hierarchies, expose domain-specific capabilities with Internal: false so other agents can orchestrate your agent. Orchestration endpoints (/orchestrate/*, /chat/stream) remain Internal: true.

// Exposed to other agents' LLM planners
a.RegisterCapability(core.Capability{
    Name:        "research_topic",
    Description: "Researches a topic using multiple tools and returns a synthesized report. " +
                 "Required: topic (research subject). Optional: depth (brief/standard/deep).",
    Endpoint:    "/research",
    Type:        core.CapabilityOrchestrator, // Extended timeout for nested orchestration + HITL
    InputTypes:  []string{"json"},
    OutputTypes: []string{"json"},
    Handler:     a.handleResearch,
    Internal:    false, // Visible to other agents
    InputSummary: &core.SchemaSummary{ /* same FieldHint pattern as above */ },
})

// Orchestration endpoint stays internal
a.RegisterCapability(core.Capability{
    Name: "orchestrate_natural", Endpoint: "/orchestrate/natural",
    Handler: a.handleNaturalOrchestration, Internal: true,
})

Set Type: core.CapabilityOrchestrator on capabilities that perform nested orchestration (their own planning → execution → synthesis cycle). Such calls receive a context deadline of HITL.DefaultTimeout + ExecutionOptions.StepTimeout to accommodate a child's approval wait. Plain tools should leave Type empty; they use the shared HTTP-client timeout and enclosing context deadlines, without a separate StepTimeout deadline. Orchestrator construction raises that shared client timeout, and the phase loop raises its phase budget, when the delegation allowance is larger. Earlier parent deadlines and the child's server timeout still apply. See Execution Limits when sizing a complete synchronous request.

Capabilities with Internal: false enter the orchestrator's tiered tool selection, where only lightweight summaries (not full descriptions) are sent to the LLM. Front-load your Description or set an explicit Summary field — see TOOL_DEVELOPMENT_GUIDE.md — Tiered Selection and the Summary Field.

Writing Effective Capability Descriptions​

The LLM uses descriptions to select tools and generate JSON payloads. TruvaG3 uses 3-phase progressive enhancement:

PhaseAccuracyWhat You Provide
Phase 1~85-90%Description field (always required)
Phase 2~95%InputSummary with FieldHint structs (recommended for production)
Phase 3~99%Schema endpoint (optional, auto-generated from Phase 2)

Description formula: [Action] [what it does]. Required: [fields]. Optional: [fields with defaults].

InputSummary (shown in the non-streaming example above) gives the AI exact field names, types, and examples. FieldHint tips:

  • Name: Exact JSON field name your handler expects
  • Type: Standard JSON types: "string", "number", "boolean", "object", "array"
  • Example: Realistic value showing format (e.g., "2026-04-15", "JFK")
  • Description: Meaning + constraints (e.g., "0-10" for replica count)

Reference: TOOL_SCHEMA_DISCOVERY_GUIDE.md for the complete 3-phase approach including Phase 3 JSON Schema validation.

7. Step 4: Implement Handlers​

Non-Streaming Handler (ProcessRequest)​

// handlers.go
package main

import (
    "encoding/json"
    "errors"
    "fmt"
    "net/http"
    "time"

    "github.com/truvaagents/truva-g3/core"
    "github.com/truvaagents/truva-g3/telemetry"
    "go.opentelemetry.io/otel/attribute"
)

// OrchestrationResponse is the JSON response for non-streaming requests.
type OrchestrationResponse struct {
    RequestID     string                 `json:"request_id"`
    Request       string                 `json:"request"`
    Response      string                 `json:"response"`
    ToolsUsed     []string               `json:"tools_used"`
    ExecutionTime string                 `json:"execution_time"`
    Confidence    float64                `json:"confidence"`
    Metadata      map[string]interface{} `json:"metadata,omitempty"`
}

func (a *YourAgent) handleNaturalOrchestration(w http.ResponseWriter, r *http.Request) {
    startTime := time.Now()
    ctx := r.Context()

    // 1. Add span event for Jaeger visibility
    telemetry.AddSpanEvent(ctx, "request_received",
        attribute.String("operation", "natural_orchestration"),
    )

    // 2. Log with trace context for correlation
    a.Logger.InfoWithContext(ctx, "Processing orchestration request", map[string]interface{}{
        "operation": "natural_orchestration",
        "method":    r.Method,
    })

    // 3. Parse and validate request
    var req struct {
        Request  string                 `json:"request"`
        Metadata map[string]interface{} `json:"metadata,omitempty"`
    }
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        telemetry.RecordSpanError(ctx, err)
        a.Logger.ErrorWithContext(ctx, "Failed to decode request", map[string]interface{}{
            "operation": "natural_orchestration",
            "error":     err.Error(),
        })
        writeError(w, http.StatusBadRequest, "Invalid request format", err)
        return
    }

    if req.Request == "" {
        writeError(w, http.StatusBadRequest, "Request field is required", nil)
        return
    }

    // 4. Check orchestrator availability
    a.mu.RLock()
    orch := a.orchestrator
    a.mu.RUnlock()

    if orch == nil {
        writeError(w, http.StatusServiceUnavailable, "Orchestrator initializing", nil)
        return
    }

    // 5. Process through AI orchestrator
    telemetry.AddSpanEvent(ctx, "orchestration_started",
        attribute.String("request", req.Request),
    )

    result, err := orch.ProcessRequest(ctx, req.Request, req.Metadata)
    if err != nil {
        // Check if the error originated from an AI provider (e.g., 400 Bad Request, 429 Rate Limit).
        // core.ProviderError carries the original HTTP status, provider name, and model —
        // surface the real status instead of masking everything as 500.
        var pe core.ProviderError
        if errors.As(err, &pe) && pe.StatusCode() >= 400 && pe.StatusCode() < 500 {
            a.Logger.WarnWithContext(ctx, "LLM provider returned client error", map[string]interface{}{
                "operation":   "natural_orchestration",
                "error":       pe.Error(),
                "status_code": pe.StatusCode(),
                "provider":    pe.Provider(),
                "model":       pe.Model(),
                "is_transient": pe.IsTransient(),
                "duration_ms": time.Since(startTime).Milliseconds(),
            })
            telemetry.RecordSpanError(ctx, err)
            writeError(w, pe.StatusCode(), pe.Error(), err)
            return
        }

        a.Logger.ErrorWithContext(ctx, "Orchestration failed", map[string]interface{}{
            "operation":   "natural_orchestration",
            "error":       err.Error(),
            "duration_ms": time.Since(startTime).Milliseconds(),
        })
        telemetry.RecordSpanError(ctx, err)
        writeError(w, http.StatusInternalServerError, "Orchestration failed", err)
        return
    }

    // 6. Build and send response
    response := &OrchestrationResponse{
        RequestID:     result.RequestID,
        Request:       req.Request,
        Response:      result.Response,
        ToolsUsed:     result.AgentsInvolved,
        ExecutionTime: time.Since(startTime).String(),
        Confidence:    result.Confidence,
        Metadata:      result.Metadata,
    }

    // 7. Record metrics
    durationMs := float64(time.Since(startTime).Milliseconds())
    telemetry.RecordRequest(telemetry.ModuleOrchestration, "natural_request", durationMs, "success")

    // 8. Add completion span event
    telemetry.AddSpanEvent(ctx, "orchestration_completed",
        attribute.String("request_id", result.RequestID),
        attribute.Int("tools_used", len(result.AgentsInvolved)),
        attribute.Float64("confidence", result.Confidence),
    )

    // 9. Log completion
    a.Logger.InfoWithContext(ctx, "Orchestration completed", map[string]interface{}{
        "operation":   "natural_orchestration",
        "request_id":  result.RequestID,
        "tools_used":  len(result.AgentsInvolved),
        "duration_ms": time.Since(startTime).Milliseconds(),
        "status":      "success",
    })

    writeJSON(w, http.StatusOK, response)
}

Error Propagation: Why This Pattern Matters​

The core.ProviderError check above is not just cosmetic — it's the final link in a chain that starts inside the AI provider layer and flows through the ChainClient and orchestrator:

AI Provider (e.g., OpenAI returns 429 Rate Limit)
  ↓  providerError{StatusCode: 429, Provider: "openai", Model: "gpt-5.6-sol"}
ChainClient.isClientError() — inspects via errors.As()
  ↓  429 excluded from client errors → tries next provider (Anthropic)
  ↓  Anthropic also fails → error propagates up
Orchestrator — wraps with fmt.Errorf("...: %w", err)
  ↓  ProviderError preserved through %w chain
Your Handler — errors.As(err, &pe) finds the ProviderError
  ↓  Surfaces pe.StatusCode() = 429 as HTTP response

Critical rule: always use %w when wrapping errors. This preserves core.ProviderError through the wrapping chain so every layer can inspect it:

// CORRECT — preserves core.ProviderError
return fmt.Errorf("streaming orchestration failed: %w", err)

// WRONG — destroys the type, errors.As() won't find ProviderError
return fmt.Errorf("failed: %s", err)   // string conversion
return errors.New(err.Error())          // new error, original type lost

If you break the %w chain:

  • ChainClient can't classify the error → defaults to failover (safe but wastes API calls for true client errors like malformed prompts)
  • Your handler can't extract the status → returns 500 for what was actually a 400 (misleading to callers, breaks client retry logic)
  • LLM debug store can't extract provider/model → error records have empty metadata

Reference: AI_PROVIDERS_SETUP_GUIDE.md — How Errors Propagate Through the Stack for the complete error flow diagram.

Streaming Handler (ProcessWithStreaming)​

For streaming agents, the orchestration logic lives in the agent struct (not in the SSE handler), because it needs access to the session store and orchestrator:

// chat_agent.go (continued)

// addConversationHistoryMetadata passes raw turns to orchestration.
// The framework builds <conversation_history> before planning and reuses
// the prepared value across continuation and synthesis phases.
// This example intentionally maps its application session UUID to the
// framework conversation ID; applications may choose a different mapping.
func (a *YourChatAgent) addConversationHistoryMetadata(
    metadata map[string]interface{},
    sessionID string,
    history []Message,
) map[string]interface{} {
    if metadata == nil {
        metadata = make(map[string]interface{})
    }
    // This example deliberately maps one session UUID to one conversation.
    if sessionID != "" {
        metadata[orchestration.MetadataConversationID] = sessionID
    }
    if len(history) == 0 {
        return metadata
    }

    turns := make([]core.ConversationTurn, 0, len(history))
    for _, msg := range history {
        turns = append(turns, core.ConversationTurn{
            Role:    msg.Role,
            Content: msg.Content,
        })
    }

    metadata[orchestration.MetadataConversationTurns] = turns
    return metadata
}

// Conversation identity is set before the history guard so the first turn is
// correlated even when no earlier messages exist.

// ProcessWithStreaming processes a query and streams progress via SSE callback.
func (a *YourChatAgent) ProcessWithStreaming(
    ctx context.Context, sessionID, query string, callback StreamCallback,
) error {
    startTime := time.Now()

    a.mu.RLock()
    orch := a.orchestrator
    a.mu.RUnlock()

    if orch == nil {
        return fmt.Errorf("orchestrator not initialized")
    }

    // 1. Load conversation history for context continuity
    history := a.sessionStore.GetHistory(sessionID)
    metadata := a.addConversationHistoryMetadata(nil, sessionID, history)

    // 2. Log with trace context
    a.Logger.InfoWithContext(ctx, "Processing chat request", map[string]interface{}{
        "operation":     "process_chat",
        "session_id":    sessionID,
        "history_turns": len(history),
    })

    // 3. Send planning status to client
    callback.SendStatus("planning", "Analyzing your request...")

    // 4. Add span event
    telemetry.AddSpanEvent(ctx, "orchestration.started",
        attribute.String("session_id", sessionID),
        attribute.Int("history_turns", len(history)),
    )

    // 5. Attach per-step callback for real-time tool progress
    globalStepCounter := 0
    ctx = orchestration.WithStepCallback(ctx,
        func(stepIndex, totalSteps int, step orchestration.RoutingStep, stepResult orchestration.StepResult) {
            globalStepCounter++
            callback.SendStep(
                fmt.Sprintf("step_%d", globalStepCounter),
                step.AgentName,
                stepResult.Success,
                stepResult.Duration.Milliseconds(),
            )
        },
    )

    // 6. Stream the orchestration response
    result, err := orch.ProcessRequestStreaming(ctx, query, metadata,
        func(chunk core.StreamChunk) error {
            // Phase-complete chunks are progress indicators, not content
            if chunk.Metadata != nil && chunk.Metadata["type"] == "phase_complete" {
                phaseNum, _ := chunk.Metadata["phase"].(int)
                callback.SendStatus("phase_complete",
                    fmt.Sprintf("Phase %d complete. Planning next phase...", phaseNum))
                return nil
            }
            // Forward content chunks to SSE
            if chunk.Content != "" {
                callback.SendChunk(chunk.Content)
            }
            return nil
        },
    )
    if err != nil {
        a.Logger.ErrorWithContext(ctx, "Streaming orchestration failed", map[string]interface{}{
            "operation":   "process_chat",
            "error":       err.Error(),
            "duration_ms": time.Since(startTime).Milliseconds(),
        })
        telemetry.RecordSpanError(ctx, err)
        return fmt.Errorf("streaming orchestration failed: %w", err)
    }

    // 7. Send token usage stats
    if result.Usage != nil {
        callback.SendUsage(
            result.Usage.PromptTokens,
            result.Usage.CompletionTokens,
            result.Usage.TotalTokens,
        )
    }

    // 8. Send finish reason if available
    if result.FinishReason != "" {
        callback.SendFinish(result.FinishReason)
    }

    // 9. Add completion span event
    telemetry.AddSpanEvent(ctx, "orchestration.completed",
        attribute.String("request_id", result.RequestID),
        attribute.Int("agents_used", len(result.AgentsInvolved)),
        attribute.Float64("confidence", result.Confidence),
    )

    // 10. Send completion event
    callback.SendDone(result.RequestID, result.AgentsInvolved,
        time.Since(startTime).Milliseconds())

    // 11. Store assistant response in session history
    a.sessionStore.AddMessage(sessionID, Message{
        Role:      "assistant",
        Content:   result.Response,
        Timestamp: time.Now(),
        Metadata: map[string]interface{}{
            "request_id": result.RequestID,
            "tools_used": result.AgentsInvolved,
        },
    })

    // 12. Log completion
    a.Logger.InfoWithContext(ctx, "Chat request completed", map[string]interface{}{
        "operation":   "process_chat",
        "session_id":  sessionID,
        "request_id":  result.RequestID,
        "tools_used":  len(result.AgentsInvolved),
        "duration_ms": time.Since(startTime).Milliseconds(),
        "status":      "success",
    })

    return nil
}

Tier 1 conversation-history protection is automatic once your agent passes raw turns in metadata and creates the orchestrator through CreateOrchestrator(...). If you want Tier 2 recursive compaction, use the Layer 2 helper and inject the resulting preparer:

config := orchestration.DefaultConfig()
preparer, err := orchestration.BuildCompactionEnabledConversationHistoryPreparer(
    config,
    a.AI,
    orchestration.WithConversationSummaryCache(myCache),   // optional override
    orchestration.WithConversationCompactor(myCompactor),  // optional override
)
if err != nil {
    return fmt.Errorf("failed to build conversation history preparer: %w", err)
}

deps := orchestration.OrchestratorDependencies{
    Discovery:                   discovery,
    AIClient:                    a.AI,
    Logger:                      a.Logger,
    Telemetry:                   telemetry.GetTelemetryProvider(),
    ConversationHistoryPreparer: preparer,
}

That helper creates the default cache and LLM compactor for you, then applies any supplied overrides last, so you can customize one concern without dropping to full direct construction.

For a dedicated walkthrough of Tier 1 defaults, Tier 2 recursive compaction, and full Layer 3 construction, see CONVERSATION_HISTORY_GUIDE.md.

Health Handler (Common to Both Types)​

func (a *YourAgent) handleHealth(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()

    health := map[string]interface{}{
        "status":    "healthy",
        "timestamp": time.Now().Unix(),
    }

    // Check Redis/Discovery
    if discovery := a.GetDiscovery(); discovery != nil {
        if _, err := discovery.Discover(ctx, core.DiscoveryFilter{}); err != nil {
            health["status"] = "degraded"
            health["redis"] = "unavailable"
            a.Logger.WarnWithContext(ctx, "Health check: Redis unavailable", map[string]interface{}{
                "error": err.Error(),
            })
        } else {
            health["redis"] = "healthy"
        }
    }

    // Check orchestrator
    a.mu.RLock()
    orch := a.orchestrator
    a.mu.RUnlock()

    if orch != nil {
        metrics := orch.GetMetrics()
        health["orchestrator"] = map[string]interface{}{
            "status":              "active",
            "total_requests":      metrics.TotalRequests,
            "successful_requests": metrics.SuccessfulRequests,
        }
    } else {
        health["orchestrator"] = "initializing"
    }

    // Check AI
    if a.AI != nil {
        health["ai_provider"] = "connected"
    } else {
        health["ai_provider"] = "not configured"
    }

    statusCode := http.StatusOK
    if health["status"] == "degraded" || health["status"] == "unhealthy" {
        statusCode = http.StatusServiceUnavailable
    }

    setCORSHeaders(w)
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(statusCode)
    json.NewEncoder(w).Encode(health)
}

// Helper functions

func setCORSHeaders(w http.ResponseWriter) {
    w.Header().Set("Access-Control-Allow-Origin", "*")
    w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Accept, X-User-ID")
    w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, OPTIONS")
}

func writeJSON(w http.ResponseWriter, statusCode int, data interface{}) {
    setCORSHeaders(w)
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(statusCode)
    json.NewEncoder(w).Encode(data)
}

func writeError(w http.ResponseWriter, statusCode int, message string, err error) {
    response := map[string]interface{}{
        "error":   message,
        "status":  statusCode,
        "success": false,
    }
    if err != nil {
        response["details"] = err.Error()
    }
    writeJSON(w, statusCode, response)
}

8. Step 5: Add SSE Streaming (Streaming Agents Only)​

StreamCallback Interface​

Define a clean interface for SSE event types:

// sse_handler.go
package main

import (
    "encoding/json"
    "errors"
    "fmt"
    "net/http"
    "time"

    "github.com/truvaagents/truva-g3/core"
    "github.com/truvaagents/truva-g3/telemetry"
    "go.opentelemetry.io/otel/attribute"
)

// StreamCallback defines the interface for SSE event delivery.
type StreamCallback interface {
    SendStatus(step, message string)
    SendStep(stepID, tool string, success bool, durationMs int64)
    SendChunk(text string)
    SendDone(requestID string, toolsUsed []string, totalDurationMs int64)
    SendError(code, message string, retryable bool)
    SendUsage(promptTokens, completionTokens, totalTokens int)
    SendFinish(reason string)
}

SSECallback Implementation​

// SSECallback implements StreamCallback for HTTP SSE responses.
type SSECallback struct {
    w       http.ResponseWriter
    flusher http.Flusher
}

func NewSSECallback(w http.ResponseWriter, flusher http.Flusher) *SSECallback {
    return &SSECallback{w: w, flusher: flusher}
}

func (c *SSECallback) SendStatus(step, message string) {
    c.sendEvent("status", map[string]interface{}{"step": step, "message": message})
}

func (c *SSECallback) SendStep(stepID, tool string, success bool, durationMs int64) {
    c.sendEvent("step", map[string]interface{}{
        "step_id": stepID, "tool": tool, "success": success, "duration_ms": durationMs,
    })
}

func (c *SSECallback) SendChunk(text string) {
    c.sendEvent("chunk", map[string]interface{}{"text": text})
}

func (c *SSECallback) SendDone(requestID string, toolsUsed []string, totalDurationMs int64) {
    c.sendEvent("done", map[string]interface{}{
        "request_id": requestID, "tools_used": toolsUsed, "total_duration_ms": totalDurationMs,
    })
}

func (c *SSECallback) SendError(code, message string, retryable bool) {
    c.sendEvent("error", map[string]interface{}{
        "code": code, "message": message, "retryable": retryable,
    })
}

func (c *SSECallback) SendUsage(promptTokens, completionTokens, totalTokens int) {
    c.sendEvent("usage", map[string]interface{}{
        "prompt_tokens": promptTokens, "completion_tokens": completionTokens, "total_tokens": totalTokens,
    })
}

func (c *SSECallback) SendFinish(reason string) {
    c.sendEvent("finish", map[string]interface{}{"reason": reason})
}

func (c *SSECallback) sendEvent(eventType string, data interface{}) {
    jsonData, err := json.Marshal(data)
    if err != nil {
        return
    }
    fmt.Fprintf(c.w, "event: %s\ndata: %s\n\n", eventType, jsonData)
    c.flusher.Flush()
}

SSE Handler​

type SSEHandler struct {
    agent *YourChatAgent
}

func NewSSEHandler(agent *YourChatAgent) *SSEHandler {
    return &SSEHandler{agent: agent}
}

type ChatRequest struct {
    SessionID string                 `json:"session_id,omitempty"`
    Message   string                 `json:"message"`
    Options   map[string]interface{} `json:"options,omitempty"`
}

func (h *SSEHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()

    // 1. CORS headers
    w.Header().Set("Access-Control-Allow-Origin", "*")
    w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Accept, X-Requested-With, X-User-ID")
    w.Header().Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS")
    w.Header().Set("Access-Control-Max-Age", "86400")
    if r.Method == http.MethodOptions {
        w.WriteHeader(http.StatusOK)
        return
    }

    // 2. Verify SSE support
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "SSE not supported", http.StatusInternalServerError)
        return
    }

    // 3. Set SSE headers
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    w.Header().Set("X-Accel-Buffering", "no") // Disable Nginx buffering

    // 4. Only accept POST
    if r.Method != http.MethodPost {
        callback := NewSSECallback(w, flusher)
        callback.SendError("method_not_allowed", "Only POST supported", false)
        return
    }

    // 5. Parse request
    var req ChatRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        callback := NewSSECallback(w, flusher)
        callback.SendError("invalid_request", "Invalid JSON", false)
        return
    }

    if req.Message == "" {
        callback := NewSSECallback(w, flusher)
        callback.SendError("validation_error", "Message is required", false)
        return
    }

    // 6. Create or retrieve session
    userID := r.Header.Get("X-User-ID")
    sessionID := req.SessionID
    if sessionID == "" {
        session := h.agent.sessionStore.Create(userID, nil)
        sessionID = session.ID

        callback := NewSSECallback(w, flusher)
        callback.sendEvent("session", map[string]interface{}{"id": sessionID})
    }

    // Validate session exists (create new if expired)
    if session := h.agent.sessionStore.Get(sessionID); session == nil {
        session := h.agent.sessionStore.Create(userID, nil)
        sessionID = session.ID

        callback := NewSSECallback(w, flusher)
        callback.sendEvent("session", map[string]interface{}{"id": sessionID})
    }

    // 7. Store user message
    h.agent.sessionStore.AddMessage(sessionID, Message{
        Role:      "user",
        Content:   req.Message,
        Timestamp: time.Now(),
    })

    // 8. Check orchestrator
    if h.agent.GetOrchestrator() == nil {
        callback := NewSSECallback(w, flusher)
        callback.SendError("service_unavailable", "Orchestrator initializing", true)
        return
    }

    // 9. Process with streaming
    callback := NewSSECallback(w, flusher)
    if err := h.agent.ProcessWithStreaming(ctx, sessionID, req.Message, callback); err != nil {
        // Extract core.ProviderError to surface the real error code and retryability.
        // Provider 4xx errors (bad request, rate limit) are NOT retryable by the client.
        // Server errors and transient proxy errors ARE retryable.
        var pe core.ProviderError
        if errors.As(err, &pe) {
            retryable := pe.StatusCode() >= 500 || pe.IsTransient()
            h.agent.Logger.WarnWithContext(ctx, "Stream processing: provider error", map[string]interface{}{
                "operation":   "chat_stream",
                "session_id":  sessionID,
                "error":       pe.Error(),
                "status_code": pe.StatusCode(),
                "provider":    pe.Provider(),
                "retryable":   retryable,
            })
            callback.SendError(fmt.Sprintf("provider_%d", pe.StatusCode()), pe.Error(), retryable)
            return
        }

        h.agent.Logger.ErrorWithContext(ctx, "Stream processing failed", map[string]interface{}{
            "operation":  "chat_stream",
            "session_id": sessionID,
            "error":      err.Error(),
        })
        callback.SendError("processing_failed", err.Error(), true)
    }
}

SSE Event Protocol​

The SSE protocol used by TruvaG3 streaming agents:

EventWhenData Fields
sessionNew session createdid
statusPlanning/phase transitionsstep, message
stepTool call completedstep_id, tool, success, duration_ms
chunkAI token generatedtext
usageToken usage statsprompt_tokens, completion_tokens, total_tokens
finishAI finish reasonreason
doneRequest completerequest_id, tools_used, total_duration_ms
errorError occurredcode, message, retryable

9. Step 6: Add Session Management (Chat Agents Only)​

For streaming chat agents, use a Redis-backed session store to persist conversation history across requests. This is what enables multi-turn conversations where the LLM understands references to previous messages.

Session and Message Types​

// session.go
package main

type Session struct {
    ID        string                 `json:"id"`
    UserID    string                 `json:"user_id"`
    Title     string                 `json:"title"`
    CreatedAt time.Time              `json:"created_at"`
    UpdatedAt time.Time              `json:"updated_at"`
    Messages  []Message              `json:"messages"`
    Metadata  map[string]interface{} `json:"metadata,omitempty"`
}

type Message struct {
    ID        string                 `json:"id"`
    Role      string                 `json:"role"` // "user" or "assistant"
    Content   string                 `json:"content"`
    Timestamp time.Time              `json:"timestamp"`
    Metadata  map[string]interface{} `json:"metadata,omitempty"`
}

SessionStore​

// SessionStore provides Redis/Valkey-backed session management in the
// versioned DB-0 sessions subspace.
type SessionStore struct {
    client      *core.RedisClient
    ttl         time.Duration
    maxMessages int
    logger      core.Logger
}

func NewSessionStore(connection core.RedisConnectionConfig, keyspace core.RedisKeyspace, ttl time.Duration, maxMessages int, logger core.Logger) (*SessionStore, error) {
    client, err := core.NewRedisClientWithConnection(core.RedisClientConnectionOptions{
        Connection: connection,
        Namespace:  keyspace.Plain("sessions"),
        Logger:     logger,
    })
    if err != nil {
        return nil, fmt.Errorf("failed to create Redis client for sessions: %w", err)
    }
    return &SessionStore{client: client, ttl: ttl, maxMessages: maxMessages, logger: logger}, nil
}

Key SessionStore Methods​

MethodPurpose
Create(userID, metadata)Create new session, add to user index
Get(sessionID)Retrieve session by ID
AddMessage(sessionID, msg)Append message, auto-title from first user message, trim to sliding window
GetHistory(sessionID)Get full conversation history for context
List(userID, offset, limit)Paginated session list with lazy cleanup of expired entries
UpdateTitle(sessionID, title)Update session title
Delete(sessionID)Remove session and index entry

Key design points: shared DB 0 with a versioned deployment/session namespace, sliding window of 50 messages, and auto-titling from the first user message (57 chars + "..."). See examples/travel-chat-agent/session.go for complete implementation.

10. Step 7: Create the Main Entry Point​

The main.go follows a precise initialization order that's critical for telemetry and orchestrator setup.

Complete main.go​

The numbered comments below show the critical initialization order — telemetry must come before agent creation (for AI spans), and the orchestrator must init in background (after framework.Run() enables Discovery).

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "os"
    "os/signal"
    "strconv"
    "syscall"
    "time"

    "github.com/truvaagents/truva-g3/core"
    "github.com/truvaagents/truva-g3/telemetry"

    // Import AI providers for auto-detection
    _ "github.com/truvaagents/truva-g3/ai/providers/anthropic"
    _ "github.com/truvaagents/truva-g3/ai/providers/openai"
)

func main() {
    if err := run(); err != nil {
        // run has already executed its deferred cleanup.
        log.Fatalf("Agent failed: %v", err)
    }
}

func run() error {
    // 1. Validate configuration (fail fast)
    if err := validateConfig(); err != nil {
        return fmt.Errorf("configuration: %w", err)
    }

    // 2. Set component type for telemetry service_type labels
    core.SetCurrentComponentType(core.ComponentTypeAgent)

    // 3. Initialize telemetry BEFORE creating agent (critical for AI spans)
    initTelemetry("your-agent")
    defer func() {
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()
        telemetry.Shutdown(ctx)
    }()

    // 4. Create agent AFTER telemetry
    agent, err := NewYourAgent()
    if err != nil {
        return fmt.Errorf("create agent: %w", err)
    }

    // 5. Create framework
    middlewareConfig := &telemetry.TracingMiddlewareConfig{
        ExcludedPaths: []string{"/health", "/metrics", "/ready", "/live"},
    }

    framework, err := core.NewFramework(agent,
        core.WithName("your-agent"),
        core.WithPort(getPort()),
        core.WithNamespace(os.Getenv("NAMESPACE")),
        core.WithRedisURL(os.Getenv("REDIS_URL")),
        core.WithDiscovery(true, "redis"),
        core.WithCORSDefaults(),
        core.WithMiddleware(telemetry.TracingMiddlewareWithConfig("your-agent", middlewareConfig)),
    )
    if err != nil {
        return fmt.Errorf("create framework: %w", err)
    }

    // 6. Register the application's discovery-dependent initializer.
    // See agent-with-orchestration for the runnable and synchronized accessors.
    framework.RegisterRunnable(&orchestratorInitializer{agent: agent})
    defer agent.stopOrchestrator() // after HTTP and runnable draining

    // 7. Graceful shutdown
    ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer cancel()

    // 8. Run framework (blocking)
    log.Printf("Starting agent on port %d", getPort())
    if err := framework.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
        return fmt.Errorf("run framework: %w", err)
    }
    // Built-in HTTP draining is complete (or returned its timeout error).
    // Deferred cleanup of application-owned clients/telemetry runs afterwards.
    return nil
}

// validateConfig(), initTelemetry(), getPort() — standard helpers.
// See examples/travel-chat-agent/main.go for configuration helpers, and
// examples/agent-with-orchestration for the managed initializer and lifecycle.

When discovery retry is enabled, BaseAgent.Initialize can return before Redis becomes available. A one-shot InitializeOrchestrator(agent.Discovery) therefore cannot provide recovery. The application initializer implements core.Runnable: it checks ctx.Err(), reads agent.GetDiscovery(), and waits cancellably using Config.Discovery.RetryInterval until discovery is available. It initializes the orchestrator with that context and publishes the successfully started instance through a synchronized accessor used by handlers. While waiting, orchestration execution endpoints return unavailable and health checks remain accessible.

Use GetDiscovery() for runtime reads; direct Discovery assignment is for configuration before initialization. Stop the application-owned orchestrator after Framework.Run returns, and guard against a late initializer publishing after cleanup has begun. The complete implementation is in agent-with-orchestration/research_agent.go. This pattern requires the framework release containing GetDiscovery and the joined HTTP lifecycle; local workspace builds can use the implementation before that release is published.

WithCORSDefaults() is convenient for local browser-facing examples because it accepts the headers used by the reference chat UI. For production, restrict origins and configure only the methods and headers your application needs; do not treat the permissive development default as a production security policy.

Background Jobs: core.Runnable and framework.RegisterRunnable​

Some agents need to do work in the background, parallel to serving HTTP requests — periodic tasks, queue consumers, expiry processors, scheduled jobs. Don't hand-roll a goroutine in main.go and try to manage its lifecycle by hand. Use core.Runnable instead.

A Runnable is anything that satisfies a single-method interface (core/interfaces.go):

type Runnable interface {
    Start(ctx context.Context) error
}

Implement Start(ctx), register the instance with the framework, and the framework starts it in a goroutine alongside the HTTP server, drains it on shutdown, and emits structured lifecycle logs under the framework_register_runnable, framework_runnable_start, framework_runnable_exit, and framework_runnable_drain operations (see LOGGING_IMPLEMENTATION_GUIDE.md).

Minimal example — a periodic background job:

type MyJob struct {
    interval time.Duration
    logger   core.Logger
}

func (j *MyJob) Start(ctx context.Context) error {
    ticker := time.NewTicker(j.interval)
    defer ticker.Stop()
    j.logger.Info("MyJob started", map[string]interface{}{
        "operation": "my_job",
        "interval":  j.interval.String(),
    })
    for {
        select {
        case <-ticker.C:
            // Do work. Log errors internally; don't propagate them
            // unless they're truly fatal — fail-open keeps the job running.
            _ = j.runOnce(ctx)
        case <-ctx.Done():
            j.logger.Info("MyJob stopping (context cancelled)", map[string]interface{}{
                "operation": "my_job",
            })
            return nil
        }
    }
}

Wire it up in main.go after creating the framework but before framework.Run:

job := &MyJob{interval: 1 * time.Hour, logger: agent.Logger}
framework.RegisterRunnable(job)

// framework.Run now starts the HTTP server AND the runnable in parallel,
// and drains the runnable on ctx cancel.
if err := framework.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
    log.Fatalf("Framework error: %v", err)
}

That's the entire wiring. Three lines instead of the goroutine + waitgroup + defer-stop dance you'd otherwise write. The framework handles all of:

  • Starting the runnable in a goroutine
  • Logging when it starts and when it exits
  • Distinguishing clean exit (nil or context.Canceled) from runtime error (everything else, logged at ERROR with error_type=runnable_exit)
  • Draining all registered runnables after the HTTP server stops, with a configurable timeout
  • Logging a warning if the drain timeout fires (error_type=runnable_drain_timeout)

The contract you must honour:

  1. Start(ctx) blocks until ctx is cancelled. This is the entire interface — no Stop(), no Cancel(), no companion methods. ctx cancellation drives shutdown.
  2. You must respond to ctx.Done() within TRUVAG3_FRAMEWORK_RUNNABLE_DRAIN_TIMEOUT (default 10s, raise via env var if your job needs a longer grace period). Buggy runnables that ignore ctx will leak goroutines until process exit — Go provides no mechanism for forcibly terminating them.
  3. Return nil on graceful shutdown, return an error on startup or runtime failure. context.Canceled is treated as clean exit.
  4. Don't call Start yourself. The framework calls it exactly once per registered runnable when Run(ctx) is invoked.
  5. Register before Run. Calling RegisterRunnable after Run has started is undefined behaviour.

Reference implementation: memory.ReflectionJob is the in-tree reference core.Runnable — a periodic background job that distills episodic events into long-term knowledge. It's registered in the devops-chat-agent, qa-agent, and event-driven-agent examples (the event-driven-agent also shows the worker-mode pattern where the runnable is started in a goroutine tied directly to the worker context, since worker mode has no core.Framework instance). See AGENT_MEMORY_USER_GUIDE.md — Long-Term Knowledge Retention for the conceptual overview and examples/devops-chat-agent/main.go for the production wiring.

Use cases that should be a Runnable:

  • Periodic jobs (every interval, do X)
  • Queue consumers (block on a channel or external queue, process messages)
  • Schedulers (look up the next due item, sleep until then, run it)
  • Expiry processors (sweep expired entries from a store)
  • Lease renewal loops
  • Health-driven reconciliation loops

Use cases that should NOT be a Runnable:

  • Per-request work — use a handler
  • Async work spawned by a request — use a goroutine inside the handler with context.Background() if it must outlive the request
  • One-shot startup tasks — call them synchronously before framework.Run
  • One-shot shutdown tasks — defer them in main.go

Receiving Scheduled Tasks​

If your agent needs to receive tasks scheduled for the future (e.g., "check this service again in 1 hour"), wire the scheduled endpoint before framework.Run():

// Lazy orchestrator resolution — supports async init
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()})
}

This mounts /api/v1/scheduled on your agent. When a scheduled task fires, the centralized scheduled-executor POSTs to this endpoint with the instruction, and your agent's orchestrator plans and executes it using whatever tools are available.

Note: This is different from Runnables. A Runnable is a background job your agent owns end-to-end (like a periodic sweep). Receiving scheduled tasks is an HTTP endpoint where work arrives from the external scheduled-executor service. Use Runnables for agent-internal periodic work; use RegisterScheduledEndpoint for receiving externally-scheduled tasks.

For the full story on scheduling -- architecture, delivery semantics, observability, troubleshooting, and writing custom backends -- see the Scheduled Tasks Guide.

11. Step 8: Add Deployment Files​

go.mod​

The basic orchestration agent in this guide directly depends on four framework modules: ai, core, orchestration, and telemetry.

module github.com/truvaagents/truva-g3/examples/your-agent

go 1.27.2

require (
    github.com/google/uuid v1.6.0
    github.com/truvaagents/truva-g3/ai v0.6.0
    github.com/truvaagents/truva-g3/core v0.6.0
    github.com/truvaagents/truva-g3/orchestration v0.6.0
    github.com/truvaagents/truva-g3/telemetry v0.6.0
    go.opentelemetry.io/otel v1.47.0
)

// Use local workspace modules for development
replace (
    github.com/truvaagents/truva-g3/ai => ../../ai
    github.com/truvaagents/truva-g3/core => ../../core
    github.com/truvaagents/truva-g3/orchestration => ../../orchestration
    github.com/truvaagents/truva-g3/resilience => ../../resilience
    github.com/truvaagents/truva-g3/telemetry => ../../telemetry
)

Those are the minimum framework modules for this orchestration example. memory is optional, but an agent that directly uses memory APIs or registers memory.ReflectionJob must add the matching module and local replacement:

require github.com/truvaagents/truva-g3/memory v0.5.0

replace github.com/truvaagents/truva-g3/memory => ../../memory

.env​

# Required: Redis for service discovery and session storage
REDIS_URL=redis://localhost:6379

# Required: HTTP server port (see examples/README.md for port allocation)
PORT=8357

# AI Provider (auto-detected — set at least one)
OPENAI_API_KEY=sk-...
# ANTHROPIC_API_KEY=sk-ant-...

# Optional: Model alias overrides (pattern: TRUVAG3_{PROVIDER}_MODEL_{ALIAS})
# These override the portable model aliases (default, fast, smart, premium, code, vision)
# without requiring code changes. See AI_PROVIDERS_SETUP_GUIDE.md for full reference.
# TRUVAG3_OPENAI_MODEL_DEFAULT=gpt-5.6-terra
# TRUVAG3_OPENAI_MODEL_FAST=gpt-5.6-luna
# TRUVAG3_OPENAI_MODEL_SMART=gpt-5.6-sol
# TRUVAG3_ANTHROPIC_MODEL_DEFAULT=claude-sonnet-5

# Telemetry
APP_ENV=development
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318

# Debug
DEV_MODE=true
TRUVAG3_LOG_LEVEL=debug

Streaming Chat Agent .env Checklist​

Streaming chat agents that serve a frontend (e.g., chat-ui) require additional environment variables beyond the basics. Use this checklist when creating a new agent to avoid missing configuration:

VariableRequiredDefaultPurpose
REDIS_URLYes—Service discovery and session storage
PORTYes—HTTP server port (port allocation)
AI provider key(s)Yes (at least one)—OPENAI_API_KEY, ANTHROPIC_API_KEY, OPENROUTER_API_KEY, GROQ_API_KEY, etc.
TRUVAG3_CORS_HEADERSYes (chat agents)Content-Type,AuthorizationMust include X-User-ID for chat-ui: Content-Type,Authorization,X-User-ID,X-Requested-With
TRUVAG3_SYNTHESIS_MAX_TOKENSRecommended30000Max output tokens for LLM synthesis. Reasoning and visible text share this cap
TRUVAG3_EXECUTION_DEBUG_STORE_ENABLEDRecommendedfalseStores orchestration DAGs in the versioned DB 0 execution-debug subspace for Registry Viewer inspection
TRUVAG3_LLM_DEBUG_ENABLEDRecommendedfalseStores LLM request/response payloads for debugging
DEV_MODENofalseEnables verbose logging
TRUVAG3_LOG_LEVELNoinfoLog verbosity: debug, info, warn, error
TRUVAG3_DEBUGNofalseEnables debug mode (sets log level to debug)
APP_ENVNodevelopmentTelemetry profile: development, staging, production
OTEL_EXPORTER_OTLP_ENDPOINTNo—OpenTelemetry collector for tracing/metrics
TRUVAG3_{PROVIDER}_MODEL_{ALIAS}NoProvider defaultModel alias overrides (e.g., TRUVAG3_OPENAI_MODEL_DEFAULT=gpt-5.6-terra)

Common mistakes:

  • Missing TRUVAG3_CORS_HEADERS — the chat-ui frontend sends X-User-ID via a custom header; without this setting, CORS blocks the request silently
  • Low TRUVAG3_SYNTHESIS_MAX_TOKENS — the default is 30000 shared output tokens; explicit smaller budgets can truncate complex responses
  • Missing TRUVAG3_EXECUTION_DEBUG_STORE_ENABLED — without this, the Registry Viewer's "Execution DAG" tab shows no data, making orchestration debugging harder

Dockerfile.workspace, k8-deployment.yaml, setup.sh​

Follow the same patterns as tools. See TOOL_DEVELOPMENT_GUIDE.md - Step 6: Add Deployment Files for templates — including the Observability Identity subsection that covers the three-way alignment rule between pod app: label, OTEL_SERVICE_NAME, and the logger's service field. Agents are subject to the same rule.

Key differences for agents:

AspectToolAgent
COPY in Dockerfilecore/, telemetry/ai/, core/, orchestration/, resilience/, telemetry/; add memory/ when the agent imports that module
K8s component labeltoolagent
K8s resource limitsLower (50m-100m CPU)Higher (100m-500m CPU) — AI calls are compute-intensive
SecretsTool-specific API keysAI provider API keys

Agent-owned skill packages​

For a skill-enabled example, keep each reviewable package at:

skills/packages/<namespace>/<skill-name>.json

Source files in Git are the deployable definition, and the published store is the runtime source read by agent replicas. The store is derived state because setup can reconstruct it from the reviewed packages. Source the shared k8-deployment/setup-env-lib.sh as other examples do, then add thin wrappers to the agent's setup.sh:

prepare_skills() {
    truvag3_prepare_agent_skills "$SCRIPT_DIR/skills/packages"
}

sync_skills() {
    truvag3_sync_agent_skills "$SCRIPT_DIR/skills/packages"
}

check_skills() {
    truvag3_check_agent_skills "$SCRIPT_DIR/skills/packages"
}

Call prepare_skills after shared infrastructure is ready and before creating or restarting the workload in every state-changing deployment path, including full-deploy, deploy, rebuild, and rollout. It is best-effort: it warns when validation or the management API is unavailable but does not block the agent deployment. Expose skills-sync and skills-check as explicit setup commands; those commands are strict so operators and continuous integration can detect publication failures or drift.

The helper discovers every valid package automatically and skips publication when the published content is already equivalent. A missing or empty package directory is a successful no-op, so a reusable setup scaffold can include this integration before an agent owns any skills.

Use TRUVAG3_SKILLS_API_URL when the management API is not available at the local example default, http://registry.localhost/api/v1/skills. On a setup host that intentionally cannot reach the API, set TRUVAG3_SKIP_SKILLS_SYNC=true to skip only the automatic deployment attempt. The explicit skills-sync and skills-check commands ignore this switch and remain strict. These variables control the example setup process; they do not configure the orchestrator's runtime skill selection.

The management API writer and the agent's SkillRegistry reader must address the same logical data store. With the included adapter, keep the Redis/Valkey topology and TRUVAG3_REDIS_NAMESPACE aligned between the management host and every skill-enabled agent. Setup synchronization uses the HTTP management API; it must not write provider keys directly.

After this one-time setup integration, adding another skill needs only a new package file and a matching runtime binding. Do not add one shell command per package. Removing a source file does not automatically delete a published revision; remove the agent binding and use the management lifecycle when deletion is intentional. See examples/AGENTS.md for the repository rules and the Agent Skills Guide for the full authoring, publication, runtime, and troubleshooting lifecycle.

12. Logging and Observability​

The Logger Interface​

Every agent gets a Logger from core.BaseAgent. It provides both basic and context-aware methods:

a.Logger.InfoWithContext(ctx, "message", fields)  // In handlers (includes trace_id, span_id)
a.Logger.Info("message", fields)                   // Startup/shutdown only (no request context)
// Also: Warn, Error, Debug — each has a WithContext variant

When to Use Each Level​

LevelWhenExample
DebugDetailed internal state, request payloads"Request payload", "Streaming stats"
InfoNormal operations, request lifecycle"Processing request", "Request completed"
WarnRecoverable issues, degraded operation"Discovery not available", "AI fallback activated"
ErrorFailures requiring attention"Orchestration failed", "Redis unavailable"

Standard Log Fields​

Use consistent field names across all handlers:

FieldTypeWhenDescription
operationstringEvery logThe capability name (e.g., "process_chat", "natural_orchestration")
session_idstringChat agentsSession identifier
request_idstringAfter orchestrationOrchestrator-generated request ID
statusstringCompletion"success" or "failure"
duration_msint64CompletionOperation duration in milliseconds
errorstringError logsError message
tools_usedint/[]stringCompletionNumber or list of tools invoked

Context-Aware Logging Pattern​

Always use WithContext in handlers for automatic trace-log correlation:

// GOOD — trace_id and span_id are automatically injected
a.Logger.InfoWithContext(ctx, "Processing request", map[string]interface{}{
    "operation":  "process_chat",
    "session_id": sessionID,
})

// BAD — loses trace correlation
a.Logger.Info("Processing request", map[string]interface{}{
    "operation":  "process_chat",
    "session_id": sessionID,
})

When logs include trace context, you can search in your log aggregator (Loki, CloudWatch) by trace_id and jump directly to the corresponding Jaeger trace.

Environment Configuration​

VariableDefaultDescription
TRUVAG3_LOG_LEVELinfoLog level: debug, info, warn, error
TRUVAG3_LOG_FORMATjsonOutput format: json or text (auto-detects json in K8s)
TRUVAG3_DEBUGfalseEnable debug mode (sets level to debug)

Key Rules​

  1. WithContext in handlers, basic methods in startup/shutdown only
  2. Always include operation field — your primary filter key
  3. Never log API keys — use "has_key": apiKey != "" instead
  4. Log both success and failure with duration_ms
  5. Init telemetry before agent — otherwise AI calls produce silent logs
  6. log.Fatalf only before framework — use agent.Logger after core.NewFramework()

Logs auto-include a component field (agent/<name>, framework/orchestration, framework/ai) for filtering. With telemetry enabled, logs auto-emit Prometheus metrics using operation, status, error_type as labels.

Reference: LOGGING_IMPLEMENTATION_GUIDE.md for component filtering, metric emission, HITL tracing, and production recipes.

13. Distributed Tracing​

Tracing is handled by three layers — all already shown in the code above:

  1. Server-side — TracingMiddlewareWithConfig in main.go extracts/creates trace context
  2. Client-side — TracedHTTPClientWithTransport propagates trace context to downstream tools
  3. Manual enrichment — telemetry.AddSpanEvent() and telemetry.RecordSpanError() add business context to spans

Key APIs for handlers:

// Add span events at key lifecycle points
telemetry.AddSpanEvent(ctx, "orchestration.started", attribute.String("session_id", sessionID))

// Record errors on spans (appears as red in Jaeger)
telemetry.RecordSpanError(ctx, err)

// Record metrics for dashboards
telemetry.RecordRequest(telemetry.ModuleOrchestration, "natural_request", durationMs, "success")

Critical init order (already shown in main.go): SetCurrentComponentType → initTelemetry → NewAgent. If reversed, AI calls won't generate spans.

Reference: DISTRIBUTED_TRACING_GUIDE.md for Jaeger setup, cross-service correlation, span event patterns, and troubleshooting.

14. Testing Your Agent​

Prerequisites​

# 1. Start Redis
docker run -d -p 6379:6379 redis:alpine

# 2. Start Jaeger (optional, for tracing)
docker run -d --name jaeger \
  -p 16686:16686 -p 4318:4318 \
  -e COLLECTOR_OTLP_ENABLED=true \
  jaegertracing/all-in-one:latest

# 3. Deploy tools your agent will orchestrate
# (e.g., weather-tool, currency-tool, devops-tool)

Build and Run​

# Using setup.sh (recommended)
./setup.sh run

# Or manually
cd examples/your-agent
set -a && source .env && set +a
GOWORK=off go build -o your-agent .
./your-agent

Test Non-Streaming Agent​

# Health check
curl http://localhost:8357/health | jq .

# Discover tools
curl http://localhost:8357/discover | jq .

# Natural language orchestration
curl -X POST http://localhost:8357/orchestrate/natural \
  -H "Content-Type: application/json" \
  -d '{"request": "What is the cluster status?"}'

Test Streaming Agent​

# Create session
curl -X POST http://localhost:8357/chat/session \
  -H "Content-Type: application/json" \
  -H "X-User-ID: test-user" | jq .

# SSE streaming chat
curl -N -X POST http://localhost:8357/chat/stream \
  -H "Content-Type: application/json" \
  -H "X-User-ID: test-user" \
  -d '{"message": "What pods are running in the default namespace?"}'

# Get conversation history
curl http://localhost:8357/chat/session/{session_id}/history | jq .

Test Agent Skills​

For a skill-enabled Kubernetes example, verify both deployment state and runtime behavior:

# Confirm that every Git-authored package matches the published representation.
./setup.sh skills-check

Then send at least one request that should activate each auto skill and one nearby request that should not. Confirm that always skills activate on every ordinary request that reaches skill preparation, and test trusted application activation separately for any explicit binding. If the agent uses multi-phase planning or HITL, also exercise continuation or resume and verify that the exact version stays pinned while boundary-relevant resources are selected again.

When stored execution evidence is enabled, open Registry Viewer → Executions, choose the request, and open its Skills tab. Verify the ordered Pin, Activate, Select resources, Load, and Project records. Use the LLM Calls or LLM Debug view only when you need to inspect the effective prompt; execution skill records deliberately exclude instruction and resource bodies. In Jaeger, verify the skill lifecycle spans and selector calls under the same request trace. The full evidence setup and expected fields are documented in Agent Skills Guide §13.

Viewing Traces​

Open Jaeger at http://localhost:16686, select your agent's service name, and search by operation (e.g., POST /chat/stream) or request_id from the API response.

15. Adding Human-in-the-Loop (HITL) Approval​

If your agent performs high-stakes operations (deployments, ticket creation, sending messages, financial transactions), you should add HITL checkpoints that pause execution and wait for human approval before proceeding.

15.1 When to Use HITL​

Add HITL approval when your agent's tools can:

  • Modify external systems — Kubernetes rollouts, database writes, API calls with side effects
  • Send communications — Slack messages, emails, ticket creation
  • Incur costs — Cloud resource provisioning, paid API calls
  • Make irreversible changes — Data deletion, production deployments

Mark these capabilities as sensitive in your .env:

TRUVAG3_HITL_ENABLED=true
TRUVAG3_HITL_STEP_SENSITIVE_CAPABILITIES=rollout_restart,scale_deployment,send_message,create_ticket

# Required when multiple HITL-enabled agents share the same Redis instance.
# Without it, agents share a single pending-checkpoint index and the registry
# viewer's HITL list will show duplicate rows. K8s manifests set this via the
# Deployment env; for local dev with more than one HITL agent, set it explicitly.
# See HUMAN_IN_THE_LOOP_USER_GUIDE.md#agent-isolation for the full mechanism.
TRUVAG3_AGENT_NAME=my-agent

The orchestrator will automatically pause before executing any step that uses a sensitive capability, creating a checkpoint that stores the full execution state (plan, completed steps, resolved parameters). At construction, the framework logs a startup WARN HITL checkpoint store using shared key prefix if no isolating identifier was supplied — treat that log line as a configuration error, not noise.

15.2 Resume Context: The Critical Contract​

Managed requests using lease-aware shared memory release investigation handles when a segment suspends or otherwise exits. BuildResumeContext restores checkpoint metadata; the shared request runner passes it to pre-planning hooks with explicit adapter metadata taking precedence. Those hooks acquire fresh ownership for the resumed segment. Keep acquisition handles private rather than placing them in checkpoint metadata. Direct hook or ExecutePlan* composition does not supply this lifecycle. The joined cleanup bound defaults to 5s and is configurable with TRUVAG3_EXECUTION_CLEANUP_TIMEOUT / WithExecutionCleanupTimeout; explicit invalid values fail construction through error-returning factories. Legacy errorless helpers preserve the failure and reject Start/execution instead of silently using defaults. Prefer CreateOrchestratorFromEnvironment or CreateResolvedOrchestrator and check the returned error before serving requests.

Resume through ResumeCoordinator. It claims the approved checkpoint, restores context, renews ownership during work, and saves the actual terminal outcome. The application supplies its normal processing adapter:

executor := orchestration.ResumeExecutorFunc(func(
    ctx context.Context, checkpoint *orchestration.ExecutionCheckpoint,
) (*orchestration.ExecutionResult, error) {
    // Use the real terminal result; successful tool steps alone are not success.
    _, execution, err := orchestrator.ProcessRequestWithExecution(
        ctx, checkpoint.OriginalRequest, checkpoint.UserContext,
    )
    return execution, err
})
config, err := orchestration.LoadResumeCoordinatorRuntimeConfigFromEnvironment(
    orchestration.DefaultResumeCoordinatorRuntimeConfig(), os.LookupEnv,
)
if err != nil { return err }
coordinator, err := orchestration.NewResumeCoordinator(
    checkpointStore, executor, config, orchestration.WithResumeLogger(logger),
)
if err != nil { return err }

// At the request boundary:
_, err = coordinator.ResumeExecution(ctx, checkpointID)
return err

For streaming, call ResumeWithExecutor with a request-local callback and ProcessRequestStreamingWithExecution. Forward progress, but send done or a new checkpoint only after the coordinator finalizes. Before SSE headers, return the status/code from HITLErrorResponse; afterward send an error event with retryable: false. Never infer success from a nil error alone.

HTTP/SSE work follows caller cancellation. Accepted tasks use their worker context independently of the submitting HTTP request. A failed task is not automatically re-enqueued; inspect effects before an explicit retry.

The coordinator calls the following context helper once; applications do not call it a second time or use it to bypass claims.

BuildResumeContext (hitl_helpers.go) sets up the full resume contract in a single call:

What it setsWhy it matters
WithResumeMode(ctx, checkpointID)Prevents re-triggering the same checkpoint
WithPlanOverride(ctx, plan)Reuses the stored plan (step IDs must match for skip logic)
WithCompletedSteps(ctx, results)Skips already-executed steps
WithPreResolvedParams(ctx, params, stepID)Uses the human-approved parameter values
WithRequestMode(ctx, mode)Preserves streaming/non-streaming expiry behavior
Linked hitl.resume spanLinks the resumed execution to the original trace; the returned cleanup function must be called
Validated conversation contextRestores canonical conversation_id to core context, checkpoint metadata, and metric-ineligible W3C Baggage when present
WithMetadata(ctx, userContext)Preserves session metadata (session ID, user info, etc.)

It also validates that the checkpoint has a resumable status (approved or expired_approved), returning an error for terminal or pending checkpoints.

15.3 Common Pitfall: Manual Context Setup​

Do not manually call individual With* helpers in your resume handler:

// BAD — fragile, will drift as the framework evolves
ctx = orchestration.WithResumeMode(ctx, checkpointID)
// Missing WithPlanOverride → orchestrator replans from scratch
// Missing WithCompletedSteps → orchestrator re-executes already-completed steps
// Missing WithPreResolvedParams → human-approved parameter values are ignored
// Missing WithRequestMode, WithMetadata → expiry behavior and session context lost

The orchestrator's executePhaseLoop checks GetPlanOverride(ctx) and GetCompletedSteps(ctx) to restore accumulated phase state. If WithPlanOverride is absent, the orchestrator triggers a fresh LLM planning call and generates a new plan with different step IDs. If WithCompletedSteps is absent, the executor sees no previously completed work and re-runs every step. Either omission causes a full or partial replay of work that already ran. Missing WithPreResolvedParams is subtler — the approved parameter values are silently dropped and the orchestrator substitutes AI-generated values instead.

This is a context drift problem: as the framework grows the resume contract (from 3 to 6 context values), any manual With* call list becomes stale without a compiler error or test failure to warn you. BuildResumeContext is the single source of truth and automatically applies the full contract as it evolves.

Deep dive: For the complete HITL setup — checkpoint stores, expiry callbacks, SSE streaming integration, status lifecycle, and configuration reference — see HUMAN_IN_THE_LOOP_USER_GUIDE.md.

Reference implementations:

16. Best Practices​

  1. Orchestration endpoints Internal: true — /orchestrate/* and /chat/stream must be internal to prevent recursive self-calls. Domain-specific endpoints may use Internal: false for multi-agent hierarchies
  2. Separate prompt responsibilities — keep identity and universal behavior in PromptConfig; use explicitly bound skills for conditional, reusable, or versioned procedures
  3. Deferred orchestrator init — use a registered core.Runnable that waits cancellably for GetDiscovery(); publish the started orchestrator through a synchronized accessor
  4. Check orchestrator availability before processing — return 503 if still initializing
  5. Surface AI provider errors correctly — Use errors.As(err, &pe) with core.ProviderError to extract the real HTTP status from AI provider errors. Return the original status code (e.g., 400, 429) instead of wrapping everything as 500. See the handler example in Step 4
  6. Extended AI timeouts — 240s+ for reasoning models; cap session history at ~50 messages
  7. Never log API keys — use env vars for all credentials, restrict CORS in production
  8. Backend role isolation — use one Redis/Valkey DB 0 and the versioned deployment/subsystem keyspaces; keep the topology and TRUVAG3_REDIS_NAMESPACE aligned across management and runtime readers
  9. Keep skill eligibility explicit — publishing a package does not grant access; bind every skill the agent may use through code or deployment configuration

17. Troubleshooting​

ProblemSymptomsCheck
Orchestrator not initializingHealth shows "orchestrator": "initializing"Redis connectivity (redis-cli ping), REDIS_URL format, AI provider key
Recursive self-callsAgent calling its own endpointsAll orchestration capabilities must be Internal: true
AI 4xx errors returned as 500Client sees 500 for provider bad-request/rate-limitAdd core.ProviderError check with errors.As() before generic 500 — see Step 4
No AI providers"no providers detected" on startupAPI key env vars set, blank import present (_ ".../ai/providers/openai")
Wrong modelLogs show unexpected model nameCheck env | grep TRUVAG3_ for alias overrides
Skill is not selectedRequest succeeds without expected skill guidanceCheck that the package is published, namespace/name match the binding, skills are enabled, and the binding activation mode matches the request
Skill setup warns during deployAgent deploys but package reconciliation did not completeRun ./setup.sh skills-check, then use strict ./setup.sh skills-sync after repairing API reachability or package validation
No traces in JaegerRequests succeed, no tracesOTEL_EXPORTER_OTLP_ENDPOINT set, init order correct, TracingMiddleware added
Missing trace_id in logsLogs lack trace_id/span_idUse WithContext methods, ctx from r.Context()
SSE events not arrivingConnection opens, no eventshttp.Flusher supported, X-Accel-Buffering: no for Nginx, flusher.Flush() called

See also: AI_PROVIDERS_SETUP_GUIDE.md for AI provider troubleshooting, DISTRIBUTED_TRACING_GUIDE.md for trace debugging.

18. Quick Reference​

Minimal Lifecycle Wiring​

Use the agent and framework construction from Step 7 above. After registering the HTTP capabilities and before calling Run, add the managed initializer:

// Fragment inside run(); agent and framework have already been constructed.
framework.RegisterRunnable(&orchestratorInitializer{agent: agent})
defer agent.stopOrchestrator()

ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer cancel()
if err := framework.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
    return fmt.Errorf("run framework: %w", err)
}
return nil

orchestratorInitializer and stopOrchestrator are application types/methods, defined in the complete agent-with-orchestration implementation. The initializer uses the framework discovery retry interval and GetDiscovery(); handlers use a synchronized orchestrator accessor and return HTTP 503 while it is unavailable. Cleanup runs after HTTP draining and prevents a late initializer from publishing an orchestrator. This keeps startup recovery, request handling, and shutdown consistent with the lifecycle described in Step 7.

Environment Variables Reference​

VariableRequiredDescription
REDIS_URLYesRedis connection URL
PORTYesHTTP server port
OPENAI_API_KEYOne AI key requiredOpenAI API key
ANTHROPIC_API_KEYOne AI key requiredAnthropic API key
TRUVAG3_CORS_HEADERSStreaming agentsAllowed CORS headers (include X-User-ID for chat-ui)
TRUVAG3_SYNTHESIS_MAX_TOKENSNoMax tokens for synthesis (default: 30000; reasoning and final output share the cap)
TRUVAG3_EXECUTION_DEBUG_STORE_ENABLEDNoStore orchestration DAGs in the versioned DB 0 execution-debug subspace for Registry Viewer
TRUVAG3_LLM_DEBUG_ENABLEDNoStore LLM request/response payloads for debugging
DEV_MODENoEnable verbose logging
TRUVAG3_LOG_LEVELNodebug, info, warn, error
TRUVAG3_DEBUGNoEnable debug mode (sets level to debug)
APP_ENVNodevelopment, staging, production
OTEL_EXPORTER_OTLP_ENDPOINTNoOpenTelemetry collector endpoint
TRUVAG3_PLAN_MAX_TOKENSNoMax tokens for planning (default: 20000)
TRUVAG3_{PROVIDER}_MODEL_{ALIAS}NoOverride model alias (e.g., TRUVAG3_OPENAI_MODEL_SMART=gpt-5.6-sol)
TRUVAG3_SKILLS_ENABLEDNoEnable skill processing; the effective binding list must also be nonempty
TRUVAG3_SKILL_BINDINGS_JSONNoReplace the complete code-owned skill binding list with deployment configuration
TRUVAG3_REDIS_NAMESPACENoSelect the deployment namespace for all versioned DB 0 keys; management and runtime must use the same value
TRUVAG3_SKILLS_API_URLNoSetup-only override for the Skills management API base URL
TRUVAG3_SKIP_SKILLS_SYNCNoSetup-only true switch that skips automatic best-effort synchronization; strict commands ignore it

For additional AI provider keys (GROQ_API_KEY, DEEPSEEK_API_KEY, GEMINI_API_KEY), base URL overrides, and Ollama setup, see AI_PROVIDERS_SETUP_GUIDE.md. For the full streaming chat agent .env checklist, see Step 8: Streaming Chat Agent .env Checklist.

See Also​

Core Documentation​

Advanced Features​

Telemetry & Observability​

Reference Implementations​

AgentTypeKey Features
examples/travel-chat-agentStreamingSSE streaming, session management, conversation history, multi-phase planning
examples/agent-with-orchestrationNon-streamingNatural language + workflow orchestration, predefined DAGs, AI synthesis toggle
examples/agent-with-human-approvalStreaming + HITLHuman approval checkpoints, auto-resume timeouts, approval audit trail
examples/event-driven-agentAsync + HITLWebhook ingestion, alert dedup, async task queue, HITL approval for high-stakes ops