callbacks

package
v0.3.0-alpha.1 Latest Latest
Warning

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

 Go to latest Published: May 16, 2026 License: MIT Imports: 17 Imported by: 0

Documentation

Overview

Package callbacks provides a callback/hook system for workforce lifecycle events. This enables observability integrations (OpenTelemetry, logging, metrics) without coupling core logic to specific tools.

Index

Constants

View Source 
const (
	// InstrumentationName is the OpenTelemetry instrumentation name
	InstrumentationName = "github.com/cloud-shuttle/drover"
	// TracerName is the tracer name for Drover
	TracerName = "drover-tracer"

	// Span names
	SpanTaskCreated   = "task.created"
	SpanTaskAssigned  = "task.assigned"
	SpanTaskStarted   = "task.started"
	SpanTaskCompleted = "task.completed"
	SpanTaskFailed    = "task.failed"
	SpanTaskBlocked   = "task.blocked"
	SpanTaskUnblocked = "task.unblocked"
	SpanTaskRecovered = "task.recovered"
	SpanWorkerStarted = "worker.started"
	SpanWorkerStopped = "worker.stopped"
	SpanWorkerStalled = "worker.stalled"
)

Variables

View Source 
var GlobalRegistry = NewRegistry()

GlobalRegistry is the default global callback registry

Functions

func ContextWithSpan 

func ContextWithSpan(ctx context.Context, span trace.Span) context.Context

ContextWithSpan returns a context with the given span

func GetSpanFromContext 

func GetSpanFromContext(ctx context.Context) trace.Span

GetSpanFromContext extracts the current span from context if available

Types

type Callback 

type Callback interface {

	// OnTaskCreated is called when a new task is created
	OnTaskCreated(ctx *TaskEventContext) error

	// OnTaskAssigned is called when a task is claimed by a worker
	OnTaskAssigned(ctx *TaskEventContext) error

	// OnTaskStarted is called when a task begins execution
	OnTaskStarted(ctx *TaskEventContext) error

	// OnTaskCompleted is called when a task completes successfully
	OnTaskCompleted(ctx *TaskEventContext) error

	// OnTaskFailed is called when a task fails
	OnTaskFailed(ctx *TaskEventContext) error

	// OnTaskBlocked is called when a task is blocked by dependencies
	OnTaskBlocked(ctx *TaskEventContext) error

	// OnTaskUnblocked is called when a blocked task becomes ready
	OnTaskUnblocked(ctx *TaskEventContext) error

	// OnTaskRecovered is called when a failed task is recovered
	OnTaskRecovered(ctx *RecoveryEventContext) error

	// OnWorkerStarted is called when a worker starts
	OnWorkerStarted(ctx *WorkerEventContext) error

	// OnWorkerStopped is called when a worker stops
	OnWorkerStopped(ctx *WorkerEventContext) error

	// OnWorkerStalled is called when a worker is unable to claim tasks
	OnWorkerStalled(ctx *WorkerEventContext) error
}

Callback is the interface that all callback implementations must implement. Each method corresponds to a specific lifecycle event type. Callbacks should be idempotent where possible and should not panic.

Performance requirements: - Each callback invocation should complete in < 1ms - Callbacks are invoked synchronously by default - For long-running operations, callbacks should spawn goroutines

func ChainMiddleware 

func ChainMiddleware(callback Callback, middlewares ...Middleware) Callback

ChainMiddleware chains multiple middleware together

type CallbackFunc 

type CallbackFunc struct {
	OnTaskCreatedFunc   func(*TaskEventContext) error
	OnTaskAssignedFunc  func(*TaskEventContext) error
	OnTaskStartedFunc   func(*TaskEventContext) error
	OnTaskCompletedFunc func(*TaskEventContext) error
	OnTaskFailedFunc    func(*TaskEventContext) error
	OnTaskBlockedFunc   func(*TaskEventContext) error
	OnTaskUnblockedFunc func(*TaskEventContext) error
	OnTaskRecoveredFunc func(*RecoveryEventContext) error
	OnWorkerStartedFunc func(*WorkerEventContext) error
	OnWorkerStoppedFunc func(*WorkerEventContext) error
	OnWorkerStalledFunc func(*WorkerEventContext) error
}

CallbackFunc is a convenience type for simple function-based callbacks. Only implement the events you care about; unimplemented methods return nil.

func (*CallbackFunc) OnTaskAssigned 

func (c *CallbackFunc) OnTaskAssigned(ctx *TaskEventContext) error

OnTaskAssigned implements Callback

func (*CallbackFunc) OnTaskBlocked 

func (c *CallbackFunc) OnTaskBlocked(ctx *TaskEventContext) error

OnTaskBlocked implements Callback

func (*CallbackFunc) OnTaskCompleted 

func (c *CallbackFunc) OnTaskCompleted(ctx *TaskEventContext) error

OnTaskCompleted implements Callback

func (*CallbackFunc) OnTaskCreated 

func (c *CallbackFunc) OnTaskCreated(ctx *TaskEventContext) error

OnTaskCreated implements Callback

func (*CallbackFunc) OnTaskFailed 

func (c *CallbackFunc) OnTaskFailed(ctx *TaskEventContext) error

OnTaskFailed implements Callback

func (*CallbackFunc) OnTaskRecovered 

func (c *CallbackFunc) OnTaskRecovered(ctx *RecoveryEventContext) error

OnTaskRecovered implements Callback

func (*CallbackFunc) OnTaskStarted 

func (c *CallbackFunc) OnTaskStarted(ctx *TaskEventContext) error

OnTaskStarted implements Callback

func (*CallbackFunc) OnTaskUnblocked 

func (c *CallbackFunc) OnTaskUnblocked(ctx *TaskEventContext) error

OnTaskUnblocked implements Callback

func (*CallbackFunc) OnWorkerStalled 

func (c *CallbackFunc) OnWorkerStalled(ctx *WorkerEventContext) error

OnWorkerStalled implements Callback

func (*CallbackFunc) OnWorkerStarted 

func (c *CallbackFunc) OnWorkerStarted(ctx *WorkerEventContext) error

OnWorkerStarted implements Callback

func (*CallbackFunc) OnWorkerStopped 

func (c *CallbackFunc) OnWorkerStopped(ctx *WorkerEventContext) error

OnWorkerStopped implements Callback

type ComponentHealth 

type ComponentHealth struct {
	Name      string                 `json:"name"`
	Status    HealthStatus           `json:"status"`
	Message   string                 `json:"message,omitempty"`
	CheckedAt time.Time              `json:"checked_at"`
	Metadata  map[string]interface{} `json:"metadata,omitempty"`
}

ComponentHealth represents the health of a single component

type CompositeCallback 

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

CompositeCallback executes multiple callbacks in sequence

func NewCompositeCallback 

func NewCompositeCallback(callbacks ...Callback) *CompositeCallback

NewCompositeCallback creates a new composite callback

func (*CompositeCallback) Add 

func (cc *CompositeCallback) Add(callback Callback)

Add adds a callback to the composite

func (*CompositeCallback) Count 

func (cc *CompositeCallback) Count() int

Count returns the number of registered callbacks

func (*CompositeCallback) GetCallbacks 

func (cc *CompositeCallback) GetCallbacks() []Callback

GetCallbacks returns all registered callbacks

func (*CompositeCallback) OnTaskAssigned 

func (cc *CompositeCallback) OnTaskAssigned(ctx *TaskEventContext) error

OnTaskAssigned implements Callback

func (*CompositeCallback) OnTaskBlocked 

func (cc *CompositeCallback) OnTaskBlocked(ctx *TaskEventContext) error

OnTaskBlocked implements Callback

func (*CompositeCallback) OnTaskCompleted 

func (cc *CompositeCallback) OnTaskCompleted(ctx *TaskEventContext) error

OnTaskCompleted implements Callback

func (*CompositeCallback) OnTaskCreated 

func (cc *CompositeCallback) OnTaskCreated(ctx *TaskEventContext) error

OnTaskCreated implements Callback

func (*CompositeCallback) OnTaskFailed 

func (cc *CompositeCallback) OnTaskFailed(ctx *TaskEventContext) error

OnTaskFailed implements Callback

func (*CompositeCallback) OnTaskRecovered 

func (cc *CompositeCallback) OnTaskRecovered(ctx *RecoveryEventContext) error

OnTaskRecovered implements Callback

func (*CompositeCallback) OnTaskStarted 

func (cc *CompositeCallback) OnTaskStarted(ctx *TaskEventContext) error

OnTaskStarted implements Callback

func (*CompositeCallback) OnTaskUnblocked 

func (cc *CompositeCallback) OnTaskUnblocked(ctx *TaskEventContext) error

OnTaskUnblocked implements Callback

func (*CompositeCallback) OnWorkerStalled 

func (cc *CompositeCallback) OnWorkerStalled(ctx *WorkerEventContext) error

OnWorkerStalled implements Callback

func (*CompositeCallback) OnWorkerStarted 

func (cc *CompositeCallback) OnWorkerStarted(ctx *WorkerEventContext) error

OnWorkerStarted implements Callback

func (*CompositeCallback) OnWorkerStopped 

func (cc *CompositeCallback) OnWorkerStopped(ctx *WorkerEventContext) error

OnWorkerStopped implements Callback

func (*CompositeCallback) Remove 

func (cc *CompositeCallback) Remove(callback Callback)

Remove removes a callback from the composite

func (*CompositeCallback) SetLogger 

func (cc *CompositeCallback) SetLogger(logger *log.Logger)

SetLogger sets the logger for the composite callback

func (*CompositeCallback) SetStopOnError 

func (cc *CompositeCallback) SetStopOnError(stop bool)

SetStopOnError sets whether to stop execution on first error

type ContextKey 

type ContextKey struct{}

ContextKey is used for storing trace context in task execution context

type EventType 

type EventType string

EventType represents the type of lifecycle event 

const (
	// Task lifecycle events
	EventTaskCreated   EventType = "task.created"
	EventTaskAssigned  EventType = "task.assigned"
	EventTaskStarted   EventType = "task.started"
	EventTaskCompleted EventType = "task.completed"
	EventTaskFailed    EventType = "task.failed"
	EventTaskBlocked   EventType = "task.blocked"
	EventTaskUnblocked EventType = "task.unblocked"
	EventTaskRecovered EventType = "task.recovered"

	// Worker lifecycle events
	EventWorkerStarted EventType = "worker.started"
	EventWorkerStopped EventType = "worker.stopped"
	EventWorkerStalled EventType = "worker.stalled"
)

type FilterMiddleware 

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

FilterMiddleware filters which callbacks are executed based on predicates

func NewFilterMiddleware 

func NewFilterMiddleware() *FilterMiddleware

NewFilterMiddleware creates a new filter middleware

func (*FilterMiddleware) FilterByEpic 

func (fm *FilterMiddleware) FilterByEpic(epics ...string)

FilterByEpic filters events by epic ID

func (*FilterMiddleware) FilterByTaskType 

func (fm *FilterMiddleware) FilterByTaskType(types ...string)

FilterByTaskType filters events by task type

func (*FilterMiddleware) FilterByWorkerID 

func (fm *FilterMiddleware) FilterByWorkerID(workers ...string)

FilterByWorkerID filters events by worker ID

func (*FilterMiddleware) SetLogger 

func (fm *FilterMiddleware) SetLogger(logger *log.Logger)

SetLogger sets the logger for the filter middleware

func (*FilterMiddleware) SetRecoveryEventFilter 

func (fm *FilterMiddleware) SetRecoveryEventFilter(filter func(*RecoveryEventContext) bool)

SetRecoveryEventFilter sets the filter for recovery events

func (*FilterMiddleware) SetTaskEventFilter 

func (fm *FilterMiddleware) SetTaskEventFilter(filter func(*TaskEventContext) bool)

SetTaskEventFilter sets the filter for task events

func (*FilterMiddleware) SetWorkerEventFilter 

func (fm *FilterMiddleware) SetWorkerEventFilter(filter func(*WorkerEventContext) bool)

SetWorkerEventFilter sets the filter for worker events

func (*FilterMiddleware) Wrap 

func (fm *FilterMiddleware) Wrap(next Callback) Callback

Wrap implements Middleware

type HealthCallback 

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

HealthCallback implements Callback with health monitoring capabilities. It tracks component health and provides HTTP endpoints for health checks.

func NewHealthCallback 

func NewHealthCallback() *HealthCallback

NewHealthCallback creates a new health callback

func (*HealthCallback) GetStatus 

func (hc *HealthCallback) GetStatus() *SystemHealth

GetStatus returns the current health status without running checks

func (*HealthCallback) IsHealthy 

func (hc *HealthCallback) IsHealthy() bool

IsHealthy returns true if the system is healthy

func (*HealthCallback) IsReady 

func (hc *HealthCallback) IsReady() bool

IsReady returns true if the system is ready

func (*HealthCallback) OnTaskAssigned 

func (hc *HealthCallback) OnTaskAssigned(ctx *TaskEventContext) error

OnTaskAssigned implements Callback

func (*HealthCallback) OnTaskBlocked 

func (hc *HealthCallback) OnTaskBlocked(ctx *TaskEventContext) error

OnTaskBlocked implements Callback

func (*HealthCallback) OnTaskCompleted 

func (hc *HealthCallback) OnTaskCompleted(ctx *TaskEventContext) error

OnTaskCompleted implements Callback

func (*HealthCallback) OnTaskCreated 

func (hc *HealthCallback) OnTaskCreated(ctx *TaskEventContext) error

OnTaskCreated implements Callback

func (*HealthCallback) OnTaskFailed 

func (hc *HealthCallback) OnTaskFailed(ctx *TaskEventContext) error

OnTaskFailed implements Callback

func (*HealthCallback) OnTaskRecovered 

func (hc *HealthCallback) OnTaskRecovered(ctx *RecoveryEventContext) error

OnTaskRecovered implements Callback

func (*HealthCallback) OnTaskStarted 

func (hc *HealthCallback) OnTaskStarted(ctx *TaskEventContext) error

OnTaskStarted implements Callback

func (*HealthCallback) OnTaskUnblocked 

func (hc *HealthCallback) OnTaskUnblocked(ctx *TaskEventContext) error

OnTaskUnblocked implements Callback

func (*HealthCallback) OnWorkerStalled 

func (hc *HealthCallback) OnWorkerStalled(ctx *WorkerEventContext) error

OnWorkerStalled implements Callback

func (*HealthCallback) OnWorkerStarted 

func (hc *HealthCallback) OnWorkerStarted(ctx *WorkerEventContext) error

OnWorkerStarted implements Callback

func (*HealthCallback) OnWorkerStopped 

func (hc *HealthCallback) OnWorkerStopped(ctx *WorkerEventContext) error

OnWorkerStopped implements Callback

func (*HealthCallback) RegisterCheck 

func (hc *HealthCallback) RegisterCheck(name string, check HealthCheck)

RegisterCheck registers a health check for a component

func (*HealthCallback) RunChecks 

func (hc *HealthCallback) RunChecks(ctx context.Context) (*SystemHealth, error)

RunChecks executes all registered health checks

func (*HealthCallback) SetLogger 

func (hc *HealthCallback) SetLogger(logger *log.Logger)

SetLogger sets the logger for the health callback

func (*HealthCallback) StartServer 

func (hc *HealthCallback) StartServer(addr string) error

StartServer starts an HTTP server for health checks

func (*HealthCallback) StopServer 

func (hc *HealthCallback) StopServer(ctx context.Context) error

StopServer stops the health check HTTP server

func (*HealthCallback) UnregisterCheck 

func (hc *HealthCallback) UnregisterCheck(name string)

UnregisterCheck removes a health check

type HealthCheck 

type HealthCheck func(ctx context.Context) (*ComponentHealth, error)

HealthCheck defines a function that checks the health of a component

func GoroutineCheck 

func GoroutineCheck(threshold int) HealthCheck

GoroutineCheck returns a health check for goroutine count

func MemoryCheck 

func MemoryCheck(thresholdMB uint64) HealthCheck

MemoryCheck returns a health check for memory usage

type HealthStatus 

type HealthStatus string

HealthStatus represents the health status of a component 

const (
	StatusHealthy   HealthStatus = "healthy"
	StatusDegraded  HealthStatus = "degraded"
	StatusUnhealthy HealthStatus = "unhealthy"
	StatusUnknown   HealthStatus = "unknown"
)

type LifecycleCallback 

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

LifecycleCallback implements Callback with lifecycle orchestration capabilities. It provides hooks for pre/post task execution and state machine management.

func NewLifecycleCallback 

func NewLifecycleCallback() *LifecycleCallback

NewLifecycleCallback creates a new lifecycle callback

func (*LifecycleCallback) GetStalledTasks 

func (lc *LifecycleCallback) GetStalledTasks(timeout time.Duration) []string

GetStalledTasks returns tasks that have been in a non-terminal phase too long

func (*LifecycleCallback) GetStateMachine 

func (lc *LifecycleCallback) GetStateMachine() *TaskStateMachine

GetStateMachine returns the state machine

func (*LifecycleCallback) GetTasksByPhase 

func (lc *LifecycleCallback) GetTasksByPhase(phase LifecyclePhase) []string

GetTasksByPhase returns all tasks in a specific phase

func (*LifecycleCallback) OnTaskAssigned 

func (lc *LifecycleCallback) OnTaskAssigned(ctx *TaskEventContext) error

OnTaskAssigned implements Callback

func (*LifecycleCallback) OnTaskBlocked 

func (lc *LifecycleCallback) OnTaskBlocked(ctx *TaskEventContext) error

OnTaskBlocked implements Callback

func (*LifecycleCallback) OnTaskCompleted 

func (lc *LifecycleCallback) OnTaskCompleted(ctx *TaskEventContext) error

OnTaskCompleted implements Callback

func (*LifecycleCallback) OnTaskCreated 

func (lc *LifecycleCallback) OnTaskCreated(ctx *TaskEventContext) error

OnTaskCreated implements Callback

func (*LifecycleCallback) OnTaskFailed 

func (lc *LifecycleCallback) OnTaskFailed(ctx *TaskEventContext) error

OnTaskFailed implements Callback

func (*LifecycleCallback) OnTaskRecovered 

func (lc *LifecycleCallback) OnTaskRecovered(ctx *RecoveryEventContext) error

OnTaskRecovered implements Callback

func (*LifecycleCallback) OnTaskStarted 

func (lc *LifecycleCallback) OnTaskStarted(ctx *TaskEventContext) error

OnTaskStarted implements Callback

func (*LifecycleCallback) OnTaskUnblocked 

func (lc *LifecycleCallback) OnTaskUnblocked(ctx *TaskEventContext) error

OnTaskUnblocked implements Callback

func (*LifecycleCallback) OnWorkerStalled 

func (lc *LifecycleCallback) OnWorkerStalled(ctx *WorkerEventContext) error

OnWorkerStalled implements Callback

func (*LifecycleCallback) OnWorkerStarted 

func (lc *LifecycleCallback) OnWorkerStarted(ctx *WorkerEventContext) error

OnWorkerStarted implements Callback

func (*LifecycleCallback) OnWorkerStopped 

func (lc *LifecycleCallback) OnWorkerStopped(ctx *WorkerEventContext) error

OnWorkerStopped implements Callback

func (*LifecycleCallback) RegisterPostHook 

func (lc *LifecycleCallback) RegisterPostHook(eventType string, hook func(*TaskEventContext) error)

RegisterPostHook registers a function to be called after a task event

func (*LifecycleCallback) RegisterPreHook 

func (lc *LifecycleCallback) RegisterPreHook(eventType string, hook func(*TaskEventContext) error)

RegisterPreHook registers a function to be called before a task event

func (*LifecycleCallback) ResumeAll 

func (lc *LifecycleCallback) ResumeAll(ctx context.Context) error

ResumeAll transitions all suspended tasks back to running

func (*LifecycleCallback) SetLogger 

func (lc *LifecycleCallback) SetLogger(logger *log.Logger)

SetLogger sets the logger for the lifecycle callback

func (*LifecycleCallback) ShutdownAll 

func (lc *LifecycleCallback) ShutdownAll(ctx context.Context) error

ShutdownAll transitions all active tasks to terminated

func (*LifecycleCallback) SuspendAll 

func (lc *LifecycleCallback) SuspendAll(ctx context.Context) error

SuspendAll transitions all running tasks to suspended state

type LifecyclePhase 

type LifecyclePhase string

LifecyclePhase represents the current phase of task or worker lifecycle 

const (
	PhaseInitializing LifecyclePhase = "initializing"
	PhaseRunning      LifecyclePhase = "running"
	PhaseSuspending   LifecyclePhase = "suspending"
	PhaseSuspended    LifecyclePhase = "suspended"
	PhaseTerminating  LifecyclePhase = "terminating"
	PhaseTerminated   LifecyclePhase = "terminated"
)

type LifecycleState 

type LifecycleState struct {
	Phase        LifecyclePhase
	StartedAt    time.Time
	LastActivity time.Time
	Metadata     map[string]string
}

LifecycleState represents the state of a tracked lifecycle entity

type LogLevel 

type LogLevel int

LogLevel defines the verbosity level for logging 

const (
	LogLevelDebug LogLevel = iota
	LogLevelInfo
	LogLevelWarn
	LogLevelError
)

func (LogLevel) String 

func (l LogLevel) String() string

String returns the string representation of the log level

type LoggingConfig 

type LoggingConfig struct {
	// Output destination (defaults to stdout)
	Writer io.Writer
	// Minimum log level to output
	MinLevel LogLevel
	// Include timestamp in logs
	Timestamp bool
	// Include correlation ID for request tracing
	CorrelationID bool
	// Enable pretty-printing for development
	Pretty bool
}

LoggingConfig configures the structured logging callback

type Metric 

type Metric struct {
	Type  MetricType
	Name  string
	Help  string
	Value float64

	Labels map[string]string
	// contains filtered or unexported fields
}

Metric represents a single metric

type MetricType 

type MetricType string

MetricType represents the type of Prometheus metric 

const (
	MetricTypeCounter   MetricType = "counter"
	MetricTypeGauge     MetricType = "gauge"
	MetricTypeHistogram MetricType = "histogram"
)

type MetricsCallback 

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

MetricsCallback implements Callback with Prometheus-compatible metrics tracking. It maintains counters, gauges, and histograms for workforce lifecycle events.

func NewMetricsCallback 

func NewMetricsCallback() *MetricsCallback

NewMetricsCallback creates a new metrics callback

func (*MetricsCallback) Gauge 

func (m *MetricsCallback) Gauge(name string, value float64, labels map[string]string)

Gauge sets a gauge metric value

func (*MetricsCallback) GetAllMetrics 

func (m *MetricsCallback) GetAllMetrics() map[string]*Metric

GetAllMetrics returns all metrics

func (*MetricsCallback) GetMetric 

func (m *MetricsCallback) GetMetric(name string, labels map[string]string) (*Metric, bool)

GetMetric returns a metric by name and labels

func (*MetricsCallback) Histogram 

func (m *MetricsCallback) Histogram(name string, value float64, labels map[string]string)

Histogram records a value in a histogram metric

func (*MetricsCallback) Increment 

func (m *MetricsCallback) Increment(name string, labels map[string]string)

Increment increments a counter metric

func (*MetricsCallback) OnTaskAssigned 

func (m *MetricsCallback) OnTaskAssigned(ctx *TaskEventContext) error

OnTaskAssigned implements Callback

func (*MetricsCallback) OnTaskBlocked 

func (m *MetricsCallback) OnTaskBlocked(ctx *TaskEventContext) error

OnTaskBlocked implements Callback

func (*MetricsCallback) OnTaskCompleted 

func (m *MetricsCallback) OnTaskCompleted(ctx *TaskEventContext) error

OnTaskCompleted implements Callback

func (*MetricsCallback) OnTaskCreated 

func (m *MetricsCallback) OnTaskCreated(ctx *TaskEventContext) error

OnTaskCreated implements Callback

func (*MetricsCallback) OnTaskFailed 

func (m *MetricsCallback) OnTaskFailed(ctx *TaskEventContext) error

OnTaskFailed implements Callback

func (*MetricsCallback) OnTaskRecovered 

func (m *MetricsCallback) OnTaskRecovered(ctx *RecoveryEventContext) error

OnTaskRecovered implements Callback

func (*MetricsCallback) OnTaskStarted 

func (m *MetricsCallback) OnTaskStarted(ctx *TaskEventContext) error

OnTaskStarted implements Callback

func (*MetricsCallback) OnTaskUnblocked 

func (m *MetricsCallback) OnTaskUnblocked(ctx *TaskEventContext) error

OnTaskUnblocked implements Callback

func (*MetricsCallback) OnWorkerStalled 

func (m *MetricsCallback) OnWorkerStalled(ctx *WorkerEventContext) error

OnWorkerStalled implements Callback

func (*MetricsCallback) OnWorkerStarted 

func (m *MetricsCallback) OnWorkerStarted(ctx *WorkerEventContext) error

OnWorkerStarted implements Callback

func (*MetricsCallback) OnWorkerStopped 

func (m *MetricsCallback) OnWorkerStopped(ctx *WorkerEventContext) error

OnWorkerStopped implements Callback

func (*MetricsCallback) RegisterMetric 

func (m *MetricsCallback) RegisterMetric(metric *Metric)

RegisterMetric registers a new metric

func (*MetricsCallback) Reset 

func (m *MetricsCallback) Reset()

Reset clears all metrics

func (*MetricsCallback) SetLogger 

func (m *MetricsCallback) SetLogger(logger *log.Logger)

SetLogger sets the logger for the metrics callback

func (*MetricsCallback) WritePrometheus 

func (m *MetricsCallback) WritePrometheus(w io.Writer) error

WritePrometheus writes metrics in Prometheus text format

type Middleware 

type Middleware interface {
	Callback

	// Wrap returns a new Callback that wraps the given one
	Wrap(next Callback) Callback
}

Middleware wraps a Callback to add pre/post processing logic

type MiddlewareFunc 

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

MiddlewareFunc is an adapter that allows a function to be used as Middleware

func (*MiddlewareFunc) OnTaskAssigned 

func (mf *MiddlewareFunc) OnTaskAssigned(ctx *TaskEventContext) error

func (*MiddlewareFunc) OnTaskBlocked 

func (mf *MiddlewareFunc) OnTaskBlocked(ctx *TaskEventContext) error

func (*MiddlewareFunc) OnTaskCompleted 

func (mf *MiddlewareFunc) OnTaskCompleted(ctx *TaskEventContext) error

func (*MiddlewareFunc) OnTaskCreated 

func (mf *MiddlewareFunc) OnTaskCreated(ctx *TaskEventContext) error

Callback forwarding for MiddlewareFunc

func (*MiddlewareFunc) OnTaskFailed 

func (mf *MiddlewareFunc) OnTaskFailed(ctx *TaskEventContext) error

func (*MiddlewareFunc) OnTaskRecovered 

func (mf *MiddlewareFunc) OnTaskRecovered(ctx *RecoveryEventContext) error

func (*MiddlewareFunc) OnTaskStarted 

func (mf *MiddlewareFunc) OnTaskStarted(ctx *TaskEventContext) error

func (*MiddlewareFunc) OnTaskUnblocked 

func (mf *MiddlewareFunc) OnTaskUnblocked(ctx *TaskEventContext) error

func (*MiddlewareFunc) OnWorkerStalled 

func (mf *MiddlewareFunc) OnWorkerStalled(ctx *WorkerEventContext) error

func (*MiddlewareFunc) OnWorkerStarted 

func (mf *MiddlewareFunc) OnWorkerStarted(ctx *WorkerEventContext) error

func (*MiddlewareFunc) OnWorkerStopped 

func (mf *MiddlewareFunc) OnWorkerStopped(ctx *WorkerEventContext) error

func (*MiddlewareFunc) Wrap 

func (mf *MiddlewareFunc) Wrap(next Callback) Callback

Wrap implements Middleware

type OTelCallback 

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

OTelCallback implements Callback with OpenTelemetry tracing. It creates spans and propagates trace context through task execution.

This is a stub implementation that provides the basic structure for OpenTelemetry integration. A full implementation would: - Use context propagation for distributed tracing - Add more detailed span attributes - Handle span links for async operations - Integrate with metrics and logs

func NewOTelCallback 

func NewOTelCallback() *OTelCallback

NewOTelCallback creates a new OpenTelemetry callback

func (*OTelCallback) OnTaskAssigned 

func (o *OTelCallback) OnTaskAssigned(ctx *TaskEventContext) error

OnTaskAssigned implements Callback

func (*OTelCallback) OnTaskBlocked 

func (o *OTelCallback) OnTaskBlocked(ctx *TaskEventContext) error

OnTaskBlocked implements Callback

func (*OTelCallback) OnTaskCompleted 

func (o *OTelCallback) OnTaskCompleted(ctx *TaskEventContext) error

OnTaskCompleted implements Callback

func (*OTelCallback) OnTaskCreated 

func (o *OTelCallback) OnTaskCreated(ctx *TaskEventContext) error

OnTaskCreated implements Callback

func (*OTelCallback) OnTaskFailed 

func (o *OTelCallback) OnTaskFailed(ctx *TaskEventContext) error

OnTaskFailed implements Callback

func (*OTelCallback) OnTaskRecovered 

func (o *OTelCallback) OnTaskRecovered(ctx *RecoveryEventContext) error

OnTaskRecovered implements Callback

func (*OTelCallback) OnTaskStarted 

func (o *OTelCallback) OnTaskStarted(ctx *TaskEventContext) error

OnTaskStarted implements Callback

func (*OTelCallback) OnTaskUnblocked 

func (o *OTelCallback) OnTaskUnblocked(ctx *TaskEventContext) error

OnTaskUnblocked implements Callback

func (*OTelCallback) OnWorkerStalled 

func (o *OTelCallback) OnWorkerStalled(ctx *WorkerEventContext) error

OnWorkerStalled implements Callback

func (*OTelCallback) OnWorkerStarted 

func (o *OTelCallback) OnWorkerStarted(ctx *WorkerEventContext) error

OnWorkerStarted implements Callback

func (*OTelCallback) OnWorkerStopped 

func (o *OTelCallback) OnWorkerStopped(ctx *WorkerEventContext) error

OnWorkerStopped implements Callback

func (*OTelCallback) SetLogger 

func (o *OTelCallback) SetLogger(logger *log.Logger)

SetLogger sets the logger for the OTel callback

type Priority 

type Priority int

Priority defines callback execution order (lower = earlier) 

const (
	PriorityHigh   Priority = 0
	PriorityMedium Priority = 50
	PriorityLow    Priority = 100
)

type RecoveryEventContext 

type RecoveryEventContext struct {
	// Task identification
	TaskID string
	Title  string

	// Recovery decision
	Strategy   string // "retry", "decompose", "reassign", "escalate", "abandon"
	Reason     string
	Confidence float64

	// Failure context
	OriginalError string
	AttemptCount  int

	// Timing
	Timestamp time.Time

	// Additional metadata
	Metadata map[string]string
}

RecoveryEventContext provides context for task recovery events

type Registry 

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

Registry manages lifecycle event callbacks. It provides thread-safe registration, unregistration, and dispatching of callbacks for workforce lifecycle events.

The registry is designed for zero-overhead when no callbacks are registered: dispatch operations check if there are any registered callbacks before creating context or invoking handlers.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry creates a new callback registry

func (*Registry) Count 

func (r *Registry) Count(event EventType) int

Count returns the number of callbacks registered for an event type

func (*Registry) Disable 

func (r *Registry) Disable(name string)

Disable disables a callback by name

func (*Registry) Dispatch 

func (r *Registry) Dispatch(event EventType, ctx *TaskEventContext) error

Dispatch invokes all registered callbacks for a task event. Returns the first non-nil error from any callback, but continues invoking remaining callbacks even after an error. Callback errors are logged but do not propagate to crash the system.

func (*Registry) DispatchRecovery 

func (r *Registry) DispatchRecovery(event EventType, ctx *RecoveryEventContext) error

DispatchRecovery invokes all registered callbacks for a recovery event

func (*Registry) DispatchTask 

func (r *Registry) DispatchTask(event EventType, ctx *TaskEventContext) error

DispatchTask invokes all registered callbacks for a task event

func (*Registry) DispatchWorker 

func (r *Registry) DispatchWorker(event EventType, ctx *WorkerEventContext) error

DispatchWorker invokes all registered callbacks for a worker event

func (*Registry) Enable 

func (r *Registry) Enable(name string)

Enable enables a callback by name

func (*Registry) Register 

func (r *Registry) Register(callback Callback, events []EventType, priority Priority, name string)

Register registers a callback for specific event types. The callback will be invoked for all specified event types.

Priority controls execution order: lower priority callbacks run first. Use the same priority for independent callbacks where order doesn't matter.

The name is used for debugging and error logging. If a callback with the same name exists for the same event, it will be replaced.

func (*Registry) RegisteredNames 

func (r *Registry) RegisteredNames(event EventType) []string

RegisteredNames returns the names of all callbacks registered for an event type

func (*Registry) SetLogger 

func (r *Registry) SetLogger(logger *log.Logger)

SetLogger sets the logger for the registry

func (*Registry) Unregister 

func (r *Registry) Unregister(name string, events []EventType)

Unregister removes a callback by name for specific event types. If events is empty, the callback is removed from all event types.

type RetryMiddleware 

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

RetryMiddleware retries failed callback executions

func NewRetryMiddleware 

func NewRetryMiddleware(maxRetries int) *RetryMiddleware

NewRetryMiddleware creates a new retry middleware

func (*RetryMiddleware) SetBackoff 

func (rm *RetryMiddleware) SetBackoff(backoff func(attempt int) time.Duration)

SetBackoff sets the backoff strategy

func (*RetryMiddleware) SetLogger 

func (rm *RetryMiddleware) SetLogger(logger *log.Logger)

SetLogger sets the logger for the retry middleware

func (*RetryMiddleware) Wrap 

func (rm *RetryMiddleware) Wrap(next Callback) Callback

Wrap implements Middleware

type StateTransition 

type StateTransition struct {
	From      LifecyclePhase
	To        LifecyclePhase
	Timestamp time.Time
	Reason    string
}

StateTransition records a state change

type StructuredLoggingCallback 

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

StructuredLoggingCallback implements Callback with JSON-formatted logs. All events are logged as structured JSON for easy parsing by external tools.

func NewStructuredLoggingCallback 

func NewStructuredLoggingCallback(config LoggingConfig) *StructuredLoggingCallback

NewStructuredLoggingCallback creates a new structured logging callback

func (*StructuredLoggingCallback) OnTaskAssigned 

func (c *StructuredLoggingCallback) OnTaskAssigned(ctx *TaskEventContext) error

OnTaskAssigned implements Callback

func (*StructuredLoggingCallback) OnTaskBlocked 

func (c *StructuredLoggingCallback) OnTaskBlocked(ctx *TaskEventContext) error

OnTaskBlocked implements Callback

func (*StructuredLoggingCallback) OnTaskCompleted 

func (c *StructuredLoggingCallback) OnTaskCompleted(ctx *TaskEventContext) error

OnTaskCompleted implements Callback

func (*StructuredLoggingCallback) OnTaskCreated 

func (c *StructuredLoggingCallback) OnTaskCreated(ctx *TaskEventContext) error

OnTaskCreated implements Callback

func (*StructuredLoggingCallback) OnTaskFailed 

func (c *StructuredLoggingCallback) OnTaskFailed(ctx *TaskEventContext) error

OnTaskFailed implements Callback

func (*StructuredLoggingCallback) OnTaskRecovered 

func (c *StructuredLoggingCallback) OnTaskRecovered(ctx *RecoveryEventContext) error

OnTaskRecovered implements Callback

func (*StructuredLoggingCallback) OnTaskStarted 

func (c *StructuredLoggingCallback) OnTaskStarted(ctx *TaskEventContext) error

OnTaskStarted implements Callback

func (*StructuredLoggingCallback) OnTaskUnblocked 

func (c *StructuredLoggingCallback) OnTaskUnblocked(ctx *TaskEventContext) error

OnTaskUnblocked implements Callback

func (*StructuredLoggingCallback) OnWorkerStalled 

func (c *StructuredLoggingCallback) OnWorkerStalled(ctx *WorkerEventContext) error

OnWorkerStalled implements Callback

func (*StructuredLoggingCallback) OnWorkerStarted 

func (c *StructuredLoggingCallback) OnWorkerStarted(ctx *WorkerEventContext) error

OnWorkerStarted implements Callback

func (*StructuredLoggingCallback) OnWorkerStopped 

func (c *StructuredLoggingCallback) OnWorkerStopped(ctx *WorkerEventContext) error

OnWorkerStopped implements Callback

type SystemHealth 

type SystemHealth struct {
	Status     HealthStatus                `json:"status"`
	Version    string                      `json:"version,omitempty"`
	Timestamp  time.Time                   `json:"timestamp"`
	Components map[string]*ComponentHealth `json:"components"`
}

SystemHealth represents the overall health of the system

type TaskEventContext 

type TaskEventContext struct {
	// Task identification
	TaskID   string
	Title    string
	EpicID   string
	Type     string
	Priority int

	// State information
	PrevState  string
	NewState   string
	Transition string // e.g., "ready", "retry", "manual"

	// Worker information
	WorkerID    string
	WorkerIndex int

	// Timing
	Timestamp time.Time
	Duration  *time.Duration

	// Attempt information
	Attempt     int
	MaxAttempts int

	// Failure information (for failed/recovered events)
	Error         string
	ErrorType     string
	ErrorCategory string

	// Additional metadata
	Metadata map[string]string
}

TaskEventContext provides context for task-related events

type TaskStateMachine 

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

TaskStateMachine tracks state transitions for tasks

func NewTaskStateMachine 

func NewTaskStateMachine() *TaskStateMachine

NewTaskStateMachine creates a new task state machine

func (*TaskStateMachine) GetAllStates 

func (sm *TaskStateMachine) GetAllStates() map[string]*LifecycleState

GetAllStates returns all tracked states

func (*TaskStateMachine) GetState 

func (sm *TaskStateMachine) GetState(taskID string) (*LifecycleState, bool)

GetState returns the current state of a task

func (*TaskStateMachine) GetTransitions 

func (sm *TaskStateMachine) GetTransitions(taskID string) []StateTransition

GetTransitions returns the transition history for a task

func (*TaskStateMachine) Initialize 

func (sm *TaskStateMachine) Initialize(taskID string, metadata map[string]string)

Initialize adds a new task to the state machine

func (*TaskStateMachine) Remove 

func (sm *TaskStateMachine) Remove(taskID string)

Remove removes a task from the state machine

func (*TaskStateMachine) SetLogger 

func (sm *TaskStateMachine) SetLogger(logger *log.Logger)

SetLogger sets the logger for the state machine

func (*TaskStateMachine) Transition 

func (sm *TaskStateMachine) Transition(taskID string, newPhase LifecyclePhase, reason string) error

Transition moves a task to a new phase

type TimeoutMiddleware 

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

TimeoutMiddleware adds timeout to callback execution

func NewTimeoutMiddleware 

func NewTimeoutMiddleware(timeout time.Duration) *TimeoutMiddleware

NewTimeoutMiddleware creates a new timeout middleware

func (*TimeoutMiddleware) SetLogger 

func (tm *TimeoutMiddleware) SetLogger(logger *log.Logger)

SetLogger sets the logger for the timeout middleware

func (*TimeoutMiddleware) SetOnTimeout 

func (tm *TimeoutMiddleware) SetOnTimeout(fn func(event string, duration time.Duration))

SetOnTimeout sets a callback function to be called on timeout

func (*TimeoutMiddleware) Wrap 

func (tm *TimeoutMiddleware) Wrap(next Callback) Callback

Wrap implements Middleware

type WorkerEventContext 

type WorkerEventContext struct {
	// Worker identification
	WorkerID    string
	WorkerIndex int

	// Worker state
	State string // "started", "stopped", "stalled"

	// Current task (if any)
	CurrentTaskID    string
	CurrentTaskTitle string

	// Timing
	Timestamp time.Time

	// Stall information (for stalled events)
	StallReason   string
	StallDuration time.Duration

	// Additional metadata
	Metadata map[string]string
}

WorkerEventContext provides context for worker-related events

Source Files

View all Source files

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.