Messaging

For the higher-level paradigms (publish/subscribe, request/reply, FIPA performatives), see Protocols.

Messages

com.tnsai.communication.AgentMessage is immutable. Its constructor rejects a blank message id, sender or receiver and a null type with IllegalArgumentException. The content and metadata maps are copied, and a null map becomes an empty one.

Each field has a getter, such as getSenderId() or getContent().

Creating messages

create builds a new message. createReply swaps the sender and receiver of the original, sets inReplyTo to the original id and starts with empty metadata. withSpec(key, value) returns a copy with one more metadata entry and the same id and timestamp.

// Basic message
AgentMessage msg = AgentMessage.create(
    "agent-A",                    // sender
    "agent-B",                    // receiver
    CNPMessageType.CFP,           // type
    Map.of("task", "analyze data", "deadline", "2h")  // content
);

// Reply (automatically swaps sender/receiver, sets inReplyTo)
AgentMessage reply = AgentMessage.createReply(
    originalMessage,
    CNPMessageType.PROPOSAL,
    Map.of("bid", 100, "eta", "1h")
);

// Add metadata to existing message
AgentMessage withMeta = msg.withSpec("priority", "high");

Reading messages

boolean isReply = msg.isReply();
boolean isCfp = msg.isType(CNPMessageType.CFP);
boolean isCnp = msg.isCategory("CNP");
Map<String, Object> spec = msg.getSpec();  // alias for getMetadata()

isType compares the type with equals. isCategory compares the type's category, ignoring case.

Message types

Every message type enum implements com.tnsai.communication.MessageTypeMarker. Its default category() is the class name without MessageType, upper-cased: CNPMessageType has the category CNP. fullName() is the category and the constant, as in CNP.CFP.

The package ships four type enums: CNPMessageType, TaskMessageType, QueryMessageType and AckMessageType.

Contract Net types

CNPMessageType names the steps of the Contract Net Protocol, where an initiator asks for bids and awards the task to one contractor.

The enum only names the steps; AgentMessage does not enforce their order. Each reply links to the message before it:

// Initiator broadcasts CFP
AgentMessage cfp = AgentMessage.create(
    initiator.id(), contractor.id(),
    CNPMessageType.CFP,
    Map.of("task", "translate document", "language", "Japanese")
);

// Contractor submits proposal
AgentMessage proposal = AgentMessage.createReply(cfp,
    CNPMessageType.PROPOSAL,
    Map.of("bid", 50, "eta", "30min", "quality", "high")
);

// Initiator accepts
AgentMessage accept = AgentMessage.createReply(proposal,
    CNPMessageType.ACCEPT,
    Map.of("contractId", "c-123")
);

// Contractor reports result
AgentMessage result = AgentMessage.createReply(accept,
    CNPMessageType.RESULT,
    Map.of("translation", translatedText, "wordCount", 1500)
);

Sending messages

AgentCommunicationManager sends messages and keeps a history of them. new AgentCommunicationManager() uses LocalTransport, which delivers in the same JVM; pass a CommunicationTransport to the other constructor to use something else.

• registerAgent(receiver) registers an AgentCommunicationManager.MessageReceiver, which has id() and receiveMessage(message). LocalTransport throws IllegalStateException for an id that is already registered.
• sendMessage(senderId, receiverId, type, content) creates the message, adds it to the history and hands it to the transport. It returns the message.
• broadcastMessage(senderId, type, content) sends a message with the receiver ALL. LocalTransport delivers it to every registered agent, the sender included.
• getMessageHistory(agentId), getConversation(a, b) and getThread(messageId) read the history. A thread is the original message plus its direct replies, oldest first.

LocalTransport calls the receiver on the sending thread. A message for an unregistered receiver is dropped with a warning, and an exception in a receiver is logged and not passed back to the sender. The history is in memory until clearHistory().

sendMessage always builds a new message with create, so it cannot send a reply. To send one, create the transport yourself, pass it to the manager's constructor, and call its send with the result of createReply; such a message is not added to the manager's history.

Group membership

com.tnsai.agents.groups.MembershipManager tracks the members of one group: their health, capabilities and running tasks. Get a group's manager with group.getMembershipManager().

Registration

// Register an agent
Optional<MemberInfo> info = membershipManager.register(agent);

// Validate before registration
MembershipValidation validation = membershipManager.validate(agent);
if (!validation.valid()) {
    System.err.println("Rejected: " + validation.reason());
}

// Unregister
boolean removed = membershipManager.unregister(agentId, "task completed");

register returns an empty Optional when validation fails. The default manager rejects a null agent, a blank id, an agent that is already a member and a group at its maxMembers. A member's capabilities are the action names of its roles. Its priority, task limit and role come from the groups element of the agent's @AgentSpec; without it they are 5, 1 and member.

Health

// Record heartbeat
membershipManager.heartbeat(agentId);

// Check health
Optional<MemberHealth> health = membershipManager.getHealth(agentId);

// Get unhealthy members
List<MemberInfo> unhealthy = membershipManager.getUnhealthyMembers();

// Remove members whose heartbeat is too old
int removed = membershipManager.checkAndRemoveUnhealthy();

// Start/stop automatic health checking
membershipManager.startHealthCheck(Duration.ofSeconds(30));
membershipManager.stopHealthCheck();

MemberHealth has HEALTHY, DEGRADED, UNHEALTHY and UNKNOWN. In the default manager, getUnhealthyMembers() returns the members whose last heartbeat is older than the group policy's memberTimeout (30 seconds by default), whatever their health value. checkAndRemoveUnhealthy() removes them and publishes a MEMBER_FAILED group event. Starting a group starts this check, run every memberTimeout.

Queries

List<MemberInfo> all = membershipManager.getMembers();
Optional<MemberInfo> member = membershipManager.getMember(agentId);
List<MemberInfo> searchers = membershipManager.findByCapability("search");
int count = membershipManager.getMemberCount();

Task tracking

// Record task start/complete
membershipManager.recordTaskStart(agentId);
membershipManager.recordTaskComplete(agentId);

// Find available members (healthy + can accept more tasks)
List<MemberInfo> available = membershipManager.getAvailableMembers();

// Available with specific capability
List<MemberInfo> coders = membershipManager.getAvailableMembersWithCapability("coding");

recordTaskStart returns false when the agent is not a member or already runs maxConcurrentTasks tasks. The available-member lists hold healthy members below their task limit, highest priority first.

Member info

record MemberInfo(
    String agentId,
    String agentName,
    List<String> capabilities,
    Instant joinedAt,
    Instant lastHeartbeat,
    MemberHealth health,
    Map<String, Object> metadata,
    int priority,              // default 5
    int maxConcurrentTasks,    // default 1
    int activeTasks,           // current active count
    String role                // default "member"
)

Statistics and history

MembershipStats stats = membershipManager.getStats();
// stats.currentCount(), stats.healthyCount(), stats.averageMembershipDuration()

List<MembershipHistoryEntry> history = membershipManager.getHistory(50);
// Each entry: agentId, agentName, action (JOINED/LEFT/REMOVED_UNHEALTHY/REMOVED_POLICY),
//             reason, timestamp

The default manager keeps the last 100 history entries, newest first.

Leader election

com.tnsai.agents.groups.LeaderElectionStrategy picks one agent from a list.

@FunctionalInterface
public interface LeaderElectionStrategy {
    Agent elect(List<Agent> members);
    default Optional<Agent> electOptional(List<Agent> members) { ... }
}

group.electLeader() uses the strategy in the group policy, which is RANDOM by default; electLeader(strategy) uses the one you pass. The elected agent becomes the group leader, and a leader-elected event is published.

Every built-in returns null for a null or empty list. The default group passes its members in hash-map order, not join order, so FIRST_JOINED and LAST_JOINED do not follow joining time there.

Custom strategies

// Lambda-based custom strategy
LeaderElectionStrategy byCapabilities = members ->
    members.stream()
        .max(Comparator.comparingInt(a -> a.roles().size()))
        .orElse(null);

// Usage
Agent leader = strategy.elect(groupMembers);
Optional<Agent> leaderOpt = strategy.electOptional(groupMembers);

Related

• Protocols
• Group topologies
• Negotiation
• Events
• SPI reference