component

package
v1.0.0-beta.162 Latest Latest
Warning

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

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 32 Imported by: 0

README

Component Package

Core component infrastructure for SemStreams, providing explicit factory registration, immutable declaration discovery, ports, schemas, and lifecycle interfaces.

Overview

The component package defines the fundamental abstractions for all SemStreams components, enabling dynamic discovery, registration, and management of input, processor, output, storage, and gateway components. This package follows explicit registration patterns with dependency injection through structured configuration.

Components in SemStreams are self-describing units whose declarations support boot composition and diagram validation. Their configuration is validated through schemas, and ComponentManager owns their lifecycle. The package supports five types of components: inputs (data sources), processors (data transformers), outputs (data sinks), storage (persistence), and gateways (query surfaces).

The Registry stores factory metadata and immutable declarations admitted during boot. ComponentManager is the sole owner of live component handles and lifecycle control. After boot admission, Registry is sealed.

Installation

import "github.com/c360/semstreams/component"

Architecture

Component Registration Flow

SemStreams uses EXPLICIT registration rather than init() self-registration:

flowchart TB
    subgraph Packages["Component Packages"]
        UDP[pkg/input/udp.go]
        WS[pkg/output/websocket.go]
        Graph[pkg/processor/graph]
        Robotics[pkg/processor/robotics]
        ObjStore[pkg/storage/objectstore]
    end

    subgraph Orchestration["componentregistry Package"]
        RegisterAll[RegisterAll function]
    end

    subgraph Main["main.go"]
        CreateReg[Create Registry]
        CallRegAll[Call RegisterAll]
        Ready[Components Ready]
    end

    UDP -->|"Register(registry)"| RegisterAll
    WS -->|"Register(registry)"| RegisterAll
    Graph -->|"Register(registry)"| RegisterAll
    Robotics -->|"Register(registry)"| RegisterAll
    ObjStore -->|"Register(registry)"| RegisterAll

    CreateReg --> CallRegAll
    RegisterAll --> CallRegAll
    CallRegAll --> Ready

    style Packages fill:#e1f5ff
    style Orchestration fill:#d4edda
    style Main fill:#fff3cd
Registration Pattern

Each component package exports a Register() function:

sequenceDiagram
    participant Main as main.go
    participant CR as componentregistry
    participant Reg as Registry
    participant UDP as pkg/input
    participant Graph as pkg/processor/graph

    Main->>Reg: NewRegistry()
    Main->>CR: RegisterAll(registry)
    CR->>UDP: Register(registry)
    UDP->>Reg: RegisterInput("udp", factory, ...)
    CR->>Graph: Register(registry)
    Graph->>Reg: RegisterProcessor("graph-processor", factory, ...)
    CR-->>Main: All components registered
    Main->>Reg: Internal boot admission
    Reg-->>Main: immutable declaration
    Main->>Reg: SealComposition()
Why Explicit Registration?
Aspect init() Self-Registration Explicit Registration
Testability ❌ Global state, hard to isolate ✅ Create isolated test registries
Explicitness ❌ Hidden dependencies via imports ✅ Clear dependency graph
Control ❌ Automatic on import ✅ Application controls what/when
Side Effects ❌ Package import modifies globals ✅ No side effects from imports
Debugging ❌ Registration order unclear ✅ Deterministic, explicit order
FlowGraph Component

The FlowGraph component (flowgraph/) provides static analysis and validation of component interconnections. It is used by the flow validator to analyze saved or draft diagrams before explicit publication.

Purpose: Build and validate connectivity graphs from component port definitions

Key Responsibilities:

  • Build connectivity graphs from component port definitions
  • Auto-discover connections via pattern matching (NATS subjects, KV buckets)
  • Detect orphaned ports and disconnected components
  • Validate interface contracts between connected ports
  • Identify resource conflicts (e.g., network port binding)

Important: FlowGraph is a validation tool, not a runtime component. It creates temporary graph structures for diagram analysis and is discarded after validation completes.

Relationship to Flow Infrastructure:

Flow Service (HTTP API)
    ↓ uses
Flow Engine (Validation + Compilation)
    ↓ validation → FlowGraph (Static Analysis)
    ↓ explicit publication → Config Manager (next boot)
  • FlowGraph: "Can these components connect?" (static graph analysis)
  • Flow Engine: "Validate and compile this diagram" (authoring operation)
  • Flow Service: "Save, validate, observe, or explicitly publish this diagram" (REST API layer)

Each layer has distinct, non-overlapping responsibilities.

Quick Start

Boot composition

Applications explicitly register factories before process composition:

registry := component.NewRegistry()
if err := componentregistry.Register(registry); err != nil {
    return err
}

ComponentManager performs instance creation through the framework-internal admission boundary, captures immutable declarations, retains every live handle, and seals Registry. External callers do not create or recover runtime component handles through Registry.

Implementing a Component
package mycomponent

import (
    "encoding/json"

    "github.com/c360/semstreams/component"
)

// Component implementation
type MyInput struct {
    config MyConfig
    deps   component.Dependencies
}

func (m *MyInput) Meta() component.Metadata {
    return component.Metadata{
        Name:        "my-input",
        Type:        "input",
        Description: "My custom input component",
        Version:     "1.0.0",
    }
}

func (m *MyInput) InputPorts() []component.Port { return nil }

func (m *MyInput) OutputPorts() []component.Port {
    return []component.Port{
        {
            Name:      "output",
            Direction: component.DirectionOutput,
            Required:  true,
            Config:    component.NATSPort{Subject: "my.output"},
        },
    }
}

func (m *MyInput) ConfigSchema() component.ConfigSchema {
    return component.ConfigSchema{
        Properties: map[string]component.PropertySchema{
            "interval": {Type: "duration", Description: "Poll interval"},
        },
    }
}

func (m *MyInput) Health() component.HealthStatus {
    return component.HealthStatus{Healthy: true}
}

func (m *MyInput) DataFlow() component.FlowMetrics {
    return component.FlowMetrics{}
}

// Factory function
func CreateMyInput(rawConfig json.RawMessage, deps component.Dependencies) (component.Discoverable, error) {
    var config MyConfig
    if err := json.Unmarshal(rawConfig, &config); err != nil {
        return nil, err
    }

    return &MyInput{
        config: config,
        deps:   deps,
    }, nil
}

// IMPORTANT: Export Register() function, NOT init()
func Register(registry *component.Registry) error {
    return registry.RegisterWithConfig(component.RegistrationConfig{
        Name:        "my-input",
        Factory:     CreateMyInput,
        Schema:      myInputSchema,
        Type:        "input",
        Protocol:    "custom",
        Domain:      "network",
        Description: "My custom input component",
        Version:     "1.0.0",
    })
}

Then add to pkg/componentregistry/register.go:

import "github.com/yourorg/semstreams/pkg/mycomponent"

func registercore (registry *component.Registry) error {
    // ... existing registrations

    if err := mycomponent.Register(registry); err != nil {
        return err
    }

    return nil
}

core Concepts

Discoverable Interface

Every component must implement:

type Discoverable interface {
    Meta() Metadata                  // Component metadata
    InputPorts() []Port              // Input port definitions
    OutputPorts() []Port             // Output port definitions
    ConfigSchema() ConfigSchema      // Configuration schema
    Health() HealthStatus            // Current health status
    DataFlow() FlowMetrics          // Data flow metrics
}
Dependencies

Dependency injection structure:

type Dependencies struct {
    NATSClient      *natsclient.Client      // Required: messaging
    ObjectStore     ObjectStore             // Optional: persistence
    MetricsRegistry *metric.MetricsRegistry // Optional: Prometheus
    Logger          *slog.Logger            // Optional: logging
    Platform        PlatformMeta            // Required: identity
}
Port Types

Components declare ports using strongly-typed configurations:

// NATS Pub/Sub
component.NATSPort{Subject: "data.output"}

// JetStream durable streaming
component.JetStreamPort{Stream: "EVENTS", Subject: "events.>"}

// KV bucket watch
component.KVWatchPort{Bucket: "CONFIG", Keys: []string{"app.*"}}

// KV bucket write
component.KVWritePort{
    Bucket: "ENTITY_STATES",
    Interface: &component.InterfaceContract{
        Type:    "graph.EntityState",
        Version: "v1",
    },
}

// Network binding
component.NetworkPort{Protocol: "udp", Port: 14550, Bind: "0.0.0.0"}

API Reference

Registry
NewRegistry() *Registry

Creates an empty registry for explicit factory registration and boot admission.

RegisterWithConfig(config RegistrationConfig) error

Registers a component factory and its static metadata before composition seals.

ListAvailable() map[string]Info

Returns defensive component-type metadata. Registry declaration reads likewise return defensive values and never expose a live component handle.

Component creation, declaration admission, and sealing require the framework-internal admission token and are not adopter APIs.

Types
Factory
type Factory func(rawConfig json.RawMessage, deps Dependencies) (Discoverable, error)

Factory function signature for component creation.

Error Handling

Error Types
ErrFactoryAlreadyExists // Duplicate factory registration
ErrInvalidFactory       // Invalid factory registration
ErrFactoryNotFound      // Unknown factory name during internal boot admission
ErrComponentCreation    // Factory execution failed during internal boot admission
Error Detection
err := registry.RegisterWithConfig(registration)
if errors.Is(err, component.ErrFactoryAlreadyExists) {
    // Registration was repeated before composition sealed.
}

Testing

Isolated Test Registries
func TestMyComponent(t *testing.T) {
    // Create isolated registry for this test
    registry := component.NewRegistry()

    // Register only components needed
    if err := mycomponent.Register(registry); err != nil {
        t.Fatal(err)
    }

    factories := registry.ListAvailable()
    assert.Contains(t, factories, "my-input")
}
Testing Patterns
  • ✅ Use real NATS via natsclient.NewTestClient() for integration tests
  • ✅ Create isolated registries per test to avoid global state
  • ✅ Mock external dependencies that cannot be containerized
  • ✅ Test component behavior through Discoverable interface
  • ✅ Verify public factory metadata and schemas through defensive values

Performance

Registry Operations
Operation Complexity Thread-Safe
Factory lookup O(1) Yes (read lock)
Component creation O(1) + factory time Yes (read lock)
Factory registration O(1) Yes (write lock)
List operations O(n) Yes (read lock)
Concurrency
  • Multiple goroutines can create components concurrently
  • Factory registration blocks component creation temporarily
  • No deadlocks due to ordered lock acquisition
  • Components maintain references until explicitly unregistered

Architecture Decisions

Explicit Registration vs init()

Decision: Use explicit Register() functions

Rationale:

  • Testability: Can create isolated registries without global state
  • Explicitness: Clear component dependency graph in componentregistry
  • Control: Application controls what gets registered and when
  • No side effects: Package imports don't modify global state
  • Deterministic: Registration order is explicit and controllable

Tradeoffs:

  • Requires componentregistry orchestration package
  • Registration must be explicitly called in main()
  • New components must update componentregistry.RegisterAll()
Dependency Injection via Struct

Decision: Use Dependencies struct

Rationale:

  • Avoids parameter proliferation
  • Easy to add dependencies without breaking factories
  • Enables testing with mock dependencies
  • Follows service architecture patterns
Factory Pattern

Decision: Components parse their own configuration

Rationale:

  • Enables flexible validation per component
  • Matches service constructor patterns
  • Centralizes configuration knowledge in component packages

Documentation

Overview

Package component defines the Discoverable interface and related types

Package component defines SemStreams component factories, lifecycle interfaces, dependencies, configuration schemas, ports, and immutable registry declarations.

Applications register component factories explicitly during process composition. Factories are side-effect free: they construct a Discoverable, while I/O begins in Start and all owned work joins Stop.

Component creation is a framework-internal boot operation. ComponentManager constructs every enabled component from its constructor-captured configuration, captures the component's declared ports and resources, and admits an immutable declaration to Registry. ComponentManager retains the concrete handle as the sole lifecycle owner.

Registry read surfaces return defensive values and never expose a live component handle. Once boot admission is complete, Registry is sealed; later factory or component admission attempts fail until another process boot creates a fresh Registry.

Discoverable components describe metadata, health, flow metrics, schemas, and typed ports. Port declarations support core NATS subjects, JetStream, KV watches and writes, request/reply, network listeners, APIs, and related framework patterns. FlowGraph consumes declaration values for static diagram validation without owning or mutating runtime lifecycle.

Package component provides base types and utilities for SemStreams components.

Package component provides port configuration and management for component connections.

Package component provides schema validation and helper functions

Package component provides schema tag parsing and generation for component configuration.

The schema tag system eliminates duplication between Config structs and ConfigSchema definitions by auto-generating schemas from struct tags. This provides a single source of truth for configuration metadata and follows Go stdlib patterns (similar to json tags).

Basic Usage

Define configuration with schema tags: 

type MyConfig struct {
    Name string `json:"name" schema:"type:string,description:Component name,category:basic"`
    Port int    `json:"port" schema:"type:int,description:Port,min:1,max:65535,default:8080"`
}

Generate schema at init time: 

var schema = component.GenerateConfigSchema(reflect.TypeOf(MyConfig{}))

Tag Syntax

Tags use comma-separated directives with colon-separated key-value pairs:

  • type:string - Field data type (required)
  • description:text - Field description (recommended)
  • category:basic - UI organization (basic or advanced)
  • default:value - Default value
  • min:N, max:N - Numeric constraints
  • enum:a|b|c - Valid enum values (pipe-separated)
  • readonly, editable - Boolean flags for PortDefinition fields
  • required, hidden - Boolean flags for validation and UI

Performance

Schema generation uses reflection but is designed for init-time execution:

  • Call GenerateConfigSchema once at package init
  • Cache result in package-level variable
  • Zero reflection cost at runtime

Error Handling

Invalid tags result in graceful degradation:

  • Fields with invalid tags are skipped
  • Errors are wrapped with context using pkg/errors
  • Missing descriptions use field names as fallback

See docs/architecture/SCHEMA_TAG_SPEC.md for complete specification.

Index

Examples

Constants

View Source 
const (
	MaxStringLength = 1024          // Maximum length for string values
	MaxJSONSize     = 1024 * 1024   // Maximum JSON size (1MB)
	MinPort         = 1             // Minimum valid port number
	MaxPort         = 65535         // Maximum valid port number
	MaxInt          = math.MaxInt32 // Maximum safe integer value
	MinInt          = math.MinInt32 // Minimum safe integer value
)

Config validation constants - security limits

View Source 
const (
	// DepModelRegistry signals that a component consumes
	// Dependencies.ModelRegistry at construction or run time. Components
	// that declare this consume the model registry selected at boot.
	// Later model_registry writes require process restart.
	DepModelRegistry = "model-registry"
)

Dependency identifiers used by Registration.Dependencies. Components declare these at registration to opt into framework-driven behavior (e.g., restart when a named runtime dependency changes). Kept as typed constants so refactors are compiler-checked.

Variables

This section is empty.

Functions

func BenchmarkLifecycleMethods 

func BenchmarkLifecycleMethods(b *testing.B, factory LifecycleFactory)

BenchmarkLifecycleMethods provides benchmark tests for lifecycle operations

func GenerateCacheFieldSchema 

func GenerateCacheFieldSchema() map[string]CacheFieldInfo

GenerateCacheFieldSchema generates metadata for cache.Config fields. This describes which fields in cache.Config are editable and their constraints, enabling the UI to render appropriate controls for cache configuration.

The function examines cache.Config struct tags to determine:

  • Field types (for appropriate UI controls)
  • Editability (whether users can modify the field)
  • Enum values (for strategy field)
  • Numeric constraints (for size limits)

All cache.Config fields are marked as "editable" to allow runtime configuration.

This metadata is included in ConfigSchema for fields with type "cache", allowing the frontend to correctly render cache configuration forms.

Returns:

  • Map of field names to CacheFieldInfo with type, editability, and constraint metadata

func GeneratePortFieldSchema 

func GeneratePortFieldSchema() map[string]PortFieldInfo

GeneratePortFieldSchema returns the canonical common-envelope metadata. Kind-specific data remains inside config and is validated by the closed port binding table; schema generation does not reflect retired flat fields.

This metadata is included in ConfigSchema for fields with type "ports", allowing the frontend to correctly render port configuration forms.

Returns:

  • Map of field names to PortFieldInfo with type and editability metadata

func GetBool 

func GetBool(config map[string]any, key string, defaultValue bool) bool

GetBool safely extracts a boolean value from config with a default fallback and validation

func GetFloat64 

func GetFloat64(config map[string]any, key string, defaultValue float64) float64

GetFloat64 safely extracts a float64 value from config with a default fallback and validation

func GetInt 

func GetInt(config map[string]any, key string, defaultValue int) int

GetInt safely extracts an integer value from config with a default fallback and bounds checking

func GetProperties 

func GetProperties(schema ConfigSchema, category string) map[string]PropertySchema

GetProperties filters schema properties by category for UI organization.

Components can categorize their configuration properties as "basic" (shown by default) or "advanced" (hidden in collapsible section). This function extracts properties belonging to a specific category.

Parameters:

  • schema: The component's configuration schema
  • category: Filter by "basic" or "advanced", or empty string for all properties

Properties without an explicit Category field default to "advanced".

Returns a map of property names to PropertySchema definitions matching the category.

Example: 

schema := component.ConfigSchema{
    Properties: map[string]component.PropertySchema{
        "port":        {Type: "int", Category: "basic"},
        "buffer_size": {Type: "int", Category: "advanced"},
        "timeout":     {Type: "int"}, // Defaults to "advanced"
    },
}

basicProps := component.GetProperties(schema, "basic")
// Returns: map["port": {...}]

advancedProps := component.GetProperties(schema, "advanced")
// Returns: map["buffer_size": {...}, "timeout": {...}]

func GetPropertyValue 

func GetPropertyValue(config map[string]any, key string) (any, bool)

GetPropertyValue safely extracts a property value from a configuration map.

Returns the value and true if the key exists, or nil and false if the key is not present in the map. This function is nil-safe - passing a nil config will return (nil, false).

Example: 

config := map[string]any{"port": 8080, "host": "localhost"}
if port, exists := component.GetPropertyValue(config, "port"); exists {
    fmt.Printf("Port: %v\n", port)
}

func GetString 

func GetString(config map[string]any, key string, defaultValue string) string

GetString safely extracts a string value from config with a default fallback and validation

func IsComplexType 

func IsComplexType(propType string) bool

IsComplexType returns true if a property type requires complex rendering.

Complex types (object, array) cannot be rendered as simple form inputs and require specialized UI components like JSON editors or nested form builders.

Currently identifies "object" and "array" types as complex. In MVP, the UI falls back to a JSON editor for these types.

Example: 

if component.IsComplexType(propSchema.Type) {
    // Use JSON editor fallback
    renderJSONEditor(propSchema)
} else {
    // Render type-specific input field
    renderInputField(propSchema)
}

func IsLifecycleComponent 

func IsLifecycleComponent(comp Discoverable) bool

IsLifecycleComponent checks if a component supports lifecycle management

func ResolveSubject 

func ResolveSubject(ports []PortDefinition, portName, suffix string) (string, error)

ResolveSubject returns the configured NATS subject for one uniquely named port with a trailing wildcard replaced by suffix.

func SafeUnmarshal 

func SafeUnmarshal(rawConfig json.RawMessage, target any) error

SafeUnmarshal performs validated unmarshaling into a target struct It validates the JSON first, then unmarshals with additional type checking

func SortedPropertyNames 

func SortedPropertyNames(schema ConfigSchema) []string

SortedPropertyNames returns property names in UI display order.

Properties are sorted by: 1. Category: "basic" properties first, then "advanced" properties 2. Alphabetically within each category

This ensures consistent, predictable ordering in configuration UIs. Properties without an explicit Category default to "advanced".

Example: 

schema := component.ConfigSchema{
    Properties: map[string]component.PropertySchema{
        "port":         {Category: "basic"},
        "bind_address": {Category: "basic"},
        "buffer_size":  {Category: "advanced"},
        "timeout":      {}, // Defaults to "advanced"
    },
}

names := component.SortedPropertyNames(schema)
// Returns: ["bind_address", "port", "buffer_size", "timeout"]
//          ^-- basic (alpha) --^  ^---- advanced (alpha) ----^

func StandardLifecycleTests 

func StandardLifecycleTests(t *testing.T, factory LifecycleFactory)

StandardLifecycleTests verifies the portable LifecycleComponent floor. Resource-specific drain ordering, blocked joins, and partial-acquisition rollback remain the responsibility of focused owner tests.

func TestErrorInjection 

func TestErrorInjection(t *testing.T, factory LifecycleFactory)

TestErrorInjection tests components with injected errors

func ValidateAndPersistComponentConfig 

func ValidateAndPersistComponentConfig(
	ctx context.Context,
	logger *slog.Logger,
	registry ConfigComponentRegistry,
	persister ConfigPersister,
	componentName, componentType string,
	configJSON json.RawMessage,
) error

ValidateAndPersistComponentConfig validates a component config and persists it to KV via the supplied persister. Returns validation errors as the first error if validation fails, or a wrapped persistence error if KV write fails.

Persister parameter (typically a *natsclient.KVStore from config.Manager) is taken explicitly so component doesn't import config — that import direction is what created the cycle this helper exists to avoid.

func ValidateComponentName 

func ValidateComponentName(name string) error

ValidateComponentName validates component/instance names for security

func ValidateConfigKey 

func ValidateConfigKey(key string) error

ValidateConfigKey checks if a configuration key is valid

func ValidateFactoryConfig 

func ValidateFactoryConfig(rawConfig json.RawMessage) error

ValidateFactoryConfig performs validation before passing to factory This is the main security gate for all component configurations

func ValidateJSONSize 

func ValidateJSONSize(data json.RawMessage) error

ValidateJSONSize checks if JSON input is within safe limits

func ValidateNetworkConfig 

func ValidateNetworkConfig(port int, bindAddr string) error

ValidateNetworkConfig validates network configuration including port and bind address

func ValidatePortNumber 

func ValidatePortNumber(port int) error

ValidatePortNumber validates port numbers are within valid range

Types

type CacheFieldInfo 

type CacheFieldInfo struct {
	Type     string   `json:"type"`
	Editable bool     `json:"editable"`
	Enum     []string `json:"enum,omitempty"` // For strategy field
	Min      *int     `json:"min,omitempty"`  // For numeric fields
}

CacheFieldInfo describes metadata for cache.Config fields

type ConfigComponentRegistry 

type ConfigComponentRegistry interface {
	GetComponentSchema(componentType string) (ConfigSchema, error)
}

ConfigComponentRegistry defines the interface needed for schema validation. Local-package interface so the validator helpers don't depend on the broader Registry surface.

type ConfigPersister 

type ConfigPersister interface {
	Put(ctx context.Context, key string, value []byte) (uint64, error)
}

ConfigPersister is the minimal kvStore-shaped surface ValidateAndPersistComponentConfig needs. *natsclient.KVStore satisfies it. Defined here so config.Manager can pass its kvStore through without component importing config (which would cycle: component → config → component).

type ConfigSchema 

type ConfigSchema struct {
	Properties map[string]PropertySchema `json:"properties"`
	Required   []string                  `json:"required"`
}

ConfigSchema describes the configuration parameters for a component

func GenerateConfigSchema 

func GenerateConfigSchema(configType reflect.Type) ConfigSchema

GenerateConfigSchema generates a ConfigSchema from a struct type using reflection. This function performs one-time reflection at initialization to extract schema metadata from struct field tags, eliminating the need for manual schema definitions.

Usage Pattern: 

type MyComponentConfig struct {
    Name string `json:"name" schema:"type:string,description:Name,category:basic"`
    Port int    `json:"port" schema:"type:int,description:Port,min:1,max:65535"`
}

var schema = component.GenerateConfigSchema(reflect.TypeOf(MyComponentConfig{}))

Field Processing:

  • Only exported fields with both 'json' and 'schema' tags are included
  • json:"-" fields are skipped
  • Fields without schema tags are skipped
  • Invalid schema tags result in skipped fields (graceful degradation)

Special Handling:

  • Fields with type "ports" automatically include PortFieldSchema metadata
  • Default values are converted from strings to appropriate types
  • Required fields are added to the schema's Required list

Performance:

  • Call once at init() time - reflection cost is paid only once
  • Generated schemas are cached in package-level variables
  • Zero reflection overhead at runtime

Parameters:

  • configType: The reflect.Type of the config struct (use reflect.TypeOf(ConfigStruct{})) Pointer types are automatically dereferenced

Returns:

  • ConfigSchema with Properties map and Required list populated from struct tags
  • Empty schema for non-struct types
Example

ExampleGenerateConfigSchema demonstrates how to use schema tags to auto-generate configuration schemas from struct definitions 

package main

import (
	"encoding/json"
	"fmt"
	"reflect"

	"github.com/c360studio/semstreams/component"
)

func main() {
	// Define a configuration struct with schema tags
	type ComponentConfig struct {
		// Basic configuration
		Name    string `json:"name"    schema:"type:string,description:Component name,category:basic"`
		Port    int    `json:"port"    schema:"type:int,description:Listen port,min:1,max:65535,default:8080,category:basic"`
		Enabled bool   `json:"enabled" schema:"type:bool,description:Enable component,default:true,category:basic"`

		// Advanced configuration
		Timeout  string `json:"timeout"   schema:"type:string,description:Request timeout,default:30s,category:advanced"`
		LogLevel string `json:"log_level" schema:"type:enum,description:Logging level,enum:debug|info|warn|error,default:info,category:advanced"`

		// Required field
		APIKey string `json:"api_key" schema:"required,type:string,description:Authentication API key"`
	}

	// Generate the schema at init time (one-time reflection cost)
	schema := component.GenerateConfigSchema(reflect.TypeOf(ComponentConfig{}))

	// The generated schema can be used for validation, UI generation, etc.
	schemaJSON, _ := json.MarshalIndent(schema, "", "  ")
	fmt.Println(string(schemaJSON))

	// Output will show the generated schema with all properties
}

type ConfigValidator 

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

ConfigValidator provides secure validation for component configurations

func NewConfigValidator 

func NewConfigValidator() *ConfigValidator

NewConfigValidator creates a validator with secure defaults

func (*ConfigValidator) ValidateConfig 

func (v *ConfigValidator) ValidateConfig(rawConfig json.RawMessage) error

ValidateConfig performs comprehensive validation on raw JSON config This prevents injection attacks, resource exhaustion, and malformed input

type ConsumerConfig 

type ConsumerConfig struct {
	DeliverPolicy     string
	AckPolicy         string
	MaxDeliver        int
	AckWait           time.Duration
	HeartbeatInterval time.Duration
	MaxAckPending     int // 0 = inherited/default/capped server policy; -1 = unlimited outstanding acks (gh#480)
}

ConsumerConfig holds extracted JetStream consumer configuration.

Duration fields (AckWait, HeartbeatInterval) are zero-valued when the port-level string is empty; consumers may apply their component default. Invalid declarations are rejected before a ConsumerConfig is returned.

func GetConsumerConfig 

func GetConsumerConfig(port Port) (ConsumerConfig, error)

GetConsumerConfig validates a JetStream port through the canonical facts projection and extracts its consumer configuration. A non-JetStream or invalid port returns an error rather than silently receiving defaults. Unset fields receive these defaults: - DeliverPolicy: "new" (safe default - don't replay historical messages) - AckPolicy: "explicit" - MaxDeliver: 3 - AckWait, HeartbeatInterval: zero (caller applies per-component default)

type DebugStatusProvider 

type DebugStatusProvider interface {
	// DebugStatus returns extended debug information for the component.
	// The returned value should be JSON-serializable.
	DebugStatus() any
}

DebugStatusProvider is an optional interface for components that can provide extended debug information beyond basic health status.

type Dependencies 

type Dependencies struct {
	NATSClient      *natsclient.Client        // NATS client for messaging
	MetricsRegistry *metric.MetricsRegistry   // Metrics registry for Prometheus (can be nil)
	Logger          *slog.Logger              // Structured logger (can be nil, defaults to slog.Default())
	Platform        PlatformMeta              // Platform identity (organization and platform)
	Security        security.Config           // Platform-wide security configuration
	ModelRegistry   model.RegistryReader      // Boot-selected model registry (can be nil)
	ToolRegistry    ToolRegistryReader        // Shared tool executor registry (can be nil; agentic-tools requires it)
	PayloadRegistry *payloadregistry.Registry // Shared payload registry (can be nil; components unmarshaling BaseMessage require it)

	// LifecycleManager is the shared pkg/lifecycle.Manager that
	// owns workflow-shaped entity instances (ADR-047). Apps that
	// declare Participant-implementing entities (drone missions,
	// sensor lifecycles, manufacturing batches, scenario executions)
	// build the Manager in main.go, call Manager.Register for each
	// app workflow, and pass it through Dependencies. Both the rule
	// processor (lifecycle_* actions + $entity.lifecycle.* condition
	// fields) and the lifecycle-gateway (operator HTTP API) read
	// from this field.
	//
	// Concrete type (not an interface) per the PayloadRegistry
	// precedent — pkg/lifecycle is a framework-owned leaf package,
	// not an external/pluggable surface, and consumers narrow to
	// their own minimum-surface interfaces locally (see
	// processor/rule.LifecycleManager) when they want to abstract
	// for testing.
	//
	// Can be nil — apps without any lifecycle-managed entity types
	// pay zero cost, and the consumers that DO read this field
	// loud-fail with a wiring error rather than silently no-op'ing.
	LifecycleManager *lifecycle.Manager

	// StoreRegistry is the shared {StorageInstance → storage.StreamableStore}
	// resolver (ADR-063). The ComponentManager populates it from the boot-composed
	// storage components' store-provide ports at Start and clears entries at Stop;
	// content-fetch consumers (graph-embedding, fusion) resolve a StorageRef's
	// StorageInstance through it, lazily per-fetch. Concrete framework-leaf type
	// per the PayloadRegistry / LifecycleManager precedent.
	//
	// Can be nil — deployments with no offloaded-content fetch pay zero cost. A
	// consumer that receives a StorageReference without this exact-name authority
	// reports content-unresolved and excludes the body; it never selects another
	// store or degrades solely because the name is unresolved.
	StoreRegistry *storeregistry.Registry
}

Dependencies provides all external dependencies needed by components.

PayloadRegistry uses the concrete *payloadregistry.Registry rather than an interface (the way ToolRegistry uses ToolRegistryReader) on purpose: payloadregistry is a leaf package this package already imports, so there's no cycle to dodge, and message.NewDecoder also requires the concrete type. An interface here would force callers to type-assert at every Decoder construction site — pure friction for no abstraction win.

func (*Dependencies) GetLogger 

func (d *Dependencies) GetLogger() *slog.Logger

GetLogger returns the configured logger or a default logger if none is provided

func (*Dependencies) GetLoggerWithComponent 

func (d *Dependencies) GetLoggerWithComponent(componentName string) *slog.Logger

GetLoggerWithComponent returns a logger configured with component context

type Direction 

type Direction string

Direction identifies whether data enters or leaves a component. 

const (
	DirectionInput  Direction = "input"
	DirectionOutput Direction = "output"
)

Direction constants for port data flow.

type Discoverable 

type Discoverable interface {
	// Meta returns basic component information
	Meta() Metadata

	// InputPorts returns the ports this component accepts data on
	InputPorts() []Port

	// OutputPorts returns the ports this component produces data on
	OutputPorts() []Port

	// ConfigSchema returns the configuration schema for this component
	ConfigSchema() ConfigSchema

	// Health returns current health status
	Health() HealthStatus

	// DataFlow returns current data flow metrics
	DataFlow() FlowMetrics
}

Discoverable defines the interface for components that can be discovered and inspected by the management layer. This interface enables dynamic discovery of component capabilities, configuration, and health status.

Components implementing this interface can be: - Input components: Accept external data (UDP, TCP, HTTP) - Processor components: Transform data (plugins) - Output components: Send data to external systems - Storage components: Store and retrieve data (ObjectStore, KV)

type ErrorInjectingComponent 

type ErrorInjectingComponent struct {
	LifecycleComponent
	// contains filtered or unexported fields
}

ErrorInjectingComponent wraps a component to inject errors for testing

func NewErrorInjectingComponent 

func NewErrorInjectingComponent(comp LifecycleComponent) *ErrorInjectingComponent

NewErrorInjectingComponent creates a component wrapper that can inject errors for testing

func (*ErrorInjectingComponent) Initialize 

func (e *ErrorInjectingComponent) Initialize() error

Initialize initializes the component, returning injected error if configured

func (*ErrorInjectingComponent) InjectInitializeError 

func (e *ErrorInjectingComponent) InjectInitializeError(err error)

InjectInitializeError configures the component to return an error on Initialize

func (*ErrorInjectingComponent) InjectStartError 

func (e *ErrorInjectingComponent) InjectStartError(err error)

InjectStartError configures the component to return an error on Start

func (*ErrorInjectingComponent) InjectStopError 

func (e *ErrorInjectingComponent) InjectStopError(err error)

InjectStopError configures the component to return an error on Stop

func (*ErrorInjectingComponent) Start 

func (e *ErrorInjectingComponent) Start(ctx context.Context) error

Start starts the component, returning injected error if configured

func (*ErrorInjectingComponent) Stop 

func (e *ErrorInjectingComponent) Stop(ctx context.Context) error

Stop stops the component, returning injected error if configured

type Factory 

type Factory func(rawConfig json.RawMessage, deps Dependencies) (Discoverable, error)

Factory creates a component instance from configuration following service pattern The factory function receives raw JSON configuration and dependencies, parses its own config, and returns a properly initialized component that implements the Discoverable interface. All I/O operations should be performed in the component's Start() method, not in the factory. This pattern matches service constructors: func(rawConfig json.RawMessage, deps Dependencies) (Service, error)

type FilePort 

type FilePort struct {
	Path    string `json:"path"`
	Pattern string `json:"pattern,omitempty"`
}

FilePort - File system access

func (FilePort) IsExclusive 

func (f FilePort) IsExclusive() bool

IsExclusive returns false as multiple components can read files

func (FilePort) Kind 

func (f FilePort) Kind() PortKind

Kind returns the canonical port kind.

func (FilePort) ResourceID 

func (f FilePort) ResourceID() string

ResourceID returns unique identifier for file ports

type FlowMetrics 

type FlowMetrics struct {
	MessagesPerSecond float64   `json:"messages_per_second"`
	BytesPerSecond    float64   `json:"bytes_per_second"`
	ErrorRate         float64   `json:"error_rate"`
	LastActivity      time.Time `json:"last_activity"`
}

FlowMetrics describes the current data flow through a component

type HTTPClientPort 

type HTTPClientPort struct {
	Method        string             `json:"method,omitempty"`         // Required HTTP method
	URLPattern    string             `json:"url_pattern"`              // target endpoint/resource pattern; NO inline creds, NO query-string secrets
	TriggerPort   string             `json:"trigger_port,omitempty"`   // NAME of a sibling TimerPort driving cadence (empty = event-driven/internal)
	AuthRef       string             `json:"auth_ref,omitempty"`       // credential key name resolved at runtime; NEVER a secret value; empty = unauthenticated
	ContactPolicy string             `json:"contact_policy,omitempty"` // public User-Agent/contact identity (feeds often require it); not a secret
	Interface     *InterfaceContract `json:"interface,omitempty"`      // message-interface contract this port produces
}

HTTPClientPort describes an outbound HTTP-client / polling input dependency: a relationship where the component initiates connections to an external HTTP resource (REST API, alert feed, snapshot endpoint) rather than binding a listener (NetworkPort) or consuming a NATS subject (NATSPort).

It is a DESCRIPTOR, not a runtime — like TimerPort it declares the dependency shape so lifecycle/config/health/flowgraph/ownership/telemetry tooling can SEE it; the polling/HTTP execution stays in the component. Cadence is NOT owned here: when polling is timer-driven the component also declares a TimerPort and this port references it by name via TriggerPort (single source of truth for the interval).

Secrets never appear in this descriptor. AuthRef carries a credential REFERENCE (a key name resolved at runtime from the component's secret source), never a value. No field holds a secret.

func (HTTPClientPort) IsExclusive 

func (h HTTPClientPort) IsExclusive() bool

IsExclusive returns false: an outbound client relationship is shareable (multiple components may poll the same endpoint), unlike a NetworkPort listener.

func (HTTPClientPort) Kind 

func (h HTTPClientPort) Kind() PortKind

Kind returns the canonical port kind.

func (HTTPClientPort) ResourceID 

func (h HTTPClientPort) ResourceID() string

ResourceID returns a unique identifier for the HTTP client dependency. Two ports with the same method and URL pattern identify the same external resource.

type HealthStatus 

type HealthStatus struct {
	Healthy    bool          `json:"healthy"`
	LastCheck  time.Time     `json:"last_check"`
	ErrorCount int           `json:"error_count"`
	LastError  string        `json:"last_error,omitempty"`
	Uptime     time.Duration `json:"uptime"`
	Status     string        `json:"status"`
}

HealthStatus describes the current health state of a component

type Info 

type Info struct {
	Type        string `json:"type"`        // "input", "processor", "output", "storage"
	Protocol    string `json:"protocol"`    // Technical protocol (udp, tcp, mavlink, etc.)
	Domain      string `json:"domain"`      // Business domain (robotics, semantic, network, storage)
	Description string `json:"description"` // Human-readable description
	Version     string `json:"version"`     // Component version
}

Info holds metadata about an available component type

type InteractionPattern 

type InteractionPattern string

InteractionPattern is the canonical communication behavior of a resolved port. 

const (
	PatternTimer      InteractionPattern = "timer"
	PatternNetwork    InteractionPattern = "network"
	PatternHTTPClient InteractionPattern = "http-client"
	PatternStream     InteractionPattern = "stream"
	PatternRequest    InteractionPattern = "request"
	PatternWatch      InteractionPattern = "watch"
	PatternRead       InteractionPattern = "read"
	PatternStore      InteractionPattern = "store"
)

Canonical interaction patterns projected from resolved ports.

type InterfaceContract 

type InterfaceContract struct {
	Type       string   `json:"type"`
	Version    string   `json:"version,omitempty"`
	Compatible []string `json:"compatible,omitempty"`
}

InterfaceContract defines the payload contract carried across a port.

type JetStreamPort 

type JetStreamPort struct {
	// Stream configuration (for outputs)
	StreamName      string   `json:"stream_name"`              // e.g., "ENTITY_EVENTS"
	Subjects        []string `json:"subjects"`                 // e.g., ["events.graph.entity.>"]
	Storage         string   `json:"storage,omitempty"`        // "file" or "memory" when declared
	RetentionPolicy string   `json:"retention,omitempty"`      // "limits", "interest", or "work_queue" when declared
	RetentionDays   int      `json:"retention_days,omitempty"` // Declared message retention in days
	MaxSizeGB       int      `json:"max_size_gb,omitempty"`    // Declared maximum stream size in GiB
	Replicas        int      `json:"replicas,omitempty"`       // Declared replica count

	// Consumer configuration (for inputs)
	ConsumerName  string `json:"consumer_name,omitempty"`  // Durable consumer name
	DeliverPolicy string `json:"deliver_policy,omitempty"` // Declared delivery policy
	AckPolicy     string `json:"ack_policy,omitempty"`     // Declared acknowledgement policy
	MaxDeliver    int    `json:"max_deliver,omitempty"`    // Declared maximum redelivery attempts
	// AckWait is the duration the JetStream server waits for an ack before
	// redelivering. Strings are parsed via time.ParseDuration ("90s",
	// "2m", "5m"). Empty falls through to a per-component default. The
	// component-level default is a starting point; this port-level field
	// lets operators tune ack_wait for long-running consumers (LLM model
	// calls, slow tool execution) without forking the component. Per
	// docs/operations/14-timeout-chain.md: ack_wait must comfortably
	// exceed the longest legitimate per-task wallclock budget so that
	// healthy long-tail work isn't reaped before it can ack.
	AckWait string `json:"ack_wait,omitempty"`
	// HeartbeatInterval is the cadence at which the consumer goroutine
	// fires msg.InProgress() to reset the ack clock. Strings are parsed
	// via time.ParseDuration ("60s", "90s"). Empty falls through to a
	// per-component default. Should be sized comfortably below ack_wait
	// (typical 1.5x margin) so a single missed heartbeat doesn't trigger
	// redelivery. Only honored on consumers that wrap their handler in
	// natsclient.ConsumeWithHeartbeat — pure-ack consumers ignore it.
	HeartbeatInterval string `json:"heartbeat_interval,omitempty"`
	// MaxAckPending caps the number of delivered-but-unacked messages the
	// server keeps in flight for this consumer — the consumer-side backpressure
	// lever. Empty/0 leaves policy to NATS, which may inherit a stream limit,
	// apply its default, or cap it under server/account policy; -1 is unlimited
	// outstanding acknowledgements. gh#480: there was previously no
	// config path to this at all, so operators could not tune ingest backpressure.
	MaxAckPending int `json:"max_ack_pending,omitempty"`

	// Interface contract
	Interface *InterfaceContract `json:"interface,omitempty"`
}

JetStreamPort - NATS JetStream for durable, at-least-once messaging

func (JetStreamPort) IsExclusive 

func (j JetStreamPort) IsExclusive() bool

IsExclusive returns false as JetStream manages consumer coordination

func (JetStreamPort) Kind 

func (j JetStreamPort) Kind() PortKind

Kind returns the canonical port kind.

func (JetStreamPort) ResourceID 

func (j JetStreamPort) ResourceID() string

ResourceID returns unique identifier for JetStream ports

type KVReadPort 

type KVReadPort struct {
	Bucket    string             `json:"bucket"`
	Interface *InterfaceContract `json:"interface,omitempty"`
}

KVReadPort declares exact or list access to current values in one KV bucket. It is metadata only: acquisition and missing-value policy remain component-owned.

func (KVReadPort) IsExclusive 

func (k KVReadPort) IsExclusive() bool

IsExclusive reports that concurrent readers may share a bucket.

func (KVReadPort) Kind 

func (k KVReadPort) Kind() PortKind

Kind returns the canonical port kind.

func (KVReadPort) ResourceID 

func (k KVReadPort) ResourceID() string

ResourceID returns the canonical KV read resource identity.

type KVWatchPort 

type KVWatchPort struct {
	Bucket    string             `json:"bucket"`            // e.g., "ENTITY_STATES"
	Keys      []string           `json:"keys,omitempty"`    // Keys to watch, empty = all
	History   bool               `json:"history,omitempty"` // Include historical values
	Interface *InterfaceContract `json:"interface,omitempty"`
}

KVWatchPort - NATS KV Watch for state observation

func (KVWatchPort) IsExclusive 

func (k KVWatchPort) IsExclusive() bool

IsExclusive returns false as multiple watchers are allowed

func (KVWatchPort) Kind 

func (k KVWatchPort) Kind() PortKind

Kind returns the canonical port kind.

func (KVWatchPort) ResourceID 

func (k KVWatchPort) ResourceID() string

ResourceID returns unique identifier for KV watch ports

type KVWritePort 

type KVWritePort struct {
	Bucket    string             `json:"bucket"`              // e.g., "ENTITY_STATES"
	Interface *InterfaceContract `json:"interface,omitempty"` // Data type contract
}

KVWritePort - NATS KV Write for state persistence

func (KVWritePort) IsExclusive 

func (k KVWritePort) IsExclusive() bool

IsExclusive returns false as multiple writers are allowed (with CAS handling)

func (KVWritePort) Kind 

func (k KVWritePort) Kind() PortKind

Kind returns the canonical port kind.

func (KVWritePort) ResourceID 

func (k KVWritePort) ResourceID() string

ResourceID returns unique identifier for KV write ports

type LifecycleComponent 

type LifecycleComponent interface {
	Discoverable
	Initialize() error
	Start(ctx context.Context) error
	Stop(ctx context.Context) error
}

LifecycleComponent defines the portable lifecycle contract for components. Initialize performs setup without runtime authority. Start accepts the context that owns continuing work. Stop uses its exact caller context to bound the component's terminal admission fence, cancellation, join, and cleanup; the resource owner determines that sequence's protocol-specific ordering. Exact resource ordering preserves admitted callback authority where native drain requires it; no universal cancel-before-drain order is implied.

During controlled shutdown, callers keep the accepted Start context live until a separately bounded Stop drains, joins, and finalizes the owner and returns nil. Ending the Start context first is abort cancellation. Stop then makes bounded best-effort progress under its exact caller authority and may return accurate native cleanup or deadline errors. Abort cleanup does not invent replacement authority, detach cleanup, or promise a second rejoin; if the bound wins, the portable contract makes no complete-join or leak-freedom claim.

Nil Start and Stop contexts are rejected before action. A completed repeated Stop is a no-op. Concurrent lifecycle calls, result replay, later rejoin after a Stop bound wins, reinitialization, and same-instance restart are not portable guarantees.

func AsLifecycleComponent 

func AsLifecycleComponent(comp Discoverable) (LifecycleComponent, bool)

AsLifecycleComponent safely casts a component to LifecycleComponent

type LifecycleFactory 

type LifecycleFactory func() LifecycleComponent

LifecycleFactory creates a new instance of a LifecycleComponent for testing

type LogEntry 

type LogEntry struct {
	Timestamp string   `json:"timestamp"` // RFC3339 format
	Level     LogLevel `json:"level"`
	Component string   `json:"component"`
	FlowID    string   `json:"flow_id"`
	Message   string   `json:"message"`
	Stack     string   `json:"stack,omitempty"` // Stack trace for errors
}

LogEntry represents a structured log entry that can be published to NATS and consumed by the Flow Builder SSE endpoint.

type LogLevel 

type LogLevel string

LogLevel represents the severity level of a log entry 

const (
	// LogLevelDebug represents debug-level logs
	LogLevelDebug LogLevel = "DEBUG"
	// LogLevelInfo represents informational logs
	LogLevelInfo LogLevel = "INFO"
	// LogLevelWarn represents warning logs
	LogLevelWarn LogLevel = "WARN"
	// LogLevelError represents error logs
	LogLevelError LogLevel = "ERROR"
)

type ManagedComponent 

type ManagedComponent struct {
	// Component is the actual component instance
	Component Discoverable

	// State tracks the current lifecycle state
	State State

	// Config is the effective component configuration this instance was
	// created from. The ComponentManager retains it so a per-component runtime
	// config update can restart the component only when the effective config
	// actually changed — a no-op update is skipped instead of stop/start-cycling
	// a healthy running component (gh#520).
	Config types.ComponentConfig

	// StartOrder tracks the order components were started for reverse shutdown
	StartOrder int

	// LastError tracks the last error that occurred during lifecycle operations
	LastError error
}

ManagedComponent tracks a component and its lifecycle state This is used by ComponentManager to properly manage component lifecycle

type Metadata 

type Metadata struct {
	Name        string `json:"name"`
	Type        string `json:"type"` // "input", "processor", "output", "storage"
	Description string `json:"description"`
	Version     string `json:"version"`
}

Metadata describes what a component is

type NATSPort 

type NATSPort struct {
	Subject   string             `json:"subject"`
	Queue     string             `json:"queue,omitempty"`
	Interface *InterfaceContract `json:"interface,omitempty"`
}

NATSPort - NATS pub/sub

func (NATSPort) IsExclusive 

func (n NATSPort) IsExclusive() bool

IsExclusive returns false as multiple components can subscribe

func (NATSPort) Kind 

func (n NATSPort) Kind() PortKind

Kind returns the canonical port kind.

func (NATSPort) ResourceID 

func (n NATSPort) ResourceID() string

ResourceID returns unique identifier for NATS ports

type NATSRequestPort 

type NATSRequestPort struct {
	Subject   string             `json:"subject"`
	Timeout   string             `json:"timeout,omitempty"` // Duration string e.g. "1s", "500ms"
	Retries   int                `json:"retries,omitempty"`
	Interface *InterfaceContract `json:"interface,omitempty"`
}

NATSRequestPort - NATS Request/Response pattern for synchronous operations

func (NATSRequestPort) IsExclusive 

func (n NATSRequestPort) IsExclusive() bool

IsExclusive returns false as multiple components can handle requests

func (NATSRequestPort) Kind 

func (n NATSRequestPort) Kind() PortKind

Kind returns the canonical port kind.

func (NATSRequestPort) ResourceID 

func (n NATSRequestPort) ResourceID() string

ResourceID returns unique identifier for NATS request ports

type NetworkFacts 

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

NetworkFacts is the immutable listener binding portion of PortFacts.

func (NetworkFacts) Host 

func (f NetworkFacts) Host() string

Host returns the configured bind host.

func (NetworkFacts) Port 

func (f NetworkFacts) Port() int

Port returns the configured bind port.

func (NetworkFacts) Protocol 

func (f NetworkFacts) Protocol() string

Protocol returns the network protocol.

type NetworkPort 

type NetworkPort struct {
	Protocol string `json:"protocol"` // "tcp", "udp"
	Host     string `json:"host"`     // "0.0.0.0", "localhost"
	Port     int    `json:"port"`     // 14550, 8080
}

NetworkPort - TCP/UDP network bindings

func (NetworkPort) IsExclusive 

func (n NetworkPort) IsExclusive() bool

IsExclusive returns true as network ports are exclusive

func (NetworkPort) Kind 

func (n NetworkPort) Kind() PortKind

Kind returns the canonical port kind.

func (NetworkPort) ResourceID 

func (n NetworkPort) ResourceID() string

ResourceID returns unique identifier for network ports

type PlatformMeta 

type PlatformMeta = types.PlatformMeta

PlatformMeta provides platform identity to components. Type alias to avoid import cycles while maintaining compatibility.

type Port 

type Port struct {
	Name        string    `json:"name"`
	Direction   Direction `json:"direction"`
	Required    bool      `json:"required,omitempty"`
	Description string    `json:"description,omitempty"`
	Config      Portable  `json:"config"`
}

Port is a resolved component I/O declaration.

func (Port) Facts 

func (p Port) Facts() (PortFacts, error)

Facts revalidates the current Port value and returns its immutable semantic projection.

func (Port) MarshalJSON 

func (p Port) MarshalJSON() ([]byte, error)

MarshalJSON encodes Port with the same config.kind envelope as PortDefinition.

func (*Port) UnmarshalJSON 

func (p *Port) UnmarshalJSON(data []byte) error

UnmarshalJSON strictly decodes and resolves a runtime port.

type PortConfig 

type PortConfig struct {
	Inputs  []PortDefinition `json:"inputs,omitempty"`
	Outputs []PortDefinition `json:"outputs,omitempty"`
}

PortConfig groups semantic declarations by data-flow direction.

func MergePortConfig 

func MergePortConfig(defaults, overrides PortConfig) (PortConfig, error)

MergePortConfig applies complete named replacements to default declarations. Inputs and outputs are independent, ordering is stable, and returned data is cloned.

func (*PortConfig) UnmarshalJSON 

func (p *PortConfig) UnmarshalJSON(data []byte) error

UnmarshalJSON rejects unknown top-level lanes such as the retired kv_write lane.

type PortDefinition 

type PortDefinition struct {
	Name        string   `json:"name" schema:"readonly,type:string,description:Port identifier"`
	Required    bool     `json:"required,omitempty" schema:"readonly,type:bool,description:Whether port connection is required"`
	Description string   `json:"description,omitempty" schema:"readonly,type:string,description:Human-readable port description"`
	Config      Portable `json:"config" schema:"editable,type:object,description:Typed semantic port configuration"`
}

PortDefinition is a configuration-shaped semantic port declaration.

func (PortDefinition) MarshalJSON 

func (p PortDefinition) MarshalJSON() ([]byte, error)

MarshalJSON writes the canonical common port envelope.

func (PortDefinition) Resolve 

func (d PortDefinition) Resolve(direction Direction) (Port, error)

Resolve validates and resolves a configuration declaration for one direction.

func (*PortDefinition) UnmarshalJSON 

func (p *PortDefinition) UnmarshalJSON(data []byte) error

UnmarshalJSON strictly decodes the canonical common port envelope.

type PortFacts 

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

PortFacts is an immutable semantic projection of one resolved Port.

func (PortFacts) ConnectionIDs 

func (f PortFacts) ConnectionIDs() []string

ConnectionIDs returns defensive copies of the identifiers used for flow matching.

func (PortFacts) InteractionPattern 

func (f PortFacts) InteractionPattern() InteractionPattern

InteractionPattern returns the canonical communication behavior.

func (PortFacts) Interface 

func (f PortFacts) Interface() (InterfaceContract, bool)

Interface returns a defensive copy of the semantic interface contract.

func (PortFacts) IsExclusive 

func (f PortFacts) IsExclusive() bool

IsExclusive reports whether two ports may claim the same resource.

func (PortFacts) KVReadBucket 

func (f PortFacts) KVReadBucket() (string, bool)

KVReadBucket returns the declared KV bucket for a kv-read port, so a component OBSERVES the bucket it was bound to instead of predicting the name with a constant. Reports false for every other port kind: the second return is the whole answer to "is this a KV read port", and a caller must never treat an empty bucket as a default.

This is the projection route for the bucket; concrete port config types are only interpreted by the canonical projection owners (this file and port_codec.go).

func (PortFacts) Kind 

func (f PortFacts) Kind() PortKind

Kind returns the canonical port kind.

func (PortFacts) NATSSubjects 

func (f PortFacts) NATSSubjects() []string

NATSSubjects returns defensive copies of declared NATS subject families.

func (PortFacts) Network 

func (f PortFacts) Network() (NetworkFacts, bool)

Network returns the immutable network binding when the port declares one.

func (PortFacts) ResourceID 

func (f PortFacts) ResourceID() string

ResourceID returns the canonical resource identity.

func (PortFacts) StoreReadBucket 

func (f PortFacts) StoreReadBucket() (string, bool)

StoreReadBucket returns the configured content-store bucket for a store-read port.

func (PortFacts) Stream 

func (f PortFacts) Stream() (StreamFacts, bool)

Stream returns the immutable JetStream projection when the port declares one.

type PortFieldInfo 

type PortFieldInfo struct {
	Type                 string                   `json:"type"`
	Editable             bool                     `json:"editable"`
	Minimum              *int                     `json:"minimum,omitempty"`
	Enum                 []string                 `json:"enum,omitempty"`
	Required             []string                 `json:"required,omitempty"`
	AnyRequired          [][]string               `json:"anyRequired,omitempty"`
	RequiredByDirection  map[Direction][]string   `json:"requiredByDirection,omitempty"`
	Directions           []Direction              `json:"directions,omitempty"`
	Properties           map[string]PortFieldInfo `json:"properties,omitempty"`
	Variants             map[string]PortFieldInfo `json:"variants,omitempty"`
	Items                *PortFieldInfo           `json:"items,omitempty"`
	AdditionalProperties *bool                    `json:"additionalProperties,omitempty"`
	// contains filtered or unexported fields
}

PortFieldInfo describes metadata for PortDefinition fields

func (PortFieldInfo) ZeroIsOmitted 

func (p PortFieldInfo) ZeroIsOmitted() bool

ZeroIsOmitted reports whether numeric zero has omission semantics for this field. It is intentionally absent from discovery JSON: editors use the canonical Directions metadata, while schema projection uses this predicate to preserve an explicitly supplied zero as semantic absence.

type PortKind 

type PortKind string

PortKind is the closed vocabulary of semantic component port kinds. 

const (
	PortKindTimer        PortKind = "timer"
	PortKindNetwork      PortKind = "network"
	PortKindFile         PortKind = "file"
	PortKindHTTPClient   PortKind = "http-client"
	PortKindNATS         PortKind = "nats"
	PortKindNATSRequest  PortKind = "nats-request"
	PortKindJetStream    PortKind = "jetstream"
	PortKindKVWatch      PortKind = "kv-watch"
	PortKindKVRead       PortKind = "kv-read"
	PortKindKVWrite      PortKind = "kv-write"
	PortKindStoreRead    PortKind = "store-read"
	PortKindStoreProvide PortKind = "store-provide"
)

Canonical port kinds. Alternate spellings and custom kinds are not accepted.

type Portable 

type Portable interface {
	ResourceID() string
	IsExclusive() bool
	Kind() PortKind
}

Portable is the common semantic contract implemented by every port config.

type ProcessorMetrics 

type ProcessorMetrics struct {
	// EventsProcessed counts total events processed, labeled by operation type
	EventsProcessed *prometheus.CounterVec

	// EventsErrors counts total processing errors, labeled by error type
	EventsErrors *prometheus.CounterVec

	// KVOperations counts KV bucket operations, labeled by operation (get/put/delete/watch)
	KVOperations *prometheus.CounterVec

	// ProcessingDuration measures processing latency in seconds
	ProcessingDuration prometheus.Histogram
	// contains filtered or unexported fields
}

ProcessorMetrics provides standard Prometheus metrics for processor components. Each processor component should create its own instance with a unique subsystem name.

func NewProcessorMetrics 

func NewProcessorMetrics(registry *metric.MetricsRegistry, subsystem string) *ProcessorMetrics

NewProcessorMetrics creates and registers processor metrics with the given subsystem name. The subsystem name should be the component name with underscores (e.g., "graph_ingest"). If registry is nil, metrics are created but not registered (useful for testing).

func (*ProcessorMetrics) ObserveDuration 

func (m *ProcessorMetrics) ObserveDuration(seconds float64)

ObserveDuration records a processing duration

func (*ProcessorMetrics) RecordError 

func (m *ProcessorMetrics) RecordError(errorType string)

RecordError increments the error counter for the given error type

func (*ProcessorMetrics) RecordEvent 

func (m *ProcessorMetrics) RecordEvent(operation string)

RecordEvent increments the events processed counter for the given operation

func (*ProcessorMetrics) RecordKVOperation 

func (m *ProcessorMetrics) RecordKVOperation(operation string)

RecordKVOperation increments the KV operations counter

type PropertySchema 

type PropertySchema struct {
	Type        string                    `json:"type"` // "string", "int", "bool", "float", "enum", "array", "object", "ports", "cache"
	Description string                    `json:"description"`
	Default     any                       `json:"default,omitempty"`
	Enum        []string                  `json:"enum,omitempty"`        // Valid string values
	Minimum     *int                      `json:"minimum,omitempty"`     // For numeric types
	Maximum     *int                      `json:"maximum,omitempty"`     // For numeric types
	MinLength   *int                      `json:"minLength,omitempty"`   // Minimum string length
	MaxLength   *int                      `json:"maxLength,omitempty"`   // Maximum string length
	Pattern     string                    `json:"pattern,omitempty"`     // Regular expression for string values
	Category    string                    `json:"category,omitempty"`    // "basic" or "advanced" for UI organization
	PortFields  map[string]PortFieldInfo  `json:"portFields,omitempty"`  // Metadata for port fields (when type is "ports")
	CacheFields map[string]CacheFieldInfo `json:"cacheFields,omitempty"` // Metadata for cache fields (when type is "cache")
	Properties  map[string]PropertySchema `json:"properties,omitempty"`  // Nested properties for object types
	// AdditionalProperties controls whether object fields outside Properties
	// are accepted. nil preserves JSON Schema's permissive default.
	AdditionalProperties *bool           `json:"additionalProperties,omitempty"`
	Required             []string        `json:"required,omitempty"` // Required nested fields for object types
	Items                *PropertySchema `json:"items,omitempty"`    // Item schema for array types
}

PropertySchema describes a single configuration property

type Registerable 

type Registerable interface {
	Registration() Registration
}

Registerable allows components to self-describe for registry registration

type Registration 

type Registration struct {
	Name         string       `json:"name"`         // Factory name (e.g., "udp-input")
	Type         string       `json:"type"`         // Component type (input/processor/output/storage)
	Protocol     string       `json:"protocol"`     // Technical protocol (udp, mavlink, websocket, etc.)
	Domain       string       `json:"domain"`       // Business domain (robotics, semantic, network, storage)
	Description  string       `json:"description"`  // Human-readable description
	Version      string       `json:"version"`      // Component version
	Schema       ConfigSchema `json:"schema"`       // Schema as static metadata (Feature 011)
	Factory      Factory      `json:"-"`            // Factory function (not serializable)
	Dependencies []string     `json:"dependencies"` // Boot dependencies such as DepModelRegistry
}

Registration holds factory and metadata for a component type

type RegistrationConfig 

type RegistrationConfig struct {
	Name         string       // Component name (e.g., "udp", "websocket", "graph-processor")
	Factory      Factory      // Factory function to create component instances
	Schema       ConfigSchema // Configuration schema for validation and discovery
	Type         string       // Component type: "input", "processor", "output", "storage"
	Protocol     string       // Technical protocol (udp, tcp, websocket, file, etc.)
	Domain       string       // Business domain (network, storage, processing, robotics, semantic)
	Description  string       // Human-readable description of the component
	Version      string       // Component version (semver recommended)
	Dependencies []string     // Runtime deps declared via constants like DepModelRegistry
}

RegistrationConfig provides a clean API for component registration. This config struct replaces the previous 7-8 parameter function signatures. It maps 1:1 to Registration struct fields for simplicity.

type Registry 

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

Registry manages component factories and immutable admitted declarations. Runtime component handles remain private to ComponentManager.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry creates a new empty component registry

func (*Registry) CreateComponent 

func (r *Registry) CreateComponent(
	_ componentadmission.Access,
	instanceName string,
	config types.ComponentConfig,
	deps Dependencies,
	prepare func(Discoverable) error,
) (Discoverable, error)

CreateComponent is the sole framework-internal boot-admission seam. prepare must finish all fallible owner-local setup before the Registry publishes the immutable declaration. Downstream adopters cannot obtain the internal access token and configure components through ComponentManager.

func (*Registry) GetComponentSchema 

func (r *Registry) GetComponentSchema(name string) (ConfigSchema, error)

GetComponentSchema retrieves a component's schema directly from Registration metadata This method retrieves schemas without component instantiation (Feature 011 - Option 1) Schema is stored as static metadata during registration, avoiding dependency validation issues

func (*Registry) GetFactory 

func (r *Registry) GetFactory(name string) (Factory, bool)

GetFactory returns a specific factory by name Unlike ListFactories, this returns the actual Factory function for creating components

func (*Registry) ListAvailable 

func (r *Registry) ListAvailable() map[string]Info

ListAvailable returns information about all available component types This provides metadata about what types of components can be created.

func (*Registry) ListComponentTypes 

func (r *Registry) ListComponentTypes() []string

ListComponentTypes returns all registered component factory type names This returns factory names (e.g., "udp-input", "websocket-output") not instance names

func (*Registry) ListFactories 

func (r *Registry) ListFactories() map[string]*Registration

ListFactories returns all registered component factories This provides information about what types of components can be created.

func (*Registry) RegisterFactory 

func (r *Registry) RegisterFactory(name string, registration *Registration) error

RegisterFactory registers a component factory with the given name Returns an error if a factory with the same name is already registered.

func (*Registry) RegisterWithConfig 

func (r *Registry) RegisterWithConfig(config RegistrationConfig) error

RegisterWithConfig registers a component using a configuration struct. This is the recommended registration method that replaces the multi-parameter functions.

Example usage: 

registry.RegisterWithConfig(component.RegistrationConfig{
    Name:        "udp",
    Factory:     CreateUDPInput,
    Schema:      udpSchema,
    Type:        "input",
    Protocol:    "udp",
    Domain:      "network",
    Description: "UDP input component for receiving network data",
    Version:     "1.0.0",
})

func (*Registry) SealComposition 

func (r *Registry) SealComposition(_ componentadmission.Access)

SealComposition closes boot admission for the current process. The internal access token prevents downstream callers from treating the seal as a public runtime-composition control.

func (*Registry) Snapshot 

func (r *Registry) Snapshot(
	_ componentadmission.Access, instanceName string,
) (

	declarationSnapshot,
	bool,
)

Snapshot returns one defensive admitted-declaration view to a root-internal framework consumer.

func (*Registry) Snapshots 

func (r *Registry) Snapshots(
	_ componentadmission.Access,
) []declarationSnapshot

Snapshots returns a deterministic defensive complete admission set to a root-internal framework consumer.

type SchemaDirectives 

type SchemaDirectives struct {
	// core (required)
	Type        string // REQUIRED - field type
	Description string // REQUIRED (warning if missing)

	// UI Organization
	Category string // "basic" or "advanced"
	ReadOnly bool   // For PortDefinition fields
	Editable bool   // For PortDefinition fields
	Hidden   bool   // Hide from UI

	// Constraints
	Default  any      // Type-specific default value (stored as string, converted during schema generation)
	Required bool     // Field must be provided
	Min      *int     // Numeric minimum
	Max      *int     // Numeric maximum
	Enum     []string // Valid enum values

	// Future extensions (stored but not used yet)
	Help        string
	Placeholder string
	Pattern     string
	Format      string
}

SchemaDirectives represents parsed schema tag directives

func ParseSchemaTag 

func ParseSchemaTag(tag string) (SchemaDirectives, error)

ParseSchemaTag parses a schema struct tag into directives.

Tag Syntax:

  • Directives are comma-separated
  • Key-value pairs use colon: "key:value"
  • Boolean flags have no colon: "readonly", "required"
  • Enum values are pipe-separated: "enum:val1|val2|val3"
  • Whitespace is trimmed from all values

Required Directives:

  • type: Field data type (string, int, bool, float, enum, array, object, ports)

Recommended Directives:

  • description: Human-readable field description (used for UI and documentation)

Optional Directives:

  • category: UI organization (basic, advanced)
  • default: Default value (converted to appropriate type)
  • min/max: Numeric constraints
  • enum: Valid values for enum types (pipe-separated)
  • readonly: Field is read-only (boolean flag)
  • editable: Field is user-editable (boolean flag)
  • hidden: Field is hidden from UI (boolean flag)
  • required: Field must be provided (boolean flag)

Example Tags: 

schema:"type:string,description:Component name,category:basic"
schema:"type:int,description:Port,min:1,max:65535,default:8080"
schema:"type:enum,description:Level,enum:debug|info|warn,default:info"
schema:"required,type:string,description:API key"
schema:"readonly,type:string,description:System ID"

Returns an error if:

  • Tag is empty
  • Type directive is missing
  • Type value is invalid
  • Directive syntax is malformed
  • Numeric values cannot be parsed

See SCHEMA_TAG_SPEC.md for complete specification.

Example

ExampleParseSchemaTag demonstrates parsing individual schema tags 

package main

import (
	"fmt"

	"github.com/c360studio/semstreams/component"
)

func main() {
	// Parse a simple field tag
	tag := "type:int,description:Port number,min:1,max:65535,default:8080"
	directives, err := component.ParseSchemaTag(tag)
	if err != nil {
		fmt.Printf("Error: %v\n", err)
		return
	}

	fmt.Printf("Type: %s\n", directives.Type)
	fmt.Printf("Description: %s\n", directives.Description)
	fmt.Printf("Min: %d\n", *directives.Min)
	fmt.Printf("Max: %d\n", *directives.Max)
	fmt.Printf("Default: %s\n", directives.Default)

}

Output:
Type: int
Description: Port number
Min: 1
Max: 65535
Default: 8080
Example (Enum)

ExampleParseSchemaTag_enum demonstrates parsing enum tags 

package main

import (
	"fmt"

	"github.com/c360studio/semstreams/component"
)

func main() {
	tag := "type:enum,description:Log level,enum:debug|info|warn|error,default:info"
	directives, _ := component.ParseSchemaTag(tag)

	fmt.Printf("Type: %s\n", directives.Type)
	fmt.Printf("Description: %s\n", directives.Description)
	fmt.Printf("Enum values: %v\n", directives.Enum)
	fmt.Printf("Default: %s\n", directives.Default)

}

Output:
Type: enum
Description: Log level
Enum values: [debug info warn error]
Default: info
Example (Flags)

ExampleParseSchemaTag_flags demonstrates boolean flags 

package main

import (
	"fmt"

	"github.com/c360studio/semstreams/component"
)

func main() {
	tag := "required,readonly,type:string,description:System identifier"
	directives, _ := component.ParseSchemaTag(tag)

	fmt.Printf("Type: %s\n", directives.Type)
	fmt.Printf("Required: %v\n", directives.Required)
	fmt.Printf("ReadOnly: %v\n", directives.ReadOnly)

}

Output:
Type: string
Required: true
ReadOnly: true

type SimpleMockComponent 

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

SimpleMockComponent is a simple component implementation for testing

func (*SimpleMockComponent) ConfigSchema 

func (m *SimpleMockComponent) ConfigSchema() ConfigSchema

ConfigSchema returns the component's configuration schema

func (*SimpleMockComponent) DataFlow 

func (m *SimpleMockComponent) DataFlow() FlowMetrics

DataFlow returns the component's data flow metrics (zeros for mock)

func (*SimpleMockComponent) Health 

func (m *SimpleMockComponent) Health() HealthStatus

Health returns the component's health status (always healthy for mock)

func (*SimpleMockComponent) InputPorts 

func (m *SimpleMockComponent) InputPorts() []Port

InputPorts returns the component's input ports (none for mock)

func (*SimpleMockComponent) Meta 

func (m *SimpleMockComponent) Meta() Metadata

Meta returns the component metadata

func (*SimpleMockComponent) OutputPorts 

func (m *SimpleMockComponent) OutputPorts() []Port

OutputPorts returns the component's output ports (none for mock)

type State 

type State int

State represents the current lifecycle state of a component 

const (
	// StateCreated indicates component was created but not initialized
	StateCreated State = iota
	// StateInitialized indicates component was initialized but not started
	StateInitialized
	// StateStarted indicates component is running
	StateStarted
	// StateStopped indicates component was stopped
	StateStopped
	// StateFailed indicates component failed during lifecycle operation
	StateFailed
)

func (State) String 

func (cs State) String() string

String returns a string representation of the component state

type StoreProvidePort 

type StoreProvidePort struct {
	Instance string `json:"instance"` // StorageInstance name this component owns
}

StoreProvidePort declares that a storage component OWNS (provides) a store instance, addressable by its StorageInstance name (ADR-063). It is the flowgraph marker for store ownership and complements the StoreProvider interface the ComponentManager reads to populate the shared StoreRegistry.

Non-exclusive by design. Duplicate live Store ownership is owner-local runtime state and remains enforced by storeregistry.Register; it is not a declaration-derived exclusive-resource claim.

func (StoreProvidePort) IsExclusive 

func (s StoreProvidePort) IsExclusive() bool

IsExclusive returns false — ownership conflicts are caught at registry-population time, not here (see the type doc).

func (StoreProvidePort) Kind 

func (s StoreProvidePort) Kind() PortKind

Kind returns the canonical port kind.

func (StoreProvidePort) ResourceID 

func (s StoreProvidePort) ResourceID() string

ResourceID returns a unique identifier for store provide ports.

type StoreProvider 

type StoreProvider interface {
	ProvidedStores() map[string]storage.StreamableStore
}

StoreProvider is implemented by storage components that own one or more stores addressable by StorageInstance name (ADR-063). For the boot-composed component set, ComponentManager reads this AFTER a component Starts to populate the shared StoreRegistry and clears those entries when the component Stops. A desired-state write does not swap a handle in the running process.

The map is keyed by the StorageInstance name each store STAMPS into refs (store.InstanceName()), so the registry key, the store-provide port token, and the ref's StorageInstance are the same value by construction. Returns nil/empty before Start (no store yet) or for a component that provides no store.

type StoreReadPort 

type StoreReadPort struct {
	Bucket    string             `json:"bucket"`              // Storage bucket name (e.g., "MESSAGES")
	Interface *InterfaceContract `json:"interface,omitempty"` // Optional content type contract
}

StoreReadPort declares streaming read access to a content storage bucket. Components use this to read large content (documents, images, video) from storage backends (NATS ObjectStore, filesystem, etc.) without coupling to a specific backend implementation.

func (StoreReadPort) IsExclusive 

func (s StoreReadPort) IsExclusive() bool

IsExclusive returns false as multiple readers are allowed

func (StoreReadPort) Kind 

func (s StoreReadPort) Kind() PortKind

Kind returns the canonical port kind.

func (StoreReadPort) ResourceID 

func (s StoreReadPort) ResourceID() string

ResourceID returns unique identifier for store read ports

type StreamFacts 

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

StreamFacts is the immutable JetStream-specific portion of PortFacts.

func (StreamFacts) AckPolicy 

func (f StreamFacts) AckPolicy() string

AckPolicy returns the declared consumer acknowledgement policy.

func (StreamFacts) AckWait 

func (f StreamFacts) AckWait() string

AckWait returns the declared acknowledgement timeout.

func (StreamFacts) ConsumerName 

func (f StreamFacts) ConsumerName() string

ConsumerName returns the declared durable consumer name.

func (StreamFacts) DeliverPolicy 

func (f StreamFacts) DeliverPolicy() string

DeliverPolicy returns the declared consumer delivery policy.

func (StreamFacts) HeartbeatInterval 

func (f StreamFacts) HeartbeatInterval() string

HeartbeatInterval returns the declared consumer heartbeat interval.

func (StreamFacts) MaxAckPending 

func (f StreamFacts) MaxAckPending() int

MaxAckPending returns the declared unacknowledged-message ceiling.

func (StreamFacts) MaxDeliver 

func (f StreamFacts) MaxDeliver() int

MaxDeliver returns the declared maximum delivery attempts.

func (StreamFacts) MaxSizeGB 

func (f StreamFacts) MaxSizeGB() int

MaxSizeGB returns the declared stream size limit in GiB.

func (StreamFacts) Name 

func (f StreamFacts) Name() string

Name returns the declared stream name.

func (StreamFacts) Replicas 

func (f StreamFacts) Replicas() int

Replicas returns the declared stream replica count.

func (StreamFacts) RetentionDays 

func (f StreamFacts) RetentionDays() int

RetentionDays returns the declared stream retention duration in days.

func (StreamFacts) RetentionPolicy 

func (f StreamFacts) RetentionPolicy() string

RetentionPolicy returns the declared JetStream retention policy.

func (StreamFacts) Storage 

func (f StreamFacts) Storage() string

Storage returns the declared JetStream storage mode.

func (StreamFacts) Subjects 

func (f StreamFacts) Subjects() []string

Subjects returns a defensive copy of the declared subjects.

type TimerPort 

type TimerPort struct {
	Interval  string             `json:"interval"` // Duration string e.g. "30s", "1m"
	Interface *InterfaceContract `json:"interface,omitempty"`
}

TimerPort represents a periodic timer trigger port

func (TimerPort) IsExclusive 

func (t TimerPort) IsExclusive() bool

IsExclusive returns false as multiple timers can run independently

func (TimerPort) Kind 

func (t TimerPort) Kind() PortKind

Kind returns the canonical port kind.

func (TimerPort) ResourceID 

func (t TimerPort) ResourceID() string

ResourceID returns unique identifier for timer ports

type ToolRegistryReader 

type ToolRegistryReader interface {
	// Execute implementations that may cause external effects MUST use
	// ToolCall.ID as the downstream idempotency key. agentic-tools persists only
	// COMPLETED outcomes, so a crash or transient Create failure after the
	// external effect but before durable completion is inherently ambiguous and
	// may redeliver the same call to Execute.
	Execute(ctx context.Context, call agentic.ToolCall) (agentic.ToolResult, error)
	ListTools() []agentic.ToolDefinition
}

ToolRegistryReader is the dependency-side surface of the agentic- tools executor registry. *agentictools.ExecutorRegistry satisfies it implicitly. Defined here in the component package so all component consumers can refer to it through Dependencies without importing agentic-tools — mirrors how model.RegistryReader is wired.

On a tool miss, Execute returns a wrapped agentic.ErrToolNotFound sentinel. Callers detect via errors.Is — the previous string-match fallback in agentic-tools/component.go was the source of repeated extension friction and is gone with this contract.

type Validatable 

type Validatable interface {
	Validate() error
}

Validatable interface for configs that can self-validate

type ValidationError 

type ValidationError struct {
	Field   string `json:"field"`   // Name of the field that failed validation
	Message string `json:"message"` // Human-readable error message
	Code    string `json:"code"`    // Machine-readable error code (see above)
}

ValidationError represents a validation error for a specific configuration field. It provides structured error information that can be displayed to users and mapped to specific form fields in the UI.

Error codes are standardized across frontend and backend:

  • "required": Field is required but missing
  • "min": Numeric value below minimum threshold
  • "max": Numeric value above maximum threshold
  • "enum": Value not in allowed enum values
  • "type": Value doesn't match expected type (string, int, bool, etc.)
  • "pattern": String doesn't match required pattern (future use)

func ValidateComponentConfig 

func ValidateComponentConfig(
	ctx context.Context,
	logger *slog.Logger,
	registry ConfigComponentRegistry,
	componentType string,
	configJSON json.RawMessage,
) []ValidationError

ValidateComponentConfig validates a component configuration from KV format. Convenience wrapper that handles JSON unmarshaling.

func ValidateConfig 

func ValidateConfig(config map[string]any, schema ConfigSchema) []ValidationError

ValidateConfig validates a configuration map against a ConfigSchema. It checks required fields, type constraints, min/max bounds, and enum values.

The validation is lenient - unknown fields are allowed to support backward compatibility and future schema evolution. Only explicitly defined properties are validated against their schema constraints.

Returns a slice of ValidationError containing all validation failures found. An empty slice indicates the configuration is valid.

Example usage: 

schema := component.ConfigSchema{
    Properties: map[string]component.PropertySchema{
        "port": {
            Type:     "int",
            Minimum:  ptrInt(1),
            Maximum:  ptrInt(65535),
            Category: "basic",
        },
    },
    Required: []string{"port"},
}

config := map[string]any{"port": 99999}
errors := component.ValidateConfig(config, schema)
if len(errors) > 0 {
    // Handle validation errors
    fmt.Printf("Validation failed: %s\n", errors[0].Message)
}

func ValidateWithSchema 

func ValidateWithSchema(
	ctx context.Context,
	logger *slog.Logger,
	registry ConfigComponentRegistry,
	componentType string,
	config map[string]any,
) []ValidationError

ValidateWithSchema validates component configuration against its schema. Free function — takes logger explicitly so it doesn't need to live on config.Manager and create the config→component→agentic→ message→config cycle.

Returns validation errors if the config doesn't meet schema requirements. Should be called before persisting configuration to KV.

Source Files

View all Source files

Directories

Path Synopsis
Package flowgraph provides flow graph analysis and validation for component connections.
 Package flowgraph provides flow graph analysis and validation for component connections.

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.