events

package
v0.8.0 Latest Latest
Warning

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

 Go to latest Published: Jul 20, 2026 License: MIT Imports: 5 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrBusClosed       error = errBusClosed{}
	ErrScopedBusClosed error = errScopedBusClosed{}
)

Sentinel errors returned by EventBus methods. @sk-task event-bus#T1.1: sentinel errors ErrBusClosed and ErrScopedBusClosed

Functions

This section is empty.

Types

type CallbackBridge 

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

CallbackBridge converts callback.StepInfo into typed EventBus events. @sk-task event-bus#T3.1: CallbackBridge mapping StepInfo to Event (AC-011)

func NewCallbackBridge 

func NewCallbackBridge(bus *EventBus) *CallbackBridge

NewCallbackBridge creates a bridge that publishes events to the given bus.

func (*CallbackBridge) Handler 

func (cb *CallbackBridge) Handler(existing callback.Callback) callback.Callback

Handler returns a callback.Callback wrapper that invokes existing (if non-nil) then converts StepInfo to an Event and publishes it on the bus. @sk-task event-bus#T3.1: Handler wrapper for dual-emit pattern (AC-011)

type Event 

type Event struct {
	Timestamp time.Time
	Type      EventType
	Source    string
	Metadata  map[string]string
	Payload   any
}

Event is the universal envelope wrapping any framework event. @sk-task event-bus#T1.1: Event struct with Timestamp/Type/Source/Metadata/Payload (AC-002)

type EventBus 

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

EventBus is a goroutine-safe publish/subscribe hub for typed events. @sk-task event-bus#T2.1: EventBus with Subscribe/Unsubscribe/Publish/Close/wildcard (AC-003, AC-005) @sk-task event-bus#T2.4: EventBus panic recovery and SubscribeWithMatcher (AC-007, AC-008)

func NewEventBus 

func NewEventBus(cfg EventBusConfig) *EventBus

NewEventBus creates a new EventBus with the given config. @sk-task event-bus#T2.1: NewEventBus constructor

func (*EventBus) Close 

func (b *EventBus) Close() error

Close shuts down the bus. Publish returns ErrBusClosed after close. PublishAsync silently drops. Close is idempotent. @sk-task event-bus#T2.1: Close with idempotent shutdown (AC-004b)

func (*EventBus) Publish 

func (b *EventBus) Publish(ctx context.Context, event Event) error

Publish synchronously delivers an event to all matching subscribers. Returns ErrBusClosed if the bus has been closed. @sk-task event-bus#T2.1: Publish with sync delivery (AC-003, AC-004b)

func (*EventBus) PublishAsync 

func (b *EventBus) PublishAsync(event Event)

PublishAsync asynchronously delivers an event via a goroutine pool. Silently drops the event if the bus is closed. @sk-task event-bus#T2.1: PublishAsync with goroutine pool (AC-004, AC-004b)

func (*EventBus) RegisterListener 

func (b *EventBus) RegisterListener(listener Listener)

RegisterListener registers a lifecycle listener. OnStart is called immediately. @sk-task event-bus#T2.3: Listener RegisterListener/UnregisterListener (AC-009)

func (*EventBus) Scoped 

func (b *EventBus) Scoped() *ScopedBus

Scoped creates a child ScopedBus whose subscriptions are automatically cleaned up when Close is called. @sk-task event-bus#T2.2: ScopedBus with auto-unsubscribe on Close (AC-006)

func (*EventBus) Subscribe 

func (b *EventBus) Subscribe(eventType EventType, handler EventHandler)

Subscribe registers a handler for the given event type. Pass "*" as eventType to receive all events (wildcard). Returns the handler for use with Unsubscribe.

func (*EventBus) SubscribeWithMatcher 

func (b *EventBus) SubscribeWithMatcher(eventType EventType, matcher MatcherFunc, handler EventHandler)

SubscribeWithMatcher registers a handler that is only invoked when the matcher returns true. A nil matcher matches all events of the given type. @sk-task event-bus#T2.4: SubscribeWithMatcher with MatcherFunc filtering (AC-008)

func (*EventBus) UnregisterListener 

func (b *EventBus) UnregisterListener(listener Listener)

UnregisterListener removes a lifecycle listener and calls OnStop.

func (*EventBus) Unsubscribe 

func (b *EventBus) Unsubscribe(eventType EventType, handler EventHandler)

Unsubscribe removes a previously registered handler for the given event type.

type EventBusConfig 

type EventBusConfig struct {
	ErrorHandler         func(error)
	MaxConsecutivePanics int
	GoroutinePoolSize    int
}

EventBusConfig configures an EventBus instance. @sk-task event-bus#T2.1: EventBusConfig with ErrorHandler, MaxConsecutivePanics, GoroutinePoolSize

type EventHandler 

type EventHandler func(ctx context.Context, event Event)

EventHandler is a subscriber function called synchronously for matching events.

type EventType 

type EventType string

EventType is a typed string constant identifying the kind of event. 

const (
	// --- Crew (≥8) ---
	EventCrewKickoffStarted  EventType = "crew.kickoff.started"
	EventCrewKickoffComplete EventType = "crew.kickoff.complete"
	EventCrewKickoffFailed   EventType = "crew.kickoff.failed"
	EventCrewTrainStarted    EventType = "crew.train.started"
	EventCrewTrainComplete   EventType = "crew.train.complete"
	EventCrewTrainFailed     EventType = "crew.train.failed"
	EventCrewTestStarted     EventType = "crew.test.started"
	EventCrewTestComplete    EventType = "crew.test.complete"

	// --- Agent (≥6) ---
	EventAgentExecutionStarted   EventType = "agent.execution.started"
	EventAgentExecutionComplete  EventType = "agent.execution.complete"
	EventAgentExecutionError     EventType = "agent.execution.error"
	EventAgentLiteKickoff        EventType = "agent.lite.kickoff"
	EventAgentEvaluationStarted  EventType = "agent.evaluation.started"
	EventAgentEvaluationComplete EventType = "agent.evaluation.complete"

	// --- Task (≥4) ---
	EventTaskStarted    EventType = "task.started"
	EventTaskComplete   EventType = "task.complete"
	EventTaskFailed     EventType = "task.failed"
	EventTaskEvaluation EventType = "task.evaluation"

	// --- Tool (≥6) ---
	EventToolUsageStarted     EventType = "tool.usage.started"
	EventToolUsageFinished    EventType = "tool.usage.finished"
	EventToolUsageError       EventType = "tool.usage.error"
	EventToolValidateInputErr EventType = "tool.validate_input.error"
	EventToolSelectionError   EventType = "tool.selection.error"
	EventToolExecutionError   EventType = "tool.execution.error"

	// --- Flow (≥8) ---
	EventFlowCreated        EventType = "flow.created"
	EventFlowStarted        EventType = "flow.started"
	EventFlowFinished       EventType = "flow.finished"
	EventFlowPaused         EventType = "flow.paused"
	EventFlowMethodStarted  EventType = "flow.method.started"
	EventFlowMethodFinished EventType = "flow.method.finished"
	EventFlowMethodFailed   EventType = "flow.method.failed"
	EventFlowPlot           EventType = "flow.plot"

	// --- LLM (≥4) ---
	EventLLMCallStarted   EventType = "llm.call.started"
	EventLLMCallComplete  EventType = "llm.call.complete"
	EventLLMCallFailed    EventType = "llm.call.failed"
	EventLLMStreamChunked EventType = "llm.stream.chunked"

	// --- Memory (≥6) ---
	EventMemoryQueryStarted  EventType = "memory.query.started"
	EventMemoryQueryComplete EventType = "memory.query.complete"
	EventMemoryQueryFailed   EventType = "memory.query.failed"
	EventMemorySaveStarted   EventType = "memory.save.started"
	EventMemorySaveComplete  EventType = "memory.save.complete"
	EventMemorySaveFailed    EventType = "memory.save.failed"

	// --- Knowledge (≥6) ---
	EventKnowledgeRetrievalStarted  EventType = "knowledge.retrieval.started"
	EventKnowledgeRetrievalComplete EventType = "knowledge.retrieval.complete"
	EventKnowledgeRetrievalFailed   EventType = "knowledge.retrieval.failed"
	EventKnowledgeQueryStarted      EventType = "knowledge.query.started"
	EventKnowledgeQueryComplete     EventType = "knowledge.query.complete"
	EventKnowledgeQueryFailed       EventType = "knowledge.query.failed"

	// --- MCP (≥6) ---
	EventMCPConnectionStarted  EventType = "mcp.connection.started"
	EventMCPConnectionComplete EventType = "mcp.connection.complete"
	EventMCPConnectionFailed   EventType = "mcp.connection.failed"
	EventMCPToolCallStarted    EventType = "mcp.toolcall.started"
	EventMCPToolCallComplete   EventType = "mcp.toolcall.complete"
	EventMCPToolCallFailed     EventType = "mcp.toolcall.failed"

	// --- Guardrail (≥2) ---
	EventGuardrailValidationStarted  EventType = "guardrail.validation.started"
	EventGuardrailValidationComplete EventType = "guardrail.validation.complete"

	// --- A2A (≥8) ---
	EventA2ADelegationStarted  EventType = "a2a.delegation.started"
	EventA2ADelegationComplete EventType = "a2a.delegation.complete"
	EventA2AMessageSent        EventType = "a2a.message.sent"
	EventA2AMessageReceived    EventType = "a2a.message.received"
	EventA2AStreamStarted      EventType = "a2a.stream.started"
	EventA2AStreamChunked      EventType = "a2a.stream.chunked"
	EventA2AAuthFailed         EventType = "a2a.auth.failed"
	EventA2AConnectionLost     EventType = "a2a.connection.lost"

	// --- HITL (≥4) ---
	EventHITLInputRequested    EventType = "hitl.input.requested"
	EventHITLInputReceived     EventType = "hitl.input.received"
	EventHITLFeedbackRequested EventType = "hitl.feedback.requested"
	EventHITLFeedbackReceived  EventType = "hitl.feedback.received"

	// --- Async Crew Execution (≥7) ---
	// @sk-task async-crew-execution#T1.3: async event constants (AC-014)
	EventAsyncStarted  EventType = "crew.async.started"
	EventAsyncComplete EventType = "crew.async.complete"
	EventAsyncFailed   EventType = "crew.async.failed"

	// @sk-task async-crew-execution#T1.3: for-each event constants (AC-014)
	EventForEachStarted      EventType = "crew.for_each.started"
	EventForEachComplete     EventType = "crew.for_each.complete"
	EventForEachItemComplete EventType = "crew.for_each.item_complete"
	EventForEachItemFailed   EventType = "crew.for_each.item_failed"

	// --- Replay & Fork (≥6) ---
	// @sk-task replay-system#T1.2: replay and fork event constants (AC-009)
	EventCrewReplayStarted  EventType = "crew.replay.started"
	EventCrewReplayComplete EventType = "crew.replay.complete"
	EventCrewReplayFailed   EventType = "crew.replay.failed"
	EventCrewForkStarted    EventType = "crew.fork.started"
	EventCrewForkComplete   EventType = "crew.fork.complete"
	EventCrewForkFailed     EventType = "crew.fork.failed"

	// --- Reasoning (≥4) ---
	EventReasoningStarted      EventType = "reasoning.started"
	EventReasoningComplete     EventType = "reasoning.complete"
	EventReasoningPlanRefined  EventType = "reasoning.plan.refined"
	EventReasoningGoalAchieved EventType = "reasoning.goal.achieved"
)

Event type constants — 72+ across 13 categories. @sk-task event-bus#T1.2: 72+ event type constants across 13 categories (AC-001)

type Listener 

type Listener interface {
	OnStart(bus *EventBus)
	OnStop(bus *EventBus)
	OnError(err error)
}

Listener provides lifecycle hooks for long-lived event bus subscribers. @sk-task event-bus#T2.3: Listener interface with OnStart/OnStop/OnError (AC-009)

type MatcherFunc 

type MatcherFunc func(event Event) bool

MatcherFunc returns true when an event matches custom filter criteria.

type OTELBridge 

type OTELBridge struct{}

OTELBridge is a no-op implementation when the otel build tag is not set. @sk-task event-bus#T3.2: OTELBridge noop stub (AC-010)

func NewOTELBridge 

func NewOTELBridge(bus *EventBus, tp any) *OTELBridge

NewOTELBridge returns nil when OTEL is not enabled.

func (*OTELBridge) Attach 

func (b *OTELBridge) Attach()

Attach is a no-op.

type ScopedBus 

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

ScopedBus is a child bus that auto-unsubscribes all subscriptions on Close. @sk-task event-bus#T2.2: ScopedBus implementation (AC-006)

func (*ScopedBus) Close 

func (s *ScopedBus) Close()

Close unsubscribes all recorded subscriptions from the parent bus.

func (*ScopedBus) Subscribe 

func (s *ScopedBus) Subscribe(eventType EventType, handler EventHandler)

Subscribe registers a handler on the parent bus and records it for cleanup.

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.