ipc

package
v1.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 20, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// CapabilitySend authorizes an agent to send IPC messages.
	CapabilitySend = "ipc.send"
	// CapabilityReceive authorizes an agent to receive and ack IPC messages.
	CapabilityReceive = "ipc.receive"
	// CapabilitySignal authorizes an agent to dispatch control signals.
	CapabilitySignal = "ipc.signal"
)

Variables

View Source 
var (
	// ErrInvalidAddress is returned when an address fails syntactic or semantic validation.
	ErrInvalidAddress = errors.New("ipc: invalid agent address")

	// ErrInvalidMessage is returned when a message is missing required fields or has invalid attributes.
	ErrInvalidMessage = errors.New("ipc: invalid agent message")

	// ErrCrossTenantDenied is returned when an IPC attempt crosses tenant boundaries.
	ErrCrossTenantDenied = errors.New("ipc: cross-tenant communication denied")

	// ErrCrossNamespaceDenied is returned when cross-namespace communication is not permitted by policy.
	ErrCrossNamespaceDenied = errors.New("ipc: cross-namespace communication denied by policy")

	// ErrCapabilityDenied is returned when an agent lacks required IPC capabilities (ipc.send, ipc.receive, ipc.signal).
	ErrCapabilityDenied = errors.New("ipc: capability denied")

	// ErrFenced is returned when an agent instance is fenced or has an invalid/stale fencing token.
	ErrFenced = errors.New("ipc: agent instance is fenced")

	// ErrMessageNotFound is returned when a requested message cannot be found in the mailbox.
	ErrMessageNotFound = errors.New("ipc: message not found")

	// ErrMessageExpired is returned when a message has exceeded its deadline.
	ErrMessageExpired = errors.New("ipc: message deadline expired")

	// ErrDuplicateMessage is returned when a message with the same ID already exists in the mailbox.
	ErrDuplicateMessage = errors.New("ipc: duplicate message id")

	// ErrTimeout is returned when a request-reply pattern times out waiting for a reply.
	ErrTimeout = errors.New("ipc: request timed out waiting for reply")

	// ErrInvalidReply is returned when a reply is sent without a matching request or correlation ID.
	ErrInvalidReply = errors.New("ipc: invalid reply or missing correlation id")

	// ErrPeerDenied is returned when sender is not authorized to communicate with receiver by peer policy.
	ErrPeerDenied = errors.New("ipc: peer communication denied by policy")

	// ErrReceiverDenied is returned when receiver policy rejects the sender.
	ErrReceiverDenied = errors.New("ipc: receiver policy denied communication from sender")

	// ErrUnauthorizedSignal is returned when an agent is not authorized to send signals to the target.
	ErrUnauthorizedSignal = errors.New("ipc: unauthorized signal dispatch denied")
)

Functions

This section is empty.

Types

type AgentAddress 

type AgentAddress struct {
	TenantID   string `json:"tenant_id"`
	Namespace  string `json:"namespace"`
	AgentID    string `json:"agent_id"`
	InstanceID string `json:"instance_id,omitempty"`
}

AgentAddress is the unified address of an agent in the Fenced kernel. It supports both logical agent addressing (InstanceID is empty) and specific instance addressing (InstanceID is non-empty).

func NewAddress 

func NewAddress(tenantID, namespace, agentID string) AgentAddress

NewAddress creates a logical AgentAddress.

func NewInstanceAddress 

func NewInstanceAddress(tenantID, namespace, agentID, instanceID string) AgentAddress

NewInstanceAddress creates an instance-specific AgentAddress.

func ParseAddress 

func ParseAddress(raw string) (AgentAddress, error)

ParseAddress parses a canonical agent address string in either: 

agent://<tenant>/<namespace>/<agent>[/<instance>]

or slash-delimited format: 

<tenant>/<namespace>/<agent>[/<instance>]

func (AgentAddress) EffectiveNamespace 

func (a AgentAddress) EffectiveNamespace() string

EffectiveNamespace returns the namespace, defaulting to "default" if empty.

func (AgentAddress) IsInstance 

func (a AgentAddress) IsInstance() bool

IsInstance reports whether this address targets a concrete agent instance.

func (AgentAddress) Logical 

func (a AgentAddress) Logical() AgentAddress

Logical returns the logical address with InstanceID cleared.

func (AgentAddress) MarshalJSON 

func (a AgentAddress) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler.

func (AgentAddress) MarshalText 

func (a AgentAddress) MarshalText() ([]byte, error)

MarshalText implements encoding.TextMarshaler for stable serialization.

func (AgentAddress) Matches 

func (a AgentAddress) Matches(target AgentAddress) bool

Matches reports whether this address matches a target receiver address. If either address has an empty InstanceID (logical addressing), they match as long as TenantID, Namespace, and AgentID match. If both have non-empty InstanceIDs, they must also match in InstanceID.

func (AgentAddress) String 

func (a AgentAddress) String() string

String returns the canonical, stably serializable URI representation: 

agent://<tenant>/<namespace>/<agent>[/<instance>]

func (*AgentAddress) UnmarshalJSON 

func (a *AgentAddress) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler, supporting both object and string forms.

func (*AgentAddress) UnmarshalText 

func (a *AgentAddress) UnmarshalText(text []byte) error

UnmarshalText implements encoding.TextUnmarshaler for stable deserialization.

func (AgentAddress) Validate 

func (a AgentAddress) Validate() error

Validate checks that the address has all required fields and valid characters. TenantID and AgentID are strictly required; InstanceID is optional.

type AgentMessage 

type AgentMessage struct {
	ID            string         `json:"id"`
	TenantID      string         `json:"tenant_id"`
	Namespace     string         `json:"namespace"`
	Sender        AgentAddress   `json:"sender"`
	Receiver      AgentAddress   `json:"receiver"`
	Type          MessageType    `json:"type"`
	CorrelationID string         `json:"correlation_id,omitempty"`
	Payload       []byte         `json:"payload,omitempty"`
	Status        DeliveryStatus `json:"status"`
	CreatedAt     time.Time      `json:"created_at"`
	Deadline      time.Time      `json:"deadline"`
	AckedAt       *time.Time     `json:"acked_at,omitempty"`
	TraceID       string         `json:"trace_id,omitempty"`
	SignalType    SignalType     `json:"signal_type,omitempty"`
}

AgentMessage represents an immutable or state-tracked kernel IPC message.

func NewMessage 

func NewMessage(sender, receiver AgentAddress, msgType MessageType, payload []byte, ttl time.Duration) (*AgentMessage, error)

NewMessage constructs a general AgentMessage with default pending status and UUID.

func NewReply 

func NewReply(sender, receiver AgentAddress, correlationID string, payload []byte, ttl time.Duration) (*AgentMessage, error)

NewReply creates a reply message corresponding to a prior request correlation ID.

func NewRequest 

func NewRequest(sender, receiver AgentAddress, correlationID string, payload []byte, ttl time.Duration) (*AgentMessage, error)

NewRequest creates a request message with an optional or generated correlation ID.

func NewSignal 

func NewSignal(sender, receiver AgentAddress, sigType SignalType, reason string, metadata map[string]string, ttl time.Duration) (*AgentMessage, error)

NewSignal constructs a signal message carrying standard control signals.

func (*AgentMessage) Clone 

func (m *AgentMessage) Clone() *AgentMessage

Clone creates a deep copy of the message.

func (*AgentMessage) IsExpired 

func (m *AgentMessage) IsExpired(now time.Time) bool

IsExpired checks if the message has passed its deadline relative to the given time.

func (*AgentMessage) ParseSignalPayload 

func (m *AgentMessage) ParseSignalPayload() (*SignalPayload, error)

ParseSignalPayload extracts the structured SignalPayload if this message is of type Signal.

func (*AgentMessage) Validate 

func (m *AgentMessage) Validate() error

Validate performs structural and semantic checks on the message.

type AuditRecord 

type AuditRecord struct {
	Timestamp   time.Time   `json:"timestamp"`
	MessageID   string      `json:"message_id"`
	TenantID    string      `json:"tenant_id"`
	Namespace   string      `json:"namespace"`
	Sender      string      `json:"sender"`
	Receiver    string      `json:"receiver"`
	MessageType MessageType `json:"message_type"`
	Action      string      `json:"action"` // "send", "receive", "ack", "request", "reply", "signal"
	Result      string      `json:"result"` // "success", "denied", "expired", "failed"
	ReasonCode  string      `json:"reason_code,omitempty"`
	Reason      string      `json:"reason,omitempty"`
	TraceID     string      `json:"trace_id,omitempty"`
}

AuditRecord logs security and lifecycle events for IPC operations without payload.

type Auditor 

type Auditor interface {
	LogAudit(ctx context.Context, record AuditRecord)
}

Auditor defines the sink for IPC audit logging.

type AuthorizationDecision 

type AuthorizationDecision struct {
	Allowed     bool      `json:"allowed"`
	ReasonCode  string    `json:"reason_code"` // "AUTH_PASSED", "CROSS_TENANT_DENIED", "CROSS_NAMESPACE_DENIED", "FENCED", "CAPABILITY_DENIED", "UNAUTHORIZED_SIGNAL", "PEER_DENIED", "RECEIVER_DENIED"
	Reason      string    `json:"reason"`
	Sender      string    `json:"sender"`
	Receiver    string    `json:"receiver"`
	TenantID    string    `json:"tenant_id"`
	TraceID     string    `json:"trace_id,omitempty"`
	EvaluatedAt time.Time `json:"evaluated_at"`
}

AuthorizationDecision records the complete trace and reason of an IPC authorization evaluation.

type CapabilityChecker 

type CapabilityChecker interface {
	CheckCapability(ctx context.Context, tenantID string, addr AgentAddress, capability string) error
}

CapabilityChecker verifies whether an agent address possesses a required capability.

type DefaultPolicy 

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

DefaultPolicy is the kernel-level IPC policy enforcer.

func NewDefaultPolicy 

func NewDefaultPolicy(checker CapabilityChecker, fencer Fencer, opts ...PolicyOption) *DefaultPolicy

NewDefaultPolicy constructs a DefaultPolicy with optional capability checker, fencer, and options.

func (*DefaultPolicy) AllowCrossNamespace 

func (p *DefaultPolicy) AllowCrossNamespace(tenantID, fromNS, toNS string)

AllowCrossNamespace permits communication from one namespace to another for a tenant.

func (*DefaultPolicy) AllowPeer 

func (p *DefaultPolicy) AllowPeer(sender AgentAddress, rule PeerRule)

AllowPeer adds an allowed peer rule to sender's outbound allowlist. If no PeerAuthorizer is configured yet, an in-memory default-deny authorizer is automatically initialized.

func (*DefaultPolicy) AllowReceiver 

func (p *DefaultPolicy) AllowReceiver(receiver AgentAddress, rule PeerRule)

AllowReceiver adds an allowed sender rule to receiver's inbound policy. If no PeerAuthorizer is configured yet, an in-memory default-deny authorizer is automatically initialized.

func (*DefaultPolicy) AuthorizeReceive 

func (p *DefaultPolicy) AuthorizeReceive(ctx context.Context, receiver AgentAddress) error

AuthorizeReceive verifies fencing and ipc.receive capability for a receiving agent.

func (*DefaultPolicy) AuthorizeSend 

func (p *DefaultPolicy) AuthorizeSend(ctx context.Context, msg *AgentMessage) error

AuthorizeSend checks tenant isolation, cross-namespace permissions, fencing, capabilities, and peer policy.

func (*DefaultPolicy) AuthorizeSendWithDecision 

func (p *DefaultPolicy) AuthorizeSendWithDecision(ctx context.Context, msg *AgentMessage) (*AuthorizationDecision, error)

AuthorizeSendWithDecision validates Identity, Tenant Isolation, Namespace Boundary, Fencing, Capabilities, and Peer Policy.

func (*DefaultPolicy) DisallowCrossNamespace 

func (p *DefaultPolicy) DisallowCrossNamespace(tenantID, fromNS, toNS string)

DisallowCrossNamespace revokes cross-namespace permission.

func (*DefaultPolicy) PeerAuthorizer 

func (p *DefaultPolicy) PeerAuthorizer() PeerAuthorizer

PeerAuthorizer returns the configured PeerAuthorizer, if any.

func (*DefaultPolicy) SetPeerAuthorizer 

func (p *DefaultPolicy) SetPeerAuthorizer(authorizer PeerAuthorizer)

SetPeerAuthorizer assigns a PeerAuthorizer to the policy.

type DeliveryStatus 

type DeliveryStatus string

DeliveryStatus represents the lifecycle delivery state of a mailbox message. 

const (
	StatusPending   DeliveryStatus = "PENDING"
	StatusDelivered DeliveryStatus = "DELIVERED"
	StatusAcked     DeliveryStatus = "ACKED"
	StatusExpired   DeliveryStatus = "EXPIRED"
)

type DurableMemoryMailbox 

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

DurableMemoryMailbox is an in-memory, thread-safe implementation of Mailbox that supports snapshots and crash-recovery simulation.

func NewDurableMemoryMailbox 

func NewDurableMemoryMailbox() *DurableMemoryMailbox

NewDurableMemoryMailbox constructs a new in-memory durable mailbox.

func NewDurableMemoryMailboxWithClock 

func NewDurableMemoryMailboxWithClock(clock func() time.Time) *DurableMemoryMailbox

NewDurableMemoryMailboxWithClock allows setting an injected clock for testing.

func (*DurableMemoryMailbox) Ack 

func (m *DurableMemoryMailbox) Ack(ctx context.Context, receiver AgentAddress, messageIDs []string) error

Ack marks messages as ACKED. Acking is idempotent.

func (*DurableMemoryMailbox) GetByCorrelationID 

func (m *DurableMemoryMailbox) GetByCorrelationID(ctx context.Context, tenantID, correlationID string) (*AgentMessage, error)

GetByCorrelationID retrieves the latest message with the specified correlation ID.

func (*DurableMemoryMailbox) GetMessage 

func (m *DurableMemoryMailbox) GetMessage(ctx context.Context, id string) (*AgentMessage, error)

GetMessage retrieves a message by its ID.

func (*DurableMemoryMailbox) Receive 

func (m *DurableMemoryMailbox) Receive(ctx context.Context, receiver AgentAddress, limit int) ([]*AgentMessage, error)

Receive retrieves pending messages intended for the specified receiver.

func (*DurableMemoryMailbox) Restore 

func (m *DurableMemoryMailbox) Restore(data []byte) error

Restore reloads mailbox data from a snapshot.

func (*DurableMemoryMailbox) Send 

func (m *DurableMemoryMailbox) Send(ctx context.Context, msg *AgentMessage) error

Send stores a validated message into the mailbox with StatusPending.

func (*DurableMemoryMailbox) SimulateCrash 

func (m *DurableMemoryMailbox) SimulateCrash() (*DurableMemoryMailbox, error)

SimulateCrash creates a new DurableMemoryMailbox instance restoring the snapshot.

func (*DurableMemoryMailbox) Snapshot 

func (m *DurableMemoryMailbox) Snapshot() ([]byte, error)

Snapshot serializes mailbox data into bytes.

type Fencer 

type Fencer interface {
	ValidateInstance(ctx context.Context, addr AgentAddress) error
}

Fencer validates whether an agent instance is active or has been fenced.

type Mailbox 

type Mailbox interface {
	Send(ctx context.Context, msg *AgentMessage) error
	Receive(ctx context.Context, receiver AgentAddress, limit int) ([]*AgentMessage, error)
	Ack(ctx context.Context, receiver AgentAddress, messageIDs []string) error
	GetMessage(ctx context.Context, id string) (*AgentMessage, error)
	GetByCorrelationID(ctx context.Context, tenantID, correlationID string) (*AgentMessage, error)
}

Mailbox defines the storage interface for kernel-level Agent IPC messages.

type MemoryAuditor 

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

MemoryAuditor is an in-memory thread-safe auditor suitable for testing and verification.

func NewMemoryAuditor 

func NewMemoryAuditor() *MemoryAuditor

NewMemoryAuditor creates a new MemoryAuditor.

func (*MemoryAuditor) Count 

func (a *MemoryAuditor) Count() int

Count returns the number of audit records logged.

func (*MemoryAuditor) LogAudit 

func (a *MemoryAuditor) LogAudit(ctx context.Context, record AuditRecord)

LogAudit appends an audit record.

func (*MemoryAuditor) Records 

func (a *MemoryAuditor) Records() []AuditRecord

Records returns a slice of all recorded audit records.

type MemoryCapabilityChecker 

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

MemoryCapabilityChecker implements in-memory capability authorization for tests and runtime checks.

func NewMemoryCapabilityChecker 

func NewMemoryCapabilityChecker() *MemoryCapabilityChecker

NewMemoryCapabilityChecker creates a new in-memory capability checker.

func (*MemoryCapabilityChecker) CheckCapability 

func (c *MemoryCapabilityChecker) CheckCapability(ctx context.Context, tenantID string, addr AgentAddress, capability string) error

CheckCapability tests if the address has the requested capability.

func (*MemoryCapabilityChecker) Grant 

func (c *MemoryCapabilityChecker) Grant(addr AgentAddress, capability string)

Grant assigns a capability to an agent address. If addr has no instance, the capability is granted to the logical agent (all instances).

func (*MemoryCapabilityChecker) Revoke 

func (c *MemoryCapabilityChecker) Revoke(addr AgentAddress, capability string)

Revoke removes a capability grant.

type MemoryFencer 

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

MemoryFencer provides an in-memory fencer registry for testing.

func NewMemoryFencer 

func NewMemoryFencer() *MemoryFencer

NewMemoryFencer creates a new MemoryFencer.

func (*MemoryFencer) FenceInstance 

func (f *MemoryFencer) FenceInstance(addr AgentAddress)

FenceInstance marks an agent instance as fenced.

func (*MemoryFencer) UnfenceInstance 

func (f *MemoryFencer) UnfenceInstance(addr AgentAddress)

UnfenceInstance removes the fenced flag from an instance.

func (*MemoryFencer) ValidateInstance 

func (f *MemoryFencer) ValidateInstance(ctx context.Context, addr AgentAddress) error

ValidateInstance returns ErrFenced if the instance address has been fenced.

type MemoryPeerAuthorizer 

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

MemoryPeerAuthorizer provides an in-memory implementation of sender allowlist and receiver policies.

func NewMemoryPeerAuthorizer 

func NewMemoryPeerAuthorizer(defaultDeny bool) *MemoryPeerAuthorizer

NewMemoryPeerAuthorizer creates a new in-memory peer authorizer.

func (*MemoryPeerAuthorizer) AllowPeer 

func (a *MemoryPeerAuthorizer) AllowPeer(sender AgentAddress, rule PeerRule)

AllowPeer adds an allowed peer rule to sender's outbound allowlist.

func (*MemoryPeerAuthorizer) AllowReceiver 

func (a *MemoryPeerAuthorizer) AllowReceiver(receiver AgentAddress, rule PeerRule)

AllowReceiver adds an allowed sender rule to receiver's inbound policy.

func (*MemoryPeerAuthorizer) AuthorizePeer 

func (a *MemoryPeerAuthorizer) AuthorizePeer(ctx context.Context, sender, receiver AgentAddress) (*AuthorizationDecision, error)

AuthorizePeer evaluates sender allowlist and receiver policy.

func (*MemoryPeerAuthorizer) SetAllowedPeers 

func (a *MemoryPeerAuthorizer) SetAllowedPeers(sender AgentAddress, rules []PeerRule)

SetAllowedPeers replaces sender's outbound allowlist.

func (*MemoryPeerAuthorizer) SetReceiverPolicy 

func (a *MemoryPeerAuthorizer) SetReceiverPolicy(receiver AgentAddress, rules []PeerRule)

SetReceiverPolicy replaces receiver's inbound policy rules.

type MessageType 

type MessageType string

MessageType represents the classification of an Agent IPC message. 

const (
	MessageTypeRequest MessageType = "request"
	MessageTypeReply   MessageType = "reply"
	MessageTypeSignal  MessageType = "signal"
)

type Metrics 

type Metrics interface {
	RecordSent(ctx context.Context, tenantID string, msgType MessageType)
	RecordReceived(ctx context.Context, tenantID string, msgType MessageType)
	RecordAcked(ctx context.Context, tenantID string, count int)
	RecordExpired(ctx context.Context, tenantID string)
	RecordDenied(ctx context.Context, tenantID string, reason string)
	RecordDeliveryDuration(ctx context.Context, tenantID string, duration time.Duration)
}

Metrics captures telemetry counters and durations for IPC operations.

type PeerAuthorizer 

type PeerAuthorizer interface {
	AuthorizePeer(ctx context.Context, sender, receiver AgentAddress) (*AuthorizationDecision, error)
}

PeerAuthorizer evaluates whether a sender is authorized to communicate with a receiver.

type PeerRule 

type PeerRule struct {
	Namespace string `json:"namespace,omitempty"` // Target namespace; empty means same namespace as sender, "*" means any permitted namespace
	Agent     string `json:"agent"`               // Target agent ID, or "*" for any agent in that namespace
}

PeerRule defines a target pattern for allowed peer communication.

func (PeerRule) Matches 

func (r PeerRule) Matches(senderNS string, target AgentAddress) bool

Matches checks whether the rule matches the target receiver address given sender's namespace.

type Policy 

type Policy interface {
	AuthorizeSend(ctx context.Context, msg *AgentMessage) error
	AuthorizeReceive(ctx context.Context, receiver AgentAddress) error
}

Policy enforces tenant isolation, cross-namespace rules, capabilities, and peer rules for IPC operations.

type PolicyOption 

type PolicyOption func(*DefaultPolicy)

PolicyOption configures a DefaultPolicy.

func WithPeerAuthorizer 

func WithPeerAuthorizer(authorizer PeerAuthorizer) PolicyOption

WithPeerAuthorizer installs a custom PeerAuthorizer.

type PostgresMailbox 

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

PostgresMailbox implements Mailbox on top of PostgreSQL agent_ipc_messages table.

func NewPostgresMailbox 

func NewPostgresMailbox(pool *pgxpool.Pool) *PostgresMailbox

NewPostgresMailbox creates a new PostgreSQL backed Mailbox.

func NewPostgresMailboxWithClock 

func NewPostgresMailboxWithClock(pool *pgxpool.Pool, clock func() time.Time) *PostgresMailbox

NewPostgresMailboxWithClock allows passing a custom clock for testing.

func (*PostgresMailbox) Ack 

func (p *PostgresMailbox) Ack(ctx context.Context, receiver AgentAddress, messageIDs []string) error

Ack updates the specified messages to StatusAcked.

func (*PostgresMailbox) GetByCorrelationID 

func (p *PostgresMailbox) GetByCorrelationID(ctx context.Context, tenantID, correlationID string) (*AgentMessage, error)

GetByCorrelationID retrieves the newest message matching the correlation ID.

func (*PostgresMailbox) GetMessage 

func (p *PostgresMailbox) GetMessage(ctx context.Context, id string) (*AgentMessage, error)

GetMessage retrieves a message by its UUID string.

func (*PostgresMailbox) Receive 

func (p *PostgresMailbox) Receive(ctx context.Context, receiver AgentAddress, limit int) ([]*AgentMessage, error)

Receive retrieves pending messages and updates their status to DELIVERED.

func (*PostgresMailbox) Send 

func (p *PostgresMailbox) Send(ctx context.Context, msg *AgentMessage) error

Send inserts a message into agent_ipc_messages.

type Service 

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

Service is the unified kernel-level Agent IPC service.

func NewService 

func NewService(cfg ServiceConfig) *Service

NewService constructs an IPC Service.

func (*Service) Ack 

func (s *Service) Ack(ctx context.Context, receiver AgentAddress, messageIDs []string) error

Ack acknowledges received messages.

func (*Service) Receive 

func (s *Service) Receive(ctx context.Context, receiver AgentAddress, limit int) ([]*AgentMessage, error)

Receive retrieves pending messages for receiver.

func (*Service) Send 

func (s *Service) Send(ctx context.Context, msg *AgentMessage) error

Send validates, authorizes, audits, and persists an IPC message.

func (*Service) SendReply 

func (s *Service) SendReply(ctx context.Context, reply *AgentMessage) error

SendReply validates and sends a reply message to an existing request.

func (*Service) SendRequest 

func (s *Service) SendRequest(ctx context.Context, req *AgentMessage, timeout time.Duration) (*AgentMessage, error)

SendRequest sends a request and synchronously awaits the matching reply or timeout.

func (*Service) SendSignal 

func (s *Service) SendSignal(ctx context.Context, sender, receiver AgentAddress, sigType SignalType, reason string, metadata map[string]string) (*AgentMessage, error)

SendSignal dispatches a lifecycle or control signal to a target agent or instance.

type ServiceConfig 

type ServiceConfig struct {
	Mailbox Mailbox
	Policy  Policy
	Auditor Auditor
	Metrics Metrics
	Tracer  trace.Tracer
}

ServiceConfig defines options when constructing an IPC Service.

type SignalPayload 

type SignalPayload struct {
	SignalType SignalType        `json:"signal_type"`
	Reason     string            `json:"reason,omitempty"`
	Metadata   map[string]string `json:"metadata,omitempty"`
}

SignalPayload contains structured metadata for signal messages.

type SignalType 

type SignalType string

SignalType defines system-level lifecycle and control signals for agents. 

const (
	SignalStop   SignalType = "STOP"
	SignalPause  SignalType = "PAUSE"
	SignalResume SignalType = "RESUME"
	SignalWake   SignalType = "WAKE"
	SignalCustom SignalType = "CUSTOM"
)

type StandardMetrics 

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

StandardMetrics provides in-memory thread-safe metric tracking and OTel instrument integration.

func NewStandardMetrics 

func NewStandardMetrics() *StandardMetrics

NewStandardMetrics creates a new StandardMetrics collector.

func (*StandardMetrics) AckedCount 

func (m *StandardMetrics) AckedCount() int64

func (*StandardMetrics) DeniedCount 

func (m *StandardMetrics) DeniedCount() int64

func (*StandardMetrics) ExpiredCount 

func (m *StandardMetrics) ExpiredCount() int64

func (*StandardMetrics) ReceivedCount 

func (m *StandardMetrics) ReceivedCount(msgType MessageType) int64

func (*StandardMetrics) RecordAcked 

func (m *StandardMetrics) RecordAcked(ctx context.Context, tenantID string, count int)

func (*StandardMetrics) RecordDeliveryDuration 

func (m *StandardMetrics) RecordDeliveryDuration(ctx context.Context, tenantID string, duration time.Duration)

func (*StandardMetrics) RecordDenied 

func (m *StandardMetrics) RecordDenied(ctx context.Context, tenantID string, reason string)

func (*StandardMetrics) RecordExpired 

func (m *StandardMetrics) RecordExpired(ctx context.Context, tenantID string)

func (*StandardMetrics) RecordReceived 

func (m *StandardMetrics) RecordReceived(ctx context.Context, tenantID string, msgType MessageType)

func (*StandardMetrics) RecordSent 

func (m *StandardMetrics) RecordSent(ctx context.Context, tenantID string, msgType MessageType)

func (*StandardMetrics) SentCount 

func (m *StandardMetrics) SentCount(msgType MessageType) int64

Counts for testing assertions

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.