Documentation
¶
Index ¶
- func MarshalCompactionEvent(event CompactionEvent) ([]byte, error)
- func MarshalPrefixResetEvent(event PrefixResetEvent) ([]byte, error)
- type Bus
- type CacheUsageEvent
- type CompactionEvent
- type CompactionEventParams
- type Delivery
- type Event
- type Handler
- type HandlerFunc
- type Identity
- type Kind
- type MetricsAdapter
- type PrefixResetEvent
- type PrefixResetEventParams
- type TokenUsageEvent
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func MarshalCompactionEvent ¶
func MarshalCompactionEvent(event CompactionEvent) ([]byte, error)
MarshalCompactionEvent serializes only the typed compaction shape. Generic events cannot be passed to this boundary, so summary/content envelopes have no conversion path into a compaction progress event.
func MarshalPrefixResetEvent ¶
func MarshalPrefixResetEvent(event PrefixResetEvent) ([]byte, error)
MarshalPrefixResetEvent serializes only the typed prefix-reset shape. Generic events cannot be passed to this boundary, so content envelopes have no conversion path into a prefix-reset event.
Types ¶
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus is an asynchronous, in-process event bus. Publishers call Publish and events are enqueued to per-subscriber bounded queues. Each subscriber has a dedicated delivery goroutine that calls HandleEvent in FIFO order, so Publish never blocks on a handler.
Overflow policy: drop-oldest. Each subscriber tracks a drop counter via Drops(). Handlers receive a cancellable context tied to bus shutdown.
Bus is safe for concurrent use. All exported methods are goroutine-safe.
Handlers must not call Unsubscribe, Flush, or Close directly from inside HandleEvent: those methods wait on the delivery goroutine that is running the handler, which deadlocks (self-join). A handler that needs to manage the bus from inside HandleEvent must use the Delivery handle obtained via DeliveryFrom(ctx).
func (*Bus) Close ¶
func (b *Bus) Close()
Close marks the bus as closed, cancels the shutdown context (which all handlers receive), and waits for all delivery goroutines to drain their queues and exit. After Close, Subscribe is a no-op and Publish is a silent no-op.
Close is idempotent and safe to call multiple times.
Close blocks until every delivery goroutine has exited, including one that is mid-handler. Handlers must NOT call Close from inside HandleEvent (the wait includes the handler's own goroutine, which cannot exit until the handler returns). Use DeliveryFrom(ctx).Close() instead.
The sync.Once body only marks the bus closed and cancels the shutdown context; the wait for delivery goroutines runs AFTER Do returns. Running wg.Wait inside Do would let a concurrent caller parked inside Do (e.g. a handler calling Delivery.Close) hold a delivery goroutine hostage: that goroutine cannot exit until its handler returns, and the handler cannot return until Do completes.
func (*Bus) Flush ¶
func (b *Bus) Flush()
Flush blocks until all events that were published BEFORE the Flush call have been delivered to their handlers. It does this by sending a barrier event to each active subscription and waiting for the delivery goroutine to process it. Safe to call concurrently with Publish.
Use in tests and teardown paths where you need to guarantee handler state reflects all prior Publish calls.
Flush blocks for as long as a handler is still running: the barrier for a subscription cannot be acknowledged until its delivery goroutine drains the events published before the barrier. Handlers must NOT call Flush from inside HandleEvent (the barrier waits on the handler's own goroutine, which cannot ack until the handler returns). Use DeliveryFrom(ctx).Flush() instead.
func (*Bus) Publish ¶
Publish delivers an event to all handlers subscribed to the event's Kind. Events are enqueued to each subscriber's bounded queue; Publish never blocks on a handler. Publishing on a closed Bus is a no-op (safe).
Per-subscriber ordering is preserved: each subscriber's delivery goroutine processes events in FIFO order from its queue.
func (*Bus) Subscribe ¶
Subscribe registers a handler for the given event Kind with a bounded queue (default 256). Subscribing a nil handler is a no-op.
func (*Bus) SubscribeMany ¶
SubscribeMany registers a handler for multiple event Kinds at once.
func (*Bus) Unsubscribe ¶
Unsubscribe removes a specific handler from the given Kind's subscriber list. It stops the handler's delivery goroutine after draining any remaining queued events. If the handler was never subscribed, this is a no-op. Comparison uses pointer identity for comparable handler types and function code-pointer identity for HandlerFunc (two closures of one literal compare equal — best effort, strictly better than a runtime panic on interface equality). Unsubscribe never panics, so the bus lock is always released even for uncomparable handler types.
Unsubscribe blocks until the target's queued events have been drained and its delivery goroutine has exited, for every caller. Handlers must NOT call Unsubscribe from inside HandleEvent: joining the delivery goroutine that is running the handler deadlocks. Use DeliveryFrom(ctx).Unsubscribe() instead.
type CacheUsageEvent ¶
type CacheUsageEvent struct {
Provider string `json:"provider"`
Model string `json:"model"`
Style string `json:"style"`
InputTokens int `json:"input_tokens"`
CachedInputTokens int `json:"cached_input_tokens"`
CacheWriteTokens int `json:"cache_write_tokens"`
// contains filtered or unexported fields
}
CacheUsageEvent is the sealed, content-free progress payload for provider-reported prompt-cache accounting on one completion turn. It intentionally carries no message content - only provider/model attribution and token counts.
func NewCacheUsageEvent ¶
func NewCacheUsageEvent(provider, model, style string, inputTokens, cachedInputTokens, cacheWriteTokens int) (CacheUsageEvent, error)
NewCacheUsageEvent constructs the only valid cache usage event. Style is a bounded free-form string (not a shared enum with provider.CacheStyle) so this package stays independent of internal/provider; a future style value there needs no matching update here.
func (CacheUsageEvent) HitPercent ¶
func (e CacheUsageEvent) HitPercent() int
HitPercent returns the cache hit rate as an integer percent. It guards the division: zero input tokens reads as 0.
func (CacheUsageEvent) Validate ¶
func (e CacheUsageEvent) Validate() error
type CompactionEvent ¶
type CompactionEvent struct {
Trigger string `json:"trigger"`
BeforeTokens int `json:"before_tokens"`
AfterTokens int `json:"after_tokens"`
ElidedMessages int `json:"elided_messages"`
ElidedBytes int `json:"elided_bytes"`
SourceRange contextstate.SourceRange `json:"source_range"`
SummaryVersion uint32 `json:"summary_version"`
// Summarized reports whether an LLM summary of the dropped messages was
// actually produced. A compaction can succeed structurally with no
// summary at all (the workspace never configured one, or the summary
// call degraded), and SummaryVersion cannot express that: the validator
// requires it to be non-zero, so it was hardcoded to 1 and claimed a
// summary existed either way. A renderer that shows a clean "compacted"
// banner for a summary-less compaction is why an operator sees an
// instant, LLM-free compact and concludes compaction is broken.
Summarized bool `json:"summarized"`
// Reason names why Summarized is false, as a fixed, classified,
// content-free string (e.g. "no summarizer is configured for this
// session") - never a raw provider/library error, which could carry
// prompt or response fragments onto this sealed wire contract. Empty
// when Summarized is true. Without this a renderer could only ever tell
// an operator to "enable" summarization, even when it was already
// enabled and something else (a missing credential, an unresolved
// endpoint, a failed provider call) was the real cause.
Reason string `json:"reason,omitempty"`
// contains filtered or unexported fields
}
CompactionEvent is the sealed, content-free progress payload for context compaction. It intentionally has no generic content/input/output fields.
func NewCompactionEvent ¶
func NewCompactionEvent(p CompactionEventParams) (CompactionEvent, error)
NewCompactionEvent constructs the only valid compaction event. Callers get a value, not a pointer, so the event bus cannot mutate the constructor's private state through a shared object.
func RehydrateCompactionEvent ¶
func RehydrateCompactionEvent(c CompactionEvent) *CompactionEvent
RehydrateCompactionEvent seals a compaction payload reconstructed from a cross-process wire projection (e.g. internal/hub's WireCompaction). The original publisher validated the values through NewCompactionEvent; re-validating here is impossible without the sealed flag and re-deriving it would drop a relayed event for rules the sender already met. An unsealed reconstruction fails its own Validate, which is the trap this exists to close for later consumers.
func UnmarshalCompactionEvent ¶
func UnmarshalCompactionEvent(data []byte) (CompactionEvent, error)
UnmarshalCompactionEvent restores the constructor seal only after all wire fields have been decoded and validated.
func (CompactionEvent) Validate ¶
func (e CompactionEvent) Validate() error
type CompactionEventParams ¶
type CompactionEventParams struct {
Trigger string
BeforeTokens int
AfterTokens int
ElidedMessages int
ElidedBytes int
SourceRange contextstate.SourceRange
SummaryVersion uint32
Summarized bool
Reason string
}
CompactionEventParams is the only constructor input for CompactionEvent. ElidedMessages and ElidedBytes are optional content-free aggregates.
type Delivery ¶
type Delivery struct {
// contains filtered or unexported fields
}
Delivery is the re-entrant handle a handler uses to manage its own subscription from inside HandleEvent. Obtain it with DeliveryFrom(ctx), using the context the Bus passes to the handler.
Calling Bus.Unsubscribe, Bus.Flush, or Bus.Close directly from a handler deadlocks: each waits on the delivery goroutine that is running the handler, and that goroutine cannot make progress until the handler returns. Delivery methods are safe from inside the handler because they never wait on the caller's own delivery goroutine.
func DeliveryFrom ¶
DeliveryFrom returns the Delivery ticket for the handler currently running. It reports ok=false when ctx does not carry a ticket (the caller is not inside HandleEvent on a live subscription, or the context comes from elsewhere).
func (*Delivery) Close ¶
func (d *Delivery) Close()
Close shuts the bus down from inside a handler. It marks the bus closed, cancels the shutdown context, and clears the subscription map. It then waits for every OTHER subscription's delivery goroutine that is NOT currently running a handler to drain and exit. It does NOT wait for the caller's own goroutine: that goroutine is running this handler and exits on its own once the handler returns (its context is cancelled, so it drains and terminates).
Delivery.Close NEVER waits on a subscription whose delivery goroutine is currently running a handler, and never waits on one whose goroutine starts delivering after the delivering check: in both cases its done channel cannot close until the handler returns. A delivering goroutine may itself be parked inside a concurrent close (waiting on the shared sync.Once.Do until the first close body returns) or inside a Delivery.Flush that barriers this caller's own subscription, so waiting on its done channel would deadlock; it drains and exits on its own once its handler returns, so skipping it loses no liveness. The wait is TOCTOU-free: under deliveringMu Close snapshots the subscription's deliveringChange channel (closed on every delivering false->true transition in handle()), re-checks delivering, then selects on {done, changed} — the moment the goroutine starts running a handler, the select abandons the wait. The shared Once body itself only marks the bus closed and cancels the shutdown context: it must not wait or mutate b.subs, because a concurrent close attempt parks inside Do until that body returns.
func (*Delivery) Flush ¶
func (d *Delivery) Flush()
Flush blocks until all events published before the call have been delivered to every subscription OTHER than the caller's own. The caller's subscription is skipped: its delivery goroutine is running this handler, so a flush barrier for it could never be acknowledged until the handler returns.
func (*Delivery) Unsubscribe ¶
func (d *Delivery) Unsubscribe()
Unsubscribe removes the caller's own subscription from the bus and stops its delivery goroutine without waiting for it. Joining here would deadlock: the goroutine is currently running this handler. The subscription is marked stopped and its context is cancelled. Once the handler returns, the delivery goroutine exits at the top of its loop, without re-invoking the handler for events queued behind the current one.
type Event ¶
type Event struct {
Kind Kind
Timestamp time.Time
SessionID string
TurnID string
ToolCallID string
Name string
Detail string
Content string
Input string
Output string
Metadata map[string]string
Err error
// Agent attribution: which subagent produced this event (empty for the
// session's root loop). Flat fields keep this package free of an
// agent-package dependency.
AgentTask string // runtime request/task id - the attribution key
AgentName string // dispatched subagent/skill name
AgentDepth int // nesting depth (root loop = 0)
// Identity is the typed, allowlisted runtime identity. It never carries
// prompts, paths, digests, tools, content, errors, or arbitrary metadata.
Identity *Identity
// PrefixReset is present only for the typed prefix-stability reset event
// (KindPrefixReset). It is not copied into generic content/input/output
// envelopes and carries no prompt or digest content (INV-68-7).
PrefixReset *PrefixResetEvent
// Compaction is present only for the typed context-compaction progress
// event (KindCompaction). It carries the content-free payload
// (events.CompactionEvent) so bus consumers - the cross-process hub, a
// --json sidecar - get the real before/after numbers instead of parsing
// Detail. Nil on every other kind.
Compaction *CompactionEvent
}
Event is the universal event type for the event bus.
func NewEventFromAgentParts ¶
func NewEventFromAgentParts(kind Kind, toolCallID, name, detail, content, input, output string) Event
NewEventFromAgentParts creates an events.Event from the individual fields of an agent.Event, without importing the agent package.
type HandlerFunc ¶
HandlerFunc is an adapter that allows ordinary functions to be used as Handler implementations.
func (HandlerFunc) HandleEvent ¶
func (f HandlerFunc) HandleEvent(ctx context.Context, ev Event)
HandleEvent implements the Handler interface.
type Identity ¶
type Identity struct {
DefinitionName string
DefinitionSource string
InstanceID string
ModelGeneration uint64
}
Identity separates definition, disposable execution instance, and model binding generation for operator-facing lifecycle events.
type Kind ¶
type Kind string
Kind is the type for event kinds.
const ( // Agent loop events (mirror agent.EventKind values). KindAssistant Kind = "assistant" KindToolStart Kind = "tool_start" KindToolEnd Kind = "tool_end" KindStep Kind = "step" // KindHeartbeat mirrors agent.EventHeartbeat: a wall-clock progress // tick during model thinking, tool batches, and batch shaping. The // root loop publishes the bare string "heartbeat". Use this constant // to subscribe to it. KindHeartbeat Kind = "heartbeat" KindPrune Kind = "prune" KindToolParallel Kind = "tool_parallel" KindSubagentStart Kind = "subagent_start" KindSubagentEnd Kind = "subagent_end" KindSubagentHeartbeat Kind = "subagent_heartbeat" // KindSubagentDone mirrors agent.EventSubagentDone: the run-level // terminal signal for one subagent, not the end of a nested tool call. KindSubagentDone Kind = "subagent_done" KindThinking Kind = "thinking" KindCompaction Kind = "compaction" // KindCacheUsage reports provider-supplied prompt-cache accounting for // one completion turn. See CacheUsageEvent. KindCacheUsage Kind = "cache_usage" // KindTokenUsage reports provider-supplied input/output token counts // for one completion turn. See TokenUsageEvent. KindTokenUsage Kind = "token_usage" // KindPrefixReset reports that the session's byte-prefix stability // identity changed at a binding switch or agent-surface publication, so a // provider-implicit prompt-cache prefix is no longer reusable for the next // request. See PrefixResetEvent. KindPrefixReset Kind = "prefix_reset" // Session/turn lifecycle events. KindSessionStart Kind = "session_start" KindSessionEnd Kind = "session_end" KindTurnStart Kind = "turn_start" KindTurnEnd Kind = "turn_end" // Workflow and invocation observability events. Run, step, and task // identifiers ride in Event.Metadata; no Event fields are added. // KindWorkflowRunStarted reports the start of one workflow run. KindWorkflowRunStarted Kind = "workflow_run_started" // KindWorkflowStepStarted reports the start of one workflow step. KindWorkflowStepStarted Kind = "workflow_step_started" // KindWorkflowStepHeartbeat is the progress tick of a running step. KindWorkflowStepHeartbeat Kind = "workflow_step_heartbeat" // KindWorkflowStepCompleted reports the completion of one workflow step. KindWorkflowStepCompleted Kind = "workflow_step_completed" // KindWorkflowGateResult reports the start of one workflow gate: the gate // begin is published at gate_started time; the gate's outcome is published // as step_completed when the attempt reaches its terminal status. KindWorkflowGateResult Kind = "workflow_gate_result" // KindWorkflowApprovalRequested reports a workflow approval request. KindWorkflowApprovalRequested Kind = "workflow_approval_requested" // KindWorkflowRunFinished reports the end of one workflow run. KindWorkflowRunFinished Kind = "workflow_run_finished" // KindWorkflowDeliveryStage reports one delivery stage of a workflow. KindWorkflowDeliveryStage Kind = "workflow_delivery_stage" // KindInvocationStarted reports the start of one invocation. KindInvocationStarted Kind = "invocation_started" // KindInvocationCompleted reports the completion of one invocation. KindInvocationCompleted Kind = "invocation_completed" // KindInvocationRetrying reports one retry of an invocation. KindInvocationRetrying Kind = "invocation_retrying" // UI/system events. KindUIResize Kind = "ui_resize" KindUserInput Kind = "user_input" KindUIReady Kind = "ui_ready" KindConfigChange Kind = "config_change" // Error events. KindError Kind = "error" )
type MetricsAdapter ¶
type MetricsAdapter struct {
// contains filtered or unexported fields
}
MetricsAdapter collects per-kind event counts. Implements events.Handler. Safe for concurrent use. Call Subscribe() to attach to a Bus. Call Close() to detach.
func NewMetricsAdapter ¶
func NewMetricsAdapter() *MetricsAdapter
NewMetricsAdapter creates a MetricsAdapter with empty counters. Does NOT subscribe. Call Subscribe() to attach to a Bus.
func (*MetricsAdapter) Close ¶
func (m *MetricsAdapter) Close()
Close unsubscribes from the bus (all subscribed kinds) and resets counters. Idempotent - safe to call multiple times. Safe to call after Bus.Close(). Close is terminal, matching Bus.Close(): a Subscribe that races or follows Close is a permanent no-op.
Unsubscribe synchronously joins the subscription's delivery goroutine (stop() waits for it to drain and exit). The delivery goroutine calls HandleEvent, which needs m.mu - so m.mu must NOT be held across the Unsubscribe loop, or a delivery goroutine draining queued events waits on m.mu forever while we wait on it: a deadlock. Snapshot the subscription list under the lock, unsubscribe lock-free, then re-lock to clear state.
func (*MetricsAdapter) HandleEvent ¶
func (m *MetricsAdapter) HandleEvent(ctx context.Context, ev Event)
HandleEvent implements events.Handler. Increments per-kind counter. Recovers from panics to avoid crashing the publisher goroutine.
func (*MetricsAdapter) Reset ¶
func (m *MetricsAdapter) Reset()
Reset zeros all counters. Does NOT unsubscribe from the bus.
func (*MetricsAdapter) Snapshot ¶
func (m *MetricsAdapter) Snapshot() (counts map[string]uint64, totalEvents uint64)
Snapshot returns a consistent snapshot of all per-kind event counts and the total event count across all kinds. Returns a new map on each call - safe to iterate.
func (*MetricsAdapter) Subscribe ¶
func (m *MetricsAdapter) Subscribe(bus *Bus)
Subscribe subscribes the adapter to all known event kinds on the given Bus. Idempotent - safe to call multiple times (subsequent calls are no-op). After (or concurrently with) Close() it is a no-op, matching Bus.Close() semantics: Close is terminal. Stores subscribed kinds for Close() to unsubscribe.
type PrefixResetEvent ¶
type PrefixResetEvent struct {
Categories []string `json:"categories"`
OutgoingModelGeneration uint64 `json:"outgoing_model_generation"`
IncomingModelGeneration uint64 `json:"incoming_model_generation"`
OutgoingSurfaceGeneration uint64 `json:"outgoing_surface_generation"`
IncomingSurfaceGeneration uint64 `json:"incoming_surface_generation"`
// contains filtered or unexported fields
}
PrefixResetEvent is the sealed, content-free payload reporting that the session's byte-prefix stability identity changed. It intentionally has no generic content/input/output fields: it carries only allowlisted changed category names and the outgoing/incoming generation counters, never prompt content, digest preimages, tool-schema bodies, or tool-argument values (INV-68-7).
func NewPrefixResetEvent ¶
func NewPrefixResetEvent(p PrefixResetEventParams) (PrefixResetEvent, error)
NewPrefixResetEvent constructs the only valid prefix-reset event. Callers get a value, not a pointer, so the event bus cannot mutate the constructor's private state through a shared object.
func UnmarshalPrefixResetEvent ¶
func UnmarshalPrefixResetEvent(data []byte) (PrefixResetEvent, error)
UnmarshalPrefixResetEvent restores the constructor seal only after all wire fields have been decoded and validated, so a wire payload with empty, unknown, duplicate, oversized, or control-character categories is rejected through re-validation (INV-68-7).
func (PrefixResetEvent) Validate ¶
func (e PrefixResetEvent) Validate() error
type PrefixResetEventParams ¶
type PrefixResetEventParams struct {
Categories []string
OutgoingModelGeneration uint64
IncomingModelGeneration uint64
OutgoingSurfaceGeneration uint64
IncomingSurfaceGeneration uint64
}
PrefixResetEventParams is the only constructor input for PrefixResetEvent. The generation counters ride as observability only: a republish that changes only a monotonic counter is byte-stable and must not emit a reset (INV-68-2, test-plan correction 4).
type TokenUsageEvent ¶
type TokenUsageEvent struct {
Provider string `json:"provider"`
Model string `json:"model"`
InputTokens int `json:"input_tokens"`
OutputTokens int `json:"output_tokens"`
EstimatedTokens int `json:"estimated_tokens"`
CalibrationRatio float64 `json:"calibration_ratio"`
// contains filtered or unexported fields
}
TokenUsageEvent is the sealed progress payload for provider-reported input/output token counts, carrying estimate-vs-actual drift metrics so operators can see when the len(s)/4 heuristic diverges.
func NewTokenUsageEvent ¶
func NewTokenUsageEvent(provider, model string, inputTokens, outputTokens, estimatedTokens int, calibrationRatio float64) (TokenUsageEvent, error)
NewTokenUsageEvent constructs the only valid token usage event. Callers get a value, not a pointer, so the event bus cannot mutate the constructor's private state through a shared object.
func (TokenUsageEvent) Validate ¶
func (e TokenUsageEvent) Validate() error
Source Files