Documentation
¶
Overview ¶
Package live is an in-process pub/sub hub for real-time activity events that feed the /admin/live console: every gateway tool-call decision, token mint, and revocation is published here and fanned out to connected SSE clients.
The hub is deliberately best-effort: publishing never blocks the request path, and a slow consumer drops events rather than applying backpressure. It is also nil-safe — emit sites may hold a nil *Hub when the live console is disabled, and calling Publish on it is a no-op.
Index ¶
Constants ¶
const Channel = "legant_live"
Channel is the Postgres NOTIFY channel that carries live events between processes (the server, the gateway, and every replica) and is consumed by the server's Listen loop.
Variables ¶
This section is empty.
Functions ¶
func Listen ¶
Listen holds a dedicated Postgres connection, LISTENs on the live channel, and republishes every received event into the hub for local SSE subscribers. It reconnects with capped exponential backoff until ctx is cancelled, so a transient database blip doesn't permanently silence the console. Run it in a goroutine for the lifetime of the process.
Types ¶
type Edge ¶
type Edge struct {
ID string `json:"id"` // delegation_chains.id
Parent string `json:"parent,omitempty"` // parent delegation id (re-delegation)
From string `json:"from"` // delegator node id
To string `json:"to"` // delegatee node id
Depth int `json:"depth"`
}
Edge is one active delegation: delegator --grants--> delegatee agent.
type Event ¶
type Event struct {
Type string `json:"type"` // decision | mint | revoke
Time time.Time `json:"ts"` // when it happened
Decision string `json:"decision,omitempty"` // ALLOW | DENY | REVOKE | MINT
Subject string `json:"subject,omitempty"` // root user, e.g. "user:alice"
Actor string `json:"actor,omitempty"` // acting agent, e.g. "agent:conductor"
Provenance string `json:"provenance,omitempty"` // "user:alice → agent:x → agent:y"
Tool string `json:"tool,omitempty"` // tool/method on a decision
Upstream string `json:"upstream,omitempty"` // gateway upstream slug / audience
Reason string `json:"reason,omitempty"` // why a decision was DENY
Delegation string `json:"delegation,omitempty"` // delegation id (mint/revoke)
Count int `json:"count,omitempty"` // e.g. revoked-subtree size
}
Event is a single real-time activity record. Only the fields relevant to the event Type are populated; the rest are omitted from the JSON.
type Graph ¶
Graph is a snapshot of the current active authority graph.
func Snapshot ¶
Snapshot returns the current active delegation graph: every delegation that is active and unexpired, as nodes (users + agents) and edges (delegations). This is the exact active-ness predicate the delegation service uses everywhere: active = true AND (expires_at IS NULL OR expires_at > now()).
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub fans Events out to all current subscribers and keeps a small replay ring so a newly-connected console can render recent history immediately.
func (*Hub) Publish ¶
Publish records the event in the replay ring and fans it out to every subscriber without blocking. A subscriber whose buffer is full drops the event — the console is observability, never a source of backpressure on the request path. Publish is nil-safe.
func (*Hub) Subscribe ¶
Subscribe registers a new subscriber and returns its event channel plus an unsubscribe function. The channel is buffered; events arriving while it is full are dropped for that subscriber. The unsubscribe function is idempotent and closes the channel.
func (*Hub) Subscribers ¶
Subscribers reports the number of currently-connected subscribers.
type Node ¶
type Node struct {
ID string `json:"id"` // namespaced stable id: "user:<uuid>" / "agent:<uuid>"
Label string `json:"label"` // display: "user:alice" / "agent:assistant"
Kind string `json:"kind"` // "user" | "agent"
}
Node is a principal in the authority graph: a user or an agent.
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher broadcasts events to the live console across processes via Postgres NOTIFY. It is the single emit path used by the token-exchange mint, the delegation revoker, the revocation store, and the MCP gateway — so one feed shows activity from every process and replica.
Publish is non-blocking and best-effort: it enqueues to a bounded buffer drained by a background worker and drops events if the buffer is full, so a busy or stalled feed never adds latency to a mint or a tool call. nil-safe.
func NewPublisher ¶
NewPublisher starts a background worker that drains queued events to Postgres NOTIFY. The worker stops when ctx is cancelled.
Source Files