agentstreams

type Ack struct {
	Accepted   bool
	Error      string
	AcceptedAt time.Time
	SessionID  string
	Revision   uint64
}

type ConfigStatusHandler func(report ConfigStatusReport)

type ConfigStatusReport struct {
	Node      agentcontrol.AgentNode
	SessionID string
	Status    *agentv1pb.ConfigStatus
	Verdict   string
}

type ConfigStreams interface {
	// ConfigNegotiated reports whether the node's current session
	// negotiated config.v1.
	ConfigNegotiated(node agentcontrol.AgentNode) bool
	// PushConfig sends snapshot on the node's session and returns the
	// session's id. It does not wait for the agent's ConfigStatus. Errors
	// wrap ErrNotConnected or ErrCapabilityMissing. A snapshot older than
	// one the session was already sent is not sent again (sent is false):
	// the agent has a newer configuration.
	PushConfig(ctx context.Context, node agentcontrol.AgentNode, snapshot *agentv1pb.ConfigSnapshot) (sessionID string, sent bool, err error)
	// OnConfigStatus registers a handler for every ConfigStatus, of every
	// node kind.
	OnConfigStatus(handler ConfigStatusHandler)
}

type DiagnosticDispatch struct {
	// MessageID identifies the dispatch on the transport: the WebSocket
	// message id, or the stream operation id.
	MessageID string
	// AckReceived is true when the agent acknowledged the task.
	AckReceived bool
	Ack         Ack
	// RawAck is the transport's own acknowledgement object, for answers
	// that show it as the legacy routes do.
	RawAck any
	// DispatchError is why the acknowledged dispatch failed.
	DispatchError error
	// LegacyFallback is true when, after DispatchError, the task went out
	// in the legacy task message without an acknowledgement.
	LegacyFallback bool
	// FallbackError is why the legacy fallback failed too.
	FallbackError error
}

type DiagnosticTask struct {
	ID      string         `json:"id"`
	Type    string         `json:"type"`
	Action  string         `json:"action"`
	Params  map[string]any `json:"params"`
	Timeout int            `json:"timeout"`
}

type DiagnosticTransport interface {
	DispatchDiagnostic(ctx context.Context, task DiagnosticTask) DiagnosticDispatch
}

type ObservedHandler func(node agentcontrol.AgentNode, observed *agentv1pb.ObservedState)

type Session struct {
	Node         agentcontrol.AgentNode
	Transport    Transport
	SessionID    string
	AgentVersion string
	InstanceID   string
	Capabilities []string
	ConnectedAt  time.Time
	LastSeen     time.Time
	// DesiredRevision and ObservedRevision are the stream's operation
	// revisions; zero on the WebSocket.
	DesiredRevision  uint64
	ObservedRevision uint64
	// System is what a WebSocket agent reported about its host.
	System map[string]any
	// Identity is the session's authenticated identity: the agent's SPIFFE
	// ID, or IdentityAPIKey.
	Identity string
	// Authentication is AuthenticationMTLS or AuthenticationAPIKey; empty
	// on the WebSocket, which authenticates by API key only.
	Authentication string
	// Certificate is the client certificate of an mTLS stream session.
	Certificate *SessionCertificate
	// NegotiatedCapabilities are the data-plane capabilities in use on a
	// stream session (both its Hello and HelloAck list them), as
	// name.version.
	NegotiatedCapabilities []string
	// AgentMetrics are the Agent's own health metrics (agent_control_*,
	// agent_identity_*, agent_dataplane_*) from the stream session's latest
	// heartbeat that carried any, reported at AgentMetricsAt.
	AgentMetrics   map[string]float64
	AgentMetricsAt *time.Time
}

type SessionCertificate struct {
	Serial   string    `json:"serial"`
	NotAfter time.Time `json:"not_after"`
	SPIFFEID string    `json:"spiffe_id"`
}

type StatusConnections interface {
	StatusConnection(node agentcontrol.AgentNode) (any, bool)
}

type Streams interface {
	// Session returns the node's current stream session.
	Session(node agentcontrol.AgentNode) (Session, bool)
	// Sessions lists every stream session, in no particular order.
	Sessions() []Session
	// Observed returns the last observed state the node's agent reported.
	Observed(node agentcontrol.AgentNode) (*agentv1pb.ObservedState, bool)
	// Dispatch sends a desired operation on the node's stream and waits for
	// the agent's acknowledgement, or for ctx to end. A revision of zero
	// takes the node's next one; the acknowledgement carries it.
	Dispatch(ctx context.Context, node agentcontrol.AgentNode, operation *agentv1pb.DesiredOperation) (*agentv1pb.OperationAck, error)
	// Cancel asks the agent to stop an operation it holds.
	Cancel(ctx context.Context, node agentcontrol.AgentNode, operationID string, revision uint64) error
	// OnObserved registers a handler for every observed state, of every
	// node kind.
	OnObserved(handler ObservedHandler)
}

type Transport string

type WebSockets interface {
	// Sessions lists the connected WebSocket agents.
	Sessions() []Session
	// Monitor returns the last monitor snapshot a proxy node's agent posted.
	Monitor(nodeID uint) (system map[string]any, receivedAt time.Time, ok bool)
	// DiagnosticTransport returns the WebSocket of a connected proxy node's
	// agent, for diagnostic tasks.
	DiagnosticTransport(nodeID uint) (DiagnosticTransport, bool)
}