Compare commits

2 Commits

Author SHA1 Message Date
hsiegeln
c3b4f70913 feat(#116): update command hooks for synchronous group response
Some checks failed
CI / cleanup-branch (push) Has been skipped
CI / build (push) Failing after 30s
CI / docker (push) Has been skipped
CI / deploy (push) Has been skipped
CI / deploy-feature (push) Has been skipped
Add CommandGroupResponse and ConfigUpdateResponse types. Switch
useSendGroupCommand and useSendRouteCommand from openapi-fetch to authFetch
returning CommandGroupResponse. Update useUpdateApplicationConfig to return
ConfigUpdateResponse and fix all consumer onSuccess callbacks to access
saved.config.version instead of saved.version.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-02 19:01:06 +02:00
hsiegeln
027e45aadf feat(#116): synchronous group command dispatch with multi-agent response collection
Add addGroupCommandWithReplies() to AgentRegistryService that sends commands
to all LIVE agents in a group and returns CompletableFuture per agent for
collecting replies. Update sendGroupCommand() and pushConfigToAgents() to
wait with a shared 10-second deadline, returning CommandGroupResponse with
per-agent status, timeouts, and overall success. Config update endpoint now
returns ConfigUpdateResponse wrapping both the saved config and push result.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-02 19:00:56 +02:00
10 changed files with 173 additions and 54 deletions

View File

@@ -3,6 +3,7 @@ package com.cameleer3.server.app.controller;
import com.cameleer3.server.app.agent.SseConnectionManager;
import com.cameleer3.server.app.dto.CommandAckRequest;
import com.cameleer3.server.app.dto.CommandBroadcastResponse;
import com.cameleer3.server.app.dto.CommandGroupResponse;
import com.cameleer3.server.app.dto.CommandRequest;
import com.cameleer3.server.app.dto.CommandSingleResponse;
import com.cameleer3.server.app.dto.ReplayRequest;
@@ -109,32 +110,60 @@ public class AgentCommandController {
@PostMapping("/groups/{group}/commands")
@Operation(summary = "Send command to all agents in a group",
description = "Sends a command to all LIVE agents in the specified group")
@ApiResponse(responseCode = "202", description = "Commands accepted")
description = "Sends a command to all LIVE agents in the specified group and waits for responses")
@ApiResponse(responseCode = "200", description = "Commands dispatched and responses collected")
@ApiResponse(responseCode = "400", description = "Invalid command payload")
public ResponseEntity<CommandBroadcastResponse> sendGroupCommand(@PathVariable String group,
@RequestBody CommandRequest request,
HttpServletRequest httpRequest) throws JsonProcessingException {
public ResponseEntity<CommandGroupResponse> sendGroupCommand(@PathVariable String group,
@RequestBody CommandRequest request,
HttpServletRequest httpRequest) throws JsonProcessingException {
CommandType type = mapCommandType(request.type());
String payloadJson = request.payload() != null ? objectMapper.writeValueAsString(request.payload()) : "{}";
List<AgentInfo> agents = registryService.findAll().stream()
.filter(a -> a.state() == AgentState.LIVE)
.filter(a -> group.equals(a.applicationId()))
.toList();
Map<String, CompletableFuture<CommandReply>> futures =
registryService.addGroupCommandWithReplies(group, type, payloadJson);
List<String> commandIds = new ArrayList<>();
for (AgentInfo agent : agents) {
AgentCommand command = registryService.addCommand(agent.instanceId(), type, payloadJson);
commandIds.add(command.id());
if (futures.isEmpty()) {
auditService.log("broadcast_group_command", AuditCategory.AGENT, group,
java.util.Map.of("type", request.type(), "agentCount", 0),
AuditResult.SUCCESS, httpRequest);
return ResponseEntity.ok(new CommandGroupResponse(true, 0, 0, List.of(), List.of()));
}
// Wait with shared 10-second deadline
long deadline = System.currentTimeMillis() + 10_000;
List<CommandGroupResponse.AgentResponse> responses = new ArrayList<>();
List<String> timedOut = new ArrayList<>();
for (var entry : futures.entrySet()) {
long remaining = deadline - System.currentTimeMillis();
if (remaining <= 0) {
timedOut.add(entry.getKey());
entry.getValue().cancel(false);
continue;
}
try {
CommandReply reply = entry.getValue().get(remaining, TimeUnit.MILLISECONDS);
responses.add(new CommandGroupResponse.AgentResponse(
entry.getKey(), reply.status(), reply.message()));
} catch (TimeoutException e) {
timedOut.add(entry.getKey());
entry.getValue().cancel(false);
} catch (Exception e) {
responses.add(new CommandGroupResponse.AgentResponse(
entry.getKey(), "ERROR", e.getMessage()));
}
}
boolean allSuccess = timedOut.isEmpty() &&
responses.stream().allMatch(r -> "SUCCESS".equals(r.status()));
auditService.log("broadcast_group_command", AuditCategory.AGENT, group,
java.util.Map.of("type", request.type(), "agentCount", agents.size()),
java.util.Map.of("type", request.type(), "agentCount", futures.size(),
"responded", responses.size(), "timedOut", timedOut.size()),
AuditResult.SUCCESS, httpRequest);
return ResponseEntity.status(HttpStatus.ACCEPTED)
.body(new CommandBroadcastResponse(commandIds, agents.size()));
return ResponseEntity.ok(new CommandGroupResponse(
allSuccess, futures.size(), responses.size(), responses, timedOut));
}
@PostMapping("/commands")

View File

@@ -1,13 +1,14 @@
package com.cameleer3.server.app.controller;
import com.cameleer3.common.model.ApplicationConfig;
import com.cameleer3.server.app.dto.CommandGroupResponse;
import com.cameleer3.server.app.dto.ConfigUpdateResponse;
import com.cameleer3.server.app.dto.TestExpressionRequest;
import com.cameleer3.server.app.dto.TestExpressionResponse;
import com.cameleer3.server.app.storage.PostgresApplicationConfigRepository;
import com.cameleer3.server.core.admin.AuditCategory;
import com.cameleer3.server.core.admin.AuditResult;
import com.cameleer3.server.core.admin.AuditService;
import com.cameleer3.server.core.agent.AgentCommand;
import com.cameleer3.server.core.agent.AgentInfo;
import com.cameleer3.server.core.agent.AgentRegistryService;
import com.cameleer3.server.core.agent.AgentState;
@@ -27,6 +28,7 @@ import org.springframework.http.ResponseEntity;
import org.springframework.security.core.Authentication;
import org.springframework.web.bind.annotation.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -88,23 +90,25 @@ public class ApplicationConfigController {
@Operation(summary = "Update application config",
description = "Saves config and pushes CONFIG_UPDATE to all LIVE agents of this application")
@ApiResponse(responseCode = "200", description = "Config saved and pushed")
public ResponseEntity<ApplicationConfig> updateConfig(@PathVariable String application,
@RequestBody ApplicationConfig config,
Authentication auth,
HttpServletRequest httpRequest) {
public ResponseEntity<ConfigUpdateResponse> updateConfig(@PathVariable String application,
@RequestBody ApplicationConfig config,
Authentication auth,
HttpServletRequest httpRequest) {
String updatedBy = auth != null ? auth.getName() : "system";
config.setApplication(application);
ApplicationConfig saved = configRepository.save(application, config, updatedBy);
int pushed = pushConfigToAgents(application, saved);
log.info("Config v{} saved for '{}', pushed to {} agent(s)", saved.getVersion(), application, pushed);
CommandGroupResponse pushResult = pushConfigToAgents(application, saved);
log.info("Config v{} saved for '{}', pushed to {} agent(s), {} responded",
saved.getVersion(), application, pushResult.total(), pushResult.responded());
auditService.log("update_app_config", AuditCategory.CONFIG, application,
Map.of("version", saved.getVersion(), "agentsPushed", pushed),
Map.of("version", saved.getVersion(), "agentsPushed", pushResult.total(),
"responded", pushResult.responded(), "timedOut", pushResult.timedOut().size()),
AuditResult.SUCCESS, httpRequest);
return ResponseEntity.ok(saved);
return ResponseEntity.ok(new ConfigUpdateResponse(saved, pushResult));
}
@GetMapping("/{application}/processor-routes")
@@ -172,24 +176,50 @@ public class ApplicationConfigController {
}
}
private int pushConfigToAgents(String application, ApplicationConfig config) {
private CommandGroupResponse pushConfigToAgents(String application, ApplicationConfig config) {
String payloadJson;
try {
payloadJson = objectMapper.writeValueAsString(config);
} catch (JsonProcessingException e) {
log.error("Failed to serialize config for push", e);
return 0;
return new CommandGroupResponse(false, 0, 0, List.of(), List.of());
}
List<AgentInfo> agents = registryService.findAll().stream()
.filter(a -> a.state() == AgentState.LIVE)
.filter(a -> application.equals(a.applicationId()))
.toList();
Map<String, CompletableFuture<CommandReply>> futures =
registryService.addGroupCommandWithReplies(application, CommandType.CONFIG_UPDATE, payloadJson);
for (AgentInfo agent : agents) {
registryService.addCommand(agent.instanceId(), CommandType.CONFIG_UPDATE, payloadJson);
if (futures.isEmpty()) {
return new CommandGroupResponse(true, 0, 0, List.of(), List.of());
}
return agents.size();
// Wait with shared 10-second deadline
long deadline = System.currentTimeMillis() + 10_000;
List<CommandGroupResponse.AgentResponse> responses = new ArrayList<>();
List<String> timedOut = new ArrayList<>();
for (var entry : futures.entrySet()) {
long remaining = deadline - System.currentTimeMillis();
if (remaining <= 0) {
timedOut.add(entry.getKey());
entry.getValue().cancel(false);
continue;
}
try {
CommandReply reply = entry.getValue().get(remaining, TimeUnit.MILLISECONDS);
responses.add(new CommandGroupResponse.AgentResponse(
entry.getKey(), reply.status(), reply.message()));
} catch (TimeoutException e) {
timedOut.add(entry.getKey());
entry.getValue().cancel(false);
} catch (Exception e) {
responses.add(new CommandGroupResponse.AgentResponse(
entry.getKey(), "ERROR", e.getMessage()));
}
}
boolean allSuccess = timedOut.isEmpty() &&
responses.stream().allMatch(r -> "SUCCESS".equals(r.status()));
return new CommandGroupResponse(allSuccess, futures.size(), responses.size(), responses, timedOut);
}
private static ApplicationConfig defaultConfig(String application) {

View File

@@ -0,0 +1,13 @@
package com.cameleer3.server.app.dto;
import java.util.List;
public record CommandGroupResponse(
boolean success,
int total,
int responded,
List<AgentResponse> responses,
List<String> timedOut
) {
public record AgentResponse(String agentId, String status, String message) {}
}

View File

@@ -0,0 +1,8 @@
package com.cameleer3.server.app.dto;
import com.cameleer3.common.model.ApplicationConfig;
public record ConfigUpdateResponse(
ApplicationConfig config,
CommandGroupResponse pushResult
) {}

View File

@@ -6,6 +6,7 @@ import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -323,6 +324,28 @@ public class AgentRegistryService {
return future;
}
/**
* Send a command to all LIVE agents in a group and return futures for collecting replies.
* Returns a map of agentId -> CompletableFuture&lt;CommandReply&gt;.
*/
public Map<String, CompletableFuture<CommandReply>> addGroupCommandWithReplies(
String group, CommandType type, String payload) {
Map<String, CompletableFuture<CommandReply>> results = new LinkedHashMap<>();
List<AgentInfo> liveAgents = findByApplication(group).stream()
.filter(a -> a.state() == AgentState.LIVE)
.toList();
for (AgentInfo agent : liveAgents) {
AgentCommand cmd = addCommand(agent.instanceId(), type, payload);
CompletableFuture<CommandReply> future = new CompletableFuture<>();
pendingReplies.put(cmd.id(), future);
future.whenComplete((r, ex) -> pendingReplies.remove(cmd.id()));
results.put(agent.instanceId(), future);
}
return results;
}
/**
* Complete a pending reply future for a command.
* Called when an agent ACKs a command that was registered via {@link #addCommandWithReply}.

View File

@@ -1,5 +1,4 @@
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
import { api } from '../client'
import { useAuthStore } from '../../auth/auth-store'
// ── Application Config ────────────────────────────────────────────────────
@@ -65,6 +64,11 @@ export function useApplicationConfig(application: string | undefined) {
})
}
export interface ConfigUpdateResponse {
config: ApplicationConfig
pushResult: CommandGroupResponse
}
export function useUpdateApplicationConfig() {
const queryClient = useQueryClient()
return useMutation({
@@ -75,10 +79,10 @@ export function useUpdateApplicationConfig() {
body: JSON.stringify(config),
})
if (!res.ok) throw new Error('Failed to update config')
return res.json() as Promise<ApplicationConfig>
return res.json() as Promise<ConfigUpdateResponse>
},
onSuccess: (saved) => {
queryClient.setQueryData(['applicationConfig', saved.application], saved)
onSuccess: (result) => {
queryClient.setQueryData(['applicationConfig', result.config.application], result.config)
queryClient.invalidateQueries({ queryKey: ['applicationConfig', 'all'] })
},
})
@@ -98,6 +102,16 @@ export function useProcessorRouteMapping(application?: string) {
})
}
// ── Group Command Response ───────────────────────────────────────────────
export interface CommandGroupResponse {
success: boolean
total: number
responded: number
responses: { agentId: string; status: string; message: string }[]
timedOut: string[]
}
// ── Generic Group Command (kept for non-config commands) ──────────────────
interface SendGroupCommandParams {
@@ -109,12 +123,13 @@ interface SendGroupCommandParams {
export function useSendGroupCommand() {
return useMutation({
mutationFn: async ({ group, type, payload }: SendGroupCommandParams) => {
const { data, error } = await api.POST('/agents/groups/{group}/commands', {
params: { path: { group } },
body: { type, payload } as any,
const res = await authFetch(`/api/v1/agents/groups/${encodeURIComponent(group)}/commands`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ type, payload }),
})
if (error) throw new Error('Failed to send command')
return data!
if (!res.ok) throw new Error('Failed to send command')
return res.json() as Promise<CommandGroupResponse>
},
})
}
@@ -163,12 +178,13 @@ export function useSendRouteCommand() {
action: 'start' | 'stop' | 'suspend' | 'resume'
routeId: string
}) => {
const { data, error } = await api.POST('/agents/groups/{group}/commands', {
params: { path: { group: application } },
body: { type: 'route-control', payload: { routeId, action, nonce: crypto.randomUUID() } } as any,
const res = await authFetch(`/api/v1/agents/groups/${encodeURIComponent(application)}/commands`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ type: 'route-control', payload: { routeId, action, nonce: crypto.randomUUID() } }),
})
if (error) throw new Error('Failed to send route command')
return data!
if (!res.ok) throw new Error('Failed to send route command')
return res.json() as Promise<CommandGroupResponse>
},
})
}

View File

@@ -155,7 +155,7 @@ export default function AppConfigDetailPage() {
updateConfig.mutate(updated, {
onSuccess: (saved) => {
setEditing(false);
toast({ title: 'Config saved', description: `${appId} updated to v${saved.version}`, variant: 'success' });
toast({ title: 'Config saved', description: `${appId} updated to v${saved.config.version}`, variant: 'success' });
},
onError: () => {
toast({ title: 'Save failed', description: 'Could not update configuration', variant: 'error' });

View File

@@ -141,7 +141,7 @@ function AppConfigDetail({ appId, onClose }: { appId: string; onClose: () => voi
if (!config || !form) return;
const updated = { ...config, ...form, tracedProcessors: tracedDraft, routeRecording: routeRecordingDraft } as ApplicationConfig;
updateConfig.mutate(updated, {
onSuccess: (saved) => { setEditing(false); toast({ title: 'Config saved', description: `${appId} updated to v${saved.version}`, variant: 'success' }); },
onSuccess: (saved) => { setEditing(false); toast({ title: 'Config saved', description: `${appId} updated to v${saved.config.version}`, variant: 'success' }); },
onError: () => { toast({ title: 'Save failed', description: 'Could not update configuration', variant: 'error' }); },
});
}

View File

@@ -145,7 +145,7 @@ export default function AgentHealth() {
onSuccess: (saved) => {
setConfigEditing(false);
setConfigDraft({});
toast({ title: 'Config updated', description: `${appId} (v${saved.version})`, variant: 'success' });
toast({ title: 'Config updated', description: `${appId} (v${saved.config.version})`, variant: 'success' });
},
onError: () => {
toast({ title: 'Config update failed', variant: 'error' });

View File

@@ -216,7 +216,7 @@ function DiagramPanel({ appId, routeId, exchangeId, onCorrelatedSelect, onClearS
if (!updatedConfig) return;
updateConfig.mutate(updatedConfig, {
onSuccess: (saved) => {
toast({ title: 'Tap configuration saved', description: `Pushed to agents (v${saved.version})`, variant: 'success' });
toast({ title: 'Tap configuration saved', description: `Pushed to agents (v${saved.config.version})`, variant: 'success' });
},
onError: () => {
toast({ title: 'Tap update failed', description: 'Could not save configuration', variant: 'error' });
@@ -229,7 +229,7 @@ function DiagramPanel({ appId, routeId, exchangeId, onCorrelatedSelect, onClearS
const taps = appConfig.taps.filter(t => t.tapId !== tap.tapId);
updateConfig.mutate({ ...appConfig, taps }, {
onSuccess: (saved) => {
toast({ title: 'Tap deleted', description: `${tap.attributeName} removed (v${saved.version})`, variant: 'success' });
toast({ title: 'Tap deleted', description: `${tap.attributeName} removed (v${saved.config.version})`, variant: 'success' });
},
onError: () => {
toast({ title: 'Tap delete failed', description: 'Could not save configuration', variant: 'error' });
@@ -256,7 +256,7 @@ function DiagramPanel({ appId, routeId, exchangeId, onCorrelatedSelect, onClearS
tracedProcessors,
}, {
onSuccess: (saved) => {
toast({ title: `Tracing ${enabled ? 'enabled' : 'disabled'}`, description: `${nodeId} — pushed to agents (v${saved.version})`, variant: 'success' });
toast({ title: `Tracing ${enabled ? 'enabled' : 'disabled'}`, description: `${nodeId} — pushed to agents (v${saved.config.version})`, variant: 'success' });
},
onError: () => {
useTracingStore.getState().toggleProcessor(appId, nodeId);