# Messaging

Agents in `tnsai-core` exchange `AgentMessage` objects. A message carries a sender, a receiver, a type and a payload; an `AgentCommunicationManager` delivers it. The same module holds the group contracts that track who is in a group and who leads it: `MembershipManager` and `LeaderElectionStrategy`. `tnsai-coordination` implements them in its group classes.

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

## 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.

| Field | Type | Meaning |
|-------|------|---------|
| `messageId` | `String` | A random UUID from `create` and `createReply` |
| `senderId` | `String` | The sending agent |
| `receiverId` | `String` | The receiving agent |
| `messageType` | `MessageTypeMarker` | The message type, for example a `CNPMessageType` value |
| `content` | `Map<String, Object>` | The payload |
| `timestamp` | `long` | Creation time in epoch milliseconds |
| `inReplyTo` | `String` | The id of the message this one answers, or null |
| `metadata` | `Map<String, Object>` | Extra data, read with `getSpec()` |

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.

<!-- java-contract: src/main/java/com/example/tnsai/docs/AgentMessageCreationExample.java -->
```java
// 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

```java
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.

| Value | Step | Meaning |
|-------|------|---------|
| `CFP` | Call for proposals | The initiator describes the task |
| `PROPOSAL` | Response | A contractor bids |
| `REFUSE` | Response | A contractor declines |
| `ACCEPT` | Selection | The initiator accepts a bid |
| `REJECT` | Selection | The initiator rejects a bid |
| `COUNTER_OFFER` | Negotiation | Either side proposes changed terms |
| `AWARD` | Finalization | The initiator awards the contract |
| `CONFIRM` | Finalization | The contractor confirms |
| `CANCEL` | Finalization | Either side cancels |
| `RESULT` | Execution | The contractor reports the result |
| `FAILURE` | Execution | The contractor reports a failure |

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

```java
// 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

```java
// 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

```java
// 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

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

### Task tracking

```java
// 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

```java
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"
)
```

| Method | Returns |
|--------|---------|
| `isHealthy()` | Whether `health` is `HEALTHY` |
| `canAcceptTask()` | Whether `activeTasks` is below `maxConcurrentTasks` |
| `getMembershipDuration()` | The time since `joinedAt` |
| `withTaskStarted()`, `withTaskCompleted()` | A copy with one more or one fewer active task, never below 0 |
| `withHeartbeat(instant)` | A copy with the new heartbeat and health `HEALTHY` |
| `withHealth(health)` | A copy with the new health |

### Statistics and history

```java
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.

```java
@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.

| Constant | Picks |
|----------|-------|
| `RANDOM` | A random member |
| `FIRST_JOINED` | The first agent in the list |
| `LAST_JOINED` | The last agent in the list |
| `ROUND_ROBIN` | The same as `FIRST_JOINED`; it keeps no rotation state |
| `ROLE_BASED` | The agent with the most roles; on a tie, the earliest in the list |
| `NONE` | Nobody (null) |

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

```java
// 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](protocols.md)
- [Group topologies](topologies.md)
- [Negotiation](negotiation.md)
- [Events](../agents/events.md)
- [SPI reference](../reference/spi.md)