events

type Bus ¶

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

Bus is a concurrency-safe in-process pub/sub bus. The zero value is not usable; construct with NewBus.

func NewBus() *Bus

func (b *Bus) Close()

func (b *Bus) Dropped() int64

func (b *Bus) Publish(topic string, payload any)

func (b *Bus) Subscribe(topic string) (<-chan Event, func())

func (b *Bus) SubscribeBuffered(topic string, buffer int) (<-chan Event, func())

type Event ¶

type Event struct {
	Topic     string
	Timestamp time.Time
	Payload   any
}

Event is a single message delivered to subscribers of a topic. Payload is topic-specific and opaque to the bus; subscribers type-assert it.

type EventBus ¶ added in v0.5.0

type EventBus interface {
	// Publish delivers payload to all current subscribers of topic.
	// Non-blocking for the publisher (G3); MAY drop events for a slow
	// subscriber per the bounded-buffer / drop-oldest policy. Publishing
	// on a topic with no subscribers succeeds. Returns an error
	// satisfying errors.Is(err, ErrClosed) after Close.
	Publish(topic string, payload any) error
	// Subscribe returns a receive-only stream for topic plus an
	// unsubscribe func. The buffer is bounded; drop-oldest on overflow.
	// The unsubscribe func closes the stream and is safe to call more
	// than once. Returns an error satisfying errors.Is(err, ErrClosed)
	// after Close.
	Subscribe(topic string) (<-chan Event, func(), error)
	// Close drains and releases backend resources, closing every
	// subscriber stream. Close is idempotent.
	Close() error
}

EventBus is the transport seam for service events (spec D4.1). Publishers and subscribers — the scheduler tasks, the transport layer, R2's SSE fan-out, R5's collection endpoint — bind to this interface, never to the concrete channel implementation, so an external backend (Kafka, NATS, Redis) can later be swapped in as a config-selected adapter without a rewrite of any publisher or subscriber.

The interface promises ONLY the D4.2 floor; callers MUST NOT assume any stronger guarantee a particular backend happens to provide:

• G1 — at-most-once, best-effort delivery. Events are "wake up and look at the new state on disk"; a missed event is recovered by re-reading the canonical sidecar, never from the bus.
• G2 — per-topic ordering is best-effort, not guaranteed. Code that needs strict order re-derives it from sidecar state.
• G3 — backpressure is a bounded buffer with drop-oldest; publishers never block on a slow subscriber.
• G4 — no cross-publish atomicity; every Publish is an independent fire-and-forget.

Backends are free to be stronger; the floor is what app code may assume. Conformance is proven mechanically by the eventbustest suite (spec D4.6).

type InProcBus ¶ added in v0.5.0

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

InProcBus adapts the channel-based Bus into the builtin EventBus backend (spec D4.3, dotagents-builtin:eventbus/inproc). It is the default — and in v1 the only — backend: bounded buffer per subscriber, drop-oldest on slow consumers, no durable storage, no cross-process IPC.

The adapter adds the error-reporting shape the D4.1 seam requires on top of the concrete Bus: after Close, Publish and Subscribe return ErrClosed instead of silently no-opping.

func NewInProcBus() *InProcBus

func (p *InProcBus) Close() error

func (p *InProcBus) Dropped() int64

func (p *InProcBus) Publish(topic string, payload any) error

func (p *InProcBus) Subscribe(topic string) (<-chan Event, func(), error)

type IterationScored ¶ added in v0.5.0

type IterationScored struct {
	Iteration   int     `json:"iter"`
	Score       float64 `json:"score"`
	Band        string  `json:"band"`
	SidecarPath string  `json:"sidecar_path"`
}

IterationScored is the payload published on TopicIterationScored after an iteration log entry has been scored and its score sidecar written. Subscribers treat it as a wake-up: the sidecar at SidecarPath is the canonical state (guarantee G1 — a dropped event is recovered by re-reading disk, never from the bus).

type RescoreDone ¶ added in v0.5.0

type RescoreDone struct {
	FromVersion string `json:"from_version"`
	ToVersion   string `json:"to_version"`
	IterCount   int    `json:"iter_count"`
}

RescoreDone is the payload published on TopicRescoreDone after a rubric version bump has driven a full-log rescore and the refreshed per-iteration and per-session sidecars are on disk. Subscribers treat it as a wake-up: the sidecars (and the rescore watermark, persisted before this publish) are the canonical state per guarantee G1.