diff --git a/mateclaw-server/src/main/java/vip/mate/agent/runtime/AgentRuntimeAggregator.java b/mateclaw-server/src/main/java/vip/mate/agent/runtime/AgentRuntimeAggregator.java index a7593330..e926a034 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/runtime/AgentRuntimeAggregator.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/runtime/AgentRuntimeAggregator.java @@ -10,6 +10,7 @@ import vip.mate.channel.web.ChatStreamTracker; import vip.mate.channel.web.ChatStreamTracker.RunSnapshot; import java.util.ArrayList; +import java.util.Collection; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; @@ -116,18 +117,26 @@ public class AgentRuntimeAggregator { ) {} public RuntimeSnapshot snapshot() { + return snapshot(null); + } + + public RuntimeSnapshot snapshot(Long workspaceId) { List rawRuns = streamTracker.getAllSnapshot(); Set agentIds = rawRuns.stream() .map(RunSnapshot::agentId) .filter(java.util.Objects::nonNull) .collect(Collectors.toSet()); - for (var rec : subagentRegistry.allActive()) { + Collection rawSubagents = subagentRegistry.allActive(); + for (var rec : rawSubagents) { if (rec.agentId() != null) agentIds.add(rec.agentId()); } Map agentInfo = resolveAgents(agentIds); Map subagentCountByParent = new HashMap<>(); - for (var rec : subagentRegistry.allActive()) { + for (var rec : rawSubagents) { + if (!belongsToWorkspace(rec.agentId(), agentInfo, workspaceId)) { + continue; + } String parent = rec.parentConversationId(); if (parent != null) { subagentCountByParent.merge(parent, 1L, Long::sum); @@ -141,6 +150,7 @@ public class AgentRuntimeAggregator { int runningCount = 0; for (RunSnapshot s : rawRuns) { if (s.done()) continue; + if (!belongsToWorkspace(s.agentId(), agentInfo, workspaceId)) continue; runningCount++; String stuckReason = computeStuckReason(s); boolean orphan = s.subscriberCount() == 0; @@ -181,7 +191,8 @@ public class AgentRuntimeAggregator { return Long.compare(b.msSinceLastEvent(), a.msSinceLastEvent()); }); - List subCards = subagentRegistry.allActive().stream() + List subCards = rawSubagents.stream() + .filter(rec -> belongsToWorkspace(rec.agentId(), agentInfo, workspaceId)) .map(rec -> { long now = System.currentTimeMillis(); AgentEntity ag = rec.agentId() == null ? null : agentInfo.get(rec.agentId()); @@ -216,6 +227,33 @@ public class AgentRuntimeAggregator { return new RuntimeSnapshot(summary, cards, subCards, System.currentTimeMillis()); } + public boolean runBelongsToWorkspace(String conversationId, Long workspaceId) { + if (conversationId == null || workspaceId == null) { + return false; + } + List rawRuns = streamTracker.getAllSnapshot(); + for (RunSnapshot run : rawRuns) { + if (run.done() || !conversationId.equals(run.conversationId())) { + continue; + } + AgentEntity agent = resolveAgent(run.agentId()); + return agent != null && workspaceId.equals(agent.getWorkspaceId()); + } + return false; + } + + public boolean subagentBelongsToWorkspace(String subagentId, Long workspaceId) { + if (subagentId == null || workspaceId == null) { + return false; + } + return subagentRegistry.get(subagentId) + .map(rec -> { + AgentEntity agent = resolveAgent(rec.agentId()); + return agent != null && workspaceId.equals(agent.getWorkspaceId()); + }) + .orElse(false); + } + /** * Returns null when the run looks healthy. The returned tag is a stable * machine-readable code (not a translated label) so the frontend can @@ -244,4 +282,25 @@ public class AgentRuntimeAggregator { } return out; } + + private AgentEntity resolveAgent(Long id) { + if (id == null) { + return null; + } + try { + return agentService.getAgent(id); + } catch (Exception e) { + log.debug("agent lookup failed for id={}: {}", id, e.getMessage()); + return null; + } + } + + private static boolean belongsToWorkspace(Long agentId, Map agentInfo, + Long workspaceId) { + if (workspaceId == null) { + return true; + } + AgentEntity agent = agentId == null ? null : agentInfo.get(agentId); + return agent != null && workspaceId.equals(agent.getWorkspaceId()); + } } diff --git a/mateclaw-server/src/main/java/vip/mate/agent/runtime/AgentRuntimeController.java b/mateclaw-server/src/main/java/vip/mate/agent/runtime/AgentRuntimeController.java index e46f21f9..1c3d0b81 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/runtime/AgentRuntimeController.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/runtime/AgentRuntimeController.java @@ -22,11 +22,10 @@ import vip.mate.workspace.core.annotation.RequireGlobalAdmin; import vip.mate.agent.runtime.dsh.DshRuntimeService; /** - * Admin-only live runtime surface: the global view of every in-flight agent + * Admin-only live runtime surface: the workspace view of every in-flight agent * turn plus the controls to friendly-stop, force-recycle, or sweep stuck * runs. Distinct from {@code /api/v1/subagents/...} which is per-conversation - * owner-scoped — this controller is intentionally cross-tenant for the - * operator role. + * owner-scoped. */ @Slf4j @Tag(name = "Agent Runtime (Live)") @@ -46,9 +45,12 @@ public class AgentRuntimeController { @Operation(summary = "Snapshot of every in-flight agent turn") @GetMapping("/snapshot") @RequireGlobalAdmin - public R snapshot(Authentication auth) { + public R snapshot( + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId, + Authentication auth) { requireAdmin(auth); - return R.ok(aggregator.snapshot()); + requireWorkspace(workspaceId); + return R.ok(aggregator.snapshot(workspaceId)); } @Operation(summary = "DSH runtime availability and capability diagnostics") @@ -63,8 +65,10 @@ public class AgentRuntimeController { @PostMapping("/runs/{conversationId}/stop") @RequireGlobalAdmin public R> stopFriendly(@PathVariable String conversationId, + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId, Authentication auth) { requireAdmin(auth); + requireRunInWorkspace(conversationId, workspaceId); boolean ok = streamTracker.requestStop(conversationId); recordAudit(auth, "agent-runtime.stop", conversationId, Map.of("result", ok)); return R.ok(Map.of("stopped", ok)); @@ -74,8 +78,10 @@ public class AgentRuntimeController { @PostMapping("/runs/{conversationId}/recycle") @RequireGlobalAdmin public R> recycle(@PathVariable String conversationId, + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId, Authentication auth) { requireAdmin(auth); + requireRunInWorkspace(conversationId, workspaceId); boolean ok = streamTracker.forceRecycle(conversationId); if (ok) { finalizeRecycledConversation(conversationId); @@ -88,8 +94,10 @@ public class AgentRuntimeController { @PostMapping("/subagents/{subagentId}/interrupt") @RequireGlobalAdmin public R> interruptSubagent(@PathVariable String subagentId, + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId, Authentication auth) { requireAdmin(auth); + requireSubagentInWorkspace(subagentId, workspaceId); boolean ok = subagentRegistry.interrupt(subagentId); recordAudit(auth, "agent-runtime.subagent.interrupt", subagentId, Map.of("result", ok)); return R.ok(Map.of("interrupted", ok)); @@ -103,9 +111,12 @@ public class AgentRuntimeController { @Operation(summary = "Recycle every run currently flagged as stuck") @PostMapping("/sweep") @RequireGlobalAdmin - public R> sweep(Authentication auth) { + public R> sweep( + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId, + Authentication auth) { requireAdmin(auth); - AgentRuntimeAggregator.RuntimeSnapshot snap = aggregator.snapshot(); + requireWorkspace(workspaceId); + AgentRuntimeAggregator.RuntimeSnapshot snap = aggregator.snapshot(workspaceId); List ids = snap.runs().stream() .filter(r -> r.stuckReason() != null) .map(AgentRuntimeAggregator.RunCard::conversationId) @@ -164,6 +175,26 @@ public class AgentRuntimeController { } } + private void requireWorkspace(Long workspaceId) { + if (workspaceId == null) { + throw new MateClawException(400, "workspace id required"); + } + } + + private void requireRunInWorkspace(String conversationId, Long workspaceId) { + requireWorkspace(workspaceId); + if (!aggregator.runBelongsToWorkspace(conversationId, workspaceId)) { + throw new MateClawException(404, "runtime run not found in workspace"); + } + } + + private void requireSubagentInWorkspace(String subagentId, Long workspaceId) { + requireWorkspace(workspaceId); + if (!aggregator.subagentBelongsToWorkspace(subagentId, workspaceId)) { + throw new MateClawException(404, "subagent not found in workspace"); + } + } + private void recordAudit(Authentication auth, String action, String resourceId, Map detail) { try { diff --git a/mateclaw-server/src/test/java/vip/mate/agent/runtime/AgentRuntimeAggregatorTest.java b/mateclaw-server/src/test/java/vip/mate/agent/runtime/AgentRuntimeAggregatorTest.java new file mode 100644 index 00000000..19dea60f --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/runtime/AgentRuntimeAggregatorTest.java @@ -0,0 +1,77 @@ +package vip.mate.agent.runtime; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import vip.mate.agent.AgentService; +import vip.mate.agent.delegation.SubagentRegistry; +import vip.mate.agent.model.AgentEntity; +import vip.mate.channel.web.ChatStreamTracker; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class AgentRuntimeAggregatorTest { + + private static final long WORKSPACE_A = 10L; + private static final long WORKSPACE_B = 20L; + + @Test + void workspaceSnapshotExcludesRunsAndSubagentsFromOtherWorkspaces() { + ChatStreamTracker tracker = new ChatStreamTracker(new ObjectMapper()); + tracker.register("conv-a"); + tracker.bindRunMeta("conv-a", 100L, "alice"); + tracker.register("conv-b"); + tracker.bindRunMeta("conv-b", 200L, "bob"); + tracker.register("conv-unknown"); + + SubagentRegistry subagents = new SubagentRegistry(); + subagents.register("conv-a", "child-a", 100L, "task a", null); + subagents.register("conv-b", "child-b", 200L, "task b", null); + + AgentService agents = mock(AgentService.class); + when(agents.getAgent(100L)).thenReturn(agent(100L, WORKSPACE_A, "A")); + when(agents.getAgent(200L)).thenReturn(agent(200L, WORKSPACE_B, "B")); + + AgentRuntimeAggregator aggregator = new AgentRuntimeAggregator(tracker, subagents, agents); + + AgentRuntimeAggregator.RuntimeSnapshot snapshot = aggregator.snapshot(WORKSPACE_A); + + assertThat(snapshot.runs()) + .extracting(AgentRuntimeAggregator.RunCard::conversationId) + .containsExactly("conv-a"); + assertThat(snapshot.subagents()) + .extracting(AgentRuntimeAggregator.SubagentCard::childConversationId) + .containsExactly("child-a"); + assertThat(snapshot.summary().running()).isEqualTo(1); + assertThat(snapshot.summary().subagentsActive()).isEqualTo(1); + } + + @Test + void runBelongsToWorkspaceOnlyWhenAgentMetadataMatches() { + ChatStreamTracker tracker = new ChatStreamTracker(new ObjectMapper()); + tracker.register("conv-a"); + tracker.bindRunMeta("conv-a", 100L, "alice"); + tracker.register("conv-b"); + tracker.bindRunMeta("conv-b", 200L, "bob"); + + AgentService agents = mock(AgentService.class); + when(agents.getAgent(100L)).thenReturn(agent(100L, WORKSPACE_A, "A")); + when(agents.getAgent(200L)).thenReturn(agent(200L, WORKSPACE_B, "B")); + + AgentRuntimeAggregator aggregator = new AgentRuntimeAggregator( + tracker, new SubagentRegistry(), agents); + + assertThat(aggregator.runBelongsToWorkspace("conv-a", WORKSPACE_A)).isTrue(); + assertThat(aggregator.runBelongsToWorkspace("conv-b", WORKSPACE_A)).isFalse(); + assertThat(aggregator.runBelongsToWorkspace("missing", WORKSPACE_A)).isFalse(); + } + + private static AgentEntity agent(Long id, Long workspaceId, String name) { + AgentEntity agent = new AgentEntity(); + agent.setId(id); + agent.setWorkspaceId(workspaceId); + agent.setName(name); + return agent; + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/agent/runtime/AgentRuntimeControllerTest.java b/mateclaw-server/src/test/java/vip/mate/agent/runtime/AgentRuntimeControllerTest.java new file mode 100644 index 00000000..76610d26 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/runtime/AgentRuntimeControllerTest.java @@ -0,0 +1,108 @@ +package vip.mate.agent.runtime; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.security.authentication.UsernamePasswordAuthenticationToken; +import org.springframework.security.core.Authentication; +import org.springframework.security.core.authority.SimpleGrantedAuthority; +import vip.mate.agent.delegation.SubagentRegistry; +import vip.mate.audit.service.AuditEventService; +import vip.mate.channel.web.ChatStreamTracker; +import vip.mate.exception.MateClawException; +import vip.mate.i18n.I18nService; +import vip.mate.workspace.conversation.ConversationService; +import vip.mate.agent.runtime.dsh.DshRuntimeService; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class AgentRuntimeControllerTest { + + private static final long WORKSPACE_ID = 10L; + private AgentRuntimeAggregator aggregator; + private ChatStreamTracker streamTracker; + private SubagentRegistry subagentRegistry; + private AgentRuntimeController controller; + + @BeforeEach + void setUp() { + aggregator = mock(AgentRuntimeAggregator.class); + streamTracker = mock(ChatStreamTracker.class); + subagentRegistry = mock(SubagentRegistry.class); + controller = new AgentRuntimeController( + aggregator, + streamTracker, + subagentRegistry, + mock(AuditEventService.class), + mock(ConversationService.class), + mock(I18nService.class), + mock(DshRuntimeService.class)); + } + + @Test + void snapshotUsesCurrentWorkspace() { + AgentRuntimeAggregator.RuntimeSnapshot snapshot = + new AgentRuntimeAggregator.RuntimeSnapshot( + new AgentRuntimeAggregator.Summary(0, 0, 0, 0, 0), + List.of(), List.of(), 123L); + when(aggregator.snapshot(WORKSPACE_ID)).thenReturn(snapshot); + + assertEquals(snapshot, controller.snapshot(WORKSPACE_ID, admin()).getData()); + + verify(aggregator).snapshot(WORKSPACE_ID); + } + + @Test + void snapshotRequiresWorkspace() { + MateClawException ex = assertThrows(MateClawException.class, + () -> controller.snapshot(null, admin())); + + assertEquals(400, ex.getCode()); + } + + @Test + void stopRejectsRunOutsideCurrentWorkspace() { + when(aggregator.runBelongsToWorkspace("conv-b", WORKSPACE_ID)).thenReturn(false); + + MateClawException ex = assertThrows(MateClawException.class, + () -> controller.stopFriendly("conv-b", WORKSPACE_ID, admin())); + + assertEquals(404, ex.getCode()); + verify(streamTracker, never()).requestStop("conv-b"); + } + + @Test + void recycleRejectsRunOutsideCurrentWorkspace() { + when(aggregator.runBelongsToWorkspace("conv-b", WORKSPACE_ID)).thenReturn(false); + + MateClawException ex = assertThrows(MateClawException.class, + () -> controller.recycle("conv-b", WORKSPACE_ID, admin())); + + assertEquals(404, ex.getCode()); + verify(streamTracker, never()).forceRecycle("conv-b"); + } + + @Test + void interruptRejectsSubagentOutsideCurrentWorkspace() { + when(aggregator.subagentBelongsToWorkspace("sa-b", WORKSPACE_ID)).thenReturn(false); + + MateClawException ex = assertThrows(MateClawException.class, + () -> controller.interruptSubagent("sa-b", WORKSPACE_ID, admin())); + + assertEquals(404, ex.getCode()); + verify(subagentRegistry, never()).interrupt("sa-b"); + } + + private static Authentication admin() { + return new UsernamePasswordAuthenticationToken( + "admin", + "n/a", + List.of(new SimpleGrantedAuthority("ROLE_ADMIN"))); + } +}