Build SSE connection management and command push infrastructure for real-time agent communication.
Purpose: The server needs to push config-update, deep-trace, and replay commands to connected agents in real time via Server-Sent Events. This completes the bidirectional communication channel (agents POST data to server, server pushes commands via SSE).
- ConcurrentHashMap<String, SseEmitter> emitters for per-agent connections
- Inject AgentRegistryConfig for ping interval, inject AgentRegistryService (call setEventListener(this) in @PostConstruct)
-`connect(String agentId)`: Create SseEmitter(Long.MAX_VALUE). Register onCompletion/onTimeout/onError callbacks that remove the emitter ONLY if the current map value is the same instance (reference equality via == check to avoid Pitfall 3 from research). Replace existing emitter with put(), complete() old one if exists. Return new emitter.
-`sendEvent(String agentId, String eventId, String eventType, Object data)`: Get emitter from map, send SseEmitter.event().id(eventId).name(eventType).data(data, MediaType.APPLICATION_JSON). Catch IOException, remove emitter, return false. Return true on success.
-`sendPingToAll()`: Iterate emitters, send comment("ping") to each. Remove on IOException.
-`isConnected(String agentId)`: Check if emitter exists in map.
-`onCommandReady(String agentId, AgentCommand command)`: Attempt sendEvent with command.id() as eventId, command.type().name().toLowerCase().replace('_', '-') as event name (config-update, deep-trace, replay), command.payload() as data. If successful, call registryService.markDelivered(agentId, command.id()). If agent not connected, command stays PENDING (caller can re-send or it expires).
2.**Update AgentRegistryBeanConfig**: After creating AgentRegistryService bean, the SseConnectionManager (auto-scanned as @Component) will call setEventListener in @PostConstruct. No change needed in bean config if SseConnectionManager handles it. BUT -- to avoid circular dependency, SseConnectionManager should inject AgentRegistryService and call setEventListener(this) in @PostConstruct.
-`GET /{id}/events` (produces TEXT_EVENT_STREAM_VALUE): Check agent exists via registryService.findById(id). If null, return 404 (throw ResponseStatusException). Read Last-Event-ID header (optional) -- log it at debug level but do NOT replay missed events (per locked decision). Call connectionManager.connect(id), return the SseEmitter.
- Add @Tag(name = "Agent SSE") and @Operation annotations.
-`POST /{id}/commands`: Accept raw String body. Parse JSON: { "type": "config-update|deep-trace|replay", "payload": {...} }. Map type string to CommandType enum (config-update -> CONFIG_UPDATE, deep-trace -> DEEP_TRACE, replay -> REPLAY). Call registryService.addCommand(id, type, payloadJsonString). The AgentEventListener.onCommandReady in SseConnectionManager handles delivery. Return 202 with { commandId, status: "PENDING" or "DELIVERED" depending on whether agent is connected }.
-`POST /groups/{group}/commands`: Same body parsing. Find all LIVE agents in group via registryService.findAll() filtered by group. For each, call registryService.addCommand(). Return 202 with { commandIds: [...], targetCount: N }.
-`POST /commands`: Broadcast to all LIVE agents. Same pattern as group but uses registryService.findByState(LIVE). Return 202 with count.
-`POST /{id}/commands/{commandId}/ack`: Call registryService.acknowledgeCommand(id, commandId). Return 200 if true, 404 if false.
- Add @Tag(name = "Agent Commands") and @Operation annotations.
5.**Update WebConfig**: The SSE endpoint GET /api/v1/agents/{id}/events is already covered by the interceptor pattern "/api/v1/agents/**". Agents send the protocol version header on all requests (per research recommendation), so no exclusion needed. However, if the SSE GET causes issues because browsers/clients may not easily add custom headers to EventSource, add the SSE events path to excludePathPatterns: `/api/v1/agents/*/events`. This is a practical consideration -- add the exclusion to be safe.
**SSE Test Strategy** (from RESEARCH.md): Testing SSE with TestRestTemplate is non-trivial. Use one of these approaches:
- Option A (preferred): Use raw HttpURLConnection or java.net.http.HttpClient to open the SSE stream in a separate thread, read lines, and assert event format.
- Option B: Use Spring WebClient (from spring-boot-starter-webflux test dependency) -- BUT do not add webflux as a main dependency, only as test scope if needed.
- Option C: Test at the service layer by calling SseConnectionManager.connect() directly, then sendEvent(), and reading from the SseEmitter via a custom handler.
Recommend Option A (HttpClient) for true end-to-end testing without adding dependencies.
- Test SSE connect for registered agent: Register agent, open GET /{id}/events with Accept: text/event-stream. Assert 200 and content-type is text/event-stream.
- Test SSE connect for unknown agent: GET /unknown-id/events, assert 404.
- Test config-update delivery: Register agent, open SSE stream (background thread), POST /{id}/commands with {"type":"config-update","payload":{"key":"value"}}. Use Awaitility to assert SSE stream received event with name "config-update" and correct data.
- Test deep-trace delivery: Same pattern with {"type":"deep-trace","payload":{"correlationId":"test-123"}}.
- Test replay delivery: Same pattern with {"type":"replay","payload":{"exchangeId":"ex-456"}}.
- Test ping keepalive: Open SSE stream, wait for ping comment (may need to set ping interval low in test config or use Awaitility with timeout). Assert ":ping" comment received.
- Test Last-Event-ID header: Open SSE with Last-Event-ID header set. Assert connection succeeds (no replay, just acknowledges).
- All POST requests include X-Cameleer-Protocol-Version:1 header. SSE GET may need the header excluded in WebConfig (test will reveal if this is an issue).
- Use Awaitility with ignoreExceptions() for async assertions (established pattern).
- Test single agent command: Register agent, POST /{id}/commands, assert 202 with commandId.
- Test group command: Register 2 agents in same group, POST /groups/{group}/commands, assert 202 with targetCount=2.
- Test broadcast command: Register 3 agents, POST /commands, assert 202 with count of LIVE agents.
- Test command ack: Send command, POST /{id}/commands/{commandId}/ack, assert 200.
- Test ack unknown command: POST /{id}/commands/unknown-id/ack, assert 404.
- Test command to unregistered agent: POST /nonexistent/commands, assert 404.
**Test configuration**: If ping interval needs to be shorter for tests, add to test application.yml or use @TestPropertySource with agent-registry.ping-interval-ms=1000.
