agent

package
v0.0.0-...-fd33c92 Latest Latest
Warning

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

 Go to latest Published: Jul 24, 2026 License: MIT Imports: 47 Imported by: 0

Documentation

Overview

CheckPointStore implements Eino's compose.CheckpointStore using Redis (D-083).

Package agent provides the AI agent configuration and registry system.

Agents are defined using Markdown files with YAML front matter. The front matter contains configuration (ID, model, parameters), and the body is the agent's system prompt.

Agent identity is determined by exact-match lookup in the AgentRegistry; the "agent/{id}" format is a convention, not enforced by the server (D-054 revised).

Context Management

The ContextManager interface provides conversation history loading with token-based trimming and in-memory caching. DBContextManager is the default implementation, backed by the MessageStore with sync.Map caching.

Phase 4: LLM Provider and Agent Building

LLMProvider and LLMClientFactory (D-064, D-066): LLMProvider is the interface each backend (OpenAI, Claude, Ollama, Qwen) implements to construct a ChatModel. LLMClientFactory holds provider registrations and selects the right one via model-name / base-URL heuristics (detectProvider). Each provider supplies a DefaultBaseURL so agents work with zero configuration (D-064). API keys are read from environment variables and never appear in error messages or logs.

AgentBuilder and BuiltAgent: AgentBuilder wraps LLMClientFactory and produces a BuiltAgent (an Eino Runner + the config it was built from). Build performs three steps: create a ChatModel via the factory, wrap it in a ChatModelAgent with the agent's system prompt as instruction, then create a Runner with streaming enabled.

Phase 4: Streaming and Broadcasting

StreamBridge (D-051, 50ms throttle, cumulative text): StreamBridge converts Eino AsyncIterator events into StreamChunks. Each StreamChunk.Content is a cumulative text snapshot (not a delta), so receivers replace their display buffer directly without maintaining state. A 50ms throttle yields ~20fps streaming; dropped frames do not affect correctness. The event-reading goroutine wraps iter.Next() in a select so it is cancellable via context.

BroadcastHelper (D-050 ephemeral, dual broadcast): BroadcastHelper sends typing indicators and streaming updates to conversation participants via the SyncBroadcast hub. SendStreamUpdate broadcasts to both the human user and the agent user; SendTyping broadcasts to the human user only. Cumulative text snapshots use is_done=false during streaming and is_done=true on completion (D-052). Typing indicators are ephemeral: sent true before agent work begins and cleared either on the first token (D-065) or on exit.

Phase 4: Execution Pipeline

AgentExecutor (D-062, D-065, D-067 orchestration pipeline): AgentExecutor orchestrates the full execution pipeline: acquire semaphore (optional concurrency limit), apply 120s total timeout, look up agent config from registry, send typing=true, load conversation context, build the agent, convert messages to Eino schema, run the agent, bridge the stream, broadcast chunks, send is_done=true, and persist the final message. ExecuteWithErrorMessage wraps Execute and sends a user-friendly Chinese error message on failure (D-067), classifying sentinel errors into appropriate responses.

Phase 5: MQ Integration and Idempotency

AgentTaskHandler (task_handler.go): AgentTaskHandler is the MQ task handler adapter layer that converts MQ tasks to ExecutePayload. It unmarshals the AgentProcessPayload, validates required fields, checks two-phase idempotency (D-121, fail-open on Redis errors), and calls AgentExecutor.ExecuteWithErrorMessage. On normal completion the handler invalidates the conversation context cache so that retried tasks (e.g. ones that were requeued because the conversation lock was held) load fresh messages from the database. The handler returns nil for permanent errors (bad payload, agent not found — retry won't help) and returns an error for transient failures (LLM timeout, rate limit) or when the conversation lock is held by another task (Asynq retries with exponential backoff until the lock is released).

Two-phase idempotency (D-121): 

The handler uses two-phase idempotency:
  1. agent:processing:{messageID} (TTL 130s): SETNX before execution, prevents concurrent duplicates
  2. agent:processed:{messageID} (TTL 24h): SET after success, prevents replay duplicates

On crash: the processing key expires after 130s, allowing Asynq retry to re-execute.
On HITL interrupt: the processing key expires naturally, no processed key is set.
On transient error: the processing key expires naturally, error returned for Asynq retry.

Resume handler idempotency (D-121): 

The resume handler uses the same two-phase pattern:
  1. agent:resume:processing:{checkpointID} (TTL 130s): SETNX before execution
  2. agent:resume:{checkpointID} (TTL 24h): SET after successful resume

On HITL re-interrupt: the processing key is deleted to allow subsequent resume.
On transient error: the processing key is deleted to allow immediate retry.
On permanent failure: processing key deleted + processed key set (24h) to prevent replay.

RedisIdempotencyStore: Redis-based deduplication using SETNX with TTL. The IdempotencyStore interface provides atomic check-and-set semantics. The key formats are "agent:processing:{messageID}" (130s), "agent:processed:{messageID}" (24h), "agent:resume:processing:{checkpointID}" (130s), and "agent:resume:{checkpointID}" (24h). Fail-open: if Redis is unavailable, processing continues (logged but not blocked).

Phase 5: main.go Wiring

The complete agent pipeline is wired in main.go after handler.RegisterAll: 

LLMClientFactory → AgentBuilder → AgentExecutor
StreamBridge + BroadcastHelper + DBContextManager → AgentExecutor
AgentExecutor + IdempotencyStore → AgentTaskHandler → MQ registration

A dedicated redis.Client is created for the idempotency store (D-Phase5-5), separate from the node broadcaster client. The AgentExecutor is configured with maxConcurrent=10 to limit parallel LLM calls.

Phase 5: Data Flow

The end-to-end data flow for agent message processing: 

MQ task → AgentTaskHandler → two-phase idempotency check (D-121)
→ AgentExecutor.ExecuteWithErrorMessage → typing=true broadcast
→ context loading (DBContextManager) → agent building (AgentBuilder)
→ LLM streaming (Eino Runner) → stream bridge (cumulative snapshots)
→ broadcast chunks (BroadcastHelper) → is_done=true broadcast
→ persist message → return nil to MQ (D-067)

Phase 6: Client Device Function Invocation

DynamicToolProvider (dynamic_tool_provider.go): DynamicToolProvider is an Eino ChatModelAgentMiddleware that dynamically injects client-device functions as InvokableTool instances before each agent run. BeforeAgent extracts the CallerDevice from context, queries the device's registered functions via ClientFunctionProvider, applies tag and exclusion filters, creates tools via newClientFunctionTool, and merges them into runCtx.Tools. All errors are fail-open: logged and skipped, never blocking agent execution (D-072 spirit).

ClientFunctionProvider (dynamic_tool_provider.go): Interface for retrieving function declarations registered by a client device. Defined here to avoid circular dependency on the server package (D-101).

CallerDevice (context_keys.go): CallerDevice holds the UserID and DeviceID of the client device that initiated the conversation. It is injected into the context by AgentExecutor.Execute and AgentResumeHandler when DeviceID is present in the payload. DynamicToolProvider reads it from context to determine which device's functions to expose as tools (D-102).

ClientToolsConfig: Per-agent configuration for dynamic client tool injection. FunctionTags filters which functions are exposed (empty = all). ExcludedFunctions is a deny-list checked first. CallTimeout sets the default timeout for client function calls. Zero/negative CallTimeout defaults to 120 seconds.

AgentBuilder Integration: AgentBuilder exposes SetClientFunctionProvider. When configured and an agent's Middleware config has EnableClientTools=true, buildMiddleware creates a DynamicToolProvider and inserts it as the first middleware (before PatchToolCalls, Summarization, and ToolReduction) so injected tools are visible to all downstream middleware.

Client functions use the RemoteCalling interrupt-resume pattern (D-137): tool.Interrupt triggers an interrupt, executor creates a RemoteCalling record, client processes it asynchronously, and agent resumes with the result.

Phase 7: Production Hardening

Semaphore (semaphore.go): Semaphore limits concurrent agent executions using a channel-based counter. Capacity > 0 creates a bounded semaphore; capacity <= 0 returns an unlimited semaphore where Acquire always succeeds. Tracks active count, peak usage, and total acquisitions via Stats(). When the executor is created with maxConcurrent > 0, a Semaphore is attached to bound parallel LLM calls.

ConversationLock (conversation_lock.go, D-075): ConversationLock is the interface for per-conversation distributed locking. RedisConversationLock implements it via Redis SETNX with a configurable TTL (default 130s, covering the 120s total timeout plus buffer). Release uses a Lua script that verifies the lock value before deletion, preventing one owner from releasing another's lock. Fail-open: Redis errors are logged but do not block execution. When a task fails to acquire the lock it returns an error so Asynq requeues it with exponential backoff; once the active task finishes and invalidates the context cache, the retried task loads the latest messages (including those queued while the lock was held) and proceeds normally. The lock reuses the same dedicated redis.Client as the idempotency store (D-074).

LLMMetrics (monitoring.go): LLMMetrics is the interface for recording LLM call metrics (agent ID, model, duration, token counts, error). LogMetrics is the default implementation that logs each call event via the structured Logger. The executor records a metrics event around every agent Build step when WithLLMMetrics is configured.

StartCleanup (db_context_manager.go): DBContextManager.StartCleanup begins a background goroutine that periodically evicts expired entries from the in-memory conversation context cache (sync.Map). It runs until the parent context is cancelled. Cache cleanup is independent of server lifecycle and ensures stale contexts do not accumulate memory indefinitely.

Reload (registry.go, D-076, D-077): AgentRegistry.Reload re-scans the directory previously passed to Load and replaces all agent configurations atomically. The reload_agents RPC handler (D-076) invokes Reload to support runtime hot-reloading of agent configs without server restart. AgentRegistry also exposes a Dir() method returning the directory path (used by the reload handler for error messages) and a SetLogger method to wire in the server's structured logger (Phase 7 review).

Structured Logger (executor.go): Logger is a structured logging interface (Info, Error, Debug) compatible with server.Logger. All agent components (AgentExecutor, AgentRegistry, BroadcastHelper, AgentTaskHandler) accept a Logger and default to noopLogger when nil is provided. This replaces ad-hoc *log.Logger usage and provides consistent key-value log output.

Functional Options (executor.go): ExecutorOption is the functional options pattern for AgentExecutor configuration. WithTotalTimeout overrides the default 120s total execution timeout. WithTypingTimeout overrides the default 60s wait for the first LLM token. WithLLMMetrics attaches an LLMMetrics recorder. All options ignore non-positive durations (zero/negative) to preserve safe defaults.

Phase 8A: Tool System and Middleware

Tool Registry (D-078): The tools sub-package provides a Registry that maps tool names to ToolFactory functions. Built-in tools (get_weather, get_current_time, retrieve_tool_result) are registered in DefaultRegistry via init(). Agent configs reference tools by name in the YAML tools list. Unknown tool names are logged and skipped (fail-open).

Built-in Tools: get_weather returns mock weather data, get_current_time returns the current time in a given timezone, retrieve_tool_result retrieves full content of previously truncated tool results from the in-memory ToolResultStore (D-080).

ToolResultStore (D-080): Stores truncated tool results in memory with TTL (default 1 hour). UTF-8-safe rune-based truncation at 50000 characters. A background cleanup goroutine (StartCleanup) removes expired entries periodically.

Middleware Chain (D-079): buildMiddleware constructs the middleware chain from AgentConfig.Middleware settings. Fixed order: PatchToolCalls → Summarization → ToolReduction. Each middleware init failure is logged and skipped (fail-open). Summarization uses the agent's chat model; ToolReduction uses an in-memory filesystem backend.

Enhanced Build(): AgentBuilder.Build() now creates tools from the registry (when set via SetToolRegistry) and builds the middleware chain, passing both to adk.ChatModelAgentConfig.

Phase 8B: RemoteCalling Cleanup Task

RemoteCallingCleanupTask (remote_calling_cleanup.go, D-123, D-137): RemoteCallingCleanupTask is a background goroutine that periodically cleans up conversations stuck in asking_user/tool_calling status and individual expired RemoteCallings. Two-layer cleanup: (1) conversation-level scans conversations exceeding MaxAge (default 24h) and performs full cleanup (clear status, soft-delete RemoteCallings, delete checkpoint, send timeout message, broadcast agent_timeout); (2) RemoteCalling-level scans individual RemoteCallings with expires_at < NOW() and marks them expired. When all RemoteCallings for a checkpoint expire (pending=0), the task enqueues a TypeAgentResume MQ task so the agent can handle the timeout gracefully. All cleanup steps are non-fatal (D-007): errors are logged but do not interrupt processing of other conversations. Distributed locking via Redis SETNX prevents duplicate cleanup across nodes.

Phase 8C: MCP Integration

MCPBridge (D-086): MCPBridge manages connections to external MCP (Model Context Protocol) servers. It supports SSE (Server-Sent Events) and stdio transports. ConnectSSE and ConnectStdio perform the MCP handshake and tool discovery, returning Eino tool.BaseTool slices that integrate seamlessly with the agent's tool set. The bridge tracks all client connections for lifecycle management via CloseAll.

MCPServerConfig: Agent configs reference MCP servers in the mcp_servers YAML list. Each entry specifies a name, transport ("sse" or "stdio"), connection details (url for SSE, command/args/env for stdio), and optional tools filter to restrict which tools are exposed to the agent.

Build() Integration: AgentBuilder.Build() connects to all configured MCP servers during agent construction. Connection failures are logged and skipped (fail-open), ensuring the agent still starts even if an MCP server is unavailable. The MCP tools are appended to the agent's tool set.

Shutdown: main.go calls mcpBridge.CloseAll() after srv.GracefulStop() to release all MCP client connections once in-flight requests have finished.

RemoteCalling timeout cleanup background task (D-123 / D-137).

Two-layer cleanup:

  1. Conversation-level: scans conversations stuck in asking_user status and cleans up those that have exceeded the configured max age (default 24h).
  2. RemoteCalling-level: scans individual RemoteCallings with expires_at < NOW().

Cleanup steps mirror the D-122 cleanupAfterResumeFailure pattern: clear agent status, soft-delete RemoteCallings, delete Redis checkpoint, send a user-friendly timeout message, and broadcast an agent_timeout ephemeral notification.

Resume handler processes TypeAgentResume MQ tasks (Phase 8B / D-085).

Sub-agent resolution and delegation for AgentBuilder (D-081).

Index

Constants

View Source 
const DefaultClientFunctionCallTimeoutMs = 120000 // 120 seconds

DefaultClientFunctionCallTimeoutMs is the fallback timeout (in milliseconds) for client function calls when no timeout is specified in the function info or the agent config. Both client_function_tool.go and resume_handler.go use this constant to ensure consistency.

View Source 
const DefaultHITLTimeout = 24 * time.Hour

DefaultHITLTimeout is the default expiration timeout for HITL (ask_user) RemoteCallings. Used when the agent interrupts to ask the user a question without a client function. This value is intentionally long (24h) to allow users to respond at their convenience. Both executor.go and resume_handler.go use this constant for consistency.

View Source 
const MinClientFunctionCallTimeoutMs = 60000 // 60 seconds

MinClientFunctionCallTimeoutMs is the minimum allowed timeout (in milliseconds) for client function calls. Even if a function specifies a smaller timeout_ms, it will be clamped to this value. This prevents RemoteCallings from expiring before the client has a reasonable chance to process them. See: Vue测试问题 - LLM超时时间设置问题 See: TC-011 联调测试 - get_page_description 30s 超时过短问题

Variables

View Source 
var (
	ErrMissingID        = errors.New("agent: missing required field: id")
	ErrMissingName      = errors.New("agent: missing required field: name")
	ErrMissingModel     = errors.New("agent: missing required field: model")
	ErrMissingAPIKeyEnv = errors.New("agent: missing required field: api_key_env")
	// ErrAgentNotFound indicates the agent ID was not found in the registry.
	ErrAgentNotFound      = errors.New("agent: not found in registry")
	ErrInvalidFrontMatter = errors.New("agent: invalid front matter")
	ErrNoFrontMatter      = errors.New("agent: no front matter found")
	ErrContextLoad        = errors.New("agent: failed to load context")

	// LLM provider errors
	ErrAPIKeyMissing    = errors.New("agent: API key environment variable not set")
	ErrUnsupportedModel = errors.New("agent: unsupported LLM provider for model")
	ErrLLMTimeout       = errors.New("agent: LLM request timed out")
	ErrLLMRateLimited   = errors.New("agent: LLM rate limited")
	ErrStreamClosed     = errors.New("agent: stream closed unexpectedly")

	// Agent execution errors
	ErrAgentBuild = errors.New("agent: failed to build agent")
	ErrAgentRun   = errors.New("agent: agent execution failed")

	// HITL errors (Phase 8B)
	ErrHITLInterrupted    = errors.New("agent: HITL interrupted, awaiting user input")
	ErrCheckpointStoreSet = errors.New("agent: checkpoint store failed to persist")
	ErrCheckpointNotFound = errors.New("agent: checkpoint not found or expired")

	// MCP errors (Phase 8C, D-086)
	ErrInvalidMCPTransport = errors.New("mcp: transport must be 'sse' or 'stdio'")
	ErrMCPMissingURL       = errors.New("mcp: SSE transport requires url")
	ErrMCPMissingCommand   = errors.New("mcp: stdio transport requires command")
	ErrMCPMissingName      = errors.New("mcp: server name is required")
	ErrMCPDuplicateName    = errors.New("mcp: duplicate server name")
	ErrMCPUnreachable      = errors.New("agent: MCP service unreachable")
)

Sentinel errors for agent operations.

Functions

func AgentIDFromContext 

func AgentIDFromContext(ctx context.Context) (string, bool)

AgentIDFromContext extracts the agent userID from ctx. Returns ("", false) if not present.

func ContextWithAgentID 

func ContextWithAgentID(ctx context.Context, agentID string) context.Context

ContextWithAgentID returns a copy of ctx carrying the agent's userID.

func ContextWithCallerDevice 

func ContextWithCallerDevice(ctx context.Context, d CallerDevice) context.Context

ContextWithCallerDevice returns a copy of ctx carrying the CallerDevice.

func ContextWithConversationID 

func ContextWithConversationID(ctx context.Context, conversationID string) context.Context

ContextWithConversationID returns a copy of ctx carrying the conversation ID.

func ConversationIDFromContext 

func ConversationIDFromContext(ctx context.Context) (string, bool)

ConversationIDFromContext extracts the conversation ID from ctx. Returns ("", false) if not present.

func NewAgentResumeHandler 

func NewAgentResumeHandler(
	executor *AgentExecutor,
	registry *AgentRegistry,
	lock ConversationLock,
	logger Logger,
	idempotency IdempotencyStore,
) func(ctx context.Context, task *mq.Task) error

NewAgentResumeHandler returns an mq.TaskHandler-compatible function that processes TypeAgentResume tasks. It resumes a paused agent after HITL.

The handler:

  1. Unmarshals the task payload
  2. Checks idempotency (skip if checkpoint already resumed)
  3. Acquires a per-conversation lock (D-084)
  4. Looks up agent config from registry
  5. Builds the agent (with CheckPointStore)
  6. Calls Runner.ResumeWithParams with the user's answer
  7. Bridges the stream and broadcasts to the human user
  8. Persists the final message
  9. Always returns nil to MQ (D-073)

idempotency may be nil to disable the duplicate-resume guard (backward compatible). When non-nil, a Redis SETNX on "agent:resume:<checkpointID>" ensures that the same checkpoint is resumed at most once, even if multiple TypeAgentResume tasks are enqueued (e.g. client calls both send_message and agent_resume, or retries agent_resume).

func NewAgentTaskHandler 

func NewAgentTaskHandler(
	executor *AgentExecutor,
	idempotency IdempotencyStore,
	lock ConversationLock,
	logger Logger,
) func(ctx context.Context, task *mq.Task) error

NewAgentTaskHandler returns an mq.TaskHandler-compatible function that processes TypeAgentProcess tasks.

The handler:

  1. Unmarshals the task payload into AgentProcessPayload
  2. Acquires a per-conversation lock (if lock is non-nil); returns error if held so Asynq retries
  3. Checks idempotency (skip if already processed)
  4. Calls AgentExecutor.ExecuteWithErrorMessage
  5. Returns nil to MQ for permanent errors (D-067); returns transient errors for retry

On normal completion (after releasing the lock) the handler invalidates the conversation context cache so subsequent retries load fresh DB state.

HITL interrupts (D-084): when the executor returns ErrHITLInterrupted the conversation lock is intentionally NOT released so that no new task can conflict while the agent is paused. The lock expires naturally.

lock may be nil to disable conversation-level serialization (backward compatible).

func NormalizeClientFunctionTimeout 

func NormalizeClientFunctionTimeout(timeoutMs int, defaultCallTimeoutMs int) int

NormalizeClientFunctionTimeout applies the standard timeout normalization logic for client function calls. This consolidates the previously duplicated logic from executor.go, resume_handler.go, and client_function_tool.go.

The normalization rules are:

  1. If timeoutMs <= 0, use defaultCallTimeoutMs (if positive) or DefaultClientFunctionCallTimeoutMs.
  2. Clamp to MinClientFunctionCallTimeoutMs if below minimum.

Returns the normalized timeout in milliseconds.

func WithBroadcastInfo 

func WithBroadcastInfo(ctx context.Context, bh *BroadcastHelper, humanUserID, agentUserID, conversationID string) context.Context

WithBroadcastInfo returns a context enriched with broadcast metadata so that the LoggingMiddleware can emit function_call updates to clients.

func WithMessage 

func WithMessage(ctx context.Context, message *model.Message) context.Context

func WithStore 

func WithStore(ctx context.Context, s store.StoreAPI) context.Context

WithStore returns a context enriched with a StoreAPI so that the LoggingMiddleware can persist tool_calling messages. When the store is nil, the middleware falls back to ephemeral-only broadcasting (D-063 nil-safe).

Types

type AgentBuilder 

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

AgentBuilder constructs runnable agents from AgentConfig using the Eino ADK.

func NewAgentBuilder 

func NewAgentBuilder(factory *LLMClientFactory) *AgentBuilder

NewAgentBuilder creates an AgentBuilder backed by the given LLM factory.

func (*AgentBuilder) Build 

func (b *AgentBuilder) Build(ctx context.Context, config *AgentConfig) (built *BuiltAgent, err error)

Build creates a fully configured agent ready for execution.

The method performs these steps:

  1. Creates a ChatModel via the LLMClientFactory.
  2. Creates tools from the tool registry if configured (D-078).
  3. Resolves sub-agents and appends them as tools (D-081).
  4. Builds the middleware chain (D-079).
  5. Wraps everything in a ChatModelAgent with the agent's system prompt as instruction.
  6. Creates a Runner with streaming enabled and optional CheckPointStore (D-083).

func (*AgentBuilder) CheckPointStore 

func (b *AgentBuilder) CheckPointStore() compose.CheckPointStore

CheckPointStore returns the checkpoint store, or nil if not set.

func (*AgentBuilder) SetCheckPointStore 

func (b *AgentBuilder) SetCheckPointStore(store compose.CheckPointStore)

SetCheckPointStore sets the checkpoint store for HITL support (D-083). If not set, checkpoint persistence is disabled and HITL is not available.

func (*AgentBuilder) SetClientFunctionProvider 

func (b *AgentBuilder) SetClientFunctionProvider(provider ClientFunctionProvider)

SetClientFunctionProvider sets the function registry used to retrieve client device functions for dynamic tool injection (D-101). If not set, client tools are not available.

func (*AgentBuilder) SetLLMLogger 

func (b *AgentBuilder) SetLLMLogger(logger *LLMLogger)

SetLLMLogger sets the dedicated LLM call logger. When set, a LoggingMiddleware is appended to the middleware chain in Build(), recording all LLM requests, responses, tool calls, and agent events to the logger's output. When nil, no LLM logging middleware is added (default).

func (*AgentBuilder) SetLogger 

func (b *AgentBuilder) SetLogger(logger Logger)

SetLogger sets the structured logger used by middleware components (e.g. DynamicToolProvider) for diagnostic output. When nil, middleware diagnostics are silently discarded (fail-open, D-072).

func (*AgentBuilder) SetMCPBridge 

func (b *AgentBuilder) SetMCPBridge(bridge *agenttools.MCPBridge)

SetMCPBridge sets the MCP bridge used to connect to MCP servers during Build (D-086). If not set, MCP servers configured in AgentConfig are ignored.

func (*AgentBuilder) SetMessageStore 

func (b *AgentBuilder) SetMessageStore(ms *store.MessageStore)

SetMessageStore sets the MessageStore used by the summarization callback to persist summary messages and mark old messages as summarized. If not set, summarization runs without DB persistence.

func (*AgentBuilder) SetRegistry 

func (b *AgentBuilder) SetRegistry(registry *AgentRegistry)

SetRegistry sets the agent registry used for sub-agent resolution (D-081). If not set, sub-agents are not resolved.

func (*AgentBuilder) SetToolRegistry 

func (b *AgentBuilder) SetToolRegistry(registry *agenttools.Registry)

SetToolRegistry sets the tool registry used to create tools during Build. If not set, no tools are created (backward compatible).

func (*AgentBuilder) SetTracingDebugFilter 

func (b *AgentBuilder) SetTracingDebugFilter(users, devices []string)

SetTracingDebugFilter configures which users/devices get full LLM content recorded in tracing spans. When non-empty, matching callers (OR logic) have their request/response content added as span events on agent.llm.call.

func (*AgentBuilder) SetTracingEnabled 

func (b *AgentBuilder) SetTracingEnabled(enabled bool)

SetTracingEnabled controls whether a TracingMiddleware is appended to the middleware chain during Build(). When enabled, each LLM call and tool call produces an OpenTelemetry span. When disabled (the default), no tracing middleware is added and there is zero overhead.

type AgentConfig 

type AgentConfig struct {
	ID          string          `yaml:"id" json:"id"`
	Name        string          `yaml:"name" json:"name"`
	Description string          `yaml:"description" json:"description"`
	Model       string          `yaml:"model" json:"model"`
	APIKeyEnv   string          `yaml:"api_key_env" json:"api_key_env"`
	BaseURL     string          `yaml:"base_url" json:"base_url"`
	Parameters  AgentParameters `yaml:"parameters" json:"parameters"`
	Context     AgentContext    `yaml:"context" json:"context"`
	Tools       []string        `yaml:"tools" json:"tools"`
	// DynamicTools lists tool names that should be resolved from the tool
	// registry at runtime (per-execution) instead of at build time. This
	// enables the Eino framework's 0->nonzero tool transition, which triggers
	// graph rebuild and ensures tools are indexed in the ToolNode.
	DynamicTools []string          `yaml:"dynamic_tools" json:"dynamic_tools"`
	ToolConfig   map[string]any    `yaml:"tool_config" json:"tool_config"` // per-tool config passed to tool factories
	Middleware   MiddlewareConfig  `yaml:"middleware" json:"middleware"`   // middleware toggles (D-079)
	SubAgents    []string          `yaml:"sub_agents" json:"sub_agents"`   // Phase 8B placeholder (D-081)
	MCPServers   []MCPServerConfig `yaml:"mcp_servers" json:"mcp_servers"` // MCP server connections (D-086)
	SystemPrompt string            `yaml:"-" json:"system_prompt"`         // Markdown body
}

AgentConfig represents the configuration for an AI agent. Parsed from YAML front matter in agent definition files.

func ParseFrontMatter 

func ParseFrontMatter(data []byte) (*AgentConfig, error)

ParseFrontMatter parses an agent definition file. The file format is YAML front matter delimited by "---" lines, followed by a Markdown body that serves as the system prompt.

Format: 

---
id: weather-bot
name: Weather Bot
...
---
Markdown body as system prompt

func (*AgentConfig) Validate 

func (c *AgentConfig) Validate() error

Validate checks that the AgentConfig has all required fields. It also validates nested MCP server configurations (D-086).

type AgentContext 

type AgentContext struct {
	MaxTokens   int `yaml:"max_tokens" json:"max_tokens"`
	MaxMessages int `yaml:"max_messages" json:"max_messages"`
}

AgentContext holds context window configuration.

type AgentExecutor 

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

AgentExecutor orchestrates the full agent execution pipeline: context loading, agent building, LLM streaming, broadcast, and persistence.

func NewAgentExecutor 

func NewAgentExecutor(
	registry *AgentRegistry,
	contextManager ContextManager,
	agentBuilder *AgentBuilder,
	streamBridge *StreamBridge,
	broadcaster *BroadcastHelper,
	store store.StoreAPI,
	maxConcurrent int,
	logger Logger,
	opts ...ExecutorOption,
) *AgentExecutor

NewAgentExecutor creates an AgentExecutor with all dependencies. If maxConcurrent > 0, a semaphore channel is created to limit parallel executions. Optional ExecutorOption values override defaults (totalTimeout=120s, typingTimeout=60s).

func (*AgentExecutor) Execute 

func (e *AgentExecutor) Execute(ctx context.Context, payload ExecutePayload) (err error)

Execute runs the full agent execution pipeline for a single user message.

Pipeline steps:

  1. Acquire semaphore (if configured).
  2. Apply total timeout (default 120s, configurable via WithTotalTimeout).
  3. Look up agent config from registry.
  4. Send typing=true to the human user (D-065).
  5. Load conversation context.
  6. Build the agent (LLM client + Eino runner).
  7. Convert messages to Eino schema.
  8. Run the agent and bridge the stream.
  9. Broadcast streaming chunks to the human user.
  10. Send is_done=true broadcast (D-052 step 1).
  11. Persist the final message (D-052 step 2).

func (*AgentExecutor) ExecuteWithErrorMessage 

func (e *AgentExecutor) ExecuteWithErrorMessage(ctx context.Context, payload ExecutePayload) error

ExecuteWithErrorMessage wraps Execute and sends a user-friendly error message on failure (D-067). The original error is always returned to the caller. HITL interrupts (ErrHITLInterrupted) are NOT treated as errors — no error message is persisted.

func (*AgentExecutor) InvalidateContextCache 

func (e *AgentExecutor) InvalidateContextCache(conversationID string)

InvalidateContextCache removes the cached context for a conversation so that the next task (e.g. a retried one) loads fresh messages from the database. Safe to call even if contextManager is nil.

func (*AgentExecutor) SetContextManager 

func (e *AgentExecutor) SetContextManager(cm ContextManager)

SetContextManager replaces the context manager. Exported for testing only.

type AgentParameters 

type AgentParameters struct {
	Temperature float64 `yaml:"temperature,omitempty" json:"temperature,omitempty"`
	MaxTokens   int     `yaml:"max_tokens,omitempty" json:"max_tokens,omitempty"`
	TopP        float64 `yaml:"top_p,omitempty" json:"top_p,omitempty"`
}

AgentParameters holds model generation parameters.

type AgentProcessPayload 

type AgentProcessPayload struct {
	MessageID      string `json:"message_id"`
	ConversationID string `json:"conversation_id"`
	AgentID        string `json:"agent_id"`
	SenderID       string `json:"sender_id"`
	DeviceID       string `json:"device_id"` // Phase 6 (D-102)
}

AgentProcessPayload is the MQ task payload for triggering agent processing. Fields must match the agentProcessPayload in internal/handler/send_message.go.

type AgentRegistry 

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

AgentRegistry manages loaded agent configurations. It is safe for concurrent use.

func NewRegistry 

func NewRegistry() *AgentRegistry

NewRegistry creates an empty AgentRegistry.

func (*AgentRegistry) Count 

func (r *AgentRegistry) Count() int

Count returns the number of registered agents.

func (*AgentRegistry) Dir 

func (r *AgentRegistry) Dir() string

Dir returns the directory path from which agents were last loaded.

func (*AgentRegistry) Get 

func (r *AgentRegistry) Get(id string) (*AgentConfig, bool)

Get returns the AgentConfig for the given agent ID.

func (*AgentRegistry) IsAgent 

func (r *AgentRegistry) IsAgent(userID string) (*AgentConfig, bool)

IsAgent reports whether the given userID corresponds to a registered agent. It performs an exact-match lookup in the registry (D-054 revised). Returns the AgentConfig and true if found, nil and false otherwise.

func (*AgentRegistry) ListAll 

func (r *AgentRegistry) ListAll() []*AgentConfig

ListAll returns a copy of all registered agent configurations.

func (*AgentRegistry) Load 

func (r *AgentRegistry) Load(dir string) error

Load scans the given directory for .md agent config files and loads them. Existing agents are cleared before loading. If the directory does not exist, Load returns nil (optional module, D-063).

func (*AgentRegistry) Register 

func (r *AgentRegistry) Register(config *AgentConfig)

Register adds an agent config to the registry. This is primarily intended for testing; production code should use Load.

func (*AgentRegistry) Reload 

func (r *AgentRegistry) Reload() error

Reload re-scans the agents directory and reloads all configurations.

func (*AgentRegistry) SetLogger 

func (r *AgentRegistry) SetLogger(logger Logger)

SetLogger sets the structured logger for registry operations. If logger is nil, the call is ignored (the existing logger is kept).

type AgentResumePayload 

type AgentResumePayload struct {
	ConversationID string `json:"conversation_id"`
	CheckpointID   string `json:"checkpoint_id"`
	AgentID        string `json:"agent_id"`
	SenderID       string `json:"sender_id"` // human user who sent the answer
	DeviceID       string `json:"device_id"` // Phase 6 (D-102)
	MessageID      uint32 `json:"message_id"`
}

AgentResumePayload is the MQ task payload for TypeAgentResume. Answers are NOT included — they are persisted in the Question table (D-116). The resume handler reads answered Questions from DB to build the targets map.

type AgentStatusPayload 

type AgentStatusPayload struct {
	UserID         string `json:"user_id"` // agent userID
	ConversationID string `json:"conversation_id"`
	Status         string `json:"status"` // thinking/tool_calling/generating/idle/asking_user
	Timestamp      int64  `json:"timestamp"`
}

AgentStatusPayload is the JSON payload for UpdateTypeAgentStatus (D-087).

type AgentTimeoutPayload 

type AgentTimeoutPayload struct {
	UserID         string `json:"user_id"` // agent userID
	ConversationID string `json:"conversation_id"`
	Reason         string `json:"reason"`
	Timestamp      int64  `json:"timestamp"`
}

AgentTimeoutPayload is the JSON payload for UpdateTypeAgentTimeout (D-087).

type BroadcastConversationUpdateFunc 

type BroadcastConversationUpdateFunc func(
	ctx context.Context,
	conversationID string,
	memberIDs []string,
	action string,
) error

BroadcastConversationUpdateFunc is the function signature for broadcasting persisted conversation updates. This mirrors handler.BroadcastConversationUpdateFunc to avoid importing the handler package (which would create a circular dependency).

type BroadcastHelper 

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

BroadcastHelper sends streaming and typing updates to users via WebSocket (C7). All broadcasts are fire-and-forget (D-007): errors are logged but not returned to the caller.

func NewBroadcastHelper 

func NewBroadcastHelper(wsServer BroadcastServer, logger Logger) *BroadcastHelper

NewBroadcastHelper creates a BroadcastHelper backed by the given WebSocket broadcast server.

func (*BroadcastHelper) BroadcastMessageUpdate 

func (bh *BroadcastHelper) BroadcastMessageUpdate(ctx context.Context, userUpdates []model.UserUpdate)

BroadcastMessageUpdate broadcasts persisted message updates to each user with their real DB-allocated seq numbers. This is needed after the agent executor persists a message via store.SendMessage — the DB records are created, but without this broadcast the clients won't receive the update until the next sync_updates call.

userUpdates must come from store.SendMessageResult.Updates (each entry already has the correct UserID, Seq, Payload, and CreatedAt).

func (*BroadcastHelper) BroadcastRaw 

func (bh *BroadcastHelper) BroadcastRaw(targetUserID string, updates *protocol.PackageDataUpdates) error

BroadcastRaw sends a pre-built PackageDataUpdates to the given user. Used by the resume handler to deliver persisted message updates in real-time.

func (*BroadcastHelper) SendAgentStatus 

func (bh *BroadcastHelper) SendAgentStatus(ctx context.Context, humanUserID, agentUserID, conversationID, status string)

SendAgentStatus broadcasts an ephemeral agent status update (Seq=0, D-050 / D-087) to the human user. status is one of: thinking, tool_calling, generating, idle, asking_user.

func (*BroadcastHelper) SendAgentTimeout 

func (bh *BroadcastHelper) SendAgentTimeout(ctx context.Context, humanUserID, agentUserID, conversationID, reason string)

SendAgentTimeout broadcasts an ephemeral agent timeout update (Seq=0, D-050 / D-087) to the human user.

func (*BroadcastHelper) SendConversationUpdate 

func (bh *BroadcastHelper) SendConversationUpdate(ctx context.Context, humanUserID, conversationID string, updatedAt time.Time)

SendConversationUpdate broadcasts a lightweight conversation update notification (Seq=0, ephemeral). This implements the pull-on-notification pattern: the client receives a conversation_id and fetches full conversation state (including questions) via get_conversation RPC.

The updatedAt parameter is included in the payload as "updated_at" (Unix seconds) when non-zero, allowing clients to detect stale updates (D-124).

func (*BroadcastHelper) SendEphemeralMessageUpdate 

func (bh *BroadcastHelper) SendEphemeralMessageUpdate(targetUserID string, payload json.RawMessage)

SendEphemeralMessageUpdate sends an ephemeral (seq=0) message update to the given user. This bypasses the client's sync pipeline (applyChain) and triggers an instant message:added event for immediate UI feedback. Used for tool_calling status changes where the persisted update would arrive too late due to sync pipeline latency.

func (*BroadcastHelper) SendFunctionCall 

func (bh *BroadcastHelper) SendFunctionCall(ctx context.Context, humanUserID, agentUserID, conversationID, name, args, result, errStr string, durationMs int64, isDone bool)

SendFunctionCall broadcasts an ephemeral function call update (Seq=0) to the human user. It should be called twice per function invocation: once with IsDone=false before execution (carrying name and args), and once with IsDone=true after execution (carrying result or error).

func (*BroadcastHelper) SendStreamUpdate 

func (bh *BroadcastHelper) SendStreamUpdate(ctx context.Context, humanUserID, agentUserID, conversationID, streamID, text string, isDone bool)

SendStreamUpdate broadcasts an ephemeral streaming update (Seq=0, D-050 / D-051) to both the human user and the agent user so that every participant sees the streamed text in real time. The ctx parameter is reserved for future cancellation support.

func (*BroadcastHelper) SendTyping 

func (bh *BroadcastHelper) SendTyping(ctx context.Context, agentUserID, targetUserID, conversationID string, isTyping bool)

SendTyping broadcasts an ephemeral typing indicator (Seq=0, D-050 / D-065) to targetUserID — typically the human user who should see the agent typing. agentUserID is the agent's identity, placed in the payload's user_id field. The ctx parameter is reserved for future cancellation support.

func (*BroadcastHelper) SetRegistry 

func (bh *BroadcastHelper) SetRegistry(registry *AgentRegistry)

SetRegistry sets the agent registry used to determine whether a userID belongs to a registered agent (D-054 revised). When nil, IsAgent defaults to false for all payloads.

type BroadcastServer 

type BroadcastServer interface {
	BroadcastUpdates(userID string, updates *protocol.PackageDataUpdates) error
}

BroadcastServer is the subset of WebSocketServer that BroadcastHelper needs.

type BuiltAgent 

type BuiltAgent struct {
	Runner *adk.Runner
	Config *AgentConfig
	Agent  adk.Agent // underlying agent, used by NewAgentTool for sub-agents
}

BuiltAgent wraps an Eino Runner together with the config it was built from. The Agent field holds the underlying agent for sub-agent wrapping (D-081).

type CallerDevice 

type CallerDevice struct {
	UserID   string
	DeviceID string
}

CallerDevice holds the (userID, deviceID) pair of the device that initiated the agent conversation.

func CallerDeviceFromContext 

func CallerDeviceFromContext(ctx context.Context) (CallerDevice, bool)

CallerDeviceFromContext extracts the CallerDevice from ctx. Returns (zero, false) if not present.

type ClaudeProvider 

type ClaudeProvider struct{}

ClaudeProvider creates ChatModel instances for Anthropic Claude.

func (*ClaudeProvider) CreateChatModel 

func (p *ClaudeProvider) CreateChatModel(ctx context.Context, config *AgentConfig, apiKey string) (einomodel.BaseChatModel, error)

CreateChatModel builds a Claude ChatModel.

func (*ClaudeProvider) DefaultBaseURL 

func (p *ClaudeProvider) DefaultBaseURL() string

DefaultBaseURL returns the Anthropic API endpoint (D-064).

type ClientFunctionProvider 

type ClientFunctionProvider interface {
	GetFunctions(ctx context.Context, userID, deviceID string) ([]protocol.FunctionInfo, error)
	// GetFunctionsByUser returns all registered functions for a userID,
	// keyed by deviceID. Used when the agent's deviceID is unknown.
	GetFunctionsByUser(ctx context.Context, userID string) (map[string][]protocol.FunctionInfo, error)
}

ClientFunctionProvider retrieves function declarations registered by a client device. Defined here to avoid circular dependency on server package (D-101).

type ClientToolsConfig 

type ClientToolsConfig struct {
	// FunctionTags filters functions by tag. Empty = accept all functions.
	// A function matches if it has at least one tag in this list (OR semantics).
	FunctionTags []string `yaml:"function_tags" json:"function_tags"`
	// ExcludedFunctions excludes specific function names. Exact match only.
	ExcludedFunctions []string `yaml:"excluded_functions" json:"excluded_functions"`
	// CallTimeout is the default timeout for client function calls.
	// Individual functions may override via timeout_ms. Default: 120s.
	// A zero value means "use default (120s)".
	CallTimeout time.Duration `yaml:"call_timeout" json:"call_timeout"`
}

ClientToolsConfig controls how client device functions are surfaced as agent tools (Phase 6 / D-100, D-101).

type ContextManager 

type ContextManager interface {
	// GetContext returns the trimmed message history for a conversation,
	// suitable for passing to an LLM as context. Messages are returned
	// in chronological order (oldest first).
	GetContext(ctx context.Context, conversationID string, config *AgentConfig) ([]*model.Message, error)

	// InvalidateCache removes the cached context for a conversation.
	InvalidateCache(conversationID string)
}

ContextManager provides conversation history for Agent processing. It loads messages from the database, caches them, and trims to fit the Agent's context window configuration.

type ConversationLock 

type ConversationLock interface {
	// Acquire attempts to acquire the lock for the given conversation.
	// Returns (true, nil) if acquired, (false, nil) if already held,
	// (false, err) on Redis error.
	Acquire(ctx context.Context, conversationID string, ttl time.Duration) (bool, error)
	// Release releases the lock. Only the owner can release it.
	Release(ctx context.Context, conversationID string) error
}

ConversationLock provides distributed per-conversation serialization.

type DBContextManager 

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

DBContextManager implements ContextManager using the database for storage and sync.Map for in-memory caching with TTL-based expiry.

Design decisions (D-060):

  • DB-backed storage with in-memory sync.Map cache (30s default TTL)
  • Token-based trimming with message-count fallback
  • HeuristicTokenCounter (len/4) as default tokenizer (D-001)

func NewDBContextManager 

func NewDBContextManager(messageStore *store.MessageStore, opts ...DBContextOption) *DBContextManager

NewDBContextManager creates a DBContextManager backed by the given MessageStore.

func (*DBContextManager) GetContext 

func (cm *DBContextManager) GetContext(ctx context.Context, conversationID string, config *AgentConfig) ([]*model.Message, error)

GetContext returns the trimmed message history for a conversation.

The method checks an in-memory cache first (TTL-based expiry). On cache miss, it loads recent messages from the database, applies type filtering, and trims by token count (falling back to message count if MaxTokens is zero).

Messages are returned in chronological order (oldest first).

func (*DBContextManager) InvalidateCache 

func (cm *DBContextManager) InvalidateCache(conversationID string)

InvalidateCache removes the cached context for a conversation.

func (*DBContextManager) StartCleanup 

func (cm *DBContextManager) StartCleanup(ctx context.Context, interval time.Duration)

StartCleanup begins a background goroutine that removes expired cache entries at the given interval. It blocks until ctx is cancelled. If interval <= 0, defaults to 5 minutes.

Callers should launch this with: go cm.StartCleanup(ctx, interval)

type DBContextOption 

type DBContextOption func(*DBContextManager)

DBContextOption configures a DBContextManager.

func WithCacheTTL 

func WithCacheTTL(d time.Duration) DBContextOption

WithCacheTTL sets the cache time-to-live duration.

func WithTokenCounter 

func WithTokenCounter(tc TokenCounter) DBContextOption

WithTokenCounter sets the token counting implementation.

type DeletableCheckPointStore 

type DeletableCheckPointStore interface {
	compose.CheckPointStore
	Delete(ctx context.Context, key string) error
}

DeletableCheckPointStore extends compose.CheckPointStore with a Delete capability (D-112). Any store that implements Get, Set, and Delete satisfies this interface.

type DynamicToolProvider 

type DynamicToolProvider struct {
	*adk.BaseChatModelAgentMiddleware
	// contains filtered or unexported fields
}

DynamicToolProvider is an Eino ChatModelAgentMiddleware that dynamically injects client-device functions as InvokableTool instances before each agent run (Phase 6 / D-100, D-101, D-102).

func NewDynamicToolProvider 

func NewDynamicToolProvider(
	registry ClientFunctionProvider,
	cfg ClientToolsConfig,
	logger Logger,
	toolRegistry *agenttools.Registry,
	dynamicTools []string,
) *DynamicToolProvider

NewDynamicToolProvider creates a DynamicToolProvider. toolRegistry and dynamicTools enable resolution of static tools from the registry at runtime (per-execution), which is required for the Eino framework's 0->nonzero tool transition to trigger a graph rebuild.

func (*DynamicToolProvider) BeforeAgent 

func (d *DynamicToolProvider) BeforeAgent(ctx context.Context, runCtx *adk.ChatModelAgentContext) (context.Context, *adk.ChatModelAgentContext, error)

BeforeAgent implements the Eino middleware hook. It queries the calling device's registered functions and appends them as tools to runCtx.Tools. It also resolves dynamic_tools from the tool registry. Both injection paths are independent: dynamic_tools are injected even when no device context or client functions are available. Fail-open (D-072 spirit): errors in GetFunctions or tool creation are logged and skipped, never blocking agent execution.

type ExecutePayload 

type ExecutePayload struct {
	MessageID      string // ID of the triggering message
	ConversationID string // Conversation to operate in
	AgentID        string // Full userID of the agent (exact match in registry, D-054 revised)
	SenderID       string // Human user who sent the message
	DeviceID       string // Device that initiated the conversation (D-102)
}

ExecutePayload contains the parameters for a single agent execution.

type ExecutorOption 

type ExecutorOption func(*AgentExecutor)

ExecutorOption configures an AgentExecutor.

func WithBroadcastConversationUpdate 

func WithBroadcastConversationUpdate(fn BroadcastConversationUpdateFunc) ExecutorOption

WithBroadcastConversationUpdate sets the function for broadcasting persisted conversation updates to conversation members. When not set, persisted broadcast is skipped (D-063 nil-safe). This is injected from the handler package to avoid circular dependency.

func WithCheckPointStore 

func WithCheckPointStore(store DeletableCheckPointStore) ExecutorOption

WithCheckPointStore sets the checkpoint store for HITL checkpoint cleanup after resume (D-112). The store must implement Delete; when set, the resume handler removes the checkpoint from Redis after a successful resume.

func WithDebugErrors 

func WithDebugErrors(enabled bool) ExecutorOption

WithDebugErrors enables exposing raw error messages to users instead of generic friendly messages. Useful for development and debugging. WARNING: Do NOT enable in production — raw errors may leak sensitive info.

func WithLLMMetrics 

func WithLLMMetrics(m LLMMetrics) ExecutorOption

WithLLMMetrics sets the metrics recorder for LLM calls. When set, the executor records duration, model, and error information for each agent Build step. Pass nil (or omit) to disable metrics.

func WithRemoteCallingStore 

func WithRemoteCallingStore(rs *store.RemoteCallingStore) ExecutorOption

WithRemoteCallingStore sets the RemoteCallingStore for RemoteCalling persistence (D-137). When not set, RemoteCalling creation is skipped (D-063 nil-safe).

func WithTotalTimeout 

func WithTotalTimeout(d time.Duration) ExecutorOption

WithTotalTimeout sets the maximum total execution time for an agent task. Default is 120 seconds. Ignored if d <= 0.

func WithTypingTimeout 

func WithTypingTimeout(d time.Duration) ExecutorOption

WithTypingTimeout sets the maximum time to wait for the first LLM token before clearing the typing indicator. Default is 60 seconds. Ignored if d <= 0.

type FunctionCallPayload 

type FunctionCallPayload struct {
	UserID         string `json:"user_id"` // agent userID
	ConversationID string `json:"conversation_id"`
	Name           string `json:"name"`                  // function name
	Args           string `json:"args,omitempty"`        // JSON-encoded arguments
	Result         string `json:"result,omitempty"`      // function result (sent after completion)
	Error          string `json:"error,omitempty"`       // error message if call failed
	DurationMs     int64  `json:"duration_ms,omitempty"` // execution duration
	IsDone         bool   `json:"is_done"`               // true when function execution completes
	Timestamp      int64  `json:"timestamp"`
}

FunctionCallPayload is the JSON payload for UpdateTypeFunctionCall. Sent when the agent invokes a tool/function, allowing clients to display function call information in the UI.

type HeuristicTokenCounter 

type HeuristicTokenCounter struct{}

HeuristicTokenCounter uses len(text)/4 as a rough token estimate. This is the default when no external tokenizer is configured. It satisfies the D-001 zero-dependency principle.

func (*HeuristicTokenCounter) CountTokens 

func (h *HeuristicTokenCounter) CountTokens(text string) int

CountTokens returns a rough estimate of the number of tokens in text. The heuristic divides byte length by 4, which approximates English token counts. For precise counting, use a tiktoken-based implementation.

type IdempotencyStore 

type IdempotencyStore interface {
	// MarkProcessed atomically checks if key was already processed and marks it.
	// Returns true if the key was already processed (duplicate), false if this is the first time.
	// TTL controls how long the processed marker persists.
	MarkProcessed(ctx context.Context, key string, ttl time.Duration) (bool, error)
	// CheckProcessed checks if a key exists without setting it.
	// Returns (true, nil) if the key exists, (false, nil) otherwise.
	CheckProcessed(ctx context.Context, key string) (bool, error)
	// DeleteKey removes a key. Used to clean up processing markers.
	DeleteKey(ctx context.Context, key string) error
}

IdempotencyStore provides atomic check-and-set for agent task deduplication.

type InterruptInfo 

type InterruptInfo struct {
	// Question is the human-readable question the agent is asking.
	// Derived from InterruptInfo.Data when the interrupt data is a string.
	Question string

	// InterruptID is the Eino interrupt address ID that should be used as the
	// key in ResumeParams.Targets when resuming this interrupt.
	InterruptID string

	// ToolCallingMsgID
	ToolCallingMsgID uint32
}

InterruptInfo carries HITL interrupt details extracted from an AgentEvent whose Action.Interrupted is non-nil.

type LLMCallEvent 

type LLMCallEvent struct {
	AgentID      string
	Model        string
	Duration     time.Duration
	InputTokens  int
	OutputTokens int
	Error        error
}

LLMCallEvent contains metrics from a single LLM call.

type LLMClientFactory 

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

LLMClientFactory manages LLMProvider registrations and creates ChatModel instances based on AgentConfig. Provider selection uses model name and BaseURL heuristics (see detectProvider).

func NewLLMClientFactory 

func NewLLMClientFactory() *LLMClientFactory

NewLLMClientFactory creates a factory with all built-in providers registered.

func (*LLMClientFactory) Create 

func (f *LLMClientFactory) Create(ctx context.Context, config *AgentConfig) (einomodel.BaseChatModel, error)

Create resolves the appropriate provider for the config and builds a ChatModel. API keys are read from the environment variable named by config.APIKeyEnv. The API key value is never included in error messages or logs (security).

func (*LLMClientFactory) RegisterProvider 

func (f *LLMClientFactory) RegisterProvider(name string, provider LLMProvider)

RegisterProvider adds or replaces a provider under the given name.

type LLMLogger 

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

LLMLogger writes structured JSON log records to an io.Writer in JSONL format (one JSON object per line). It is safe for concurrent use; all writes are serialized through an internal mutex.

func NewLLMLogger 

func NewLLMLogger(w io.Writer, indent bool) *LLMLogger

NewLLMLogger creates an LLMLogger that writes to w. When indent is true each record is pretty-printed; production deployments should pass false to keep one record per line.

type LLMMetrics 

type LLMMetrics interface {
	Record(ctx context.Context, event LLMCallEvent)
}

LLMMetrics records LLM call metrics. Implementations must be safe for concurrent use.

type LLMProvider 

type LLMProvider interface {
	// CreateChatModel builds a ChatModel from the agent configuration.
	// The apiKey is provided separately so that providers can ignore it when
	// not needed (e.g. Ollama for local models).
	CreateChatModel(ctx context.Context, config *AgentConfig, apiKey string) (einomodel.BaseChatModel, error)

	// DefaultBaseURL returns the provider's default API endpoint (D-064).
	DefaultBaseURL() string
}

LLMProvider abstracts LLM backend creation (D-066). Each provider knows how to construct a ChatModel for its specific LLM service and supplies a default BaseURL so agents work with zero configuration (D-064).

type LogMetrics 

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

LogMetrics implements LLMMetrics using the agent Logger.

func NewLogMetrics 

func NewLogMetrics(logger Logger) *LogMetrics

NewLogMetrics creates a LogMetrics that logs LLM call events. If logger is nil, a noop logger is used so Record never panics.

func (*LogMetrics) Record 

func (m *LogMetrics) Record(ctx context.Context, event LLMCallEvent)

Record logs an LLM call event with structured key-value pairs. Errors are logged at Error level; successful calls at Info level.

type LogRecord 

type LogRecord struct {
	Timestamp  time.Time         `json:"timestamp"`
	AgentID    string            `json:"agent_id"`
	Model      string            `json:"model"`
	Iteration  int               `json:"iteration"`
	Phase      string            `json:"phase"` // "request" | "response" | "tool_call" | "tool_result" | "agent_start" | "agent_end" | "error"
	Messages   []MessageSnapshot `json:"messages,omitempty"`
	Tools      []ToolSnapshot    `json:"tools,omitempty"`
	Output     *MessageSnapshot  `json:"output,omitempty"`
	TokenUsage *TokenSnapshot    `json:"token_usage,omitempty"`
	DurationMs int64             `json:"duration_ms,omitempty"`
	ToolName   string            `json:"tool_name,omitempty"`
	ToolArgs   string            `json:"tool_args,omitempty"`
	ToolResult string            `json:"tool_result,omitempty"`
	Error      string            `json:"error,omitempty"`
}

LogRecord is a single JSONL record describing one phase of LLM interaction.

type Logger 

type Logger interface {
	Info(msg string, args ...any)
	Error(msg string, args ...any)
	Debug(msg string, args ...any)
}

Logger is a structured logger interface compatible with server.Logger. Implementations should output key-value pairs from the args slice.

type LoggingMiddleware 

type LoggingMiddleware struct {
	*adk.BaseChatModelAgentMiddleware
	// contains filtered or unexported fields
}

LoggingMiddleware is an Eino ChatModelAgentMiddleware that logs every model request/response, tool call, and agent lifecycle event to an LLMLogger. It implements adk.TypedChatModelAgentMiddleware[*schema.Message] by embedding *adk.BaseChatModelAgentMiddleware and overriding only the hooks it needs.

func NewLoggingMiddleware 

func NewLoggingMiddleware(logger *LLMLogger, agentID, model string) *LoggingMiddleware

NewLoggingMiddleware creates a LoggingMiddleware that writes records to logger for the given agentID and model name.

func (*LoggingMiddleware) AfterAgent 

func (m *LoggingMiddleware) AfterAgent(ctx context.Context, state *adk.ChatModelAgentState) (context.Context, error)

AfterAgent logs the "agent_end" phase with a summary of the final state.

func (*LoggingMiddleware) AfterModelRewriteState 

func (m *LoggingMiddleware) AfterModelRewriteState(ctx context.Context, state *adk.ChatModelAgentState, mc *adk.ModelContext) (context.Context, *adk.ChatModelAgentState, error)

AfterModelRewriteState logs the "response" phase after each model invocation. It captures the model's output (last message) and any token usage information.

func (*LoggingMiddleware) BeforeAgent 

func (m *LoggingMiddleware) BeforeAgent(ctx context.Context, runCtx *adk.ChatModelAgentContext) (context.Context, *adk.ChatModelAgentContext, error)

BeforeAgent logs the "agent_start" phase.

func (*LoggingMiddleware) BeforeModelRewriteState 

func (m *LoggingMiddleware) BeforeModelRewriteState(ctx context.Context, state *adk.ChatModelAgentState, mc *adk.ModelContext) (context.Context, *adk.ChatModelAgentState, error)

BeforeModelRewriteState logs the "request" phase before each model invocation. It captures the full message list and tool definitions that will be sent to the model.

func (*LoggingMiddleware) WrapInvokableToolCall 

func (m *LoggingMiddleware) WrapInvokableToolCall(ctx context.Context, endpoint adk.InvokableToolCallEndpoint, tCtx *adk.ToolContext) (adk.InvokableToolCallEndpoint, error)

WrapInvokableToolCall wraps tool execution to log "tool_call" before invocation and "tool_result" after completion. It also persists tool_calling messages to the database when a StoreAPI is available in the context, and broadcasts updates to clients via BroadcastHelper. When the store is not available (nil-safe D-063), it falls back to ephemeral-only broadcasting.

type MCPServerConfig 

type MCPServerConfig struct {
	Name      string   `yaml:"name" json:"name"`
	Transport string   `yaml:"transport" json:"transport"`                 // "sse" or "stdio"
	URL       string   `yaml:"url,omitempty" json:"url,omitempty"`         // SSE endpoint
	Command   string   `yaml:"command,omitempty" json:"command,omitempty"` // stdio command
	Args      []string `yaml:"args,omitempty" json:"args,omitempty"`       // stdio arguments
	Env       []string `yaml:"env,omitempty" json:"env,omitempty"`         // stdio environment variables
	Tools     []string `yaml:"tools,omitempty" json:"tools,omitempty"`     // filter specific tools (empty = all)
}

MCPServerConfig defines an MCP server connection (D-086). Transport selects the connection method: "sse" for remote servers via Server-Sent Events, "stdio" for local processes communicating over standard input/output.

type MessageSnapshot 

type MessageSnapshot struct {
	Role      string             `json:"role"`
	Content   string             `json:"content"`
	ToolCalls []ToolCallSnapshot `json:"tool_calls,omitempty"`
}

MessageSnapshot is a trimmed representation of a schema.Message.

type MiddlewareConfig 

type MiddlewareConfig struct {
	EnableSummarization   bool `yaml:"enable_summarization" json:"enable_summarization"`
	SummarizationTokens   int  `yaml:"summarization_tokens" json:"summarization_tokens"` // default: 160000
	EnableToolReduction   bool `yaml:"enable_tool_reduction" json:"enable_tool_reduction"`
	ToolReductionMaxChars int  `yaml:"tool_reduction_max_chars" json:"tool_reduction_max_chars"` // default: 50000
	EnablePatchToolCalls  bool `yaml:"enable_patch_tool_calls" json:"enable_patch_tool_calls"`
	// EnableClientTools enables dynamic injection of client device functions
	// as agent tools. Requires ClientFunctionProvider to be set on AgentBuilder.
	// Client functions use the RemoteCalling interrupt-resume pattern (D-137).
	EnableClientTools bool `yaml:"enable_client_tools" json:"enable_client_tools"`
	// ClientTools controls how client device functions are surfaced as tools.
	ClientTools ClientToolsConfig `yaml:"client_tools" json:"client_tools"`
}

MiddlewareConfig controls which Eino middleware to enable per agent. All fields are optional; when the middleware block is absent from YAML, no middleware is applied (backward compatible with Phase 1-7).

type MultiMetrics 

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

MultiMetrics fans out Record calls to multiple LLMMetrics implementations. It is safe for concurrent use.

func NewMultiMetrics 

func NewMultiMetrics(impls ...LLMMetrics) *MultiMetrics

NewMultiMetrics creates a MultiMetrics that delegates Record to all provided implementations in order.

func (*MultiMetrics) Record 

func (m *MultiMetrics) Record(ctx context.Context, event LLMCallEvent)

Record calls Record on each underlying LLMMetrics implementation.

type OllamaProvider 

type OllamaProvider struct{}

OllamaProvider creates ChatModel instances for local Ollama servers.

func (*OllamaProvider) CreateChatModel 

func (p *OllamaProvider) CreateChatModel(ctx context.Context, config *AgentConfig, _ string) (einomodel.BaseChatModel, error)

CreateChatModel builds an Ollama ChatModel. The apiKey parameter is ignored because Ollama runs locally without authentication.

func (*OllamaProvider) DefaultBaseURL 

func (p *OllamaProvider) DefaultBaseURL() string

DefaultBaseURL returns the default Ollama endpoint (D-064).

type OpenAIProvider 

type OpenAIProvider struct{}

OpenAIProvider creates ChatModel instances for OpenAI-compatible APIs.

func (*OpenAIProvider) CreateChatModel 

func (p *OpenAIProvider) CreateChatModel(ctx context.Context, config *AgentConfig, apiKey string) (einomodel.BaseChatModel, error)

CreateChatModel builds an OpenAI ChatModel.

func (*OpenAIProvider) DefaultBaseURL 

func (p *OpenAIProvider) DefaultBaseURL() string

DefaultBaseURL returns the OpenAI API endpoint (D-064).

type PrometheusMetrics 

type PrometheusMetrics struct{}

PrometheusMetrics implements LLMMetrics by recording to Prometheus. It is safe for concurrent use.

func NewPrometheusMetrics 

func NewPrometheusMetrics() *PrometheusMetrics

NewPrometheusMetrics creates a PrometheusMetrics that records LLM call events to the Prometheus metrics defined in the metrics package.

func (*PrometheusMetrics) Record 

func (m *PrometheusMetrics) Record(ctx context.Context, event LLMCallEvent)

Record updates Prometheus metrics for a single LLM call event.

type QwenProvider 

type QwenProvider struct{}

QwenProvider creates ChatModel instances for Alibaba Qwen (DashScope).

func (*QwenProvider) CreateChatModel 

func (p *QwenProvider) CreateChatModel(ctx context.Context, config *AgentConfig, apiKey string) (einomodel.BaseChatModel, error)

CreateChatModel builds a Qwen ChatModel.

func (*QwenProvider) DefaultBaseURL 

func (p *QwenProvider) DefaultBaseURL() string

DefaultBaseURL returns the DashScope compatible-mode endpoint (D-064).

type RedisCheckPointStore 

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

RedisCheckPointStore implements compose.CheckPointStore (Eino) backed by Redis. Checkpoints are stored as raw bytes under a configurable key prefix with a TTL to bound storage growth.

Design notes:

  • D-083: HITL does NOT support fail-open. All Redis errors propagate to the caller so the executor can abort the HITL flow.
  • D-074: uses an independent redis.Client (same pattern as the Pub/Sub client and idempotency client).

func NewRedisCheckPointStore 

func NewRedisCheckPointStore(client redisCheckpointClient, keyPrefix string, ttl time.Duration) *RedisCheckPointStore

NewRedisCheckPointStore creates a RedisCheckPointStore. If keyPrefix is empty it defaults to "agent:checkpoint:". If ttl <= 0 it defaults to 24 h.

func (*RedisCheckPointStore) Delete 

func (s *RedisCheckPointStore) Delete(ctx context.Context, key string) error

Delete removes a checkpoint by key. Idempotent — no error when key does not exist. Redis errors are returned to the caller (D-083: fail-closed).

func (*RedisCheckPointStore) Get 

func (s *RedisCheckPointStore) Get(ctx context.Context, key string) ([]byte, bool, error)

Get retrieves a checkpoint by its ID.

Returns (nil, false, nil) when the key does not exist (redis.Nil). Any other Redis error is returned to the caller (D-083: fail-closed).

func (*RedisCheckPointStore) Set 

func (s *RedisCheckPointStore) Set(ctx context.Context, key string, value []byte) error

Set persists a checkpoint value under the given key with the configured TTL.

type RedisConversationLock 

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

RedisConversationLock implements ConversationLock using Redis SETNX.

func NewRedisConversationLock 

func NewRedisConversationLock(client redisLockClient) *RedisConversationLock

NewRedisConversationLock creates a lock backed by Redis. The client must implement both SetNX and Eval.

func (*RedisConversationLock) Acquire 

func (l *RedisConversationLock) Acquire(ctx context.Context, conversationID string, ttl time.Duration) (bool, error)

Acquire uses SETNX with a unique token as value.

func (*RedisConversationLock) Release 

func (l *RedisConversationLock) Release(ctx context.Context, conversationID string) error

Release uses a Lua script to check token match before DEL.

type RedisIdempotencyStore 

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

RedisIdempotencyStore implements IdempotencyStore using Redis SETNX.

func NewRedisIdempotencyStore 

func NewRedisIdempotencyStore(client redisClient) *RedisIdempotencyStore

NewRedisIdempotencyStore creates a new RedisIdempotencyStore.

func (*RedisIdempotencyStore) CheckProcessed 

func (s *RedisIdempotencyStore) CheckProcessed(ctx context.Context, key string) (bool, error)

CheckProcessed checks if a key exists without setting it. Returns (true, nil) if the key exists, (false, nil) otherwise.

func (*RedisIdempotencyStore) DeleteKey 

func (s *RedisIdempotencyStore) DeleteKey(ctx context.Context, key string) error

DeleteKey removes a key. Used to clean up processing markers.

func (*RedisIdempotencyStore) MarkProcessed 

func (s *RedisIdempotencyStore) MarkProcessed(ctx context.Context, key string, ttl time.Duration) (bool, error)

MarkProcessed atomically checks if key was already processed and marks it. Returns (true, nil) if duplicate, (false, nil) if first time.

type RemoteCallingCleanupConfig 

type RemoteCallingCleanupConfig struct {
	// Interval is the time between cleanup runs. Defaults to 5 minutes.
	Interval time.Duration

	// MaxAge is the maximum time a conversation can remain in asking_user
	// status before being cleaned up. Defaults to 24 hours.
	MaxAge time.Duration

	// BatchSize is the maximum number of conversations/remote callings to process per cycle.
	// Defaults to 100.
	BatchSize int

	// LockTTL is the TTL for the per-conversation distributed lock.
	// Defaults to 30 seconds.
	LockTTL time.Duration
}

RemoteCallingCleanupConfig holds configuration for the RemoteCalling timeout cleanup task.

type RemoteCallingCleanupTask 

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

RemoteCallingCleanupTask periodically cleans up conversations stuck in asking_user status and individual expired RemoteCallings. It implements D-123 / D-137.

All cleanup steps are non-fatal (D-007): errors are logged but do not interrupt processing of other conversations.

func NewRemoteCallingCleanupTask 

func NewRemoteCallingCleanupTask(
	cfg RemoteCallingCleanupConfig,
	convStore *store.ConversationStore,
	remoteCallingStore *store.RemoteCallingStore,
	checkpointStore DeletableCheckPointStore,
	broadcaster *BroadcastHelper,
	dataStore store.StoreAPI,
	broker mq.Broker,
	lockClient redisClient,
	logger Logger,
) *RemoteCallingCleanupTask

NewRemoteCallingCleanupTask creates a RemoteCallingCleanupTask with the given dependencies. Zero-value fields in cfg are filled with sensible defaults:

  • Interval: 5 minutes
  • MaxAge: 24 hours
  • BatchSize: 100
  • LockTTL: 30 seconds

remoteCallingStore and checkpointStore may be nil (gracefully skipped). broker may be nil (agent resume enqueue skipped).

func (*RemoteCallingCleanupTask) Run 

func (t *RemoteCallingCleanupTask) Run(ctx context.Context)

Run starts the cleanup loop. It blocks until ctx is cancelled.

On each tick, Run calls cleanupOnce and logs the outcome. Panics are recovered to prevent the background goroutine from crashing the server. Cleanup failures are logged but do not interrupt the loop (D-007).

The first cleanup runs immediately on startup to avoid a 5-minute delay before stale conversations and expired RemoteCallings are cleaned up.

type Semaphore 

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

Semaphore limits concurrent agent executions with metrics tracking.

func NewSemaphore 

func NewSemaphore(capacity int) *Semaphore

NewSemaphore creates a Semaphore with the given capacity. If capacity <= 0, Acquire always succeeds immediately (unlimited).

func (*Semaphore) Acquire 

func (s *Semaphore) Acquire(ctx context.Context) error

Acquire blocks until a slot is available or ctx is cancelled. Returns nil if acquired, ctx.Err() if cancelled. For unlimited semaphores (capacity <= 0), always returns nil immediately.

func (*Semaphore) Release 

func (s *Semaphore) Release()

Release frees a slot. Safe to call on nil receiver.

func (*Semaphore) Stats 

func (s *Semaphore) Stats() SemaphoreStats

Stats returns current semaphore metrics.

type SemaphoreStats 

type SemaphoreStats struct {
	Capacity      int
	Active        int
	Peak          int
	TotalAcquired int64
}

SemaphoreStats contains a point-in-time snapshot of semaphore metrics.

type StreamBridge 

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

StreamBridge converts Eino's streaming output into Xyncra StreamChunks, applying a 50ms throttle for ~20fps streaming (D-051).

func NewStreamBridge 

func NewStreamBridge() *StreamBridge

NewStreamBridge creates a StreamBridge with the default 50ms throttle interval (D-051: 20 frames per second).

func (*StreamBridge) Bridge 

func (sb *StreamBridge) Bridge(ctx context.Context, iter *adk.AsyncIterator[*adk.AgentEvent], outCh chan<- StreamChunk)

Bridge reads events from the Eino AsyncIterator and emits cumulative StreamChunks into outCh. The method blocks until the iterator is exhausted, an error occurs, or ctx is cancelled. outCh is always closed on return.

Throttling ensures at most ~20 snapshots per second are emitted (D-051). Each StreamChunk.Content is the full accumulated text, so dropped frames do not affect correctness — the receiver simply overwrites its buffer.

func (*StreamBridge) BridgeWithInterrupt 

func (sb *StreamBridge) BridgeWithInterrupt(
	ctx context.Context,
	iter *adk.AsyncIterator[*adk.AgentEvent],
	outCh chan<- StreamChunk,
	interruptCh chan<- *InterruptInfo,
)

BridgeWithInterrupt is like Bridge but also detects HITL interrupt events. When an event with Action.Interrupted is received, the interrupt info is sent to interruptCh and the stream stops (outCh is closed normally).

If no interrupt occurs the method behaves identically to Bridge. interruptCh is always closed on return.

type StreamChunk 

type StreamChunk struct {
	Content string // Cumulative text snapshot (D-051)
	IsDone  bool   // True when stream completed normally
	Err     error  // Non-nil if stream ended with error
}

StreamChunk represents a single piece of streaming output. Content holds a cumulative text snapshot (D-051): each emitted chunk contains the full text accumulated so far, not just the delta. This lets receivers replace their display buffer directly without maintaining state.

type StreamingPayload 

type StreamingPayload struct {
	UserID         string `json:"user_id"`
	ConversationID string `json:"conversation_id"`
	StreamID       string `json:"stream_id"`
	Text           string `json:"text"`
	IsDone         bool   `json:"is_done"`
	IsAgent        bool   `json:"is_agent"`
	Timestamp      int64  `json:"timestamp"`
}

StreamingPayload is the JSON payload for UpdateTypeStreaming.

type TokenCounter 

type TokenCounter interface {
	CountTokens(text string) int
}

TokenCounter abstracts token counting for testability. Implementations range from simple heuristic estimators to precise tokenizers like tiktoken-go.

type TokenSnapshot 

type TokenSnapshot struct {
	InputTokens  int `json:"input_tokens"`
	OutputTokens int `json:"output_tokens"`
	TotalTokens  int `json:"total_tokens"`
}

TokenSnapshot captures token usage from a model response.

type ToolCallSnapshot 

type ToolCallSnapshot struct {
	Name string `json:"name"`
	Args string `json:"args"`
}

ToolCallSnapshot captures one tool call from an assistant message.

type ToolCallingPayload 

type ToolCallingPayload struct {
	Name       string `json:"name"`
	Args       string `json:"args"`
	Status     string `json:"status"`                // "executing" | "completed" | "failed"
	Result     string `json:"result,omitempty"`      // non-empty on completed
	Error      string `json:"error,omitempty"`       // non-empty on failed
	DurationMs int64  `json:"duration_ms,omitempty"` // non-zero on completed/failed
}

ToolCallingPayload is the JSON structure persisted as a tool_calling message's Content field. It captures the full lifecycle of a single tool invocation for client-side rendering and post-hoc analysis.

type ToolSnapshot 

type ToolSnapshot struct {
	Name string `json:"name"`
	Desc string `json:"desc"`
}

ToolSnapshot is a trimmed representation of a schema.ToolInfo.

type TracingMiddleware 

type TracingMiddleware struct {
	*adk.BaseChatModelAgentMiddleware
	// contains filtered or unexported fields
}

TracingMiddleware records LLM call details as OpenTelemetry spans. It mirrors the structure of LoggingMiddleware but writes spans instead of JSONL logs. Each model invocation produces an "agent.llm.call" span with model name, iteration, token usage, and duration. Each tool call produces an "agent.tool.call" span with tool name, duration, and error status.

Thread safety: Not safe for concurrent use. The Eino ADK serializes model calls per runner, so BeforeModelRewriteState/AfterModelRewriteState are always called in pairs without interleaving.

TracingMiddleware is only appended to the middleware chain when tracing is enabled (see AgentBuilder.SetTracingEnabled). When the global tracer provider is a no-op, this middleware has zero overhead because it is not added to the chain at all.

func NewTracingMiddleware 

func NewTracingMiddleware(agentID, model string) *TracingMiddleware

NewTracingMiddleware creates a TracingMiddleware for the given agent. It uses the global OpenTelemetry tracer provider (otel.Tracer).

func (*TracingMiddleware) AfterModelRewriteState 

func (m *TracingMiddleware) AfterModelRewriteState(ctx context.Context, state *adk.ChatModelAgentState, mc *adk.ModelContext) (context.Context, *adk.ChatModelAgentState, error)

AfterModelRewriteState ends the active "agent.llm.call" span, recording token usage (if available) and duration in milliseconds.

func (*TracingMiddleware) BeforeModelRewriteState 

func (m *TracingMiddleware) BeforeModelRewriteState(ctx context.Context, state *adk.ChatModelAgentState, mc *adk.ModelContext) (context.Context, *adk.ChatModelAgentState, error)

BeforeModelRewriteState starts an "agent.llm.call" span before each model invocation. The span records the model name and iteration number.

func (*TracingMiddleware) SetDebugFilter 

func (m *TracingMiddleware) SetDebugFilter(users, devices []string)

SetDebugFilter configures per-user/device debug content capture. When the caller matches (OR logic), full request/response content is recorded as span events on the agent.llm.call span.

func (*TracingMiddleware) WrapInvokableToolCall 

func (m *TracingMiddleware) WrapInvokableToolCall(ctx context.Context, endpoint adk.InvokableToolCallEndpoint, tCtx *adk.ToolContext) (adk.InvokableToolCallEndpoint, error)

WrapInvokableToolCall wraps tool execution with an "agent.tool.call" span that records the tool name, duration, and any error. When debug content capture is active, it also records tool input (argumentsInJSON) and output (result) as span events for debugging purposes.

type TypingPayload 

type TypingPayload struct {
	UserID         string `json:"user_id"`
	ConversationID string `json:"conversation_id"`
	IsTyping       bool   `json:"is_typing"`
	IsAgent        bool   `json:"is_agent"`
	Timestamp      int64  `json:"timestamp"`
}

TypingPayload is the JSON payload for UpdateTypeTyping.

Source Files

View all Source files

Directories

Path Synopsis
Package tools provides a factory-pattern tool registry and built-in tool implementations for the Xyncra Agent system.
 Package tools provides a factory-pattern tool registry and built-in tool implementations for the Xyncra Agent system.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.