protocol

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Mar 23, 2026 License: MIT Imports: 6 Imported by: 0

Documentation

Overview

Package protocol defines the agent-to-agent communication protocol.

The bus uses lock-free delivery with object pooling, bounded inboxes with back-pressure, and direct agent-to-agent channels that bypass the central router for point-to-point communication.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ReleaseEnvelope 

func ReleaseEnvelope(e *Envelope)

ReleaseEnvelope returns an envelope to the pool after clearing it.

Types

type Bus 

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

Bus is the central message router for agent-to-agent communication. Delivery is non-blocking: if a peer's inbox is full the send fails with an error rather than blocking the sender (back-pressure).

func NewBus 

func NewBus() *Bus

NewBus creates a new communication bus with default settings.

func NewBusWithConfig 

func NewBusWithConfig(cfg BusConfig) *Bus

NewBusWithConfig creates a bus with explicit resource limits.

func (*Bus) Ask 

func (b *Bus) Ask(ctx context.Context, from, to, question string) (string, error)

Ask sends a question and waits for the answer.

func (*Bus) Close 

func (b *Bus) Close()

Close shuts down the bus and all direct channels.

func (*Bus) DelegateTask 

func (b *Bus) DelegateTask(ctx context.Context, from, to, subject string, task TaskPayload) (*ResultPayload, error)

DelegateTask sends a task request and waits for the result.

func (*Bus) DirectChannelBetween 

func (b *Bus) DirectChannelBetween(agentA, agentB string, bufSize int) *DirectChannel

DirectChannelBetween returns (or creates) a dedicated channel between two agents.

func (*Bus) FindByCapability 

func (b *Bus) FindByCapability(capability string) []*Peer

FindByCapability returns agents that advertise the given capability.

func (*Bus) History 

func (b *Bus) History() []*Envelope

History returns all messages recorded by the bus (for observability/debugging).

func (*Bus) Peers 

func (b *Bus) Peers() []*Peer

Peers returns all registered agents.

func (*Bus) Register 

func (b *Bus) Register(id, name, description string, capabilities []string, handler Handler) error

Register adds an agent to the bus so it can send and receive messages.

func (*Bus) Send 

func (b *Bus) Send(ctx context.Context, env *Envelope) error

Send delivers an envelope to its recipient. For broadcasts (To=="*") the message is delivered to all peers except the sender. Returns an error immediately if the recipient's inbox is full (back-pressure).

func (*Bus) SendAndWait 

func (b *Bus) SendAndWait(ctx context.Context, env *Envelope) (*Envelope, error)

SendAndWait sends an envelope and blocks until a reply is received or the context is canceled.

func (*Bus) Unregister 

func (b *Bus) Unregister(id string)

Unregister removes an agent from the bus.

type BusConfig 

type BusConfig struct {
	InboxSize  int // per-peer inbox buffer; 0 = defaultInboxSize
	HistoryCap int // max retained history entries; 0 = defaultHistoryCap
}

BusConfig tunes Bus resource limits.

type ChatMessage 

type ChatMessage struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

ChatMessage is a message in a conversation being handed off.

type DirectChannel 

type DirectChannel struct {
	AtoB chan *Envelope
	BtoA chan *Envelope
}

DirectChannel is a dedicated bidirectional pipe between two specific agents that bypasses the central bus for minimal-latency point-to-point messaging. Each direction uses a separate buffered channel to avoid head-of-line blocking.

func NewDirectChannel 

func NewDirectChannel(bufSize int) *DirectChannel

NewDirectChannel creates a channel pair with the given buffer capacity.

func (*DirectChannel) Close 

func (dc *DirectChannel) Close()

Close drains and closes both directions.

type Envelope 

type Envelope struct {
	ID        string            `json:"id"`
	Type      MessageType       `json:"type"`
	From      string            `json:"from"`
	To        string            `json:"to"`
	ReplyTo   string            `json:"reply_to,omitempty"`
	Subject   string            `json:"subject"`
	Body      json.RawMessage   `json:"body"`
	Priority  Priority          `json:"priority"`
	Metadata  map[string]string `json:"metadata,omitempty"`
	CreatedAt time.Time         `json:"created_at"`
	ExpiresAt time.Time         `json:"expires_at,omitempty"`
}

Envelope is the unit of communication between agents.

func AcquireEnvelope 

func AcquireEnvelope() *Envelope

AcquireEnvelope returns a pooled envelope. Call ReleaseEnvelope when done.

type Handler 

type Handler func(ctx context.Context, env *Envelope) (*Envelope, error)

Handler processes an incoming envelope and optionally returns a reply.

type HandoffPayload 

type HandoffPayload struct {
	Reason       string         `json:"reason"`
	Conversation []ChatMessage  `json:"conversation,omitempty"`
	Context      map[string]any `json:"context,omitempty"`
}

HandoffPayload is the body of a TypeHandoff envelope.

type MessageType 

type MessageType string

MessageType classifies the intent of an inter-agent message. 

const (
	TypeTaskRequest MessageType = "task_request"
	TypeTaskResult  MessageType = "task_result"
	TypeQuestion    MessageType = "question"
	TypeAnswer      MessageType = "answer"
	TypeBroadcast   MessageType = "broadcast"
	TypeAck         MessageType = "ack"
	TypeError       MessageType = "error"
	TypeHandoff     MessageType = "handoff"
	TypeStatus      MessageType = "status"
)

type Peer 

type Peer struct {
	ID           string   `json:"id"`
	Name         string   `json:"name"`
	Description  string   `json:"description,omitempty"`
	Capabilities []string `json:"capabilities,omitempty"`
	// contains filtered or unexported fields
}

Peer represents a registered agent in the communication bus.

type Priority 

type Priority int

Priority controls message ordering when an agent's inbox has multiple pending messages. 

const (
	PriorityLow    Priority = 0
	PriorityNormal Priority = 1
	PriorityHigh   Priority = 2
	PriorityUrgent Priority = 3
)

type ResultPayload 

type ResultPayload struct {
	TaskID  string         `json:"task_id"`
	Success bool           `json:"success"`
	Output  map[string]any `json:"output,omitempty"`
	Error   string         `json:"error,omitempty"`
	Summary string         `json:"summary,omitempty"`
}

ResultPayload is the body of a TypeTaskResult envelope.

type StatusPayload 

type StatusPayload struct {
	TaskID   string  `json:"task_id"`
	Progress float64 `json:"progress"`
	Message  string  `json:"message"`
}

StatusPayload is the body of a TypeStatus envelope.

type TaskPayload 

type TaskPayload struct {
	Description string         `json:"description"`
	Input       map[string]any `json:"input,omitempty"`
	Constraints []string       `json:"constraints,omitempty"`
	Deadline    time.Time      `json:"deadline,omitempty"`
}

TaskPayload is the body of a TypeTaskRequest envelope.

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.