This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 50a96fb7274a CAMEL-25424: camel-jbang - the monitor's MCP tools act on
the named integration and read fresh data (#27515)
50a96fb7274a is described below
commit 50a96fb7274a8221d4a3f71f22b1c2cc22ae2523
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Oct 7 22:17:55 2026 +0200
CAMEL-25424: camel-jbang - the monitor's MCP tools act on the named
integration and read fresh data (#27515)
* CAMEL-25424: camel-jbang - camel_control in the monitor acts on the
integration it names
The monitor's camel_control ignored the name (or pid) argument and always
acted on the selected integration, so an agent asking to stop one app could
stop another. It now acts on the integration or infra service it names,
which
becomes the selected one; without a name, on the selected one or the only
one
running.
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01STT6whBgK1AqsSsUKrnE8m
* CAMEL-25426: camel-jbang - the monitor's Overview table is up to date for
an agent on any tab
On the other tabs the monitor refreshes only the selected integration, so an
agent reading the Overview table from there got stale rows for the others
(an app started a minute ago still read 0/1 ready, uptime 2s). The read now
asks for a scan of every integration and waits (at most 4 seconds) for it.
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01STT6whBgK1AqsSsUKrnE8m
* CAMEL-25426: camel-jbang - say which tabs the monitor lists to an agent
The MCP tool that lists the monitor's tabs said it lists every tab, but it
lists those that apply to the selected integration (SQL Query once it
reports a data source), or all of them when none is selected.
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01STT6whBgK1AqsSsUKrnE8m
---------
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
.../dsl/jbang/core/commands/tui/CamelMonitor.java | 1 +
.../core/commands/tui/DataRefreshService.java | 11 ++
.../dsl/jbang/core/commands/tui/McpFacade.java | 82 ++++++++++++++-
.../core/commands/tui/TuiToolDefinitions.java | 4 +-
.../jbang/core/commands/tui/TuiToolRegistry.java | 2 +-
.../core/commands/tui/McpFacadeControlTest.java | 116 +++++++++++++++++++++
.../core/commands/tui/McpFacadeFullScanTest.java | 51 +++++++++
7 files changed, 264 insertions(+), 3 deletions(-)
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/CamelMonitor.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/CamelMonitor.java
index 05a4bd1a5ea8..90441c84e7ef 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/CamelMonitor.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/CamelMonitor.java
@@ -909,6 +909,7 @@ public class CamelMonitor extends CamelCommand {
mcpFacade.setSourceValidator(tabRegistry.sourceTab().editAssist()::validateSource);
mcpFacade.setPropertyLineValidator(tabRegistry.sourceTab().editAssist()::validateSpringBootPropertyLine);
mcpFacade.setLaunchManager(actionsPopup.getLaunchManager());
+ mcpFacade.setFullScan(dataService::forceFullScan,
dataService::fullScansDone);
aiPanel.setMcpFacade(mcpFacade);
aiPanel.setOtelSpans(dataService.otelSpans());
mcpFacade.setAiActivityLog(aiPanel::getActivityLog);
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/DataRefreshService.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/DataRefreshService.java
index 426a9acbc3d1..6a8f90242af2 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/DataRefreshService.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/DataRefreshService.java
@@ -30,6 +30,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -83,6 +84,8 @@ class DataRefreshService {
// Cached PID list -- full process scan throttled to every 2 seconds (1
second in burst mode)
private volatile List<Long> cachedPids = Collections.emptyList();
private volatile long lastFullScanTime;
+ // counts the scans of every integration whose data has been published, so
a reader can wait for the next one
+ private final AtomicLong fullScansDone = new AtomicLong();
private volatile long lastLivenessCheckTime;
private volatile long forceFullScanUntil;
private volatile long burstModeUntil;
@@ -169,6 +172,11 @@ class DataRefreshService {
forceFullScanUntil = System.currentTimeMillis() + 20_000;
}
+ /** How many scans of every integration have published their data. */
+ long fullScansDone() {
+ return fullScansDone.get();
+ }
+
boolean isBurstMode() {
return System.currentTimeMillis() < burstModeUntil;
}
@@ -361,6 +369,9 @@ class DataRefreshService {
mergePhantoms(infos);
rates.retain(infos.stream().map(i ->
i.pid).collect(Collectors.toSet()));
data.set(infos);
+ if (fullScan) {
+ fullScansDone.incrementAndGet();
+ }
return fullScan;
}
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/McpFacade.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/McpFacade.java
index 23d12664c4f7..7bec98b44e7c 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/McpFacade.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/McpFacade.java
@@ -27,7 +27,9 @@ import java.util.Queue;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import java.util.function.Function;
+import java.util.function.LongSupplier;
import java.util.function.Supplier;
+import java.util.stream.Collectors;
import dev.tamboui.buffer.Buffer;
import dev.tamboui.export.ExportRequest;
@@ -218,6 +220,10 @@ class McpFacade {
// the F2 menu's launcher: starts examples (tui_run_example) and infra
services (tui_infra start). It is held
// here, not in the tool registry, so the AI panel's registry and the MCP
server's registry both see it.
private volatile LaunchManager launchManager;
+ // asks the data refresh for a scan of every integration, and counts the
scans done: on the other tabs only the
+ // selected integration is refreshed, so the Overview rows of the others
go stale while an agent reads them
+ private volatile Runnable fullScanRequest;
+ private volatile LongSupplier fullScansDone;
McpFacade(
MonitorContext ctx,
@@ -261,6 +267,34 @@ class McpFacade {
this.launchManager = launchManager;
}
+ void setFullScan(Runnable request, LongSupplier done) {
+ this.fullScanRequest = request;
+ this.fullScansDone = done;
+ }
+
+ /**
+ * Brings the data of every integration up to date before an agent reads
the Overview from another tab, waiting at
+ * most a few seconds for the next scan.
+ */
+ void awaitFullScan() {
+ Runnable request = fullScanRequest;
+ LongSupplier done = fullScansDone;
+ if (request == null || done == null) {
+ return;
+ }
+ long before = done.getAsLong();
+ request.run();
+ long deadline = System.currentTimeMillis() + FULL_SCAN_TIMEOUT_MS;
+ while (done.getAsLong() <= before && System.currentTimeMillis() <
deadline) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ }
+
LaunchManager getLaunchManager() {
return launchManager;
}
@@ -666,9 +700,14 @@ class McpFacade {
/** How long a table read waits for a tab that loads its data on demand.
The connector action timeout is 5s. */
static final long ON_DEMAND_LOAD_TIMEOUT_MS = 8_000;
+ // a full scan runs at most every 2 seconds, on the next refresh
+ static final long FULL_SCAN_TIMEOUT_MS = 4_000;
JsonObject getTableData(String tabName) {
MonitorTab tab = resolveTab(tabName);
+ if (tab != null && tab == tabRegistry.overviewTab() && bridge != null
&& bridge.activeTab() != tab) {
+ awaitFullScan();
+ }
return tab != null ? awaitTableData(tab, ON_DEMAND_LOAD_TIMEOUT_MS) :
null;
}
@@ -1052,6 +1091,24 @@ class McpFacade {
return result;
}
+ /** The pid of the integration, or else the infra service, with the given
name (alias) or pid. */
+ private String findPidByNameOrPid(String nameOrPid) {
+ for (IntegrationInfo info : data.get()) {
+ if (!info.vanishing && (nameOrPid.equals(info.name) ||
nameOrPid.equals(info.pid))) {
+ return info.pid;
+ }
+ }
+ List<InfraInfo> infras = ctx.infraData != null ? ctx.infraData.get() :
null;
+ if (infras != null) {
+ for (InfraInfo info : infras) {
+ if (!info.vanishing && (nameOrPid.equals(info.alias) ||
nameOrPid.equals(info.pid))) {
+ return info.pid;
+ }
+ }
+ }
+ return null;
+ }
+
private IntegrationInfo findIntegration(String name) {
if (name != null && !name.isEmpty()) {
for (IntegrationInfo info : data.get()) {
@@ -1474,6 +1531,14 @@ class McpFacade {
}
String controlIntegration(String action) {
+ return controlIntegration(action, null);
+ }
+
+ /**
+ * Controls the integration with the given name or pid, which becomes the
selected one, so the screen shows what the
+ * agent acts on (CAMEL-25424). Without a name: the selected integration,
or the only one running.
+ */
+ String controlIntegration(String action, String nameOrPid) {
if (action == null || action.isBlank()) {
return "Error: action is required";
}
@@ -1481,6 +1546,21 @@ class McpFacade {
bridge.stopAll();
return "Stopping all processes";
}
+ if (nameOrPid != null && !nameOrPid.isBlank()) {
+ String pid = findPidByNameOrPid(nameOrPid.trim());
+ if (pid == null) {
+ String names = data.get().stream().filter(i -> !i.vanishing)
+ .map(i -> i.name + " (pid " + i.pid +
")").collect(Collectors.joining(", "));
+ return "Error: no integration with name or pid " + nameOrPid
+ + (names.isEmpty() ? "; none is running" : ". Known: "
+ names);
+ }
+ ctx.selectedPid = pid;
+ } else if (ctx.selectedPid == null) {
+ List<IntegrationInfo> running = data.get().stream().filter(i ->
!i.vanishing && !i.phantom).toList();
+ if (running.size() == 1) {
+ ctx.selectedPid = running.get(0).pid;
+ }
+ }
if ("close".equals(action)) {
if (ctx.selectedPid == null) {
return "Error: no integration selected";
@@ -1494,7 +1574,7 @@ class McpFacade {
return "Closed project: " + info.name;
}
if (ctx.selectedPid == null) {
- return "Error: no integration selected";
+ return "Error: no integration selected; give its name or pid";
}
String name = ctx.selectedName();
return switch (action) {
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiToolDefinitions.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiToolDefinitions.java
index e2be28681706..5a80bfdb1577 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiToolDefinitions.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiToolDefinitions.java
@@ -139,7 +139,9 @@ final class TuiToolDefinitions {
List.of("keys"))));
tools.add(toToolDef(toolDef(
"tui_get_options",
- "Lists every tab with a description of the data it provides,
plus the running integrations. "
+ "Lists the tabs with a description of the data they provide,
plus the running integrations. "
+ + "The tabs are those that apply to the
selected integration (SQL Query once it "
+ + "reports a data source), or all of them
when none is selected. "
+ "Use it when unsure which tab holds the
data for a question (e.g. 'kafka offset' "
+ "-> Kafka tab), then read that tab with
tui_get_table.",
Map.of())));
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiToolRegistry.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiToolRegistry.java
index b46d2843a098..36dd757d9f8a 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiToolRegistry.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiToolRegistry.java
@@ -1448,7 +1448,7 @@ class TuiToolRegistry {
if (action == null || action.isBlank()) {
return "Error: action is required";
}
- return facade.controlIntegration(action);
+ return facade.controlIntegration(action, (String) args.get("name"));
}
private String callOpenProject(Map<String, Object> args) {
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/McpFacadeControlTest.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/McpFacadeControlTest.java
new file mode 100644
index 000000000000..bb24bc6f384d
--- /dev/null
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/McpFacadeControlTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.dsl.jbang.core.commands.tui;
+
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * {@code camel_control} acts on the integration its name (or pid) names,
which becomes the selected one, so an agent
+ * asking to stop one app does not stop the app that happens to be selected
(CAMEL-25424).
+ */
+class McpFacadeControlTest {
+
+ private final List<String> calls = new ArrayList<>();
+ private final AtomicReference<List<IntegrationInfo>> data = new
AtomicReference<>(new ArrayList<>());
+ private final AtomicReference<List<InfraInfo>> infra = new
AtomicReference<>(new ArrayList<>());
+ private final MonitorContext ctx = new MonitorContext(data, infra);
+ private final McpFacade facade = new McpFacade(
+ ctx, data, null, null, null, null, null, null, null, null, null,
null, recordingBridge());
+
+ @Test
+ void theNamedIntegrationIsStoppedNotTheSelectedOne() {
+ integration("orders", "100");
+ integration("shipping", "200");
+ ctx.selectedPid = "200";
+
+ String answer = facade.controlIntegration("stop", "orders");
+
+ assertThat(answer).isEqualTo("Stopping orders");
+ assertThat(ctx.selectedPid).isEqualTo("100");
+ assertThat(calls).containsExactly("stopProcess [false]");
+ }
+
+ @Test
+ void aPidWorksLikeAName() {
+ integration("orders", "100");
+ integration("shipping", "200");
+
+ assertThat(facade.controlIntegration("stop-routes",
"200")).isEqualTo("Routes stopped for shipping");
+ assertThat(calls).containsExactly("sendRouteCommand [200, *, stop]");
+ }
+
+ @Test
+ void anInfraServiceIsFoundByItsAlias() {
+ InfraInfo kafka = new InfraInfo();
+ kafka.alias = "kafka";
+ kafka.pid = "300";
+ infra.get().add(kafka);
+
+ assertThat(facade.controlIntegration("stop",
"kafka")).isEqualTo("Stopping kafka");
+ assertThat(ctx.selectedPid).isEqualTo("300");
+ }
+
+ @Test
+ void anUnknownNameSaysWhatIsRunning() {
+ integration("orders", "100");
+ ctx.selectedPid = "100";
+
+ assertThat(facade.controlIntegration("stop", "nope"))
+ .startsWith("Error: no integration with name or pid
nope").contains("orders (pid 100)");
+ assertThat(calls).isEmpty();
+ }
+
+ @Test
+ void withoutANameTheOnlyOneRunningIsUsed() {
+ integration("orders", "100");
+
+ assertThat(facade.controlIntegration("restart",
null)).isEqualTo("Restarting orders");
+ assertThat(calls).containsExactly("restartProcess []");
+ }
+
+ @Test
+ void withoutANameAndSeveralRunningOneMustBeNamed() {
+ integration("orders", "100");
+ integration("shipping", "200");
+
+ assertThat(facade.controlIntegration("stop", null)).contains("give its
name or pid");
+ assertThat(calls).isEmpty();
+ }
+
+ private void integration(String name, String pid) {
+ IntegrationInfo info = new IntegrationInfo();
+ info.name = name;
+ info.pid = pid;
+ data.get().add(info);
+ }
+
+ private McpFacade.MonitorBridge recordingBridge() {
+ return (McpFacade.MonitorBridge)
Proxy.newProxyInstance(getClass().getClassLoader(),
+ new Class<?>[] { McpFacade.MonitorBridge.class }, (proxy,
method, args) -> {
+ calls.add(method.getName() + " " + Arrays.toString(args !=
null ? args : new Object[0]));
+ return method.getReturnType() == boolean.class ? false :
null;
+ });
+ }
+}
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/McpFacadeFullScanTest.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/McpFacadeFullScanTest.java
new file mode 100644
index 000000000000..592eb6de7bb3
--- /dev/null
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/McpFacadeFullScanTest.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.dsl.jbang.core.commands.tui;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * An agent reading the Overview from another tab gets the data of every
integration as of now, not as of the last time
+ * the Overview was shown: the read asks for a scan of every integration and
waits for it.
+ */
+class McpFacadeFullScanTest {
+
+ @Test
+ void theReadWaitsForTheNextFullScan() {
+ AtomicLong done = new AtomicLong();
+ McpFacade facade = new McpFacade(
+ null, new AtomicReference<>(List.of()), null, null, null,
null, null, null, null, null, null, null,
+ null);
+ // the refresh scans every integration a little later, on its own
thread
+ facade.setFullScan(
+ () -> CompletableFuture.runAsync(done::incrementAndGet,
CompletableFuture.delayedExecutor(200,
+ TimeUnit.MILLISECONDS)),
+ done::get);
+
+ facade.awaitFullScan();
+
+ assertThat(done.get()).isEqualTo(1);
+ }
+}
