event

Documentation

Index

• Variables
• func Message(payload any) (string, bool)
• type Bus
  • func NewBus() *Bus
  • func (b *Bus) Close() error
  • func (b *Bus) Publish(ctx context.Context, event Event) error
  • func (b *Bus) Subscribe(eventType Type, buffer int) (Subscription, error)
  • func (b *Bus) SubscribeExecution(executionID core.ID, buffer int) (Subscription, error)
• type CapabilityCancelledPayload
  • func (p CapabilityCancelledPayload) Reason() string
• type CapabilityCompletedPayload
• type CapabilityDetachedPayload
• type CapabilityFailedPayload
  • func (p CapabilityFailedPayload) Reason() string
• type CapabilityReadyPayload
• type CapabilityRetryPayload
• type CapabilityStartedPayload
• type Event
  • func New(eventType Type, executionID, correlationID, capabilityID core.ID, payload any) Event
• type ExecutionCancelledPayload
  • func (p ExecutionCancelledPayload) Reason() string
• type ExecutionCompletedPayload
• type ExecutionFailedPayload
  • func (p ExecutionFailedPayload) Reason() string
• type ExecutionStartedPayload
• type Field
• type LogLevel
• type LogPayload
• type Metadata
• type Store
  • func NewStore(store storage.Store) *Store
  • func (s *Store) DeleteExecution(ctx context.Context, executionID core.ID) error
  • func (s *Store) List(ctx context.Context, executionID core.ID) ([]Event, error)
  • func (s *Store) ListAfter(ctx context.Context, executionID core.ID, afterEventID core.ID) ([]Event, error)
  • func (s *Store) Save(ctx context.Context, evt Event) error
• type Subscription
• type Type
  • func (t Type) IsTerminal() bool
  • func (t Type) String() string

Variables

var ErrBusClosed = errors.New("event bus is closed")

Functions

func Message

func Message(payload any) (string, bool)

Message extracts the human-readable reason from an event payload, reporting whether one was present.

Payloads travel through an `any` field, so a caller that asserted one concrete type silently discarded the reason of every other form it might legally arrive in -- a pointer, or a sibling payload type -- and substituted a generic message. That hides the actual cause from whoever has to diagnose the run, which is the opposite of what a reason is for. Keeping the extraction here also keeps the several consumers from disagreeing about what a payload means.

A reason that is present but empty is still present, and is returned as an empty string: a caller must be able to distinguish "carries no reason" from "no reason was given", because only the first warrants substituting text.

Callers keep their own fallback for the absent case, since the scheduler and the structured log deliberately report it differently.

Types

type Bus

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

func NewBus

func NewBus() *Bus

func (*Bus) Close

func (b *Bus) Close() error

func (*Bus) Publish

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

func (*Bus) Subscribe

func (b *Bus) Subscribe(eventType Type, buffer int) (Subscription, error)

func (*Bus) SubscribeExecution

func (b *Bus) SubscribeExecution(executionID core.ID, buffer int) (Subscription, error)

SubscribeExecution delivers every event belonging to the given execution ID.

type CapabilityCancelledPayload

type CapabilityCancelledPayload struct{ Message string }

CapabilityCancelledPayload explains why a capability stopped without reaching an outcome of its own. The message names the cause that ended it, not a fault in the capability.

func (CapabilityCancelledPayload) Reason

func (p CapabilityCancelledPayload) Reason() string

type CapabilityCompletedPayload

type CapabilityCompletedPayload struct{ Result map[string]any }

type CapabilityDetachedPayload

type CapabilityDetachedPayload struct{}

CapabilityDetachedPayload records a lifecycle boundary, so it carries no data of its own. The capability ID on the event identifies the boundary, and the task that now owns the work is reachable from the parent execution through the execution store.

type CapabilityFailedPayload

type CapabilityFailedPayload struct{ Message string }

CapabilityFailedPayload carries the reason a capability failed on its own account. It is distinct from CapabilityCancelledPayload because the two mean opposite things to whoever reacts to them: one is a fault to investigate, the other is work that was stopped for a reason already accounted for.

func (CapabilityFailedPayload) Reason

func (p CapabilityFailedPayload) Reason() string

type CapabilityReadyPayload

type CapabilityReadyPayload struct{ Params map[string]any }

type CapabilityRetryPayload

type CapabilityRetryPayload struct {
	// Attempt is the attempt that just failed.
	Attempt int

	// NextAttempt is the attempt about to be made.
	NextAttempt int

	// Delay is how long N.O.R.E. waited before NextAttempt.
	Delay string

	// Message is the failure that caused the retry.
	Message string
}

CapabilityRetryPayload reports a scheduled re-attempt of a capability invocation so an observer can see that a capability is still being worked on rather than stalled.

type CapabilityStartedPayload

type CapabilityStartedPayload struct{}

type Event

type Event struct {
	Type     Type
	Metadata Metadata
	Payload  any
}

func New

func New(eventType Type, executionID, correlationID, capabilityID core.ID, payload any) Event

type ExecutionCancelledPayload

type ExecutionCancelledPayload struct{ Message string }

ExecutionCancelledPayload carries why an execution was cancelled. The message is empty when the cancellation needs no explanation, so a client must treat it as optional rather than as a missing value.

func (ExecutionCancelledPayload) Reason

func (p ExecutionCancelledPayload) Reason() string

type ExecutionCompletedPayload

type ExecutionCompletedPayload struct {
	// Result is the Assembly result object produced from the blueprint's result
	// contract. It contains only the fields declared in the Assembly result
	// mapping, not the full capability result map.
	Result map[string]any
}

type ExecutionFailedPayload

type ExecutionFailedPayload struct{ Message string }

func (ExecutionFailedPayload) Reason

func (p ExecutionFailedPayload) Reason() string

type ExecutionStartedPayload

type ExecutionStartedPayload struct{ Params map[string]any }

type Field

type Field struct {
	Key   string
	Value any
}

Field is a single structured key/value pair on a log line.

type LogLevel

type LogLevel string

LogLevel is the severity of a CapabilityLog payload.

const (
	LogDebug LogLevel = "debug"
	LogInfo  LogLevel = "info"
	LogWarn  LogLevel = "warn"
	LogError LogLevel = "error"
)

type LogPayload

type LogPayload struct {
	Level   LogLevel
	Message string
	NodeID  core.ID
	Fields  []Field
}

LogPayload is carried by CapabilityLog events and represents one structured diagnostic line emitted by an executing capability. It is what capability runtimes should use instead of writing directly to stdout.

type Metadata

type Metadata struct {
	EventID       core.ID
	CorrelationID core.ID
	ExecutionID   core.ID
	CapabilityID  core.ID
	OccurredAt    time.Time
}

type Store

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

func NewStore

func NewStore(store storage.Store) *Store

func (*Store) DeleteExecution

func (s *Store) DeleteExecution(ctx context.Context, executionID core.ID) error

DeleteExecution removes every persisted event recorded for the execution.

func (*Store) List

func (s *Store) List(ctx context.Context, executionID core.ID) ([]Event, error)

func (*Store) ListAfter

func (s *Store) ListAfter(ctx context.Context, executionID core.ID, afterEventID core.ID) ([]Event, error)

func (*Store) Save

func (s *Store) Save(ctx context.Context, evt Event) error

type Subscription

type Subscription interface {
	Events() <-chan Event
	Close() error
}

type Type

type Type uint16

Type identifies an execution lifecycle event.

The numeric values are part of the persisted event format: the event store writes them as-is, so a type may only ever be appended. Reordering or removing one would silently reinterpret events already on disk.

const (
	Unknown Type = iota
	ExecutionStarted
	ExecutionCompleted
	ExecutionFailed
	ExecutionCancelled
	CapabilityReady
	CapabilityStarted
	CapabilityCompleted
	CapabilityFailed
	CapabilityLog
	// CapabilityDetached records that a capability's work was handed off to a
	// separate execution. It ends this capability's participation in the
	// execution but is not an execution-terminal event: the work itself
	// continues under the task created for it.
	CapabilityDetached
	// CapabilityRetry records a scheduled re-attempt of one capability
	// invocation. Retries are internal to a single capability execution, so
	// this is progress rather than a new execution.
	CapabilityRetry
	// CapabilityCancelled records that a capability stopped without reaching an
	// outcome of its own, because something outside it ended first: the
	// execution was cancelled, or a sibling capability failed.
	//
	// It exists so that distinction survives to consumers. Without it a
	// capability stopped as a consequence of another failure was published as a
	// failure of its own, so a correct implementation was reported as broken and
	// a client could not tell the two apart without reading the message text.
	// This is not an execution-terminal event; the execution's own terminal event
	// already records why everything stopped.
	CapabilityCancelled
)

const All Type = 0xffff

func (Type) IsTerminal

func (t Type) IsTerminal() bool

IsTerminal reports whether the type ends an execution. A terminal event is the last event an execution produces, so consumers that only wait for the outcome of an execution can stop once one has been delivered.

func (Type) String

func (t Type) String() string