events

Documentation

Index

• Constants
• func SessionScopedMetadata(base map[string]string, itemID string) map[string]string
• type BacklogChangeKind
• type BacklogItemEventPayload
• type Event
  • func NewApprovalResponseEvent(sessionID string, approved bool, context string) *Event
  • func NewBacklogItemChangedEvent(payload *BacklogItemEventPayload) *Event
  • func NewNotificationEvent(sessionID string, sessionName string, notificationID string, ...) *Event
  • func NewRemoteHealthChangedEvent(remoteName string, state, previousState sshremote.RemoteConnectionState) *Event
  • func NewSessionAcknowledgedEvent(sessionID, reason string) *Event
  • func NewSessionCreatedEvent(sess *session.Instance) *Event
  • func NewSessionDeletedEvent(sessionID string) *Event
  • func NewSessionUpdatedEvent(sess *session.Instance, updatedFields []string) *Event
  • func NewSessionUpdatedEventWithDetection(sess *session.Instance, updatedFields []string, ...) *Event
  • func NewUserInteractionEvent(sessionID, interactionType, context string) *Event
• type EventBus
  • func NewEventBus(bufferSize int) *EventBus
  • func (eb *EventBus) Close()
  • func (eb *EventBus) EventsSince(afterSeq uint64) []*Event
  • func (eb *EventBus) Publish(event *Event)
  • func (eb *EventBus) Subscribe(ctx context.Context) (<-chan *Event, string)
  • func (eb *EventBus) SubscriberCount() int
  • func (eb *EventBus) Unsubscribe(id string)
• type EventType
• type RemoteHealthEventPayload
• type Subscriber
  • func NewSubscriber(id string, events <-chan *Event, cleanup func()) *Subscriber
  • func (s *Subscriber) Close()

Constants

const (
	// MetadataKeySessionScoped marks a notification as originating from a
	// specific session (as opposed to a backlog-item-level or global
	// notification with no associated session).
	MetadataKeySessionScoped = "session_scoped"
	// MetadataKeyItemID carries the backlog item ID a session-scoped
	// notification's session is linked to, when known.
	MetadataKeyItemID = "item_id"
)

Metadata keys stamped onto session-scoped notifications (see SessionScopedMetadata). Shared here so producers (server/review_queue_manager.go) and consumers (e.g. server/notifications) agree on the exact key strings.

Functions

func SessionScopedMetadata

func SessionScopedMetadata(base map[string]string, itemID string) map[string]string

SessionScopedMetadata builds a fresh metadata map for a session-scoped notification, copying any entries from base (never mutating base — base may be a *ReviewItem.Metadata map stored unlocked and concurrently read by other goroutines, e.g. WatchReviewQueue's ReviewItemToProto) and adding the session_scoped marker plus item_id when non-empty.

Types

type BacklogChangeKind

type BacklogChangeKind string

BacklogChangeKind identifies which kind of backlog item mutation a BacklogItemEventPayload describes.

const (
	// BacklogChangeStatusTransition is emitted when an item's status changes.
	BacklogChangeStatusTransition BacklogChangeKind = "status_transition"
	// BacklogChangeVerdictRecorded is emitted when a review verdict is saved.
	BacklogChangeVerdictRecorded BacklogChangeKind = "verdict_recorded"
	// BacklogChangeSessionAttached is emitted when a session is attached to an item.
	BacklogChangeSessionAttached BacklogChangeKind = "session_attached"
	// BacklogChangeItemUpdated is emitted when item fields (title, description, etc.) change.
	BacklogChangeItemUpdated BacklogChangeKind = "item_updated"
	// BacklogChangeItemArchived is emitted when an item is archived.
	BacklogChangeItemArchived BacklogChangeKind = "item_archived"
	// BacklogChangeItemRemoved is emitted when an item is deleted.
	BacklogChangeItemRemoved BacklogChangeKind = "item_removed"
	// BacklogChangeTriageProgressUpdated is emitted when in-flight triage progress
	// is written (UpdateItemSessionTriageResult). Converts to the existing
	// BacklogItemUpdatedEvent oneof variant on the wire, not a new proto message.
	BacklogChangeTriageProgressUpdated BacklogChangeKind = "triage_progress_updated"
	// BacklogChangeActivityNoteAdded is emitted when a free-form activity note
	// is posted (AppendActivityNote, ADR-001's sibling table). Converts to the
	// dedicated BacklogItemActivityNoteAddedEvent oneof variant, never a full
	// item snapshot (ADR-002).
	BacklogChangeActivityNoteAdded BacklogChangeKind = "activity_note_added"
)

type BacklogItemEventPayload

type BacklogItemEventPayload struct {
	// Kind identifies which backlog mutation this payload describes.
	Kind BacklogChangeKind
	// Item is the current snapshot of the backlog item after the mutation.
	Item *session.BacklogItemData
	// OldStatus is the prior status for BacklogChangeStatusTransition.
	OldStatus string
	// NewStatus is the new status for BacklogChangeStatusTransition.
	NewStatus string
	// UpdatedFields lists which fields changed for BacklogChangeItemUpdated
	// (and BacklogChangeTriageProgressUpdated).
	UpdatedFields []string
	// SessionID identifies the session for BacklogChangeSessionAttached.
	SessionID string
	// ClaimantHostID is the attaching process's own stable host identifier for
	// BacklogChangeSessionAttached, mirrored from session.BacklogItemChange —
	// never derived from the session being attached.
	ClaimantHostID string
	// ArchivedAt is the archival timestamp for BacklogChangeItemArchived.
	ArchivedAt *time.Time
	// RemovedReason describes why an item was removed for BacklogChangeItemRemoved.
	RemovedReason string
	// Verdict mirrors BacklogItemChange.Verdict one-to-one; populated only
	// when Kind == BacklogChangeVerdictRecorded, copied straight through by
	// the adapter so the verdict reaches subscribers as first-class payload
	// data rather than something derived by joining item_sessions.
	Verdict *session.ReviewVerdictData
	// ActivityNote mirrors session.BacklogItemChange.ActivityNote one-to-one;
	// populated only when Kind == BacklogChangeActivityNoteAdded.
	ActivityNote *session.ActivityNoteData
	// IsSnapshot is true when this event was generated as part of an
	// initial-snapshot send (e.g. WatchBacklogItems's first batch) rather
	// than a live mutation.
	IsSnapshot bool
}

BacklogItemEventPayload carries the backlog-specific data for an EventBacklogItemChanged event. Only the fields relevant to Kind are expected to be populated.

type Event

type Event struct {
	// Seq is assigned by the EventBus when Publish is called. Zero means unpublished.
	Seq uint64
	// Type of the event
	Type EventType
	// Timestamp when the event occurred
	Timestamp time.Time
	// Session affected by the event (may be nil for delete events)
	Session *session.Instance
	// SessionID for delete events when Session is nil
	SessionID string
	// UpdatedFields tracks which fields were modified (for update events)
	UpdatedFields []string
	// OldStatus for status change events
	OldStatus session.Status
	// NewStatus for status change events
	NewStatus session.Status
	// DetectedStatus is kept for legacy compatibility; no longer serialized to wire.
	DetectedStatus string
	// DetectedContext is the human-readable context from the terminal detector
	// (e.g. "Waiting for tool approval"). Empty when DetectedStatus is empty.
	DetectedContext string
	// DetectedStatusTyped is the typed DetectedStatus for SessionUpdatedEvent.
	// Zero value (detection.StatusUnknown) means no detection data is available.
	DetectedStatusTyped detection.DetectedStatus
	// InteractionType for user interaction events
	InteractionType string
	// Approved for approval response events (true = approved, false = denied)
	Approved bool
	// Context provides additional context about the event
	Context string
	// Notification fields for notification events
	NotificationID       string
	NotificationType     int32 // Maps to sessionv1.NotificationType
	NotificationPriority int32 // Maps to sessionv1.NotificationPriority
	NotificationTitle    string
	NotificationMessage  string
	NotificationMetadata map[string]string
	// BacklogItemPayload carries backlog item change data for
	// EventBacklogItemChanged events. Nil for all other event types.
	BacklogItemPayload *BacklogItemEventPayload
	// RemoteHealthPayload carries remote connection-health transition data
	// for EventRemoteHealthChanged events. Nil for all other event types.
	RemoteHealthPayload *RemoteHealthEventPayload
}

Event represents a session state change event. This is the internal Go representation that will be converted to protobuf events.

func NewApprovalResponseEvent

func NewApprovalResponseEvent(sessionID string, approved bool, context string) *Event

NewApprovalResponseEvent creates an event for approval responses.

func NewBacklogItemChangedEvent

func NewBacklogItemChangedEvent(payload *BacklogItemEventPayload) *Event

NewBacklogItemChangedEvent creates an event for a backlog item mutation.

func NewNotificationEvent

func NewNotificationEvent(
	sessionID string,
	sessionName string,
	notificationID string,
	notificationType int32,
	priority int32,
	title string,
	message string,
	metadata map[string]string,
) *Event

NewNotificationEvent creates an event for session notifications.

func NewRemoteHealthChangedEvent

func NewRemoteHealthChangedEvent(remoteName string, state, previousState sshremote.RemoteConnectionState) *Event

NewRemoteHealthChangedEvent creates an event for a configured remote's SSH connection health-state transition (session/sshremote. RemoteHealthProber, Epic 6.4). state is the new state; previousState is the state immediately prior to this transition.

func NewSessionAcknowledgedEvent

func NewSessionAcknowledgedEvent(sessionID, reason string) *Event

NewSessionAcknowledgedEvent creates an event for session acknowledgments.

func NewSessionCreatedEvent

func NewSessionCreatedEvent(sess *session.Instance) *Event

NewSessionCreatedEvent creates an event for session creation.

func NewSessionDeletedEvent

func NewSessionDeletedEvent(sessionID string) *Event

NewSessionDeletedEvent creates an event for session deletion.

func NewSessionUpdatedEvent

func NewSessionUpdatedEvent(sess *session.Instance, updatedFields []string) *Event

NewSessionUpdatedEvent creates an event for session updates.

func NewSessionUpdatedEventWithDetection

func NewSessionUpdatedEventWithDetection(
	sess *session.Instance,
	updatedFields []string,
	detectedStatus detection.DetectedStatus,
	detectedContext string,
) *Event

NewSessionUpdatedEventWithDetection creates a session update event that includes typed detected-status information from the terminal detection layer. Use this instead of NewSessionUpdatedEvent when the detection state is known and should be propagated to frontend clients (e.g. the UpdateSession RPC path).

func NewUserInteractionEvent

func NewUserInteractionEvent(sessionID, interactionType, context string) *Event

NewUserInteractionEvent creates an event for user interactions.

type EventBus

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

EventBus provides a thread-safe pub/sub event bus for session events. It uses Go channels for event distribution and supports multiple concurrent subscribers.

func NewEventBus

func NewEventBus(bufferSize int) *EventBus

NewEventBus creates a new event bus with the specified buffer size. Buffer size determines how many events can be queued per subscriber before dropping.

func (*EventBus) Close

func (eb *EventBus) Close()

Close unsubscribes all subscribers and closes their channels. Should be called during graceful shutdown.

func (*EventBus) EventsSince

func (eb *EventBus) EventsSince(afterSeq uint64) []*Event

EventsSince returns buffered events with Seq > afterSeq in ascending order. Returns nil when afterSeq is 0 (no replay requested) or all buffered events are at or below afterSeq. Events older than one hour are not available.

func (*EventBus) Publish

func (eb *EventBus) Publish(event *Event)

Publish assigns a sequence number to the event, appends it to the ring buffer, then broadcasts it to all active subscribers. Events are sent asynchronously and non-blocking. If a subscriber's buffer is full, the event is dropped for that subscriber to prevent blocking other subscribers.

func (*EventBus) Subscribe

func (eb *EventBus) Subscribe(ctx context.Context) (<-chan *Event, string)

Subscribe creates a new subscription to the event bus. Returns a read-only channel for receiving events and a subscription ID for cleanup. The subscription is automatically cleaned up when the context is canceled.

func (*EventBus) SubscriberCount

func (eb *EventBus) SubscriberCount() int

SubscriberCount returns the current number of active subscribers. Useful for monitoring and testing.

func (*EventBus) Unsubscribe

func (eb *EventBus) Unsubscribe(id string)

Unsubscribe removes a subscriber and closes their channel. This is idempotent - calling it multiple times with the same ID is safe.

type EventType

type EventType string

EventType represents the type of session event that occurred.

const (
	// EventSessionCreated is emitted when a new session is created
	EventSessionCreated EventType = "session.created"
	// EventSessionUpdated is emitted when session properties are modified
	EventSessionUpdated EventType = "session.updated"
	// EventSessionDeleted is emitted when a session is deleted
	EventSessionDeleted EventType = "session.deleted"
	// EventUserInteraction is emitted when user interacts with a session
	EventUserInteraction EventType = "session.user_interaction"
	// EventSessionAcknowledged is emitted when user acknowledges a session
	EventSessionAcknowledged EventType = "session.acknowledged"
	// EventApprovalResponse is emitted when user responds to an approval prompt
	EventApprovalResponse EventType = "session.approval_response"
	// EventNotification is emitted when a session sends a notification
	EventNotification EventType = "session.notification"
	// EventBacklogItemChanged is emitted when a backlog item is mutated
	// (status transition, verdict recorded, session attached, item updated,
	// archived, removed, or triage progress updated).
	EventBacklogItemChanged EventType = "backlog_item_changed"
	// EventRemoteHealthChanged is emitted when a configured remote's SSH
	// connection health transitions between connected/reconnecting/
	// disconnected (session/sshremote.RemoteHealthProber, Epic 6.4).
	EventRemoteHealthChanged EventType = "remote.health_changed"
)

type RemoteHealthEventPayload

type RemoteHealthEventPayload struct {
	// RemoteName is the config.RemoteConfig.Name this health transition
	// applies to.
	RemoteName string
	// State is the remote's connection state as of this event.
	State sshremote.RemoteConnectionState
	// PreviousState is the state immediately before this transition.
	PreviousState sshremote.RemoteConnectionState
}

RemoteHealthEventPayload carries the remote-specific data for an EventRemoteHealthChanged event (session/sshremote.RemoteHealthProber, Epic 6.4).

type Subscriber

type Subscriber struct {
	ID     string
	Events <-chan *Event
	// contains filtered or unexported fields
}

Subscriber represents an active event bus subscription. Provides a convenient wrapper around the channel and subscription ID.

func NewSubscriber

func NewSubscriber(id string, events <-chan *Event, cleanup func()) *Subscriber

NewSubscriber creates a Subscriber wrapper. This is primarily for convenience and testing.

func (*Subscriber) Close

func (s *Subscriber) Close()

Close unsubscribes and cleans up the subscriber.