events

type AdminScopeUsedPayload ¶

type AdminScopeUsedPayload struct {
	SafeSealed
	Tenant       string
	User         string
	Session      string
	SubscriberID uint64
}

AdminScopeUsedPayload is emitted whenever Subscribe is called with Filter.Admin=true, regardless of whether the triple is empty or partially specified. Surfaces admin-scope use for after-the-fact auditability — Phase 61 will additionally enforce a cryptographic scope claim, but the audit emit itself is Phase 05's contribution.

type Aggregator ¶

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

Aggregator is the compiled artifact that produces time-bucketed event-type counts over a window. It is a reusable artifact in the D-025 sense: bus and clock are set once at construction and never mutated; Aggregate() holds no per-request state on the Aggregator (each request creates its own buckets slice). One Aggregator serves N concurrent requests safely (concurrent_test.go pins it under -race).

The aggregator consumes the bus's Replayer surface for the historical snapshot — runtime-side aggregation per brief 11 §CC-4 ("events are high-cardinality runtime-side; the runtime owns the index and exposes a search method"). When the bus does not implement Replayer (a forward-only driver), Aggregate returns ErrReplayUnavailable — fail loudly, never an empty series that looks like "no events."

func NewAggregator(bus EventBus, opts ...AggregatorOption) (*Aggregator, error)

func (a *Aggregator) Aggregate(ctx context.Context, req prototypes.EventAggregateRequest) (prototypes.EventAggregateResponse, error)

type AggregatorClock ¶

type AggregatorClock interface {
	Now() time.Time
}

AggregatorClock abstracts the runtime's notion of "now" so the aggregator can be tested with a fake clock. Production callers pass nil and the aggregator falls back to time.Now (UTC). The interface is intentionally narrow: only Now() — the aggregator never needs to tick. Mirrors the inmem-driver Clock seam (no second clock vocabulary added).

type AggregatorOption ¶

type AggregatorOption func(*Aggregator)

AggregatorOption configures NewAggregator at construction.

func WithAggregatorClock(c AggregatorClock) AggregatorOption

type AuditRedactionFailedPayload ¶

type AuditRedactionFailedPayload struct {
	SafeSealed
	OriginalType EventType
	Reason       string
}

AuditRedactionFailedPayload reports a Publish call whose payload the audit redactor refused. Carries the failing event's type and identity but NO original payload bytes — the failure is observable without leaking the bytes the redactor refused.

type BusDroppedPayload ¶

type BusDroppedPayload struct {
	SafeSealed
	FromSeq      uint64
	ToSeq        uint64
	DroppedCount uint64
	SubscriberID uint64
}

BusDroppedPayload reports a bounded burst of dropped events that were silently lost before the bus emitted this notification. The dropped sequence range is closed/closed: [FromSeq, ToSeq] inclusive.

The bus emits at most one BusDroppedPayload per DropWindow per subscriber; the range covers every event lost since the last emit.

type Cursor ¶

type Cursor struct {
	SessionID string
	Sequence  uint64
}

Cursor identifies the last event a subscriber has consumed for a session. Sequence is the per-bus monotonic value assigned by Publish; SessionID scopes the cursor so two subscribers on different sessions can use the same numeric Sequence without collision. The bus that issued the Sequence is the only bus the Cursor is meaningful against; cross-bus replay is not supported.

Cursor{Sequence: 0} has the special meaning "from the beginning" for Replayer.Replay — equivalent to "the caller has seen nothing yet" — and bypasses the ErrCursorTooOld check so a fresh client can read whatever the ring still retains without coordinating on its tail.

type Event ¶

type Event struct {
	Type       EventType
	Identity   identity.Quadruple
	OccurredAt time.Time
	Sequence   uint64
	Payload    EventPayload
	Extra      map[string]string
}

Event is the canonical bus record.

Sequence is per-bus monotonic and gap-free; assigned by Publish. Callers MUST NOT pre-fill Sequence (Publish rejects with ErrSequenceProvided). OccurredAt defaults to time.Now() when zero.

Extra is reserved for Phase 56's bounded low-cardinality metric labels. Phase 05 does not derive metrics; the slot exists so later phases can populate it without changing the Event shape.

type EventBus ¶

type EventBus interface {
	Publish(ctx context.Context, ev Event) error
	Subscribe(ctx context.Context, f Filter) (Subscription, error)
	Close(ctx context.Context) error
}

EventBus is the canonical pub/sub surface. Implementations MUST be safe for concurrent use by N goroutines against a single shared instance (D-025).

func From(ctx context.Context) (EventBus, bool)

func MustFrom(ctx context.Context) EventBus

func Open(_ context.Context, cfg config.EventsConfig, r audit.Redactor) (EventBus, error)

func OpenDriver(name string, cfg config.EventsConfig, r audit.Redactor) (EventBus, error)

type EventID ¶

type EventID string

EventID is a per-event idempotency key. ULID-shaped at construction time; the convention is to generate via `state.NewEventID` (or any caller-side ULID source) at the publish site. Used by `internal/distributed.BusEnvelope.EventID` to dedupe at-least-once deliveries on `(TaskID, Edge, EventID)`.

The type lives here (and not in `internal/state`) so distributed callers can reference it without importing state's persistence surface. The two namespaces are intentionally parallel: state's `EventID` is the persistence-layer idempotency key; events' `EventID` is the bus-layer correlation key. Both ULID-shaped, both caller-supplied; consumers MAY pass the same value across both layers when convenient.

type EventPayload ¶

type EventPayload interface {
	// contains filtered or unexported methods
}

EventPayload is the sealed payload interface. Concrete payload types live alongside their owning subsystems and embed Sealed to satisfy the seal. The unexported method on Sealed is the seal.

type EventType ¶

type EventType string

EventType is a string-typed exhaustive enum. Each canonical type is declared as an exported constant plus registered in init() so the registry stays the single source of truth.

Adding a new type in a later phase: declare an exported constant, extend canonicalTypes, and update the master plan / glossary if the new type is reader-facing.

const (
	// EventTypeRuntimeError — emitted by the Phase 04 logger on Error.
	EventTypeRuntimeError EventType = "runtime.error"
	// EventTypeRuntimeWarning — reserved for future runtime-warn emits.
	EventTypeRuntimeWarning EventType = "runtime.warning"
	// EventTypeBusDropped — emitted by the bus into a subscriber's
	// stream when its buffer overflowed; carries the dropped sequence
	// range. At most one per DropWindow per subscriber.
	EventTypeBusDropped EventType = "bus.dropped"
	// EventTypeBusSubscriptionIdleClosed — emitted by the reaper when
	// a subscription is cancelled for not draining within IdleTimeout.
	EventTypeBusSubscriptionIdleClosed EventType = "bus.subscription_idle_closed"
	// EventTypeAuditRedactionFailed — emitted when audit.Redactor.Redact
	// returns an error during Publish. Carries the failing event's
	// type + identity but NO payload bytes.
	EventTypeAuditRedactionFailed EventType = "audit.redaction_failed"
	// EventTypeAdminScopeUsed — emitted when a Subscribe call passes
	// Admin: true (cross-session/cross-tenant) so admin-scope use is
	// retroactively detectable.
	EventTypeAdminScopeUsed EventType = "audit.admin_scope_used"
	// EventTypeGovernanceBudgetExceeded — reserved for Phase 36b emit.
	EventTypeGovernanceBudgetExceeded EventType = "governance.budget_exceeded"
	// EventTypeGovernanceRateLimited — reserved for Phase 36b emit.
	EventTypeGovernanceRateLimited EventType = "governance.rate_limited"
	// EventTypeRuntimeRunCancelled — emitted by Engine.Cancel(runID)
	// when the cancellation was observed for an active run. Payload is
	// RunCancelledPayload (SafePayload). Phase 13.
	EventTypeRuntimeRunCancelled EventType = "runtime.run_cancelled"
	// EventTypeTopologyChanged — emitted by the engine on construction
	// and on every adjacency-set change, carrying the canonical
	// TopologyProjection (Phase 74 / D-114). Payload is
	// TopologyChangedPayload (SafePayload — the projection has no
	// secret-shaped fields, so the audit redactor bypasses it and
	// subscribers keep typed access). The paired request-side surface
	// is the `topology.snapshot` Protocol method.
	EventTypeTopologyChanged EventType = "topology.changed"
	// EventTypeMCPRawHTMLTrustToggled — emitted by the runtime (the
	// Phase 73k MCPSurface) when a Console admin flips the per-server
	// raw-HTML opt-in flag via `mcp.servers.set_raw_html_trust`. Payload
	// is MCPRawHTMLTrustToggledPayload (SafePayload by construction —
	// server name + boolean + actor identity quadruple; no upstream MCP
	// content). The default-deny posture for raw-HTML / SVG rendering
	// (brief 11 §8) makes this audit event the load-bearing record of
	// the operator's explicit opt-in. Phase 73k / D-119.
	EventTypeMCPRawHTMLTrustToggled EventType = "mcp.raw_html_trust_toggled"
	// EventTypeRunOverridesSet — emitted by the runtime (the Phase 73n
	// Runs surface) when the Console Playground page records a
	// next-message override via `runs.set_overrides`. Payload is
	// RunOverridesSetPayload (SafePayload by construction — the actor's
	// identity quadruple, the session id, and a bounded set of boolean
	// "which field was set" flags; the override VALUES themselves are
	// NOT in the payload, since a system-prompt override is
	// caller-supplied text that must not reach an audit subscriber
	// unredacted — CLAUDE.md §7). Phase 73n / D-130.
	EventTypeRunOverridesSet EventType = "runs.overrides_set"
)

func EventTypes() []EventType

type Factory ¶

type Factory func(config.EventsConfig, audit.Redactor) (EventBus, error)

Factory builds an EventBus from EventsConfig + an audit.Redactor. Drivers expose one Factory each via init() → Register.

type Filter ¶

type Filter struct {
	Tenant  string
	User    string
	Session string
	// Run, when non-empty, narrows the subscription to a single run
	// inside the (tenant, user, session) scope. An empty Run means
	// "every run in the session" (session-scoped subscription) — the
	// Phase 60 default. The wire transport carries this via the
	// optional `X-Harbor-Run` (stream/HeaderRun) carrier header.
	// PR #91 / D-082 (Wave 10 audit WARN-5).
	Run   string
	Types []EventType
	Admin bool
}

Filter is the server-enforced subscription predicate. Subscribe rejects filters that elide the identity triple (Tenant + User + Session) unless Admin is set. When Admin is set with a partial triple, the bus emits an audit.admin_scope_used event before returning the subscription.

The Admin claim is trust-based in Phase 05. Cryptographic verification arrives with Protocol auth in Phase 61; until then the audit emit on every Admin-true Subscribe makes any abuse retroactively detectable.

func (f Filter) HasFullTriple() bool

func (f Filter) Matches(ev Event) bool

type MCPRawHTMLTrustToggledPayload ¶

type MCPRawHTMLTrustToggledPayload struct {
	SafeSealed
	// Actor is the identity quadruple of the admin who toggled the flag.
	Actor identity.Quadruple
	// ServerName is the MCP server the flag was toggled on.
	ServerName string
	// Trusted is the new raw-HTML trust value.
	Trusted bool
	// OccurredAt is the wall-clock instant the toggle was applied.
	OccurredAt time.Time
}

MCPRawHTMLTrustToggledPayload is the SafePayload carried by an EventTypeMCPRawHTMLTrustToggled event (Phase 73k / D-119). The runtime publishes one on every successful `mcp.servers.set_raw_html_trust` Protocol call.

It is SafePayload by construction (D-028): the payload carries only the MCP server name, the new boolean trust value, the actor's identity quadruple, and the toggle instant — no upstream MCP content, no secret-shaped fields. The bus therefore skips the audit.Redactor for this payload and a subscriber keeps typed access.

type RedactedMap ¶

type RedactedMap struct {
	Sealed
	Data map[string]any
}

RedactedMap is the post-redaction payload form for payloads that did NOT implement SafePayload and whose Redact() output became a generic map[string]any (the audit redactor's normalised shape for reflective struct walks). Subscribers can extract redacted fields via Data lookup.

type Replayer ¶

type Replayer interface {
	// Replay returns events whose Sequence > from.Sequence and that
	// match f, in Sequence order (strictly increasing). The returned
	// slice is owned by the caller. (nil, nil) is the "nothing newer
	// to replay" case (cursor at or past the bus head). See
	// ErrCursorTooOld and ErrReplayUnavailable for failure modes.
	//
	// The same filter rules as Subscribe apply: empty-triple
	// non-admin filters are rejected with ErrIdentityScopeRequired,
	// and Admin-scope replay emits an audit.admin_scope_used event
	// before returning the snapshot so admin-scope use is observable.
	Replay(ctx context.Context, from Cursor, f Filter) ([]Event, error)
}

Replayer is the optional capability interface drivers may implement to support replay-from-cursor. EventBus.Subscribe + Replayer.Replay together give the caller a "resume cleanly after disconnect" pattern:

1. Open a fresh Subscribe with the desired filter — let the live stream begin queuing into the subscriber's buffer.
2. Call Replay(lastSeenCursor, filter) — drain the historical snapshot strictly newer than the cursor.
3. Live-tail the Subscribe channel; dedupe against the snapshot's last sequence so a Publish landing between Subscribe and Replay is not double-counted.

A driver that does not implement Replayer (or whose EventsConfig.ReplayBufferSize is 0) returns ErrReplayUnavailable. The type assertion bus.(events.Replayer) still succeeds in the configured-off case — the assertion is a compile-shaped contract; the runtime decision lives in the call.

type RunCancelledPayload ¶

type RunCancelledPayload struct {
	SafeSealed
	RunID                string
	CancelledAt          int64
	DroppedEnvelopeCount int64
}

RunCancelledPayload is emitted by Engine.Cancel(runID) when the cancellation was observed for an active run. Carries the RunID, the wall-clock CancelledAt timestamp (unix nanoseconds), and the number of envelopes the cancellation drained from the engine's channels (a coarse "how loaded was the run" metric).

SafePayload by construction — every field is internal bookkeeping (no caller-controlled bytes). Subscribers consume the typed shape directly without an audit-redactor walk. Phase 13.

type RunOverridesSetPayload ¶

type RunOverridesSetPayload struct {
	SafeSealed
	// Actor is the identity quadruple of the operator who recorded the
	// override.
	Actor identity.Quadruple
	// SessionID is the session the override applies to.
	SessionID string
	// SetReasoningEffort reports whether the reasoning-effort field was
	// set in this override.
	SetReasoningEffort bool
	// SetTemperature reports whether the temperature field was set.
	SetTemperature bool
	// SetMaxTokens reports whether the max-tokens field was set.
	SetMaxTokens bool
	// SetSystemPrompt reports whether the system-prompt-override field
	// was set.
	SetSystemPrompt bool
	// OccurredAt is the wall-clock instant the override was recorded.
	OccurredAt time.Time
}

RunOverridesSetPayload is the SafePayload carried by an EventTypeRunOverridesSet event (Phase 73n / D-130). The runtime publishes one on every successful `runs.set_overrides` Protocol call.

It is SafePayload by construction (D-028): the payload carries only the actor's identity quadruple, the target session id, the recording instant, and a bounded set of boolean "which field was set" flags. The override VALUES are deliberately NOT in the payload — a system-prompt override is caller-supplied free text, and a reasoning-effort / temperature / max-tokens value could leak intent; the bus must never carry caller-supplied bytes unredacted (CLAUDE.md §7). A subscriber that needs the values reads them from the runtime's pending-override slot under the same identity scope. The bus therefore skips the audit.Redactor for this payload and a subscriber keeps typed access.

type RuntimeErrorPayload ¶

type RuntimeErrorPayload struct {
	Sealed
	Message string
	Fields  map[string]any
}

RuntimeErrorPayload is the bus-side projection of a Logger.Error call. The telemetry/eventbus adapter constructs one of these from the redacted (msg, attrs) the Logger handed to its BusEmitter seam, so RuntimeErrorPayload arrives at the bus pre-redacted.

Even so, it is NOT marked SafePayload: a defensive contributor might later construct a RuntimeErrorPayload outside the Logger path (e.g. emitting a runtime error directly from a handler that bypassed the redactor). Running this payload through the bus redactor on every Publish is an extra walk per error event, but it preserves the audit-redactor-as-bus-boundary contract (D-020).

Fields is the slog.Attr key/value map after Logger redaction, in `map[string]any` shape so the audit redactor's reflective walk is deterministic.

type SafePayload ¶

type SafePayload interface {
	EventPayload
	// contains filtered or unexported methods
}

SafePayload marks payload types whose contents are known not to carry secrets. Bus implementations skip the audit.Redactor for these — typed access is preserved on the subscriber side.

Bus-internal payloads (BusDropped, IdleClosed, AuditRedactionFailed, AdminScopeUsed) are SafePayload by construction. External payloads MAY implement SafePayload when their declarer is confident the type carries no secret-shaped data; declarers in doubt should NOT implement SafePayload — the bus will run the value through the redactor and the subscriber-side payload becomes a RedactedMap.

type SafeSealed ¶

type SafeSealed struct{ Sealed }

SafeSealed is embedded in payload structs that are both EventPayload AND SafePayload. Composes Sealed.

type Sealed ¶

type Sealed struct{}

Sealed is embedded in concrete payload structs to satisfy EventPayload from any package. The interface stays sealed in spirit — to declare a payload, you have to import this package.

Standard Go pattern (mirrors net/netip.Addr's seal, encoding/gob stdlib types, etc.). Compile-time enforcement.

type Subscription ¶

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

Subscription delivers events to one consumer.

Events() returns a receive-only channel. The channel is closed by the bus on Cancel or Close — consumers can use the close as the termination signal.

Cancel is idempotent and safe to call from any goroutine.

type SubscriptionIdleClosedPayload ¶

type SubscriptionIdleClosedPayload struct {
	SafeSealed
	SubscriberID uint64
	IdleSeconds  float64
}

SubscriptionIdleClosedPayload reports that the reaper cancelled a subscription that had not drained its channel within IdleTimeout.

type TopologyChangedPayload ¶

type TopologyChangedPayload struct {
	SafeSealed
	// Projection is the canonical engine topology at the moment the
	// event was emitted.
	Projection types.TopologyProjection
}

TopologyChangedPayload is the SafePayload carried by an EventTypeTopologyChanged event (Phase 74 / D-114). The engine publishes one on construction and on every adjacency-set change.

It is SafePayload by construction (D-028): a TopologyProjection carries only node names, edge endpoints, kind tags, and bounded integer queue depths — no secret-shaped fields. The bus therefore skips the audit.Redactor for this payload and a subscriber keeps typed access to Projection (no RedactedMap downgrade).

type WireConversion ¶

type WireConversion struct {
	// Filter is the bus-facing predicate the caller passes to Subscribe.
	// Triple components are backfilled from the caller's identity.
	Filter Filter
	// RequiresAdminScope is true when the wire EventFilter requested a
	// cross-tenant fan-in (a TenantID set other than the caller's own,
	// or len(TenantIDs) > 1). The wire edge gates on auth.HasScope
	// before calling Subscribe; without the scope, the request is
	// rejected with CodeIdentityScopeRequired (HTTP 403).
	RequiresAdminScope bool
	// Since / Until carry the optional time-window bounds from the wire
	// filter. They are NOT enforced by Filter.Matches (the bus's
	// per-event match is identity + type only); the aggregator + any
	// post-filtering replay consult them directly.
	Since time.Time
	Until time.Time
}

WireConversion is the result of converting a wire EventFilter into a runtime Filter (see FilterFromWire). The wire shape (prototypes.EventFilter) is operator-facing and lives in internal/protocol/types; the runtime shape (events.Filter) is bus-facing and lives here. This conversion is the one place the two namespaces meet — the wire transport produces a WireConversion once per request.

Semantics of the conversion:

• Empty TenantIDs / UserIDs / SessionIDs / RunIDs on the wire are interpreted as "the caller's own component" — the caller's identity quadruple supplies the value. This preserves backward compatibility with the pre-72a wire shape (no filter struct → full triple-scoped subscription).
• A single non-caller TenantID OR len(TenantIDs) > 1 signals a cross-tenant request — the result has RequiresAdminScope == true so the wire edge can gate on auth.HasScope before calling Subscribe.
• EventTypes are converted 1:1.
• Since / Until are passed through unchanged.

The conversion NEVER silently drops a missing identity component: it returns the (best-effort) Filter alongside RequiresAdminScope; the caller is responsible for the actual rejection if RequiresAdminScope is true but the scope claim is absent. (CLAUDE.md §5 "fail loudly.")

WireConversion is a pure value type: zero package-level state, safe for concurrent use (D-025).

func FilterFromWire(
	wire prototypes.EventFilter,
	callerTenant, callerUser, callerSession string,
) WireConversion