diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java index 0d49a598..2642cde5 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java @@ -134,6 +134,19 @@ public class AgentGraphBuilder { /** PR-0b: DashScope-specific construction lives here now; we only call into it for the search-on log. */ private final vip.mate.agent.chatmodel.AgentDashScopeChatModelBuilder dashScopeBuilder; + /** + * Optional audit pipeline. Setter injection (rather than a constructor + * parameter) keeps existing constructor-based wiring + tests intact. + * When present, the executor receives it so child-agent denied-tool + * attempts can be recorded. + */ + private vip.mate.audit.service.AuditEventService auditEventService; + + @org.springframework.beans.factory.annotation.Autowired(required = false) + public void setAuditEventService(vip.mate.audit.service.AuditEventService s) { + this.auditEventService = s; + } + /** * 根据 AgentEntity 构建完整的 Agent 实例 */ @@ -347,6 +360,11 @@ public class AgentGraphBuilder { // LLM mis-calls a skill name as a tool, the response tells it // the right invocation pattern instead of a dead-end error. executor.setSkillRuntimeService(skillRuntimeService); + // Optional: route child-agent denied-tool audit events through + // the audit pipeline. Null when audit is not wired (legacy / test). + if (auditEventService != null) { + executor.setAuditEventService(auditEventService); + } PlanGenerationNode planGenerationNode = new PlanGenerationNode(chatModel, planningService, streamingHelper, conversationWindowManager, toolSet); StepExecutionNode stepExecutionNode = new StepExecutionNode(chatModel, toolSet, executor, planningService, streamTracker, reasoningEffort, streamingHelper, conversationWindowManager); PlanSummaryNode planSummaryNode = new PlanSummaryNode(chatModel, planningService, streamingHelper); @@ -478,6 +496,11 @@ public class AgentGraphBuilder { // LLM mis-calls a skill name as a tool, the response tells it // the right invocation pattern instead of a dead-end error. executor.setSkillRuntimeService(skillRuntimeService); + // Optional: route child-agent denied-tool audit events through + // the audit pipeline. Null when audit is not wired (legacy / test). + if (auditEventService != null) { + executor.setAuditEventService(auditEventService); + } // PR-1.2 (RFC-049 L1-B): propagate the bound model's capability so ReasoningNode // can gate the ThinkingLevelHolder override explicitly, rather than inferring // capability from reasoningEffort == null. diff --git a/mateclaw-server/src/main/java/vip/mate/agent/GraphEventPublisher.java b/mateclaw-server/src/main/java/vip/mate/agent/GraphEventPublisher.java index db5165f9..305e0d13 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/GraphEventPublisher.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/GraphEventPublisher.java @@ -81,10 +81,15 @@ public final class GraphEventPublisher { public static GraphEvent toolComplete(String toolCallId, String toolName, String result, boolean success) { long ts = System.currentTimeMillis(); + // Carry the full tool result; transport-layer chunking lives in + // ChatStreamTracker.broadcastChunked, which splits oversize payloads + // into ordered tool_result_chunk events when they exceed the 8 KB + // single-event budget. The previous unconditional 500-char truncation + // here destroyed data that the front-end could otherwise render in full. return new GraphEvent(EVENT_TOOL_COMPLETE, Map.of( "toolCallId", toolCallId != null ? toolCallId : "", "toolName", toolName, - "result", result != null ? truncateResult(result) : "", + "result", result != null ? result : "", "success", success, "timestamp", ts ), ts); @@ -110,9 +115,11 @@ public final class GraphEventPublisher { public static GraphEvent stepCompleted(int index, String result) { long ts = System.currentTimeMillis(); + // Full step result; broadcastChunked splits at the transport layer + // when the payload exceeds the per-event size budget. return new GraphEvent(EVENT_STEP_COMPLETED, Map.of( "index", index, - "result", result != null ? truncateResult(result) : "", + "result", result != null ? result : "", "timestamp", ts ), ts); } @@ -123,7 +130,7 @@ public final class GraphEventPublisher { return new GraphEvent(EVENT_TOOL_APPROVAL_REQUESTED, Map.of( "pendingId", pendingId, "toolName", toolName != null ? toolName : "", - "arguments", arguments != null ? truncateResult(arguments) : "", + "arguments", arguments != null ? arguments : "", "reason", reason != null ? reason : "", "timestamp", ts ), ts); @@ -140,7 +147,7 @@ public final class GraphEventPublisher { java.util.Map data = new java.util.LinkedHashMap<>(); data.put("pendingId", pendingId); data.put("toolName", toolName != null ? toolName : ""); - data.put("arguments", arguments != null ? truncateForBroadcast(arguments) : ""); + data.put("arguments", arguments != null ? arguments : ""); data.put("reason", reason != null ? reason : ""); data.put("summary", summary); data.put("maxSeverity", maxSeverity); @@ -196,14 +203,90 @@ public final class GraphEventPublisher { .orElse(List.of()); } + /** + * Pass-through; preserved for source/binary compatibility with older callers. + * Truncation at the SSE transport layer is now handled by {@code + * ChatStreamTracker.broadcastChunked} which splits oversize payloads into + * ordered chunk events instead of dropping bytes. Logs a one-time + * deprecation hint when invoked. + * + * @deprecated callers should pass full payloads and let the transport layer + * decide whether to chunk. + */ + @Deprecated private static String truncateResult(String result) { - return result.length() > 500 ? result.substring(0, 500) + "..." : result; + warnTruncateDeprecation(); + return result; } /** - * 截断字符串用于直推广播（公共方法，供 Node 直接构造广播数据时使用） + * Pass-through, kept for source compatibility with code that built broadcast + * payloads directly. Same deprecation reason as {@link #truncateResult}. + * + * @deprecated callers should pass full payloads. */ + @Deprecated public static String truncateForBroadcast(String text) { - return truncateResult(text); + warnTruncateDeprecation(); + return text; + } + + private static final java.util.concurrent.atomic.AtomicBoolean TRUNCATE_WARNED = + new java.util.concurrent.atomic.AtomicBoolean(false); + + private static void warnTruncateDeprecation() { + if (TRUNCATE_WARNED.compareAndSet(false, true)) { + org.slf4j.LoggerFactory.getLogger(GraphEventPublisher.class) + .warn("GraphEventPublisher.truncateResult/truncateForBroadcast are deprecated " + + "no-op pass-throughs; payloads are no longer truncated here. " + + "Move callers to send the full string and rely on " + + "ChatStreamTracker.broadcastChunked for transport-level chunking."); + } + } + + // ===== Iteration lifecycle events ===== + + public static final String EVENT_ITERATION_START = "iteration_start"; + public static final String EVENT_ITERATION_END = "iteration_end"; + + /** + * Marks the entry of an iteration boundary so consumers can group later + * tool / content / thinking events under a single logical step. The + * {@code scope} field distinguishes the parent agent ("parent") from a + * delegated sub-agent ("subagent"); when scope is "subagent" the + * {@code subagentId} payload field is populated by the producer. + */ + public static GraphEvent iterationStart(int index, String reason, String scope, String subagentId) { + long ts = System.currentTimeMillis(); + Map data = new java.util.LinkedHashMap<>(); + data.put("index", index); + data.put("reason", reason != null ? reason : ""); + data.put("scope", scope != null ? scope : "parent"); + if (subagentId != null && !subagentId.isEmpty()) { + data.put("subagentId", subagentId); + } + data.put("timestamp", ts); + return new GraphEvent(EVENT_ITERATION_START, Map.copyOf(data), ts); + } + + /** + * Closes the matching {@link #iterationStart} boundary. {@code contentChars} + * and {@code thinkingChars} let consumers render a compact "this turn + * produced X content / Y thinking" header without re-aggregating the + * underlying delta events. + */ + public static GraphEvent iterationEnd(int index, String scope, String subagentId, + int contentChars, int thinkingChars) { + long ts = System.currentTimeMillis(); + Map data = new java.util.LinkedHashMap<>(); + data.put("index", index); + data.put("scope", scope != null ? scope : "parent"); + if (subagentId != null && !subagentId.isEmpty()) { + data.put("subagentId", subagentId); + } + data.put("contentChars", contentChars); + data.put("thinkingChars", thinkingChars); + data.put("timestamp", ts); + return new GraphEvent(EVENT_ITERATION_END, Map.copyOf(data), ts); } } diff --git a/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentController.java b/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentController.java new file mode 100644 index 00000000..2dce4adb --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentController.java @@ -0,0 +1,189 @@ +package vip.mate.agent.delegation; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.tags.Tag; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.security.core.Authentication; +import org.springframework.web.bind.annotation.*; +import vip.mate.audit.service.AuditEventService; +import vip.mate.common.result.R; +import vip.mate.exception.MateClawException; +import vip.mate.workspace.conversation.ConversationService; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** + * REST surface for managing live sub-agents: + *
    + *
  • {@code POST /interrupt} — stop a running sub-agent.
    • + *
  • {@code POST /spawn-pause} — toggle the per-parent spawn-pause flag.
    • + *
  • {@code GET /active} — list sub-agents under one parent conversation.
    • + *
+ * + *

Every endpoint authorizes the caller against the parent conversation's + * owner before mutating or revealing anything; the {@code parentConversationId} + * query parameter on {@code /active} is mandatory so the route cannot be used + * to enumerate cross-tenant sub-agents. + * + *

Authorization mirrors the {@link vip.mate.workspace.conversation.ConversationService#isConversationOwner} + * pattern used by the chat stop / fork routes — usernames are the principal + * identity carried on {@link Authentication#getName()}, and shared "system" + * conversations are accessible to all logged-in users (matches the existing + * cron-job convention). + */ +@Slf4j +@Tag(name = "Sub-agents") +@RestController +@RequestMapping("/api/v1/subagents") +@RequiredArgsConstructor +public class SubagentController { + + private final SubagentRegistry registry; + private final ConversationService conversationService; + private final AuditEventService auditEventService; + private final ObjectMapper objectMapper; + + /** + * Resolve the record and verify the caller owns its parent conversation. + * Throws a 403-coded exception when ownership fails so the global handler + * can render a uniform JSON error envelope. + */ + private SubagentRegistry.SubagentRecord requireOwnership(String subagentId, Authentication auth) { + Optional opt = registry.get(subagentId); + if (opt.isEmpty()) { + throw new MateClawException(404, "subagent " + subagentId + " not found"); + } + SubagentRegistry.SubagentRecord rec = opt.get(); + String username = currentUsername(auth); + if (!conversationService.isConversationOwner(rec.parentConversationId(), username)) { + // Audit denial separately from the operation itself so admins can + // see what cross-tenant attempts hit the registry. Best-effort + // serialization — the audit insert is async on the service side. + auditEventService.record("subagent.interrupt.denied", "subagent", + subagentId, rec.subagentId(), + safeJson(Map.of( + "callerUsername", username, + "parent", rec.parentConversationId(), + "agentId", rec.agentId() == null ? -1L : rec.agentId() + ))); + throw new MateClawException(403, "not the owner of subagent's parent conversation"); + } + return rec; + } + + /** + * Stop a running sub-agent. The registry flips status to {@code interrupted} + * and disposes the streaming subscription if one was registered. Returns + * the {@code interrupted} flag so the caller can distinguish "we did stop + * something" from "the subagent was already finished" (404 case is handled + * separately by {@link #requireOwnership}). + */ + @Operation(summary = "Interrupt a running sub-agent") + @PostMapping("/{subagentId}/interrupt") + public R> interrupt(@PathVariable String subagentId, Authentication auth) { + SubagentRegistry.SubagentRecord rec = requireOwnership(subagentId, auth); + boolean ok = registry.interrupt(subagentId); + auditEventService.record("subagent.interrupt", "subagent", + subagentId, rec.subagentId(), + safeJson(Map.of( + "by", currentUsername(auth), + "parent", rec.parentConversationId(), + "result", ok + ))); + return R.ok(Map.of("interrupted", ok)); + } + + /** + * Toggle whether new sub-agent spawns are accepted under a parent + * conversation. Used by the operator UI to halt runaway parent agents + * mid-turn without killing the parent's own LLM call. + */ + @Operation(summary = "Set sub-agent spawn-pause for a conversation") + @PostMapping("/spawn-pause") + public R> setPaused(@RequestBody Map body, Authentication auth) { + Object parentObj = body == null ? null : body.get("parentConversationId"); + String parent = parentObj == null ? null : parentObj.toString(); + if (parent == null || parent.isBlank()) { + throw new MateClawException(400, "parentConversationId required"); + } + String username = currentUsername(auth); + if (!conversationService.isConversationOwner(parent, username)) { + throw new MateClawException(403, "not the owner of conversation " + parent); + } + boolean paused = Boolean.TRUE.equals(body.get("paused")); + registry.setSpawnPaused(parent, paused); + auditEventService.record("subagent.spawn-pause", "conversation", + parent, parent, + safeJson(Map.of( + "paused", paused, + "by", username + ))); + return R.ok(Map.of("paused", paused)); + } + + /** + * List the sub-agents currently active under {@code parentConversationId}. + * The query parameter is mandatory: returning all subagents process-wide + * would let any logged-in user enumerate other tenants' delegation trees. + */ + @Operation(summary = "List active sub-agents under a parent conversation") + @GetMapping("/active") + public R> listActive(@RequestParam(required = false) String parentConversationId, + Authentication auth) { + if (parentConversationId == null || parentConversationId.isBlank()) { + throw new MateClawException(400, "parentConversationId required"); + } + String username = currentUsername(auth); + if (!conversationService.isConversationOwner(parentConversationId, username)) { + throw new MateClawException(403, "not the owner of conversation " + parentConversationId); + } + List> snapshot = registry.snapshot(parentConversationId).stream() + .map(this::toResponseDto) + .toList(); + return R.ok(Map.of("subagents", snapshot)); + } + + /** Username from auth context; falls back to "anonymous" only when null. */ + private String currentUsername(Authentication auth) { + return auth != null ? auth.getName() : "anonymous"; + } + + /** + * DTO projection that drops the {@link reactor.core.Disposable} (not + * serializable to the wire) and exposes only the user-facing fields. + */ + private Map toResponseDto(SubagentRegistry.SubagentRecord rec) { + Map dto = new LinkedHashMap<>(); + dto.put("subagentId", rec.subagentId()); + dto.put("parentConversationId", rec.parentConversationId()); + dto.put("childConversationId", rec.childConversationId()); + dto.put("agentId", rec.agentId()); + dto.put("goal", rec.goal()); + dto.put("startedAt", rec.startedAt()); + dto.put("status", rec.status().get()); + dto.put("toolCount", rec.toolCount().get()); + dto.put("lastTool", rec.lastTool().get()); + dto.put("currentPhase", rec.currentPhase().get()); + return dto; + } + + /** + * Best-effort JSON serialization for audit detail. Falling back to a + * marker string keeps the audit row insertable when payload contains + * a non-serializable value — the alternative (throwing) would lose the + * audit record entirely. + */ + private String safeJson(Map payload) { + try { + return objectMapper.writeValueAsString(payload); + } catch (JsonProcessingException e) { + return "{\"error\":\"audit_serialization_failed\"}"; + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentHeartbeat.java b/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentHeartbeat.java new file mode 100644 index 00000000..564de2c0 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentHeartbeat.java @@ -0,0 +1,101 @@ +package vip.mate.agent.delegation; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import vip.mate.channel.web.ChatStreamTracker; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Objects; + +/** + * Periodic watchdog that flips a sub-agent's status to {@code stale} when its + * child stream stops making observable progress. + * + *

Progress is probed via {@link ChatStreamTracker#getRunningToolName} and + * {@link ChatStreamTracker#getCurrentPhase}. When neither has changed across + * the configured number of cycles, the record is marked stale and a + * {@code subagent_stale} event is broadcast on the parent conversation so the + * UI can surface the issue. Cycle count uses two separate thresholds — one + * for idle children and one for children mid-tool — because legitimately slow + * tools (large file scans, slow LLM calls) need a longer window than an idle + * model that has simply gone quiet. + * + *

The runtime tool name + phase combination is a deliberately coarse + * progress signal: it does not require introspecting LLM token deltas, which + * keeps the watchdog cheap and avoids racing with the streaming hot path. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class SubagentHeartbeat { + + private final SubagentRegistry registry; + private final SubagentHeartbeatConfig cfg; + private final ChatStreamTracker streamTracker; + + /** + * Scheduled tick. Defaults to every 30 s; controlled by + * {@code mateclaw.delegation.heartbeat.intervalSec}. + */ + @Scheduled(fixedRateString = "#{@subagentHeartbeatConfig.intervalSec * 1000L}") + public void check() { + for (var rec : registry.allActive()) { + if (!"running".equals(rec.status().get())) { + continue; + } + evaluate(rec); + } + } + + /** + * Visible for testing — apply one heartbeat tick to a single record so + * tests can drive the watchdog deterministically without scheduling. + */ + void evaluate(SubagentRegistry.SubagentRecord rec) { + // Probe child progress. We use (currentTool, currentPhase) as the + // monotonic-progress signal: any change in either implies the child + // advanced at least one observable step. We deliberately do NOT + // depend on a private apiCallCount field — the RunState does not + // expose one, and counting deltas across the streaming hot path + // would race with token emission. Tool/phase ticks are atomic + // volatile writes from the streaming layer, so reading them here + // is cheap and correct. + String currentTool = streamTracker.getRunningToolName(rec.childConversationId()); + String currentPhase = streamTracker.getCurrentPhase(rec.childConversationId()); + int phaseHash = currentPhase != null ? currentPhase.hashCode() : 0; + + boolean toolChanged = !Objects.equals(currentTool, rec.lastSeenTool().get()); + boolean phaseChanged = phaseHash != rec.lastSeenIter().get(); + + if (toolChanged || phaseChanged) { + rec.lastSeenTool().set(currentTool); + rec.lastSeenIter().set(phaseHash); + rec.staleCount().set(0); + return; + } + + int sc = rec.staleCount().incrementAndGet(); + int limit = (currentTool != null && !currentTool.isEmpty()) + ? cfg.getStaleCyclesInTool() + : cfg.getStaleCyclesIdle(); + + if (sc >= limit) { + // Atomic transition: only the first thread to flip running -> stale + // emits the event. Subsequent ticks fall through the running guard + // in check(). + if (rec.status().compareAndSet("running", "stale")) { + Map payload = new LinkedHashMap<>(); + payload.put("subagentId", rec.subagentId()); + payload.put("cycles", sc); + payload.put("lastTool", currentTool != null ? currentTool : ""); + payload.put("elapsedMs", System.currentTimeMillis() - rec.startedAt()); + streamTracker.broadcastObject(rec.parentConversationId(), "subagent_stale", payload); + log.info("[SubagentHeartbeat] subagent {} marked stale after {} idle cycles (limit={})", + rec.subagentId(), sc, limit); + } + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentHeartbeatConfig.java b/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentHeartbeatConfig.java new file mode 100644 index 00000000..b32bfd0d --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentHeartbeatConfig.java @@ -0,0 +1,70 @@ +package vip.mate.agent.delegation; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +/** + * Configuration knobs for {@link SubagentHeartbeat}. + * + *

Defaults are tuned so a wedged child surfaces visibly to the parent UI + * without firing on legitimately slow tool runs: + *

    + *
  • Idle (no tool running) → 5 cycles × 30 s = 150 s before stale.
    • + *
  • In a tool → 20 cycles × 30 s = 600 s before stale.
    • + *
+ * + *

The in-tool threshold MUST stay greater than or equal to + * {@code child_hard_timeout / intervalSec}. If it fires before the per-child + * hard cap then the cap stops being the source of truth for "this child is + * dead" and operators see ambiguous telemetry. + */ +@Component +@ConfigurationProperties("mateclaw.delegation.heartbeat") +public class SubagentHeartbeatConfig { + + /** + * Heartbeat check interval in seconds. Lower values make the parent + * transcript more responsive at the cost of scheduler overhead. + */ + private int intervalSec = 30; + + /** + * Stale threshold (in heartbeat cycles) when the child has no current + * tool in flight. With the default 30 s interval this is 150 s, tight + * enough that a wedged child does not mask a legitimate gateway timeout. + */ + private int staleCyclesIdle = 5; + + /** + * Stale threshold (in heartbeat cycles) while the child is inside a + * tool. Generous enough to tolerate slow tools (large file reads, slow + * LLM calls). Must be at least the per-child hard timeout divided by + * {@link #intervalSec}, otherwise stale fires before the hard cap and + * obscures fallback semantics. + */ + private int staleCyclesInTool = 20; + + public int getIntervalSec() { + return intervalSec; + } + + public void setIntervalSec(int intervalSec) { + this.intervalSec = intervalSec; + } + + public int getStaleCyclesIdle() { + return staleCyclesIdle; + } + + public void setStaleCyclesIdle(int staleCyclesIdle) { + this.staleCyclesIdle = staleCyclesIdle; + } + + public int getStaleCyclesInTool() { + return staleCyclesInTool; + } + + public void setStaleCyclesInTool(int staleCyclesInTool) { + this.staleCyclesInTool = staleCyclesInTool; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentRegistry.java b/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentRegistry.java new file mode 100644 index 00000000..f21e8cb4 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/delegation/SubagentRegistry.java @@ -0,0 +1,173 @@ +package vip.mate.agent.delegation; + +import org.springframework.stereotype.Component; +import reactor.core.Disposable; + +import java.security.SecureRandom; +import java.util.Collection; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Process-wide registry of live sub-agents spawned through the delegation flow. + * + *

Holds the in-memory subagent tree so the parent transcript, the heartbeat + * watcher, and the operator UI can observe / interrupt children that the parent + * conversation spawned. Records use atomic accessors throughout because the + * heartbeat thread may mutate {@code staleCount} / {@code status} concurrently + * with the spawning thread that registered the record. + * + *

The pause flag is keyed per parent conversation so two unrelated users + * cannot freeze each other's spawning by toggling a global switch. + */ +@Component +public class SubagentRegistry { + + /** + * Single live sub-agent. + * + *

Mutable counters are atomics so the heartbeat scheduler and the + * spawn / completion thread can update them without locking. Status is + * driven by external lifecycle events; allowed values are + * {@code running} / {@code completed} / {@code interrupted} / {@code stale} + * / {@code timeout}. + */ + public record SubagentRecord( + String subagentId, + String parentConversationId, + String childConversationId, + Long agentId, + String goal, + long startedAt, + AtomicReference status, + AtomicInteger toolCount, + AtomicReference lastTool, + AtomicReference currentPhase, + AtomicInteger lastSeenIter, + AtomicReference lastSeenTool, + AtomicInteger staleCount, + AtomicLong firstApiCallAt, + Disposable disposable + ) {} + + private final ConcurrentMap active = new ConcurrentHashMap<>(); + + /** + * Per-parent pause flag set: scoping prevents one user from freezing + * another user's spawning. A parent conversation appears in this set iff + * spawning is currently paused for it. + */ + private final Set pausedParents = ConcurrentHashMap.newKeySet(); + + private final SecureRandom rng = new SecureRandom(); + + /** + * Register a freshly spawned sub-agent. Returns the assigned subagentId + * which the caller must thread through to {@link #unregister(String)} on + * completion (success / failure / timeout) so the registry does not leak. + * + *

ID format {@code sa--<8 hex chars>} keeps IDs sortable by + * spawn time while the random suffix prevents collisions when many + * children spawn within the same millisecond. + */ + public String register(String parentConvId, String childConvId, Long agentId, String goal, Disposable d) { + String sid = "sa-" + System.currentTimeMillis() + "-" + nextHexSuffix(); + active.put(sid, new SubagentRecord( + sid, + parentConvId, + childConvId, + agentId, + goal, + System.currentTimeMillis(), + new AtomicReference<>("running"), + new AtomicInteger(0), + new AtomicReference<>(""), + new AtomicReference<>("starting"), + new AtomicInteger(0), + new AtomicReference<>(null), + new AtomicInteger(0), + new AtomicLong(0), + d)); + return sid; + } + + /** + * Mark a sub-agent as interrupted and dispose its underlying stream + * subscription if one was registered. Returns {@code false} when the + * subagentId is unknown (already cleaned up or never registered) so + * callers can distinguish "not running anymore" from "interrupted". + */ + public boolean interrupt(String subagentId) { + if (subagentId == null) return false; + SubagentRecord r = active.get(subagentId); + if (r == null) return false; + r.status().set("interrupted"); + Disposable d = r.disposable(); + if (d != null && !d.isDisposed()) { + d.dispose(); + } + return true; + } + + public Optional get(String subagentId) { + return subagentId == null ? Optional.empty() : Optional.ofNullable(active.get(subagentId)); + } + + /** + * Snapshot of all sub-agents whose parent matches {@code parentConvId}. + * Filtering at the registry boundary prevents callers from accidentally + * surfacing other tenants' subagents in API responses. + */ + public List snapshot(String parentConvId) { + if (parentConvId == null) return List.of(); + return active.values().stream() + .filter(r -> parentConvId.equals(r.parentConversationId())) + .toList(); + } + + public void unregister(String subagentId) { + if (subagentId == null) return; + active.remove(subagentId); + } + + public boolean isSpawnPaused(String parentConvId) { + if (parentConvId == null) return false; + return pausedParents.contains(parentConvId); + } + + /** + * Toggle the pause flag for one parent conversation. Returns the new + * paused state so the caller can echo the resulting flag without an + * extra read. + */ + public boolean setSpawnPaused(String parentConvId, boolean paused) { + if (parentConvId == null) return false; + if (paused) { + pausedParents.add(parentConvId); + } else { + pausedParents.remove(parentConvId); + } + return paused; + } + + public Collection allActive() { + return active.values(); + } + + /** Lowercase 8-hex-char suffix sourced from a SecureRandom. */ + private String nextHexSuffix() { + byte[] bytes = new byte[4]; + rng.nextBytes(bytes); + StringBuilder sb = new StringBuilder(8); + for (byte b : bytes) { + sb.append(String.format("%02x", b)); + } + return sb.toString(); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java index c222b8a8..e8f174bd 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java @@ -701,12 +701,47 @@ public class NodeStreamingChatHelper { AtomicInteger cacheReadTokens = new AtomicInteger(0); AtomicInteger cacheWriteTokens = new AtomicInteger(0); - // 重复检测器：检测 LLM 退化输出（如不断重复同一句话） - RepetitionDetector contentRepDetector = new RepetitionDetector(); - RepetitionDetector thinkingRepDetector = new RepetitionDetector(); + // Cross-call repetition detectors: scoped to the conversation rather + // than this single LLM call so the sentence-level path can catch + // adjacent-iteration loops. Tracker returns a fresh detector when + // conversationId is unknown (tests, legacy callers); mocked trackers + // without stubs may return null, so we fall back defensively. + final RepetitionDetector contentRepDetector = + pickDetector(streamTracker != null ? streamTracker.getContentRepDetector(conversationId) : null); + final RepetitionDetector thinkingRepDetector = + pickDetector(streamTracker != null ? streamTracker.getThinkingRepDetector(conversationId) : null); // 重复检测触发后设为 true，外层轮询线程据此 dispose 订阅 AtomicBoolean repetitionTriggered = new AtomicBoolean(false); + // Lifecycle events emitted at most once per call so consumers can + // pivot the UI between "thinking" and "drafting" without inspecting + // delta rates. + AtomicBoolean thinkingStartEmitted = new AtomicBoolean(false); + AtomicBoolean thinkingEndEmitted = new AtomicBoolean(false); + AtomicBoolean firstTokenSignaled = new AtomicBoolean(false); + + // Pre-stream lifecycle: tell the front-end how the prompt was sized + // and which provider/model is being asked. These events ride on the + // existing SSE bus, so the heartbeat / first_token signaling stays + // consistent. + if (broadcast && streamTracker != null && conversationId != null && !conversationId.isEmpty()) { + int messageCount = prompt.getInstructions() != null ? prompt.getInstructions().size() : 0; + int contextChars = approximatePromptChars(prompt); + streamTracker.broadcastObject(conversationId, "context_prepared", Map.of( + "messages", messageCount, + "contextChars", contextChars, + "timestamp", System.currentTimeMillis() + )); + String modelId = identifyModel(chatModel); + String providerId = primaryProviderId != null ? primaryProviderId : ""; + streamTracker.broadcastObject(conversationId, "llm_request_sent", Map.of( + "provider", providerId, + "model", modelId != null ? modelId : "", + "phase", phase != null ? phase : "", + "timestamp", System.currentTimeMillis() + )); + } + CountDownLatch latch = new CountDownLatch(1); Disposable subscription = chatModel.stream(prompt) @@ -729,9 +764,25 @@ public class NodeStreamingChatHelper { if (contentRepDetector.appendAndCheck(contentDelta)) { log.warn("[{}] Content repetition detected, will cancel stream " + "for conversation {}", phase, conversationId); + broadcastContentTruncated(conversationId, + contentRepDetector.lastTriggerReason(), + contentAccum.length()); repetitionTriggered.set(true); return; } + // First content delta closes the thinking phase if one + // was open, and arms first-token heartbeat relaxation. + if (broadcast && streamTracker != null + && firstTokenSignaled.compareAndSet(false, true)) { + streamTracker.markFirstTokenReceived(conversationId); + } + if (broadcast && thinkingAccum.length() > 0 + && thinkingEndEmitted.compareAndSet(false, true)) { + streamTracker.broadcastObject(conversationId, "thinking_end", Map.of( + "thinkingChars", thinkingAccum.length(), + "timestamp", System.currentTimeMillis() + )); + } contentAccum.append(contentDelta); if (broadcast) { broadcastDelta(conversationId, "content_delta", contentDelta); @@ -744,9 +795,29 @@ public class NodeStreamingChatHelper { if (thinkingRepDetector.appendAndCheck(thinkingDelta)) { log.warn("[{}] Thinking repetition detected, will cancel stream " + "for conversation {}", phase, conversationId); + broadcastContentTruncated(conversationId, + thinkingRepDetector.lastTriggerReason(), + thinkingAccum.length()); repetitionTriggered.set(true); return; } + // First-token signaling fires for thinking too — UI + // shows "thinking" activity before any content streams. + if (broadcast && streamTracker != null + && firstTokenSignaled.compareAndSet(false, true)) { + streamTracker.markFirstTokenReceived(conversationId); + } + // First thinking delta opens the thinking phase. We + // emit the start lazily (on first delta) rather than + // before subscription so models that never produce + // thinking don't ghost-pair an empty segment. + if (broadcast && thinkingAccum.length() == 0 + && thinkingStartEmitted.compareAndSet(false, true)) { + streamTracker.broadcastObject(conversationId, "thinking_start", Map.of( + "phase", phase != null ? phase : "", + "timestamp", System.currentTimeMillis() + )); + } thinkingAccum.append(thinkingDelta); // thinkingLevel=off 时不广播 thinking（模型仍可能产生，但前端不展示） boolean suppressThinking = "off".equalsIgnoreCase( @@ -1411,6 +1482,79 @@ public class NodeStreamingChatHelper { streamTracker.broadcast(conversationId, eventName, json); } + /** + * Broadcast a {@code content_truncated} lifecycle event so consumers can + * surface why the stream stopped early. {@code reason} is "char_pattern" + * or "sentence_repetition" depending on which detector fired. + */ + private void broadcastContentTruncated(String conversationId, String reason, int truncatedChars) { + if (streamTracker == null || conversationId == null || conversationId.isEmpty()) { + return; + } + try { + streamTracker.broadcastObject(conversationId, "content_truncated", Map.of( + "reason", reason != null ? reason : "char_pattern", + "truncatedChars", truncatedChars, + "timestamp", System.currentTimeMillis() + )); + } catch (Exception e) { + log.debug("Failed to broadcast content_truncated for {}: {}", conversationId, e.getMessage()); + } + } + + /** + * Resolve a non-null {@link RepetitionDetector}. Returns the supplied + * detector when present; otherwise creates a fresh per-call instance so + * mocked trackers (Mockito returns null for unstubbed methods) and + * legacy code paths never trip a NullPointerException. + */ + private static RepetitionDetector pickDetector(RepetitionDetector candidate) { + return candidate != null ? candidate : new RepetitionDetector(); + } + + /** + * Best-effort character count of the outbound prompt for the + * {@code context_prepared} event. Cheaper than tokenizing and only used + * for UI presentation, so an exact figure is unnecessary. + */ + private static int approximatePromptChars(Prompt prompt) { + if (prompt == null || prompt.getInstructions() == null) return 0; + int total = 0; + for (Message m : prompt.getInstructions()) { + String text = m.getText(); + if (text != null) total += text.length(); + } + return total; + } + + /** + * Pick a stable model identifier from whatever {@link ChatModel} + * implementation we received — Spring AI doesn't expose a single accessor. + * We try the well-known fields by reflection so this stays decoupled from + * concrete provider classes (Anthropic / OpenAI / DashScope all expose + * {@code defaultOptions.model} or equivalent). + */ + private static String identifyModel(ChatModel chatModel) { + if (chatModel == null) return ""; + try { + // Common Spring AI shape: getDefaultOptions().getModel() + java.lang.reflect.Method getDefaultOptions = chatModel.getClass().getMethod("getDefaultOptions"); + Object opts = getDefaultOptions.invoke(chatModel); + if (opts != null) { + try { + java.lang.reflect.Method getModel = opts.getClass().getMethod("getModel"); + Object model = getModel.invoke(opts); + if (model != null) return model.toString(); + } catch (NoSuchMethodException ignored) { + // fall through + } + } + } catch (Exception ignored) { + // fall through to class-name fallback + } + return chatModel.getClass().getSimpleName(); + } + /** * 构建 {"delta":"..."} JSON */ diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/RepetitionDetector.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/RepetitionDetector.java index 567090ef..534f389d 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/RepetitionDetector.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/RepetitionDetector.java @@ -2,14 +2,28 @@ package vip.mate.agent.graph; import lombok.extern.slf4j.Slf4j; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.regex.Pattern; + /** * 流式输出重复检测器 *

* 检测 LLM 流式输出中的退化重复模式（degenerate repetition）， * 当检测到内容在滑动窗口内高度重复时返回 true，调用方应截断 LLM 流。 *

- * 算法：维护一个滑动窗口缓冲区，每次追加新 delta 后， - * 检查窗口尾部是否存在连续重复的 n-gram 模式。 + * Two complementary detection paths run on every {@link #appendAndCheck}: + *

    + *
  1. Character-level n-gram repetition (continuous "X X X X" loops), + * which catches the classic degenerate-output failure within a single + * LLM stream.
    2. + *
  2. Sentence-level Jaccard similarity over the trailing N sentences, + * which catches "two near-identical sentences ~5 sentences apart" — + * a softer failure that the character path misses because the loop is + * not adjacent.
    3. + *
* * @author MateClaw Team */ @@ -31,8 +45,42 @@ public class RepetitionDetector { /** 已累积内容的最小长度才开始检测（避免误判短内容） */ private static final int MIN_CONTENT_LEN = 200; + // ===== Sentence-level detection ===== + + /** + * Number of trailing sentences to compare against the prior window. 3 + * tail sentences plus a 10-sentence lookback is enough to catch "ABC + * (other) ABC (other)" duplication while keeping the cost bounded. + */ + private static final int SENTENCE_TAIL_COUNT = 3; + private static final int SENTENCE_LOOKBACK = 10; + private static final double JACCARD_THRESHOLD = 0.85; + /** + * Lower bound on the per-sentence token count. Below this both the + * tail and the historical sentence are too short to make Jaccard + * meaningful (single phrases like "好的。" would otherwise false-positive). + */ + private static final int SENTENCE_MIN_TOKENS = 6; + /** + * Buffer must hold at least this many characters before sentence-level + * detection runs — keeps the cost off the early hot path. + */ + private static final int SENTENCE_MIN_BUFFER = 1500; + /** + * Sentence delimiters: full-width Chinese punctuation plus ASCII end-of + * -sentence punctuation and newline. Splitting greedily on any of these + * is good enough for Jaccard's set-of-tokens semantics. + */ + private static final Pattern SENTENCE_SPLIT = + Pattern.compile("[\\u3002\\uff01\\uff1f.!?\\n]+"); + private final StringBuilder buffer = new StringBuilder(); private boolean repetitionDetected = false; + /** + * Marks repetition detected via the sentence path so callers can tell + * "char_pattern" from "sentence_repetition" in their warning broadcasts. + */ + private boolean lastTriggerWasSentence = false; /** * 追加新的 delta 并检测是否存在重复 @@ -52,9 +100,13 @@ public class RepetitionDetector { return false; } - // 保持窗口大小 - if (buffer.length() > WINDOW_SIZE * 2) { - buffer.delete(0, buffer.length() - WINDOW_SIZE); + // Window trim policy: the char-level path only needs the last ~1024 + // chars, but the sentence path benefits from a larger horizon so it + // can compare against the full 10-sentence lookback. Keep up to + // SENTENCE_MIN_BUFFER * 2 so the sentence detector has room. + int retainCap = Math.max(WINDOW_SIZE, SENTENCE_MIN_BUFFER * 2); + if (buffer.length() > retainCap * 2) { + buffer.delete(0, buffer.length() - retainCap); } // 在窗口尾部检测重复模式 @@ -89,6 +141,7 @@ public class RepetitionDetector { } repetitionDetected = true; + lastTriggerWasSentence = false; log.warn("[RepetitionDetector] Detected degenerate repetition: " + "pattern length={}, repeats={}, pattern preview=\"{}\"", patternLen, count, @@ -97,9 +150,139 @@ public class RepetitionDetector { } } + // Sentence-level path: cheap to skip until the buffer is long enough + // to actually contain multiple sentences worth comparing. + if (buffer.length() >= SENTENCE_MIN_BUFFER && checkSentenceRepetition(window)) { + repetitionDetected = true; + lastTriggerWasSentence = true; + return true; + } + return false; } + /** + * Return true when one of the trailing {@link #SENTENCE_TAIL_COUNT} + * sentences is near-duplicate to one of the previous {@link #SENTENCE_LOOKBACK} + * sentences (Jaccard over token unigram sets). Both candidates must clear + * {@link #SENTENCE_MIN_TOKENS} so we don't false-positive on common short + * acknowledgements. + */ + private boolean checkSentenceRepetition(String window) { + String[] split = SENTENCE_SPLIT.split(window); + // Drop trailing whitespace-only segments produced by the splitter. + List sentences = new ArrayList<>(split.length); + for (String s : split) { + String trimmed = s.trim(); + if (!trimmed.isEmpty()) { + sentences.add(trimmed); + } + } + if (sentences.size() < 2) { + return false; + } + + int total = sentences.size(); + int tailStart = Math.max(0, total - SENTENCE_TAIL_COUNT); + int lookbackStart = Math.max(0, tailStart - SENTENCE_LOOKBACK); + + for (int i = tailStart; i < total; i++) { + Set tailTokens = tokenize(sentences.get(i)); + if (tailTokens.size() < SENTENCE_MIN_TOKENS) continue; + + for (int j = lookbackStart; j < tailStart; j++) { + Set earlierTokens = tokenize(sentences.get(j)); + if (earlierTokens.size() < SENTENCE_MIN_TOKENS) continue; + + double jaccard = jaccard(tailTokens, earlierTokens); + if (jaccard >= JACCARD_THRESHOLD) { + String preview = sentences.get(i); + log.warn("[RepetitionDetector] Sentence-level repetition: " + + "tailIdx={}, earlierIdx={}, jaccard={}, preview=\"{}\"", + i, j, String.format("%.2f", jaccard), + preview.length() > 60 ? preview.substring(0, 60) + "..." : preview); + return true; + } + } + } + return false; + } + + /** + * Cheap tokenizer that treats each Chinese char as its own token and + * splits ASCII / European text on whitespace. Producing token sets keeps + * Jaccard symmetric on lengths, which is the property we rely on. + */ + private static Set tokenize(String sentence) { + Set tokens = new HashSet<>(); + StringBuilder asciiWord = new StringBuilder(); + for (int i = 0; i < sentence.length(); i++) { + char c = sentence.charAt(i); + if (isCjk(c)) { + if (asciiWord.length() > 0) { + tokens.add(asciiWord.toString().toLowerCase()); + asciiWord.setLength(0); + } + tokens.add(String.valueOf(c)); + } else if (Character.isLetterOrDigit(c)) { + asciiWord.append(c); + } else { + if (asciiWord.length() > 0) { + tokens.add(asciiWord.toString().toLowerCase()); + asciiWord.setLength(0); + } + } + } + if (asciiWord.length() > 0) { + tokens.add(asciiWord.toString().toLowerCase()); + } + return tokens; + } + + private static boolean isCjk(char c) { + return (c >= 0x4E00 && c <= 0x9FFF) + || (c >= 0x3400 && c <= 0x4DBF) + || (c >= 0xF900 && c <= 0xFAFF); + } + + private static double jaccard(Set a, Set b) { + if (a.isEmpty() && b.isEmpty()) return 1.0; + int intersect = 0; + Set smaller = a.size() <= b.size() ? a : b; + Set larger = smaller == a ? b : a; + for (String t : smaller) { + if (larger.contains(t)) intersect++; + } + int union = a.size() + b.size() - intersect; + return union == 0 ? 0.0 : (double) intersect / union; + } + + /** + * Mark a logical iteration boundary. Buffer is intentionally NOT cleared + * — sentence-level detection across LLM call boundaries is the whole + * reason this detector lives on the conversation, not on a single call. + * The debug log lets fixtures observe that the boundary signal arrived. + */ + public void markIterationBoundary() { + log.debug("[RepetitionDetector] iteration boundary marked (buffer chars={})", + buffer.length()); + } + + /** + * Returns "sentence_repetition" when the most recent trigger was the + * Jaccard path; otherwise "char_pattern". Useful for warning broadcasts + * that need to differentiate the two failure modes. + */ + public String lastTriggerReason() { + if (!repetitionDetected) return null; + return lastTriggerWasSentence ? "sentence_repetition" : "char_pattern"; + } + + /** Number of characters currently held in the detector's buffer. */ + public int bufferLength() { + return buffer.length(); + } + /** * 判断 pattern 是否为装饰性字符（不应判定为退化重复）。 *

@@ -146,6 +329,7 @@ public class RepetitionDetector { public void reset() { buffer.setLength(0); repetitionDetected = false; + lastTriggerWasSentence = false; } /** diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java index a918bcb9..74f12ded 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java @@ -164,6 +164,27 @@ public class ToolExecutionExecutor { this.skillRuntimeService = s; } + /** + * Optional audit sink. When set, child-agent denied-tool attempts are + * recorded so admins can see what children are trying that gets blocked. + * Left optional because legacy constructors and tests run without an + * audit pipeline. + */ + private vip.mate.audit.service.AuditEventService auditEventService; + + public void setAuditEventService(vip.mate.audit.service.AuditEventService s) { + this.auditEventService = s; + } + + /** + * Per-turn deduplication key set for child-agent denial audit. Without + * this, a child that retries the same denied tool many times in one + * turn would write one audit row per call. Cleared at the start of + * every {@code execute(...)} so it does not retain entries across turns. + */ + private final ThreadLocal> auditedDenials = + ThreadLocal.withInitial(HashSet::new); + public ToolExecutionExecutor(AgentToolSet toolSet, ToolGuardService toolGuardService, ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker) { this(toolSet, toolGuardService, null, approvalService, streamTracker, null, null, null); @@ -284,6 +305,10 @@ public class ToolExecutionExecutor { String workspaceBasePath, ChatOrigin origin) { ChatOrigin safeOrigin = origin != null ? origin : ChatOrigin.EMPTY; + // Reset per-turn audit dedupe state. A retried denied tool inside the + // same turn writes a single audit row; the set is repopulated by the + // denial branch below. + auditedDenials.get().clear(); List allResponses = new ArrayList<>(); List events = Collections.synchronizedList(new ArrayList<>()); // RFC-052: accumulate full-text outputs from returnDirect tools so the @@ -329,6 +354,22 @@ public class ToolExecutionExecutor { if (denied.contains(toolName)) { String msg = "[安全限制] 子 Agent 不允许使用工具: " + toolName; log.info("[ToolExecutor] Child agent blocked from using tool: {}", toolName); + // Audit per (toolName, conversationId) tuple at most once + // per turn so a child retrying the same denied tool many + // times does not spam the audit table. + if (auditEventService != null && auditedDenials.get().add(toolName)) { + try { + String detail = "{\"toolName\":\"" + toolName + "\",\"conversationId\":\"" + + (conversationId != null ? conversationId : "") + + "\",\"agentId\":\"" + (agentId != null ? agentId : "") + + "\"}"; + auditEventService.record("subagent.tool.denied", "tool", + toolName, toolName, detail); + } catch (Exception auditEx) { + log.debug("[ToolExecutor] Audit write failed for denied tool {}: {}", + toolName, auditEx.getMessage()); + } + } events.add(GraphEventPublisher.toolComplete(toolCall.id(), toolName, msg, false)); allResponses.add(new org.springframework.ai.chat.messages.ToolResponseMessage.ToolResponse( toolCall.id(), toolName, msg)); diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ObservationNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ObservationNode.java index 169e2cae..7bac57ba 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ObservationNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ObservationNode.java @@ -4,6 +4,7 @@ import com.alibaba.cloud.ai.graph.OverAllState; import com.alibaba.cloud.ai.graph.action.NodeAction; import lombok.extern.slf4j.Slf4j; import org.springframework.ai.chat.messages.ToolResponseMessage; +import vip.mate.agent.GraphEventPublisher; import vip.mate.agent.graph.observation.ObservationProcessor; import vip.mate.agent.graph.state.MateClawStateAccessor; @@ -120,6 +121,17 @@ public class ObservationNode implements NodeAction { .shouldSummarize(shouldSummarize) .toolCallCount(newToolCallCount); + // Close out the iteration we just observed. We use currentIteration + // (not nextIteration) so the index pairs with whatever + // iteration_start the ReasoningNode emitted at the top of this turn. + // Char totals are best-effort: ObservationNode doesn't see the LLM + // delta stream directly, so 0/0 is acceptable for now — consumers + // that care fall back to summing the deltas themselves. + if (streamTracker == null || streamTracker.isIterationEventsEnabled()) { + builder.events(List.of( + GraphEventPublisher.iterationEnd(currentIteration, "parent", null, 0, 0))); + } + // 重复观察时标记错误，让 ObservationDispatcher 路由到 limitExceededNode if (duplicateObservation) { builder.put(ERROR, "连续 3 次工具调用返回相同结果，已强制终止循环"); diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java index 32a4792e..bb4d739b 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java @@ -364,6 +364,17 @@ public class ReasoningNode implements NodeAction { GraphEventPublisher.GraphEvent phaseEvent = GraphEventPublisher.phase("reasoning", Map.of("iteration", accessor.iterationCount())); + // Iteration boundary marker for the parent ReAct loop. Reason + // distinguishes the very first turn of the conversation from a + // mid-loop repeat for consumers grouping events into per-turn cards. + boolean iterationEventsOn = streamTracker == null || streamTracker.isIterationEventsEnabled(); + GraphEventPublisher.GraphEvent iterStartEvent = iterationEventsOn + ? GraphEventPublisher.iterationStart( + accessor.iterationCount(), + accessor.iterationCount() == 0 ? "first_turn" : "react_step", + "parent", + null) + : null; pushPhase(conversationId, "reasoning", Map.of( "iteration", accessor.iterationCount(), "llmCallCount", nextLlmCallCount @@ -481,7 +492,7 @@ public class ReasoningNode implements NodeAction { .thinkingStreamed(!result.thinking().isEmpty()) .llmCallCount(nextLlmCallCount) .mergeUsage(state, result) - .events(List.of(phaseEvent)) + .events(buildEvents(phaseEvent, iterStartEvent)) .build(); } else { String content = result.text(); @@ -491,6 +502,14 @@ public class ReasoningNode implements NodeAction { "answerChars", content != null ? content.length() : 0 )); + // Final-answer path: iteration ends in this same node because + // ReAct never re-enters the loop afterwards. + GraphEventPublisher.GraphEvent iterEndEvent = iterationEventsOn + ? GraphEventPublisher.iterationEnd(accessor.iterationCount(), + "parent", null, + content != null ? content.length() : 0, + result.thinking() != null ? result.thinking().length() : 0) + : null; return MateClawStateAccessor.output() .needsToolCall(false) .shouldSummarize(false) @@ -502,7 +521,7 @@ public class ReasoningNode implements NodeAction { .thinkingStreamed(!result.thinking().isEmpty()) .llmCallCount(nextLlmCallCount) .mergeUsage(state, result) - .events(List.of(phaseEvent)) + .events(buildEvents(phaseEvent, iterStartEvent, iterEndEvent)) .build(); } } @@ -523,6 +542,19 @@ public class ReasoningNode implements NodeAction { } } + /** + * Compose the per-call event list, dropping any null entries so the + * iteration-boundary toggle ({@code mateclaw.stream.iteration-events}) + * works without forcing every caller into branching code. + */ + private static List buildEvents(GraphEventPublisher.GraphEvent... events) { + List out = new ArrayList<>(events.length); + for (GraphEventPublisher.GraphEvent ev : events) { + if (ev != null) out.add(ev); + } + return out; + } + private void pushPhase(String conversationId, String phase, Map extra) { if (streamTracker == null || !StringUtils.hasText(conversationId)) { return; diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java index dc440e31..f0ab36ad 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java @@ -146,6 +146,15 @@ public class StepExecutionNode implements NodeAction { log.info("[StepExecution] Executing step {}/{}: {}", stepIndex + 1, steps.size(), step); List events = new ArrayList<>(); + // Iteration boundary for the plan-execute loop: each step is one + // iteration that may itself fan out to multiple LLM calls. Reason is + // "plan_step" so consumers can distinguish it from ReAct's + // "react_step" / "first_turn" markers when both stream into the + // same SSE feed. + boolean iterationEventsOn = streamTracker == null || streamTracker.isIterationEventsEnabled(); + if (iterationEventsOn) { + events.add(GraphEventPublisher.iterationStart(stepIndex, "plan_step", "parent", null)); + } events.add(GraphEventPublisher.stepStarted(stepIndex, step)); events.add(GraphEventPublisher.phase("executing", Map.of("stepIndex", stepIndex, "stepTitle", step))); @@ -344,6 +353,10 @@ public class StepExecutionNode implements NodeAction { "Plan completed via returnDirect tool: " + stepDirectOutputs.get(0).toolName()); events.add(GraphEventPublisher.stepCompleted(stepIndex, assembled)); + if (iterationEventsOn) { + events.add(GraphEventPublisher.iterationEnd(stepIndex, "parent", null, + assembled != null ? assembled.length() : 0, 0)); + } return PlanStateAccessor.output() .currentStepResult(assembled) .currentStepIndex(steps.size()) // 越界 → dispatcher 收束 @@ -376,6 +389,10 @@ public class StepExecutionNode implements NodeAction { planningService.updateSubPlanFailure(planId, stepIndex, shortError); planningService.markPlanFailed(planId, "步骤" + (stepIndex + 1) + " 执行失败：" + shortError); events.add(GraphEventPublisher.stepCompleted(stepIndex, shortError)); + if (iterationEventsOn) { + events.add(GraphEventPublisher.iterationEnd(stepIndex, "parent", null, + shortError != null ? shortError.length() : 0, 0)); + } return PlanStateAccessor.output() .currentStepResult(shortError) .currentPhase("plan_aborted") @@ -388,6 +405,11 @@ public class StepExecutionNode implements NodeAction { planningService.updateSubPlanResult(planId, stepIndex, finalResult); events.add(GraphEventPublisher.stepCompleted(stepIndex, finalResult)); + if (iterationEventsOn) { + events.add(GraphEventPublisher.iterationEnd(stepIndex, "parent", null, + finalResult != null ? finalResult.length() : 0, + stepThinking != null ? stepThinking.length() : 0)); + } log.info("[StepExecution] Step {}/{} completed: {}", stepIndex + 1, steps.size(), diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java index b0d5f223..f54713c5 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java @@ -448,6 +448,19 @@ public class ChatController { registerEmitterCallbacks(emitter, conversationId); streamTracker.attach(conversationId, emitter); + // Per-emitter "the SSE channel is open and you should reset any + // pending placeholder UI". Sent directly to the emitter rather than + // broadcast so reconnecting subscribers don't see a duplicate marker + // for an already-open conversation. + try { + sendEvent(emitter, "stream_started", Map.of( + "conversationId", conversationId, + "timestamp", System.currentTimeMillis() + )); + } catch (IOException e) { + log.debug("Failed to send stream_started event for {}: {}", conversationId, e.getMessage()); + } + // 标记 emitter 是否已结束，防止 Flux 回调再次写入已关闭的 emitter AtomicBoolean emitterDone = new AtomicBoolean(false); diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java index 5cafdaaa..68dd9745 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java @@ -3,9 +3,11 @@ package vip.mate.channel.web; import com.fasterxml.jackson.databind.ObjectMapper; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import reactor.core.Disposable; +import vip.mate.agent.graph.RepetitionDetector; import vip.mate.workspace.conversation.model.MessageContentPart; import java.io.IOException; @@ -57,10 +59,71 @@ public class ChatStreamTracker { private final ObjectMapper objectMapper; + /** + * Maximum size, in bytes, of a single SSE event JSON payload before + * {@link #broadcastChunked} splits the body into ordered + * {@code tool_result_chunk} events. + */ + static final int CHUNK_SIZE = 8192; + + // ===== Configurable knobs (mateclaw.stream.*) ===== + + /** + * Gate for chunked tool-result transport. When {@code false}, + * {@link #broadcastChunked} falls back to a single broadcast call so + * environments that prefer the legacy single-event behavior can opt out. + */ + @Value("${mateclaw.stream.chunked-tool-results:true}") + private boolean chunkedToolResultsEnabled = true; + + /** + * Gate for {@code iteration_start} / {@code iteration_end} events emitted + * from graph nodes. Off-by-default deployments can suppress them without + * touching node code. + */ + @Value("${mateclaw.stream.iteration-events:true}") + private boolean iterationEventsEnabled = true; + + /** + * Heartbeat cadence (seconds) before the first model token arrives. Short + * because pre-token gaps strand the UI on a blank "正在生成中" placeholder + * with no visible activity. + */ + @Value("${mateclaw.stream.heartbeat.pre-token-sec:2}") + private int heartbeatPreTokenSec = 2; + + /** + * Heartbeat cadence (seconds) once the model is actively streaming tokens — + * deltas themselves keep the connection warm, so heartbeats relax. + */ + @Value("${mateclaw.stream.heartbeat.streaming-sec:10}") + private int heartbeatStreamingSec = 10; + + /** + * Heartbeat cadence (seconds) while a tool call is in flight. Tools can + * take longer than streaming chunks but should still tick faster than the + * default proxy idle timeout. + */ + @Value("${mateclaw.stream.heartbeat.tool-sec:5}") + private int heartbeatToolSec = 5; + public ChatStreamTracker(ObjectMapper objectMapper) { this.objectMapper = objectMapper; } + /** Test-only setters; production paths use Spring property binding. */ + void setChunkedToolResultsEnabled(boolean enabled) { + this.chunkedToolResultsEnabled = enabled; + } + + void setIterationEventsEnabled(boolean enabled) { + this.iterationEventsEnabled = enabled; + } + + public boolean isIterationEventsEnabled() { + return iterationEventsEnabled; + } + record SseEvent(String name, String json) {} /** @@ -121,12 +184,44 @@ public class ChatStreamTracker { /** 心跳定时器 */ volatile ScheduledFuture heartbeatFuture; + /** + * Flips the first time any content/thinking delta is observed for this + * run. Heartbeat scheduling watches this flag to switch from the short + * pre-token cadence to the streaming cadence — pre-token gaps need + * frequent keep-alives because the UI has no other signal of activity. + */ + volatile boolean firstTokenReceived = false; + + /** + * Cross-call repetition detectors scoped to the conversation, not the + * single LLM call. Sharing across {@code streamLLMChat} invocations + * lets the sentence-level path catch "the model produces N near- + * identical sentences across two consecutive iterations" — a common + * failure that the per-call detectors used to miss. + */ + volatile RepetitionDetector contentRepDetector = new RepetitionDetector(); + volatile RepetitionDetector thinkingRepDetector = new RepetitionDetector(); + /** 已广播的 pending approval ID 集合（用于幂等去重） */ final java.util.Set broadcastedApprovalIds = java.util.concurrent.ConcurrentHashMap.newKeySet(); /** 创建时间（用于 stale 检测和清理） */ final long createdAt = System.currentTimeMillis(); + /** + * Wall-clock millis of the most recent meaningful event on this run. + * Updated whenever {@link #broadcast(String, String, String)} pushes a + * non-heartbeat event so a watchdog can tell "actively producing" + * apart from "alive but silent". + */ + volatile long lastEventAt = System.currentTimeMillis(); + + /** Bound agent identifier; null while not yet resolved. */ + volatile Long agentId; + + /** Username that owns this run; null for system-driven runs. */ + volatile String username; + RunState(String conversationId) { this.conversationId = conversationId; } @@ -158,6 +253,168 @@ public class ChatStreamTracker { }; } + /** + * Batching variant of {@link #addEventRelay} for sub-conversation streams + * whose tool-call chatter would flood the parent transcript. Tool start / + * complete events accumulate into a buffer; lifecycle and error events + * (subagent_*, error, tool_approval_requested, phase, done) bypass the + * buffer but flush it first so ordering is preserved. + *

+ * Buffered events are emitted as a single {@code delegation_batch} + * envelope on the parent conversation listener: + * 

+     * {
+     *   "kind":   "delegation_batch",
+     *   "scope":  "subagent",
+     *   "events": [{ "event": "tool_call_started", "data": "<json>" }, ...]
+     * }
+     *
+ * + * @param sourceConversationId conversation to listen on + * @param parentConversationId parent conversation context (currently + * forwarded only as listener metadata; the + * tracker itself does not target it) + * @param batchSize flush threshold by event count + * @param flushMs flush threshold by elapsed millis since + * first buffered event + * @return Runnable that deregisters the relay (and flushes any pending + * events first) + */ + public Runnable addBatchedEventRelay(String sourceConversationId, + String parentConversationId, + int batchSize, + long flushMs, + java.util.function.BiConsumer listener) { + BatchedRelay relay = new BatchedRelay(parentConversationId, listener, + Math.max(1, batchSize), Math.max(1, flushMs)); + java.util.function.BiConsumer wrapper = relay::accept; + eventRelays.computeIfAbsent(sourceConversationId, + k -> new java.util.concurrent.CopyOnWriteArrayList<>()) + .add(wrapper); + log.debug("Batched relay registered for conversation {} -> parent={}", + sourceConversationId, parentConversationId); + return () -> { + relay.shutdown(); + List> listeners = + eventRelays.get(sourceConversationId); + if (listeners != null) { + listeners.remove(wrapper); + if (listeners.isEmpty()) { + eventRelays.remove(sourceConversationId); + } + } + log.debug("Batched relay removed for {} -> parent={}", + sourceConversationId, parentConversationId); + }; + } + + /** + * Internal helper holding the batch buffer and the scheduled flush. Each + * relay owns its own state but reuses {@link #heartbeatScheduler} for + * timer ticks (sharing the daemon-thread scheduler avoids one-thread-per + * -relay sprawl in long agent sessions). + */ + private final class BatchedRelay { + private final String parentConversationId; + private final java.util.function.BiConsumer downstream; + private final int batchSize; + private final long flushMs; + private final List> buffer = new ArrayList<>(); + private final Object lock = new Object(); + private ScheduledFuture pendingFlush; + private volatile boolean closed; + + BatchedRelay(String parentConversationId, + java.util.function.BiConsumer downstream, + int batchSize, long flushMs) { + this.parentConversationId = parentConversationId; + this.downstream = downstream; + this.batchSize = batchSize; + this.flushMs = flushMs; + } + + void accept(String eventName, String json) { + if (closed) return; + // Pass-through (with prior flush to preserve ordering) for any + // event that conveys lifecycle or critical state. Tool call + // boundaries are the only batched class today; the explicit list + // here is the source of truth. + if (isPassThrough(eventName)) { + flushNow(); + downstream.accept(eventName, json); + return; + } + if (!"tool_call_started".equals(eventName) + && !"tool_call_completed".equals(eventName)) { + downstream.accept(eventName, json); + return; + } + + boolean shouldFlush = false; + synchronized (lock) { + Map entry = new java.util.LinkedHashMap<>(); + entry.put("event", eventName); + entry.put("data", json); + buffer.add(entry); + if (buffer.size() >= batchSize) { + shouldFlush = true; + } else if (pendingFlush == null || pendingFlush.isDone()) { + pendingFlush = heartbeatScheduler.schedule(this::flushNow, + flushMs, TimeUnit.MILLISECONDS); + } + } + if (shouldFlush) { + flushNow(); + } + } + + private boolean isPassThrough(String eventName) { + return "subagent_start".equals(eventName) + || "subagent_complete".equals(eventName) + || "error".equals(eventName) + || "tool_approval_requested".equals(eventName) + || "phase".equals(eventName) + || "done".equals(eventName); + } + + void flushNow() { + List> snapshot; + synchronized (lock) { + if (buffer.isEmpty()) { + if (pendingFlush != null) { + pendingFlush.cancel(false); + pendingFlush = null; + } + return; + } + snapshot = new ArrayList<>(buffer); + buffer.clear(); + if (pendingFlush != null) { + pendingFlush.cancel(false); + pendingFlush = null; + } + } + Map envelope = new java.util.LinkedHashMap<>(); + envelope.put("kind", "delegation_batch"); + envelope.put("scope", "subagent"); + if (parentConversationId != null && !parentConversationId.isEmpty()) { + envelope.put("parent", parentConversationId); + } + envelope.put("events", snapshot); + try { + String json = objectMapper.writeValueAsString(envelope); + downstream.accept("delegation_batch", json); + } catch (Exception e) { + log.warn("Batched relay flush failed: {}", e.getMessage()); + } + } + + void shutdown() { + closed = true; + flushNow(); + } + } + /** 心跳调度线程池（守护线程） */ private final ScheduledExecutorService heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> { @@ -269,6 +526,13 @@ public class ChatStreamTracker { boolean isAsyncTask = eventName != null && eventName.startsWith("async_task_"); boolean isHeartbeat = "heartbeat".equals(eventName); + // Stamp last activity for stuck detection. Heartbeats are excluded + // because they fire on a timer regardless of model progress; counting + // them would mask a wedged turn behind a healthy timestamp. + if (state != null && !isHeartbeat) { + state.lastEventAt = System.currentTimeMillis(); + } + if (isDone || isAsyncTask) { if (state == null) return; SseEvent ev = new SseEvent(eventName, jsonData); @@ -377,6 +641,156 @@ public class ChatStreamTracker { broadcast(conversationId, eventName, json); } + /** + * Broadcast {@code payload} as a single SSE event when its serialized form + * fits within {@link #CHUNK_SIZE}; otherwise extract the long {@code result} + * field and emit it as ordered {@code tool_result_chunk} events. + *

+ * Each chunk carries: + * 

+     * {
+     *   "kind":  "tool_result",
+     *   "scope": "parent",         // sub-agent producers will set "subagent"
+     *   "ref":   "<refKey>",
+     *   "seq":   <0..N>,
+     *   "final": <true|false>,
+     *   "delta": "<text>"
+     * }
+     *
+ * The last chunk has {@code "final": true}; consumers reassemble by + * concatenating {@code delta} in seq order keyed on {@code ref}. When the + * payload's {@code result} field cannot be located (or chunked transport + * is disabled), the entire envelope is sent unchanged. + * + * @param conversationId target conversation + * @param eventName SSE event name for the small-payload path + * @param payload envelope; the {@code result} field (or, failing + * that, the {@code arguments} field) is split + * @param refKey identifier consumers use to group chunks; usually + * {@code toolCallId} or the step index as a string + */ + public void broadcastChunked(String conversationId, String eventName, + Object payload, String refKey) { + String json; + try { + json = objectMapper.writeValueAsString(payload); + } catch (Exception e) { + log.warn("Failed to serialize chunked broadcast for event {}: {}", + eventName, e.getMessage()); + return; + } + + if (!chunkedToolResultsEnabled || json.length() <= CHUNK_SIZE) { + broadcast(conversationId, eventName, json); + return; + } + + // Find a long string field worth splitting; tool results live under + // "result", approval payloads under "arguments". Falling back to the + // whole envelope keeps the transport correct even for unknown shapes + // (the consumer can still concatenate by ref+seq and decode itself). + Map envelope = asMap(payload); + String fieldKey = null; + String longText = null; + if (envelope != null) { + Object resultField = envelope.get("result"); + Object argsField = envelope.get("arguments"); + if (resultField instanceof String s && s.length() > CHUNK_SIZE / 2) { + fieldKey = "result"; + longText = s; + } else if (argsField instanceof String s && s.length() > CHUNK_SIZE / 2) { + fieldKey = "arguments"; + longText = s; + } + } + + if (longText == null) { + // No splittable string field — emit unchanged and let the client + // handle the larger envelope as best it can. + broadcast(conversationId, eventName, json); + return; + } + + // 1. Send a header event with the long field replaced by an empty + // placeholder so consumers see the same envelope shape; the body + // arrives via the chunk events that follow. + Map headerEnvelope = new java.util.LinkedHashMap<>(envelope); + headerEnvelope.put(fieldKey, ""); + headerEnvelope.put("chunked", true); + headerEnvelope.put("chunkRef", refKey != null ? refKey : ""); + try { + String headerJson = objectMapper.writeValueAsString(headerEnvelope); + broadcast(conversationId, eventName, headerJson); + } catch (Exception e) { + log.warn("Failed to serialize chunk header for {}: {}", eventName, e.getMessage()); + broadcast(conversationId, eventName, json); + return; + } + + // 2. Stream the body in fixed-size slices. + int total = longText.length(); + int offset = 0; + int seq = 0; + // Reserve room in CHUNK_SIZE for the JSON envelope around the slice; + // 256 bytes covers kind/scope/ref/seq/final + JSON escapes. + final int sliceMax = Math.max(512, CHUNK_SIZE - 256); + while (offset < total) { + int end = Math.min(offset + sliceMax, total); + String slice = longText.substring(offset, end); + boolean isFinal = end >= total; + Map chunk = new java.util.LinkedHashMap<>(); + chunk.put("kind", "tool_result"); + chunk.put("scope", "parent"); + chunk.put("ref", refKey != null ? refKey : ""); + chunk.put("seq", seq); + chunk.put("final", isFinal); + chunk.put("delta", slice); + try { + String chunkJson = objectMapper.writeValueAsString(chunk); + broadcast(conversationId, "tool_result_chunk", chunkJson); + } catch (Exception e) { + log.warn("Failed to serialize tool_result_chunk seq={}: {}", seq, e.getMessage()); + return; + } + offset = end; + seq++; + } + } + + @SuppressWarnings("unchecked") + private static Map asMap(Object payload) { + if (payload instanceof Map m) { + try { + return (Map) m; + } catch (ClassCastException ignored) { + return null; + } + } + return null; + } + + /** + * Conversation-scoped repetition detector for content deltas. Lazily + * instantiated (a tracker created without a registered conversation + * receives a fresh detector so callers never get null). + */ + public RepetitionDetector getContentRepDetector(String conversationId) { + RunState state = runs.get(conversationId); + if (state == null) { + return new RepetitionDetector(); + } + return state.contentRepDetector; + } + + /** Conversation-scoped repetition detector for thinking deltas. */ + public RepetitionDetector getThinkingRepDetector(String conversationId) { + RunState state = runs.get(conversationId); + if (state == null) { + return new RepetitionDetector(); + } + return state.thinkingRepDetector; + } + /** * Diagnostic helper for the multi-node deployment edge case (issue #17): * tells the caller whether a {@link RunState} for this conversation @@ -559,8 +973,18 @@ public class ChatStreamTracker { // ===== Heartbeat ===== - /** 心跳间隔（秒） */ - private static final int HEARTBEAT_INTERVAL_SEC = 10; + /** + * Pick the heartbeat cadence (seconds) that matches the run's current + * phase. Pre-token gaps need fast keep-alives so the UI shows activity; + * tool execution stretches slightly; mid-stream is rate-limited because + * deltas already keep the connection warm. + */ + private int currentHeartbeatIntervalSec(RunState state) { + if (state.runningToolName != null && !state.runningToolName.isEmpty()) { + return heartbeatToolSec; + } + return state.firstTokenReceived ? heartbeatStreamingSec : heartbeatPreTokenSec; + } /** * 启动心跳定时器。在流注册后调用，定期向前端发送 heartbeat 事件。 @@ -572,6 +996,7 @@ public class ChatStreamTracker { // 避免重复启动 if (state.heartbeatFuture != null && !state.heartbeatFuture.isDone()) return; + int intervalSec = currentHeartbeatIntervalSec(state); state.heartbeatFuture = heartbeatScheduler.scheduleAtFixedRate(() -> { try { RunState s = runs.get(conversationId); @@ -605,7 +1030,38 @@ public class ChatStreamTracker { } catch (Exception e) { log.debug("Heartbeat error for {}: {}", conversationId, e.getMessage()); } - }, HEARTBEAT_INTERVAL_SEC, HEARTBEAT_INTERVAL_SEC, TimeUnit.SECONDS); + }, intervalSec, intervalSec, TimeUnit.SECONDS); + } + + /** + * Mark that the first content/thinking token has been received for this + * run and reschedule the heartbeat at the streaming cadence. + *

+ * Called from the LLM streaming layer so the heartbeat relaxes once the + * connection is naturally being kept warm by data deltas. Idempotent — a + * second call is a no-op. + */ + public void markFirstTokenReceived(String conversationId) { + RunState state = runs.get(conversationId); + if (state == null) return; + if (state.firstTokenReceived) return; + state.firstTokenReceived = true; + rescheduleHeartbeat(conversationId); + } + + /** + * Cancels the active heartbeat (if any) and starts a new one at the + * cadence currently appropriate for the run state. Public so callers that + * mutate {@code runningToolName} can request a tool-cadence heartbeat. + */ + public void rescheduleHeartbeat(String conversationId) { + RunState state = runs.get(conversationId); + if (state == null || state.done) return; + if (state.heartbeatFuture != null) { + state.heartbeatFuture.cancel(false); + state.heartbeatFuture = null; + } + startHeartbeat(conversationId); } /** @@ -637,10 +1093,39 @@ public class ChatStreamTracker { public void updateRunningTool(String conversationId, String toolName) { RunState state = runs.get(conversationId); if (state != null) { + String previous = state.runningToolName; state.runningToolName = toolName; + // Heartbeat cadence depends on whether a tool is in flight; switch + // cadences when the tool slot transitions in either direction. + boolean wasRunning = previous != null && !previous.isEmpty(); + boolean nowRunning = toolName != null && !toolName.isEmpty(); + if (wasRunning != nowRunning) { + rescheduleHeartbeat(conversationId); + } } } + /** + * Read-only accessor for the currently running tool name on a conversation. + * Returns {@code null} when no run state exists or no tool is in flight. + * Used by external observers (heartbeat watchdog, status APIs) that need + * to probe progress without mutating the run. + */ + public String getRunningToolName(String conversationId) { + RunState state = runs.get(conversationId); + return state != null ? state.runningToolName : null; + } + + /** + * Read-only accessor for the current execution phase. Returns {@code null} + * when no run state exists. Mirrors {@link #getRunningToolName(String)} so + * external observers can read both fields without touching internals. + */ + public String getCurrentPhase(String conversationId) { + RunState state = runs.get(conversationId); + return state != null ? state.currentPhase : null; + } + /** * 设置等待原因 */ @@ -1063,4 +1548,118 @@ public class ChatStreamTracker { } } } + + // ===== Runtime snapshot surface (admin Backstage) ===== + + /** + * Bind the resolved agent + owner to the active run so the runtime + * snapshot can label cards without re-querying the conversation table. + * Idempotent — overwrites are fine because both fields are observation- + * only metadata. + */ + public void bindRunMeta(String conversationId, Long agentId, String username) { + RunState s = runs.get(conversationId); + if (s == null) return; + if (agentId != null) s.agentId = agentId; + if (username != null) s.username = username; + } + + /** + * Immutable view of one in-flight run. Computed eagerly under the + * RunState lock so the receiver sees a consistent picture even if the + * underlying state mutates while it iterates. + */ + public record RunSnapshot( + String conversationId, + Long agentId, + String username, + String currentPhase, + String runningToolName, + String waitingReason, + boolean done, + boolean stopRequested, + boolean firstTokenReceived, + int subscriberCount, + int queueLen, + int activeFluxCount, + long createdAt, + long lastEventAt, + long ageMs, + long msSinceLastEvent + ) {} + + /** + * Snapshot every active run. Used by the admin Backstage to render the + * global "what are my agents doing right now" view. Returned list is a + * defensive copy — callers may freely sort / filter it. + */ + public List getAllSnapshot() { + long now = System.currentTimeMillis(); + List out = new ArrayList<>(runs.size()); + for (RunState s : runs.values()) { + int subs; + int queue; + synchronized (s.lock) { + subs = s.subscribers.size(); + queue = s.messageQueue.size(); + } + out.add(new RunSnapshot( + s.conversationId, + s.agentId, + s.username, + s.currentPhase, + s.runningToolName, + s.waitingReason, + s.done, + s.stopRequested.get(), + s.firstTokenReceived, + subs, + queue, + s.activeFluxCount, + s.createdAt, + s.lastEventAt, + now - s.createdAt, + now - s.lastEventAt + )); + } + return out; + } + + /** + * Force a wedged run to terminate. Used by the admin Backstage's + * "End it" action when the friendly stop has been observed not to take + * effect (model wedged in a tool call beyond the timeout). Sequence + * matches what {@link #onShutdown()} does for individual runs. + * + * @return true when a run was found and torn down; false if already gone + */ + public boolean forceRecycle(String conversationId) { + RunState state = runs.get(conversationId); + if (state == null) return false; + try { + state.stopRequested.set(true); + state.interruptType = InterruptType.USER_STOP; + Disposable d = state.disposable; + if (d != null && !d.isDisposed()) { + d.dispose(); + } + } catch (Exception e) { + log.warn("forceRecycle: dispose failed for {}: {}", conversationId, e.getMessage()); + } + try { + state.done = true; + stopHeartbeat(conversationId); + } catch (Exception e) { + log.warn("forceRecycle: heartbeat stop failed for {}: {}", conversationId, e.getMessage()); + } + synchronized (state.lock) { + for (SseEmitter em : state.subscribers) { + try { em.complete(); } catch (Exception ignored) {} + } + state.subscribers.clear(); + } + runs.remove(conversationId); + log.info("forceRecycle: run {} torn down", conversationId); + return true; + } } diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java index 17b26729..7e113d80 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java @@ -10,10 +10,13 @@ import org.springframework.ai.tool.annotation.Tool; import org.springframework.ai.tool.annotation.ToolParam; import org.springframework.lang.Nullable; import org.springframework.stereotype.Component; +import org.springframework.beans.factory.annotation.Value; import vip.mate.agent.AgentService; import vip.mate.agent.context.ChatOrigin; +import vip.mate.agent.delegation.SubagentRegistry; import vip.mate.agent.model.AgentEntity; import vip.mate.agent.repository.AgentMapper; +import vip.mate.audit.service.AuditEventService; import vip.mate.channel.web.ChatStreamTracker; import vip.mate.workspace.conversation.ConversationService; @@ -66,11 +69,35 @@ public class DelegateAgentTool { */ private static final int PARALLEL_TIMEOUT_SECONDS = 120; - /** Tools blocked for child agents — prevents recursion and side effects. */ - private static final Set CHILD_DENIED_TOOLS = Set.of( - "delegateToAgent", // no recursive serial delegation - "delegateParallel", // no recursive parallel delegation - "listAvailableAgents" // child agents do not need to discover other agents + /** + * Default deny list for child agents. Names are matched against the + * canonical tool names exposed by the runtime, so they MUST mirror the + * actual {@code @Tool}-annotated method names. + * + *

Categories: + *

    + *
  • Recursion guards (delegate*, listAvailableAgents) — prevent a + * child from spawning another child or enumerating sibling agents.
    • + *
  • Memory writers (remember, *_structured) — children must not + * persist into the parent's shared MEMORY.md / SOUL.md surface; + * the parent owns long-term memory.
    • + *
+ * + *

{@code execute_shell_command} is intentionally NOT in the default + * deny list because legitimate dev-tooling agents rely on shell access. + * Operators that need a stricter posture can append it via + * {@code mateclaw.delegation.child-denied-tools}. + */ + static final Set DEFAULT_CHILD_DENIED_TOOLS = Set.of( + // Recursion guards. + "delegateToAgent", + "delegateParallel", + "listAvailableAgents", + // Memory writes from children would pollute the parent's shared + // long-term memory surface. + "remember", + "remember_structured", + "forget_structured" ); /** Executor for parallel delegation — one JDK 21 virtual thread per child agent. */ @@ -82,6 +109,36 @@ public class DelegateAgentTool { private final ChatStreamTracker streamTracker; private final ConversationService conversationService; private final ObjectMapper objectMapper; + private final SubagentRegistry subagentRegistry; + private final AuditEventService auditEventService; + + /** + * Operator-supplied deny-list extension. Configured via + * {@code mateclaw.delegation.child-denied-tools} as a comma-separated + * list. Empty by default — the {@link #DEFAULT_CHILD_DENIED_TOOLS} set + * already covers the recursion + memory cases that matter for safety. + */ + @Value("${mateclaw.delegation.child-denied-tools:}") + private List additionalDeniedTools; + + /** + * Effective deny list = defaults ∪ operator additions. Computed on each + * delegation entry rather than cached because Spring applies + * {@code @Value} after construction and we want operator overrides to + * take effect on the next delegation, not on the next restart. + */ + Set deniedToolsForChild() { + if (additionalDeniedTools == null || additionalDeniedTools.isEmpty()) { + return DEFAULT_CHILD_DENIED_TOOLS; + } + Set merged = new HashSet<>(DEFAULT_CHILD_DENIED_TOOLS); + for (String name : additionalDeniedTools) { + if (name != null && !name.isBlank()) { + merged.add(name.trim()); + } + } + return Set.copyOf(merged); + } // ==================== Single-task delegation ==================== @@ -126,6 +183,14 @@ public class DelegateAgentTool { } String parentConversationId = resolveParentConversationId(); + + // Spawn-pause: when the operator paused this conversation's tree + // (via /api/v1/subagents/spawn-pause), short-circuit before creating + // child state so no conversation rows / relays / registry entries leak. + if (parentConversationId != null && subagentRegistry.isSpawnPaused(parentConversationId)) { + return "[错误] Spawning paused for this conversation; resume via /api/v1/subagents/spawn-pause"; + } + String childConversationId = createChildConv(target, parentConversationId); // RFC-03 Lane C2: optionally prepend a parent-context prefix to the task. @@ -152,16 +217,38 @@ public class DelegateAgentTool { "childAgentName", target.getName(), "task", truncate(task, 200))); } - Runnable stopRelay = hasParent ? registerRelay(childConversationId, parentConversationId, target.getName()) : null; + Runnable stopRelay = hasParent ? registerBatchedRelay(childConversationId, parentConversationId, target.getName()) : null; + + // Register the live sub-agent so the operator UI / heartbeat watchdog + // can observe it. Disposable is null in the synchronous single-task + // path because the executor blocks on AgentService#chat directly — + // there is no Flux subscription to dispose. Interrupts in this path + // are best-effort (status flip; no underlying cancel). + String subagentId = parentConversationId != null + ? subagentRegistry.register(parentConversationId, childConversationId, + target.getId(), task, null) + : null; // Execute child agent — RFC-063r §2.5 改动点 5: inherit the parent // ChatOrigin and only swap the agentId, so channel binding / // workspace / requester all flow into the child. ChatOrigin parentOrigin = ChatOrigin.from(ctx); - ChildResult result = runSingleChild(0, target, taskWithContext, parentConversationId, childConversationId, parentOrigin); - - // Cleanup relay, then broadcast final result - if (stopRelay != null) stopRelay.run(); + ChildResult result; + try { + result = runSingleChild(0, target, taskWithContext, parentConversationId, childConversationId, parentOrigin); + } finally { + // Cleanup relay + registry regardless of how the child returned + // (success / exception / interruption) so we never leak entries. + if (stopRelay != null) stopRelay.run(); + if (subagentId != null) { + subagentRegistry.get(subagentId).ifPresent(rec -> { + if ("running".equals(rec.status().get())) { + rec.status().set("completed"); + } + }); + subagentRegistry.unregister(subagentId); + } + } if (hasParent) { broadcastEnd(parentConversationId, childConversationId, target.getName(), result); } @@ -204,10 +291,19 @@ public class DelegateAgentTool { } String parentConversationId = resolveParentConversationId(); + + // Spawn-pause: short-circuit before allocating any per-child state so + // we don't leak conversation rows / relays / registry entries when an + // operator paused this conversation's tree. + if (parentConversationId != null && subagentRegistry.isSpawnPaused(parentConversationId)) { + return "[错误] Spawning paused for this conversation; resume via /api/v1/subagents/spawn-pause"; + } + boolean hasParent = parentConversationId != null && streamTracker.isRunning(parentConversationId); // 2. Main thread: validate agents, create child conversations, register relays - record PreparedChild(int index, AgentEntity agent, String task, String childConvId, Runnable stopRelay) {} + record PreparedChild(int index, AgentEntity agent, String task, String childConvId, + Runnable stopRelay, String subagentId) {} List prepared = new ArrayList<>(); List errors = new ArrayList<>(); @@ -228,8 +324,14 @@ public class DelegateAgentTool { } String childConvId = createChildConv(agent, parentConversationId); - Runnable stopRelay = hasParent ? registerRelay(childConvId, parentConversationId, agent.getName()) : null; - prepared.add(new PreparedChild(i, agent, task, childConvId, stopRelay)); + Runnable stopRelay = hasParent + ? registerBatchedRelay(childConvId, parentConversationId, agent.getName()) + : null; + String subagentId = parentConversationId != null + ? subagentRegistry.register(parentConversationId, childConvId, + agent.getId(), task, null) + : null; + prepared.add(new PreparedChild(i, agent, task, childConvId, stopRelay, subagentId)); } if (prepared.isEmpty()) { @@ -331,9 +433,20 @@ public class DelegateAgentTool { long totalDurationMs = System.currentTimeMillis() - startTime; - // 6. Stop all relays + // 6. Stop all relays + drain registry entries. Both must run for every + // prepared child regardless of whether the future succeeded, timed + // out, or threw — otherwise the registry leaks one entry per stuck + // child until the JVM restarts. for (PreparedChild p : prepared) { if (p.stopRelay != null) p.stopRelay.run(); + if (p.subagentId != null) { + subagentRegistry.get(p.subagentId).ifPresent(rec -> { + if ("running".equals(rec.status().get())) { + rec.status().set("completed"); + } + }); + subagentRegistry.unregister(p.subagentId); + } } // 7. Broadcast delegation_end with per-child structured summary @@ -440,7 +553,7 @@ public class DelegateAgentTool { private ChildResult runSingleChild(int taskIndex, AgentEntity target, String task, String parentConversationId, String childConversationId, ChatOrigin parentOrigin) { - DelegationContext.enter(parentConversationId, CHILD_DENIED_TOOLS); + DelegationContext.enter(parentConversationId, deniedToolsForChild()); try { long startTime = System.currentTimeMillis(); // RFC-063r §2.5 改动点 5: inherit parent origin, swap agentId @@ -684,6 +797,43 @@ public class DelegateAgentTool { }); } + /** + * Registers a batched relay so a chatty child does not flood the parent + * transcript with one tool-call event per LLM step. The streaming layer + * batches {@code tool_call_started} / {@code tool_call_completed} into + * envelopes (5 events / 500 ms) and flushes immediately on lifecycle + * events ({@code subagent_*}, {@code error}, {@code phase}, etc.). + * + *

The wrapper keeps the on-the-wire shape identical to + * {@link #registerRelay} so frontend consumers do not need to change + * — both batched envelopes and pass-through events surface as + * {@code delegation_progress} on the parent. + */ + private Runnable registerBatchedRelay(String childConvId, String parentConvId, String childAgentName) { + return streamTracker.addBatchedEventRelay(childConvId, parentConvId, 5, 500L, + (eventName, jsonData) -> { + if ("tool_call_started".equals(eventName) + || "tool_call_completed".equals(eventName) + || "phase".equals(eventName)) { + try { + Object parsedData; + try { + parsedData = objectMapper.readValue(jsonData, Object.class); + } catch (Exception ignored) { + parsedData = jsonData; + } + streamTracker.broadcastObject(parentConvId, "delegation_progress", Map.of( + "childConversationId", childConvId, + "childAgentName", childAgentName, + "originalEvent", eventName, + "data", parsedData)); + } catch (Exception e) { + log.debug("Batched relay error: {}", e.getMessage()); + } + } + }); + } + private void broadcastEnd(String parentConvId, String childConvId, String agentName, ChildResult result) { streamTracker.broadcastObject(parentConvId, "delegation_end", Map.of( "childConversationId", childConvId, diff --git a/mateclaw-ui/src/components/chat/MessageBubble.vue b/mateclaw-ui/src/components/chat/MessageBubble.vue index 63b391fb..ac0a61be 100644 --- a/mateclaw-ui/src/components/chat/MessageBubble.vue +++ b/mateclaw-ui/src/components/chat/MessageBubble.vue @@ -23,10 +23,26 @@

- @@ -751,6 +767,45 @@ const segments = computed(() => { /** 是否使用分段模式渲染（有 segments 数据且包含多个分段） */ const useSegmentedView = computed(() => segments.value.length > 1) +/** + * Group segments by iterationIndex so each ReAct iteration renders as its own + * thinking/tool-calls/content cluster. Falls back to a single ungrouped bucket + * for legacy messages (no iterationIndex tagged) so historical conversations + * keep rendering as before — including the existing "single-thinking reorder" + * normalization done in the `segments` computed above. + */ +const groupedIterations = computed(() => { + const segs = segments.value || [] + const anyTagged = segs.some(s => typeof s.iterationIndex === 'number') + if (!anyTagged) { + return [{ + key: 'all', + index: 0, + empty: false, + thinkings: segs.filter(s => s.type === 'thinking'), + tools: segs.filter(s => s.type === 'tool_call'), + contents: segs.filter(s => s.type === 'content'), + }] + } + const buckets = new Map() + for (const s of segs) { + const idx = s.iterationIndex ?? 0 + if (!buckets.has(idx)) buckets.set(idx, { thinkings: [], tools: [], contents: [] }) + const b = buckets.get(idx)! + if (s.type === 'thinking') b.thinkings.push(s) + else if (s.type === 'tool_call') b.tools.push(s) + else if (s.type === 'content') b.contents.push(s) + } + return [...buckets.entries()] + .sort(([a], [b]) => a - b) + .map(([index, b]) => ({ + key: `iter-${index}`, + index, + empty: b.thinkings.length === 0 && b.tools.length === 0 && b.contents.length === 0, + ...b, + })) +}) + const toolCallsMeta = computed(() => { return parsedMetadata.value?.toolCalls || [] }) @@ -859,6 +914,43 @@ watch(isGenerating, (generating) => { padding: 4px 0; } +/* Iteration "no output" chip (interrupted iteration). */ +.iter-empty-chip { + display: inline-flex; + align-items: center; + gap: 6px; + align-self: flex-start; + padding: 4px 10px; + margin: 4px 0; + font-size: 12px; + color: var(--mc-text-tertiary, #94a3b8); + background: var(--mc-bg-elevated, #f8fafc); + border: 1px dashed var(--mc-border, #e2e8f0); + border-radius: 12px; +} + +/* Inline informational banner shown when the backend trimmed a repetitive + tail off a content segment. Amber, not red — this is informational. */ +.repetition-warning { + display: flex; + align-items: center; + gap: 6px; + padding: 6px 10px; + margin: 6px 0 2px; + font-size: 12px; + color: #92400e; + background: rgba(245, 158, 11, 0.08); + border-left: 3px solid var(--mc-warning, #f59e0b); + border-radius: 4px; +} +.repetition-warning__text { + flex: 1; +} +.repetition-warning__meta { + color: var(--mc-text-tertiary, #94a3b8); + font-size: 11px; +} + .message-wrapper { display: flex; gap: 12px; diff --git a/mateclaw-ui/src/components/chat/StreamLoadingBar.vue b/mateclaw-ui/src/components/chat/StreamLoadingBar.vue index 69345878..bb831e31 100644 --- a/mateclaw-ui/src/components/chat/StreamLoadingBar.vue +++ b/mateclaw-ui/src/components/chat/StreamLoadingBar.vue @@ -28,6 +28,14 @@ import { ref, computed, watch, onBeforeUnmount } from 'vue' import { useI18n } from 'vue-i18n' import type { PhaseEventData, StreamPhase } from '@/types' +/** Pre-token lifecycle stage from useChat. Drives the loading copy in the + * window between "user sent" and "first token landed". */ +interface LifecycleStage { + stage: 'connecting' | 'started' | 'context_prepared' | 'llm_request_sent' | 'streaming' + detail?: any + since: number +} + interface Props { isLoading: boolean toolCount?: number @@ -43,6 +51,8 @@ interface Props { runningToolName?: string /** 是否有排队消息 */ hasQueued?: boolean + /** Fine-grained pre-token stage. Preferred over `phase` while no token has arrived. */ + lifecycleStage?: LifecycleStage | null } const props = withDefaults(defineProps(), { @@ -55,6 +65,7 @@ const props = withDefaults(defineProps(), { phaseInfo: null, runningToolName: '', hasQueued: false, + lifecycleStage: null, }) const { t } = useI18n() @@ -95,7 +106,35 @@ const userPhaseI18nMap: Record = { stopped: 'chat.streamStopped', } +const lifecycleI18nMap: Record = { + connecting: 'chat.streamConnecting', + started: 'chat.streamStarted', + context_prepared: 'chat.streamContextPrepared', + llm_request_sent: 'chat.streamLlmRequestSent', +} + +/** True iff the pre-token lifecycle is active (no first delta yet). */ +const inPreTokenWindow = computed(() => { + const ls = props.lifecycleStage + return !!ls && ls.stage !== 'streaming' +}) + const statusText = computed(() => { + // Prefer fine-grained pre-token text when no first delta has arrived yet. + if (inPreTokenWindow.value && props.lifecycleStage) { + const key = lifecycleI18nMap[props.lifecycleStage.stage] + if (key) { + const base = t(key) + // Append "(Xs elapsed)" once the same stage has lingered for >= 5s, so + // the user can see something is still happening even if the next stage + // is delayed (e.g. a slow context_prepared → llm_request_sent gap). + const sec = Math.floor((Date.now() - props.lifecycleStage.since) / 1000) + if (sec >= 5 && stageTickSec.value >= 5) { + return base + t('chat.streamElapsedSuffix', { sec: stageTickSec.value }) + } + return base + } + } const key = userPhaseI18nMap[userFacingPhase.value] if (!key) return '' return t(key) @@ -137,6 +176,12 @@ const phaseTextClass = computed(() => { const elapsedSeconds = ref(0) const elapsedTime = ref('0s') +/** Seconds elapsed since the current lifecycleStage became active. Used by + * statusText to decide whether to append the "(Xs elapsed)" hint. Tick at + * 1 Hz only while the pre-token window is open to keep idle cost minimal. */ +const stageTickSec = ref(0) +let stageTimer: ReturnType | null = null + let timerInterval: ReturnType | null = null @@ -166,11 +211,28 @@ watch(() => props.isLoading, (loading) => { }, { immediate: true }) +// Tick the stage-elapsed counter only while the pre-token window is open. +// Reset on every stage transition so the "(Xs elapsed)" hint reflects time +// in the *current* stage, not total time since send. +watch(() => props.lifecycleStage, (ls) => { + if (stageTimer) { clearInterval(stageTimer); stageTimer = null } + stageTickSec.value = 0 + if (ls && ls.stage !== 'streaming') { + stageTimer = setInterval(() => { + stageTickSec.value = Math.floor((Date.now() - ls.since) / 1000) + }, 1000) + } +}, { immediate: true }) + onBeforeUnmount(() => { if (timerInterval) { clearInterval(timerInterval) timerInterval = null } + if (stageTimer) { + clearInterval(stageTimer) + stageTimer = null + } }) diff --git a/mateclaw-ui/src/components/chat/ThinkingSegment.vue b/mateclaw-ui/src/components/chat/ThinkingSegment.vue index dd3e0c87..243a7a49 100644 --- a/mateclaw-ui/src/components/chat/ThinkingSegment.vue +++ b/mateclaw-ui/src/components/chat/ThinkingSegment.vue @@ -11,13 +11,16 @@ const props = defineProps<{ segment: MessageSegment }>() +// Expand while streaming so the user can watch the model think in real time; +// auto-collapse the moment streaming ends so the assistant's final answer +// stays the focal point without a long reasoning block above it. The user +// can click the header to re-expand at any time. const expanded = ref(props.segment.status === 'running') const { renderMarkdown } = useMarkdownRenderer() const renderedThinking = computed(() => renderMarkdown(props.segment.thinkingText || '')) const isRunning = computed(() => props.segment.status === 'running') -// running 结束后自动折叠 watch(() => props.segment.status, (val) => { if (val === 'completed') expanded.value = false }) diff --git a/mateclaw-ui/src/composables/chat/useChat.ts b/mateclaw-ui/src/composables/chat/useChat.ts index d4d59b78..515538ee 100644 --- a/mateclaw-ui/src/composables/chat/useChat.ts +++ b/mateclaw-ui/src/composables/chat/useChat.ts @@ -62,6 +62,16 @@ export interface UseChatReturn { queueSize: import('vue').ComputedRef /** Latest heartbeat data */ heartbeat: import('vue').Ref + /** + * Fine-grained pre-token lifecycle stage. Drives the loading bar copy in the + * window between "send pressed" and "first delta arrived". `null` once a + * delta is observed (StreamLoadingBar then falls back to `phase`-derived text). + */ + lifecycleStage: import('vue').Ref<{ + stage: 'connecting' | 'started' | 'context_prepared' | 'llm_request_sent' | 'streaming' + detail?: any + since: number + } | null> /** Send a message (can be called while generating — automatically routes to interrupt/queue) */ sendMessage: (content: string, options: SendMessageOptions) => Promise /** Stop generation (user-initiated stop; does not auto-resume queued messages) */ @@ -123,6 +133,27 @@ export function useChat(options: UseChatOptions): UseChatReturn { const segIdCounter = { value: 0 } const genSegId = () => `seg-${Date.now()}-${segIdCounter.value++}` + /** + * Fine-grained lifecycle stage exposed to the UI for the "connecting → started + * → context_prepared → llm_request_sent → streaming" loading bar. Reset on + * every new turn; transitions to `streaming` implicitly when the first + * thinking/content delta lands. + */ + const lifecycleStage = ref<{ + stage: 'connecting' | 'started' | 'context_prepared' | 'llm_request_sent' | 'streaming' + detail?: any + since: number + } | null>(null) + + /** Helper: tag a freshly-created segment with the active iteration / scope. */ + function applyIterationTags(seg: MessageSegment) { + const stash = currentSegments.value as any + const idx = stash._currentIteration + if (typeof idx === 'number') seg.iterationIndex = idx + const subId = stash._currentSubagentId + if (subId) seg.subagentId = subId + } + /** Unique ID for the current turn — prevents flushSegmentsToMessage from writing stale segments to a new message */ let activeTurnId = '' @@ -296,6 +327,11 @@ export function useChat(options: UseChatOptions): UseChatReturn { if (['thinking', 'reasoning', 'drafting_answer', 'preparing_context'].includes(streamPhase.value)) { streamPhase.value = 'streaming' } + // First visible delta — flip lifecycleStage so the loading bar drops the + // pre-stream messaging and yields to the per-phase status text. + if (lifecycleStage.value && lifecycleStage.value.stage !== 'streaming') { + lifecycleStage.value = { stage: 'streaming', since: Date.now() } + } // Segments: append to the current running content segment, or create a new one const segs = currentSegments.value let contentSeg = segs.findLast((s: MessageSegment) => s.type === 'content' && s.status === 'running') @@ -304,6 +340,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { const thinkingSeg = segs.findLast((s: MessageSegment) => s.type === 'thinking' && s.status === 'running') if (thinkingSeg) thinkingSeg.status = 'completed' contentSeg = { id: genSegId(), type: 'content', status: 'running', text: '', timestamp: Date.now() } + applyIterationTags(contentSeg) segs.push(contentSeg) flushSegmentsToMessage() // sync once when a new content segment is created } @@ -330,12 +367,17 @@ export function useChat(options: UseChatOptions): UseChatReturn { ) if (!thinkSeg) { thinkSeg = { id: genSegId(), type: 'thinking', status: 'running', thinkingText: '', timestamp: Date.now() } + applyIterationTags(thinkSeg) // Append in timeline order (interleaved with tool_calls) — old behavior unshift'd to top, // but with per-round splitting that misorders rounds 2+ relative to their tool calls. segs.push(thinkSeg) flushSegmentsToMessage() } thinkSeg.thinkingText = (thinkSeg.thinkingText || '') + (data.delta || '') + // First thinking delta — same lifecycle flip as content_delta. + if (lifecycleStage.value && lifecycleStage.value.stage !== 'streaming') { + lifecycleStage.value = { stage: 'streaming', since: Date.now() } + } } }) @@ -458,6 +500,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { : data.status === 'stopped' ? 'stopped' : 'completed' if (data.status !== 'awaiting_approval') { phaseInfo.value = null + lifecycleStage.value = null expirePendingApprovals(data.status === 'stopped' ? 'stopped' : 'completed') } @@ -533,6 +576,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { error.value = new Error(errorMessage) streamPhase.value = 'idle' phaseInfo.value = null + lifecycleStage.value = null // Clear queue on error to avoid stale state messageQueue.clear() expirePendingApprovals('failed') @@ -550,7 +594,9 @@ export function useChat(options: UseChatOptions): UseChatReturn { // ===== Agent event handlers ===== - stream.on('tool_call_started', (data) => { + // Body of tool_call_started — extracted so delegation_batch can replay the + // same behavior for buffered child events without duplicating logic. + function handleToolCallStarted(data: any) { if (isStaleEvent(data)) return streamPhase.value = 'executing_tool' if (currentAssistantId.value) { @@ -577,17 +623,20 @@ export function useChat(options: UseChatOptions): UseChatReturn { const segs = currentSegments.value const runningSeg = segs.findLast((s: MessageSegment) => s.status === 'running' && (s.type === 'thinking' || s.type === 'content')) if (runningSeg) runningSeg.status = 'completed' - segs.push({ + const toolSeg: MessageSegment = { id: genSegId(), type: 'tool_call', status: 'running', toolCallId: data.toolCallId || '', toolName: data.toolName, toolArgs: data.arguments, timestamp: data.timestamp || Date.now(), - }) + } + applyIterationTags(toolSeg) + segs.push(toolSeg) flushSegmentsToMessage() } - }) + } - stream.on('tool_call_completed', (data) => { + // Body of tool_call_completed — see handleToolCallStarted. + function handleToolCallCompleted(data: any) { if (isStaleEvent(data)) return if (currentAssistantId.value) { const msg = getMessage(currentAssistantId.value) @@ -641,7 +690,10 @@ export function useChat(options: UseChatOptions): UseChatReturn { pendingAsyncTaskIds.add(taskId) } } - }) + } + + stream.on('tool_call_started', handleToolCallStarted) + stream.on('tool_call_completed', handleToolCallCompleted) // ===== Browser action events ===== @@ -875,6 +927,123 @@ export function useChat(options: UseChatOptions): UseChatReturn { } }) + // ===== Stream lifecycle (pre-token) events ===== + // These give the loading bar substantive status text in the gap between + // "user pressed send" and "first token arrived" — eliminating the dead air + // where the user only saw a spinner with no progress signal. + stream.on('stream_started', (data) => { + if (isStaleEvent(data)) return + lifecycleStage.value = { stage: 'started', since: Date.now() } + }) + + stream.on('context_prepared', (data) => { + if (isStaleEvent(data)) return + lifecycleStage.value = { stage: 'context_prepared', detail: data, since: Date.now() } + }) + + stream.on('llm_request_sent', (data) => { + if (isStaleEvent(data)) return + lifecycleStage.value = { stage: 'llm_request_sent', detail: data, since: Date.now() } + }) + + // ===== Per-iteration boundaries (single-turn UX overhaul) ===== + stream.on('iteration_start', (data) => { + if (isStaleEvent(data)) return + // Force-close any running thinking/content segments — guarantees no segment + // belonging to the next iteration is appended onto the previous one's tail. + const segs = currentSegments.value + for (const seg of segs) { + if (seg.status === 'running' && (seg.type === 'thinking' || seg.type === 'content')) { + seg.status = 'completed' + } + } + // Stash iteration / scope on the array so segment factories pick them up. + ;(currentSegments.value as any)._currentIteration = data.index ?? 0 + ;(currentSegments.value as any)._currentScope = data.scope ?? 'parent' + ;(currentSegments.value as any)._currentSubagentId = data.subagentId + flushSegmentsToMessage() + }) + + stream.on('iteration_end', (data) => { + if (isStaleEvent(data)) return + // Sentence-level repetition warning runs on the backend (content_truncated + // event); we don't recompute it here. Just close any still-running + // thinking/content segments belonging to the iteration that's wrapping up. + const segs = currentSegments.value + for (const seg of segs) { + if (seg.status === 'running' && (seg.type === 'thinking' || seg.type === 'content')) { + seg.status = 'completed' + } + } + flushSegmentsToMessage() + }) + + stream.on('thinking_start', (data) => { + if (isStaleEvent(data)) return + // Pure analytics signal — thinking_delta will create the segment as needed. + if (currentAssistantId.value) streamPhase.value = 'thinking' + }) + + stream.on('thinking_end', (data) => { + if (isStaleEvent(data)) return + // Auto-collapse decisions belong to ThinkingSegment.vue. We deliberately + // do not flip status here — thinking_delta after thinking_end is rare but + // valid (e.g. a late provider chunk), and we want it to extend the same + // segment rather than open a new one. + }) + + stream.on('content_truncated', (data) => { + if (isStaleEvent(data)) return + // Mark the running content segment with a repetition warning so the UI + // can render an inline informational banner. + const segs = currentSegments.value + const contentSeg = segs.findLast((s: MessageSegment) => s.type === 'content' && s.status === 'running') + if (contentSeg) { + contentSeg.repetitionWarning = data.reason + contentSeg.truncatedChars = data.truncatedChars + } + flushSegmentsToMessage() + }) + + stream.on('tool_result_chunk', (data) => { + if (isStaleEvent(data)) return + // Streamed tool result delta — append to the matching tool_call segment. + // tool_call_completed still fires separately and carries the canonical + // success/result fields; this just lets large results stream in instead + // of arriving as one giant blob. + const segs = currentSegments.value + const toolSeg = segs.find((s: MessageSegment) => + s.type === 'tool_call' && s.toolCallId === data.ref) + if (toolSeg) { + toolSeg.toolResult = (toolSeg.toolResult || '') + (data.delta || '') + // Note: data.final = true just means the buffer for this tool's result + // is exhausted. Final success/error state still arrives via + // tool_call_completed, so we don't terminate the segment here. + } + flushSegmentsToMessage() + }) + + stream.on('delegation_batch', (data) => { + if (isStaleEvent(data)) return + // Buffered child events from a delegated subagent. Replay them in order + // through the same handlers as live events so segment state stays + // consistent with the rest of the timeline. + const events = Array.isArray(data?.events) ? data.events : [] + for (const ev of events) { + const evData = ev?.data ?? {} + switch (ev?.event) { + case 'tool_call_started': + handleToolCallStarted(evData) + break + case 'tool_call_completed': + handleToolCallCompleted(evData) + break + // Other event kinds (phase / thinking_delta / content_delta / etc.) + // are not currently produced inside batches; extend here when added. + } + } + }) + stream.on('plan_created', (data) => { if (isStaleEvent(data)) return if (currentAssistantId.value) { @@ -1140,6 +1309,8 @@ export function useChat(options: UseChatOptions): UseChatReturn { currentAssistantId.value = assistantMessage.id as string streamPhase.value = options.thinkingLevel?.value === 'off' ? 'streaming' : 'thinking' phaseInfo.value = null + // New turn — reset lifecycle so the loading bar shows pre-token progress. + lifecycleStage.value = { stage: 'connecting', since: Date.now() } }) // ===== Async task completion events (video / image / music generation) ===== @@ -1293,6 +1464,9 @@ export function useChat(options: UseChatOptions): UseChatReturn { streamConversationId = conversationId streamPhase.value = thinkingLevelRef?.value === 'off' ? 'streaming' : 'thinking' phaseInfo.value = null + // Begin pre-token lifecycle. Subsequent stream_started / context_prepared / + // llm_request_sent events override this; first delta clears it. + lifecycleStage.value = { stage: 'connecting', since: Date.now() } try { if (!isApprovalCommand) { @@ -1555,6 +1729,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { segIdCounter.value = 0 streamPhase.value = 'idle' phaseInfo.value = null + lifecycleStage.value = null error.value = null messageQueue.clear() if (stopFallbackTimer) { @@ -1573,6 +1748,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { hasQueued: messageQueue.hasQueued, queueSize: messageQueue.queueSize, heartbeat, + lifecycleStage, sendMessage, stopGeneration, cancelQueued, diff --git a/mateclaw-ui/src/composables/chat/useStream.ts b/mateclaw-ui/src/composables/chat/useStream.ts index 926c818a..35afdf47 100644 --- a/mateclaw-ui/src/composables/chat/useStream.ts +++ b/mateclaw-ui/src/composables/chat/useStream.ts @@ -44,6 +44,20 @@ export type SSEEventType = | 'delegation_progress' | 'delegation_end' | 'delegation_child_complete' + // Stream lifecycle + per-iteration boundaries (single-turn UX overhaul). + // The parser handles arbitrary `event:` lines via parseEvent — these names + // exist in the union purely so TypeScript callers can register handlers + // with a typed `on(event, handler)` signature. + | 'stream_started' + | 'context_prepared' + | 'llm_request_sent' + | 'thinking_start' + | 'thinking_end' + | 'iteration_start' + | 'iteration_end' + | 'content_truncated' + | 'tool_result_chunk' + | 'delegation_batch' export interface SSEEvent { type: SSEEventType diff --git a/mateclaw-ui/src/i18n/locales/en-US.ts b/mateclaw-ui/src/i18n/locales/en-US.ts index 50137f16..c598cadd 100644 --- a/mateclaw-ui/src/i18n/locales/en-US.ts +++ b/mateclaw-ui/src/i18n/locales/en-US.ts @@ -248,6 +248,15 @@ export default { streamSlowSummarizing: 'Organizing a larger tool result. This step often takes longer.', streamSlowReasoning: 'The model is doing heavier analysis here. This is not a frozen connection.', streamSlowGeneral: 'Processing is still ongoing. The time is currently being spent in internal analysis.', + // Pre-token lifecycle stages + streamConnecting: 'Connecting to server...', + streamStarted: 'Connection established, preparing context...', + streamContextPrepared: 'Context ready, calling LLM...', + streamLlmRequestSent: 'Waiting for first token...', + streamElapsedSuffix: ' ({sec}s elapsed)', + // Per-iteration grouping + iterationEmpty: 'Iteration {index} interrupted (no output)', + contentRepetitionWarning: 'Repetitive content detected near the end (model artifact)', // Approval bar approvalAllow: 'Allow', approvalExecute: 'to execute?', diff --git a/mateclaw-ui/src/i18n/locales/zh-CN.ts b/mateclaw-ui/src/i18n/locales/zh-CN.ts index 8db7ceb8..453543e2 100644 --- a/mateclaw-ui/src/i18n/locales/zh-CN.ts +++ b/mateclaw-ui/src/i18n/locales/zh-CN.ts @@ -248,6 +248,15 @@ export default { streamSlowSummarizing: '正在整理较长的工具结果。因为信息较多，这一步可能需要更久。', streamSlowReasoning: '正在做较复杂的分析和取舍，不是卡住。', streamSlowGeneral: '处理还在继续，当前主要耗时在内部分析，不代表连接中断。', + // 首 token 前的生命周期阶段 + streamConnecting: '正在连接服务器...', + streamStarted: '连接已建立，正在准备上下文...', + streamContextPrepared: '上下文已就绪，正在调用 LLM...', + streamLlmRequestSent: '等待模型首个响应...', + streamElapsedSuffix: '（已等 {sec} 秒）', + // 按轮次分组渲染 + iterationEmpty: '第 {index} 轮被中断（无输出）', + contentRepetitionWarning: '检测到内容尾部重复（疑似模型输出 artifact）', // 审批栏 approvalAllow: '允许', approvalExecute: '执行？', diff --git a/mateclaw-ui/src/types/index.ts b/mateclaw-ui/src/types/index.ts index 8c95a3fd..fb5d824d 100644 --- a/mateclaw-ui/src/types/index.ts +++ b/mateclaw-ui/src/types/index.ts @@ -162,6 +162,18 @@ export interface MessageSegment { plan?: PlanMeta /** 时间戳 */ timestamp?: number + /** + * Iteration index this segment belongs to (0-based). Set by iteration_start — + * lets MessageBubble group thinking/tool/content segments per iteration so + * the next iteration's output never appends onto the previous one's tail. + */ + iterationIndex?: number + /** Subagent / delegation child ID, when this segment was emitted under a child scope. */ + subagentId?: string + /** Backend signaled the running content was truncated to break a repetition pattern. */ + repetitionWarning?: 'char_pattern' | 'sentence_repetition' + /** Number of trailing characters dropped when the repetition guard fired. */ + truncatedChars?: number } export interface MessageMetadata { diff --git a/mateclaw-ui/src/views/ChatConsole.vue b/mateclaw-ui/src/views/ChatConsole.vue index 8d339977..716dd532 100644 --- a/mateclaw-ui/src/views/ChatConsole.vue +++ b/mateclaw-ui/src/views/ChatConsole.vue @@ -251,6 +251,7 @@ :phase-info="phaseInfo" :running-tool-name="currentRunningToolName" :has-queued="hasQueued" + :lifecycle-stage="lifecycleStage" /> @@ -608,6 +609,7 @@ const { hasQueued, queueSize, heartbeat, + lifecycleStage, sendMessage: sendChatMessage, stopGeneration: stopChatGeneration, cancelQueued,