events

Documentation

Overview

Package events defines the shared rocketclaw event bus.

Index

• Variables
• func AttachmentNamesSpeech(attachments []OutboundAttachment) string
• func MainConversationID() string
• type AudioChunk
• type Bus
  • func New(configs ...Config) *Bus
  • func (b *Bus) Audio(ctx context.Context) iter.Seq[*AudioChunk]
  • func (b *Bus) Close()
  • func (b *Bus) Inbound(ctx context.Context) iter.Seq[*InboundMessage]
  • func (b *Bus) Outbound(ctx context.Context) iter.Seq[*OutboundMessage]
  • func (b *Bus) PublishAudio(ctx context.Context, chunk *AudioChunk) error
  • func (b *Bus) PublishInbound(ctx context.Context, msg *InboundMessage) error
  • func (b *Bus) PublishOutbound(ctx context.Context, msg *OutboundMessage) error
  • func (b *Bus) StopInbound()
  • func (b *Bus) WaitInboundDequeued(ctx context.Context) error
  • func (b *Bus) WaitOutboundIdle(ctx context.Context) error
• type Config
• type DiscordReplyTarget
• type InboundAttachment
• type InboundKind
• type InboundMessage
  • func NewMainInboundMessage(source Source, kind InboundKind, label, text string, human bool) *InboundMessage
  • func (m *InboundMessage) CompleteResponse(text string, err error)
  • func (m *InboundMessage) EnableResponseWait() <-chan InboundResponse
• type InboundResponse
• type OutboundAttachment
  • func CloneOutboundAttachments(attachments []OutboundAttachment) []OutboundAttachment
• type OutboundMessage
  • func NewMainOutboundMessage(source Source, text string, targets ...OutputTarget) *OutboundMessage
  • func (m *OutboundMessage) MarkDelivered(err error)
  • func (m *OutboundMessage) WaitDelivered(ctx context.Context) error
• type OutputTarget
  • func MainOutputTargets() []OutputTarget
• type ResponseCheckpoint
• type SlackReplyTarget
• type Source

Variables

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

ErrBusClosed reports that an event was published after the bus shut down.

Functions

func AttachmentNamesSpeech

func AttachmentNamesSpeech(attachments []OutboundAttachment) string

AttachmentNamesSpeech returns a short spoken description of attachment names.

func MainConversationID

func MainConversationID() string

MainConversationID returns the stable key for the shared main session.

Types

type AudioChunk

type AudioChunk struct {
	SessionID, SpeakerID string
	Source               Source
	RTPSequence          uint16
	Timestamp, SSRC      uint32
	SampleRate, Channels int
	Format               string
	Data                 []byte
}

AudioChunk carries a connector audio frame into the transcription pipeline.

type Bus

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

Bus routes inbound text, outbound text, and audio events between components.

func New

func New(configs ...Config) *Bus

New constructs an event bus.

func (*Bus) Audio

func (b *Bus) Audio(ctx context.Context) iter.Seq[*AudioChunk]

Audio returns a single-use iterator over inbound audio chunks.

func (*Bus) Close

func (b *Bus) Close()

Close shuts down the bus and wakes all waiting consumers.

func (*Bus) Inbound

func (b *Bus) Inbound(ctx context.Context) iter.Seq[*InboundMessage]

Inbound returns a single-use iterator over inbound text messages.

func (*Bus) Outbound

func (b *Bus) Outbound(ctx context.Context) iter.Seq[*OutboundMessage]

Outbound returns a single-use iterator over outbound text messages.

func (*Bus) PublishAudio

func (b *Bus) PublishAudio(ctx context.Context, chunk *AudioChunk) error

PublishAudio publishes an audio chunk into the voice pipeline.

func (*Bus) PublishInbound

func (b *Bus) PublishInbound(ctx context.Context, msg *InboundMessage) error

PublishInbound publishes a text message into the shared input queue.

func (*Bus) PublishOutbound

func (b *Bus) PublishOutbound(ctx context.Context, msg *OutboundMessage) error

PublishOutbound publishes a text message to all output sinks.

func (*Bus) StopInbound

func (b *Bus) StopInbound()

StopInbound stops new inbound messages while allowing accepted messages to be dequeued.

func (*Bus) WaitInboundDequeued

func (b *Bus) WaitInboundDequeued(ctx context.Context) error

WaitInboundDequeued waits for accepted inbound work to leave the bus queues.

func (*Bus) WaitOutboundIdle

func (b *Bus) WaitOutboundIdle(ctx context.Context) error

WaitOutboundIdle waits until outbound work is queued nowhere and delivered everywhere.

type Config

type Config struct {
	MinimumWaitAfterHumanInteraction time.Duration
}

Config controls event bus behavior.

type DiscordReplyTarget

type DiscordReplyTarget struct {
	ChannelID, MessageID, ThreadID string
}

DiscordReplyTarget identifies the Discord message or thread that owns a streamed reply.

type InboundAttachment

type InboundAttachment struct {
	Name     string
	MIMEType string
	Data     []byte
}

InboundAttachment carries an inline attachment into the shared main-session prompt.

type InboundKind

type InboundKind string

InboundKind describes how an inbound message should be handled.

const (
	// InboundKindPrompt is a normal conversational prompt.
	InboundKindPrompt InboundKind = "prompt"
	// InboundKindInternalize is a note the session should absorb without replying.
	InboundKindInternalize InboundKind = "internalize"
)

type InboundMessage

type InboundMessage struct {
	Source                       Source
	Label, Text                  string
	VerbatimMessage              string
	VerbatimAttachments          []OutboundAttachment
	Attachments                  []InboundAttachment
	SlackReply                   *SlackReplyTarget
	DiscordReply                 *DiscordReplyTarget
	HadAttachments               bool
	HadNonImageAttachments       bool
	AttachmentWarnings           []string
	Human                        bool
	Kind                         InboundKind
	ConversationID, WebSessionID string
	Metadata                     map[string]string
	// contains filtered or unexported fields
}

InboundMessage is a message headed into the shared main-session prompt queue.

func NewMainInboundMessage

func NewMainInboundMessage(source Source, kind InboundKind, label, text string, human bool) *InboundMessage

NewMainInboundMessage constructs a message for the shared main session.

func (*InboundMessage) CompleteResponse

func (m *InboundMessage) CompleteResponse(text string, err error)

CompleteResponse marks this inbound turn result ready.

func (*InboundMessage) EnableResponseWait

func (m *InboundMessage) EnableResponseWait() <-chan InboundResponse

EnableResponseWait returns a channel that receives the final result for this inbound turn.

type InboundResponse

type InboundResponse struct {
	Text string
	Err  error
}

InboundResponse is the final plain-text result for a queued inbound turn.

type OutboundAttachment

type OutboundAttachment struct {
	Name     string
	MIMEType string
	Data     []byte
}

OutboundAttachment carries a human-visible file attachment to output sinks.

func CloneOutboundAttachments

func CloneOutboundAttachments(attachments []OutboundAttachment) []OutboundAttachment

CloneOutboundAttachments returns a deep copy of attachments.

type OutboundMessage

type OutboundMessage struct {
	Text, SlackThinking                  string
	SlackPostText                        bool
	Source                               Source
	Targets                              []OutputTarget
	ConversationID, TurnID, WebSessionID string
	Sequence                             int
	Complete                             bool
	SlackReply                           *SlackReplyTarget
	DiscordReply                         *DiscordReplyTarget
	Checkpoint                           *ResponseCheckpoint
	Attachments                          []OutboundAttachment
	// contains filtered or unexported fields
}

OutboundMessage is a text message headed to enabled connectors.

func NewMainOutboundMessage

func NewMainOutboundMessage(source Source, text string, targets ...OutputTarget) *OutboundMessage

NewMainOutboundMessage constructs an outbound message for the shared main session.

func (*OutboundMessage) MarkDelivered

func (m *OutboundMessage) MarkDelivered(err error)

MarkDelivered marks outbound delivery for this message complete.

func (*OutboundMessage) WaitDelivered

func (m *OutboundMessage) WaitDelivered(ctx context.Context) error

WaitDelivered waits until outbound delivery for this message finishes.

type OutputTarget

type OutputTarget string

OutputTarget identifies which connector should receive an outbound message.

const (
	// OutputTargetSlackMain delivers a response to the main Slack DM.
	OutputTargetSlackMain OutputTarget = "slack_main"
	// OutputTargetDiscordText delivers a response to Discord text.
	OutputTargetDiscordText OutputTarget = "discord_text"
	// OutputTargetDiscord delivers a response to Discord voice.
	OutputTargetDiscord OutputTarget = "discord"
	// OutputTargetWebUI delivers a response to the browser voice-mode client.
	OutputTargetWebUI OutputTarget = "web_ui"
)

func MainOutputTargets

func MainOutputTargets() []OutputTarget

MainOutputTargets returns the default targets for main-session replies.

type ResponseCheckpoint

type ResponseCheckpoint struct {
	ConversationID string
	SessionEntryID int64
	ResponseID     string
	Model          string
	AssistantText  string
}

ResponseCheckpoint identifies a persisted main-session turn that can seed a Slack thread.

type SlackReplyTarget

type SlackReplyTarget struct {
	ChannelID, MessageTS, ThreadTS string
}

SlackReplyTarget identifies the Slack message that owns a streamed reply.

type Source

type Source string

Source identifies where an inbound or outbound message originated.

const (
	SourceSlack        Source = "slack"
	SourceDiscordText  Source = "discord_text"
	SourceDiscordVoice Source = "discord_voice"
	SourceExternalMCP  Source = "external_mcp"
	SourceWebVoice     Source = "web_voice"
	SourceSystem       Source = "system"
)

Known inbound and outbound message source labels.