diff --git a/mateclaw-server/pom.xml b/mateclaw-server/pom.xml index a7bddb0c..d2f4f785 100644 --- a/mateclaw-server/pom.xml +++ b/mateclaw-server/pom.xml @@ -367,6 +367,20 @@ spring-boot-starter-test test + + + + com.tngtech.archunit + archunit-junit5 + 1.3.0 + test + diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java index a7886f2e..235c25e0 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java @@ -7,6 +7,8 @@ import org.springframework.context.event.EventListener; import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; import reactor.core.publisher.Flux; +import vip.mate.agent.context.ChatOrigin; +import vip.mate.agent.context.ChatOriginHolder; import vip.mate.agent.model.AgentEntity; import vip.mate.agent.repository.AgentMapper; import vip.mate.exception.MateClawException; @@ -98,31 +100,67 @@ public class AgentService { // ==================== 运行时入口 ==================== public String chat(Long agentId, String message, String conversationId) { + return chat(agentId, message, conversationId, ChatOrigin.EMPTY); + } + + /** + * RFC-063r §2.5: preferred entry — accepts the originating + * {@link ChatOrigin} so channel binding and workspace context propagate + * down to {@code @Tool} methods via Spring AI {@link org.springframework.ai.chat.model.ToolContext}. + */ + public String chat(Long agentId, String message, String conversationId, ChatOrigin origin) { memoryRecallTracker.trackRecalls(agentId, message); BaseAgent agent = getOrBuildAgent(agentId); - return withLifecycleSync(agentId, message, conversationId, - (msg, convId) -> agent.chat(msg, convId)); + ChatOriginHolder.set(origin != null ? origin : ChatOrigin.EMPTY); + try { + return withLifecycleSync(agentId, message, conversationId, + (msg, convId) -> agent.chat(msg, convId)); + } finally { + ChatOriginHolder.clear(); + } } public Flux chatStream(Long agentId, String message, String conversationId) { + return chatStream(agentId, message, conversationId, ChatOrigin.EMPTY); + } + + public Flux chatStream(Long agentId, String message, String conversationId, ChatOrigin origin) { memoryRecallTracker.trackRecalls(agentId, message); BaseAgent agent = getOrBuildAgent(agentId); - return withLifecycleFlux(agentId, message, conversationId, - (msg, convId) -> agent.chatStream(msg, convId), - chunk -> chunk); + // Capture the origin into a request-scoped holder; cleared on Flux + // termination so the next reactive subscriber doesn't inherit stale state. + ChatOrigin captured = origin != null ? origin : ChatOrigin.EMPTY; + return Flux.defer(() -> { + ChatOriginHolder.set(captured); + return withLifecycleFlux(agentId, message, conversationId, + (msg, convId) -> agent.chatStream(msg, convId), + chunk -> chunk); + }).doFinally(signal -> ChatOriginHolder.clear()); } public Flux chatStructuredStream(Long agentId, String message, String conversationId) { - return chatStructuredStream(agentId, message, conversationId, "", null); + return chatStructuredStream(agentId, message, conversationId, "", null, ChatOrigin.EMPTY); } public Flux chatStructuredStream(Long agentId, String message, String conversationId, String requesterId) { - return chatStructuredStream(agentId, message, conversationId, requesterId, null); + return chatStructuredStream(agentId, message, conversationId, requesterId, null, ChatOrigin.EMPTY); + } + + public Flux chatStructuredStream(Long agentId, String message, String conversationId, + String requesterId, ChatOrigin origin) { + return chatStructuredStream(agentId, message, conversationId, requesterId, null, origin); } public Flux chatStructuredStream(Long agentId, String message, String conversationId, String requesterId, String thinkingLevel) { + return chatStructuredStream(agentId, message, conversationId, requesterId, thinkingLevel, + ChatOrigin.EMPTY); + } + + public Flux chatStructuredStream(Long agentId, String message, String conversationId, + String requesterId, String thinkingLevel, + ChatOrigin origin) { memoryRecallTracker.trackRecalls(agentId, message); BaseAgent agent = getOrBuildAgent(agentId); @@ -139,27 +177,45 @@ public class AgentService { } } + ChatOrigin captured = origin != null ? origin : ChatOrigin.EMPTY; if (agent instanceof StructuredStreamCapable capable) { - return withLifecycleFlux(agentId, message, conversationId, - (msg, convId) -> capable.chatStructuredStream(msg, convId, - requesterId != null ? requesterId : "") - .doFinally(signal -> ThinkingLevelHolder.clear()), - StreamDelta::content); + return Flux.defer(() -> { + ChatOriginHolder.set(captured); + return withLifecycleFlux(agentId, message, conversationId, + (msg, convId) -> capable.chatStructuredStream(msg, convId, + requesterId != null ? requesterId : "") + .doFinally(signal -> ThinkingLevelHolder.clear()), + StreamDelta::content); + }) + .doFinally(signal -> ChatOriginHolder.clear()); } // 降级：不支持结构化流的 Agent，包装为纯内容流 ThinkingLevelHolder.clear(); - return withLifecycleFlux(agentId, message, conversationId, - (msg, convId) -> agent.chatStream(msg, convId) - .map(chunk -> new StreamDelta(chunk, null)), - StreamDelta::content); + return Flux.defer(() -> { + ChatOriginHolder.set(captured); + return withLifecycleFlux(agentId, message, conversationId, + (msg, convId) -> agent.chatStream(msg, convId) + .map(chunk -> new StreamDelta(chunk, null)), + StreamDelta::content); + }) + .doFinally(signal -> ChatOriginHolder.clear()); } public String execute(Long agentId, String goal, String conversationId) { + return execute(agentId, goal, conversationId, ChatOrigin.EMPTY); + } + + public String execute(Long agentId, String goal, String conversationId, ChatOrigin origin) { memoryRecallTracker.trackRecalls(agentId, goal); BaseAgent agent = getOrBuildAgent(agentId); - return withLifecycleSync(agentId, goal, conversationId, - (msg, convId) -> agent.execute(msg, convId)); + ChatOriginHolder.set(origin != null ? origin : ChatOrigin.EMPTY); + try { + return withLifecycleSync(agentId, goal, conversationId, + (msg, convId) -> agent.execute(msg, convId)); + } finally { + ChatOriginHolder.clear(); + } } /** @@ -173,10 +229,20 @@ public class AgentService { */ public String chatWithReplay(Long agentId, String userMessage, String conversationId, String toolCallPayload) { + return chatWithReplay(agentId, userMessage, conversationId, toolCallPayload, ChatOrigin.EMPTY); + } + + public String chatWithReplay(Long agentId, String userMessage, String conversationId, + String toolCallPayload, ChatOrigin origin) { memoryRecallTracker.trackRecalls(agentId, userMessage); BaseAgent agent = getOrBuildAgent(agentId); - return withLifecycleSync(agentId, userMessage, conversationId, - (msg, convId) -> agent.chatWithReplay(msg, convId, toolCallPayload)); + ChatOriginHolder.set(origin != null ? origin : ChatOrigin.EMPTY); + try { + return withLifecycleSync(agentId, userMessage, conversationId, + (msg, convId) -> agent.chatWithReplay(msg, convId, toolCallPayload)); + } finally { + ChatOriginHolder.clear(); + } } /** @@ -184,17 +250,29 @@ public class AgentService { */ public Flux chatWithReplayStream(Long agentId, String userMessage, String conversationId, String toolCallPayload) { - return chatWithReplayStream(agentId, userMessage, conversationId, toolCallPayload, ""); + return chatWithReplayStream(agentId, userMessage, conversationId, toolCallPayload, "", ChatOrigin.EMPTY); } public Flux chatWithReplayStream(Long agentId, String userMessage, String conversationId, String toolCallPayload, String requesterId) { + return chatWithReplayStream(agentId, userMessage, conversationId, toolCallPayload, requesterId, + ChatOrigin.EMPTY); + } + + public Flux chatWithReplayStream(Long agentId, String userMessage, String conversationId, + String toolCallPayload, String requesterId, + ChatOrigin origin) { memoryRecallTracker.trackRecalls(agentId, userMessage); BaseAgent agent = getOrBuildAgent(agentId); - return withLifecycleFlux(agentId, userMessage, conversationId, - (msg, convId) -> agent.chatWithReplayStream(msg, convId, toolCallPayload, - requesterId != null ? requesterId : ""), - StreamDelta::content); + ChatOrigin captured = origin != null ? origin : ChatOrigin.EMPTY; + return Flux.defer(() -> { + ChatOriginHolder.set(captured); + return withLifecycleFlux(agentId, userMessage, conversationId, + (msg, convId) -> agent.chatWithReplayStream(msg, convId, toolCallPayload, + requesterId != null ? requesterId : ""), + StreamDelta::content); + }) + .doFinally(signal -> ChatOriginHolder.clear()); } public AgentState getAgentState(Long agentId) { diff --git a/mateclaw-server/src/main/java/vip/mate/agent/context/ChannelTarget.java b/mateclaw-server/src/main/java/vip/mate/agent/context/ChannelTarget.java new file mode 100644 index 00000000..da3c769a --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/context/ChannelTarget.java @@ -0,0 +1,22 @@ +package vip.mate.agent.context; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.springframework.lang.Nullable; + +/** + * Channel-bound target identity used when an agent's response must be delivered + * back to a specific external channel (cron, IM relay, etc.). + * + *

Kept as a sub-VO of {@link ChatOrigin} so future channel-related fields do + * not pollute the top-level origin record. + * + *

Field evolution rule: only-add, do-not-rename, deprecate-for-90-days before + * physical removal — see {@link ChatOrigin}'s class doc. + */ +@JsonIgnoreProperties(ignoreUnknown = true) +public record ChannelTarget( + @Nullable String targetId, + @Nullable String threadId, + @Nullable String accountId +) { +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/context/ChatOrigin.java b/mateclaw-server/src/main/java/vip/mate/agent/context/ChatOrigin.java new file mode 100644 index 00000000..56c1a01e --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/context/ChatOrigin.java @@ -0,0 +1,102 @@ +package vip.mate.agent.context; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.springframework.ai.chat.model.ToolContext; +import org.springframework.lang.Nullable; + +import java.util.Map; + +/** + * Immutable value object that travels alongside an agent invocation describing + * where the request came from — channel, conversation, requester, + * workspace, and optional delivery target. + * + *

Replaces ad-hoc ThreadLocal threading (RFC-063 v1) with explicit Spring AI + * {@link ToolContext} carriage (RFC-063r §2.1). The wither-style API enables + * the agent runtime to enrich the origin (agentId, workspace) without mutation. + * + *

Field evolution rule

+ *
    + *
  • Only add — never delete; deprecate at least 90 days (covers approval TTL) + * before physical removal.
    • + *
  • Never rename — add a new field plus deprecate-old-field, double-write + * during the migration window.
    • + *
  • {@link JsonIgnoreProperties#ignoreUnknown()} guards forward/backward + * compatibility when older approval rows are deserialized after upgrades.
    • + *
+ */ +@JsonIgnoreProperties(ignoreUnknown = true) +public record ChatOrigin( + @Nullable Long agentId, + @Nullable String conversationId, + @Nullable String requesterId, + @Nullable Long workspaceId, + @Nullable String workspaceBasePath, + @Nullable Long channelId, + @Nullable ChannelTarget channelTarget +) { + + /** Key used when this origin is wrapped into a Spring AI {@link ToolContext}. */ + public static final String CTX_KEY = "mateclaw.chatOrigin"; + + /** Sentinel used by AgentService default overloads where no origin is supplied. */ + public static final ChatOrigin EMPTY = + new ChatOrigin(null, null, "", null, null, null, null); + + // ---------------- Factories per entry point ---------------- + + public static ChatOrigin web(@Nullable String conversationId, + @Nullable String requesterId, + @Nullable Long workspaceId, + @Nullable String workspaceBasePath) { + return new ChatOrigin(null, conversationId, + requesterId != null ? requesterId : "", + workspaceId, workspaceBasePath, null, null); + } + + public static ChatOrigin cron(@Nullable String conversationId, + @Nullable Long workspaceId, + @Nullable String workspaceBasePath, + @Nullable Long channelId, + @Nullable ChannelTarget target) { + return new ChatOrigin(null, conversationId, "system", + workspaceId, workspaceBasePath, channelId, target); + } + + // ---------------- Wither-style updates ---------------- + + public ChatOrigin withAgent(@Nullable Long newAgentId) { + return new ChatOrigin(newAgentId, conversationId, requesterId, + workspaceId, workspaceBasePath, channelId, channelTarget); + } + + public ChatOrigin withWorkspace(@Nullable Long newWorkspaceId, + @Nullable String newWorkspaceBasePath) { + return new ChatOrigin(agentId, conversationId, requesterId, + newWorkspaceId, newWorkspaceBasePath, channelId, channelTarget); + } + + public ChatOrigin withConversationId(@Nullable String newConversationId) { + return new ChatOrigin(agentId, newConversationId, requesterId, + workspaceId, workspaceBasePath, channelId, channelTarget); + } + + // ---------------- Spring AI ToolContext interop ---------------- + + /** Wrap this origin into a Spring AI {@link ToolContext} the runtime can pass to tools. */ + public ToolContext toToolContext() { + return new ToolContext(Map.of(CTX_KEY, this)); + } + + /** + * Read a {@link ChatOrigin} stored under {@link #CTX_KEY} in the given + * {@link ToolContext}. Returns {@link #EMPTY} when {@code ctx} is null, has + * no entry, or the value is not a ChatOrigin (defensive — keeps single-tool + * callers safe even if wiring is partial). + */ + public static ChatOrigin from(@Nullable ToolContext ctx) { + if (ctx == null) return EMPTY; + Object v = ctx.getContext().get(CTX_KEY); + return v instanceof ChatOrigin co ? co : EMPTY; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/context/ChatOriginHolder.java b/mateclaw-server/src/main/java/vip/mate/agent/context/ChatOriginHolder.java new file mode 100644 index 00000000..86d3db8f --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/context/ChatOriginHolder.java @@ -0,0 +1,40 @@ +package vip.mate.agent.context; + +/** + * Request-scoped {@link ChatOrigin} bridge between {@code AgentService}'s + * public entry points and the StateGraph's {@code buildInitialState}. + * + *

RFC-063r §2.5 carries the origin end-to-end via Spring AI {@code ToolContext} + * once it lands in graph state. This holder is the small bridge that gets the + * origin from the AgentService method invocation into the graph's initial + * state map — the holder lifecycle is bounded by the AgentService method + * call (set on entry, cleared in {@code finally}). Once written into the + * graph state under {@link vip.mate.agent.graph.state.MateClawStateKeys#CHAT_ORIGIN}, + * the rest of the runtime reads via the typed accessor — no further ThreadLocal + * access. Mirrors {@link vip.mate.agent.ThinkingLevelHolder}. + */ +public final class ChatOriginHolder { + + private static final ThreadLocal HOLDER = new ThreadLocal<>(); + + private ChatOriginHolder() { + } + + /** Set the origin for the current AgentService invocation. */ + public static void set(ChatOrigin origin) { + HOLDER.set(origin); + } + + /** + * @return the origin set for the current invocation, or {@link ChatOrigin#EMPTY} + * when no entry path has supplied one (legacy callers). + */ + public static ChatOrigin get() { + ChatOrigin v = HOLDER.get(); + return v != null ? v : ChatOrigin.EMPTY; + } + + public static void clear() { + HOLDER.remove(); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java index ebc4b2e4..64c65d4c 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java @@ -410,6 +410,19 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC inputs.put(RUNTIME_MODEL_NAME, modelName != null ? modelName : ""); inputs.put(RUNTIME_PROVIDER_ID, runtimeProviderId != null ? runtimeProviderId : ""); inputs.put(TRACE_ID, UUID.randomUUID().toString().substring(0, 8)); + + // RFC-063r §2.5: enrich the originating ChatOrigin with this agent's id + // and workspace, then write it into graph state so ActionNode + + // StepExecutionNode can forward it to ToolExecutionExecutor → ToolContext. + vip.mate.agent.context.ChatOrigin origin = vip.mate.agent.context.ChatOriginHolder.get(); + Long parsedAgentIdForOrigin = null; + try { parsedAgentIdForOrigin = agentId != null ? Long.valueOf(agentId) : null; } catch (Exception ignored) {} + if (parsedAgentIdForOrigin != null) { + origin = origin.withAgent(parsedAgentIdForOrigin); + } + origin = origin.withConversationId(conversationId) + .withWorkspace(origin.workspaceId(), workspaceBasePath); + inputs.put(CHAT_ORIGIN, origin); return inputs; } 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 869bd574..afca2a30 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 @@ -4,10 +4,12 @@ import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.ai.chat.messages.AssistantMessage; import org.springframework.ai.chat.messages.ToolResponseMessage; +import org.springframework.ai.chat.model.ToolContext; import org.springframework.ai.tool.ToolCallback; import vip.mate.tool.builtin.ToolExecutionContext; import vip.mate.agent.AgentToolSet; import vip.mate.agent.GraphEventPublisher; +import vip.mate.agent.context.ChatOrigin; import vip.mate.agent.graph.state.DirectToolOutput; import vip.mate.approval.ApprovalWorkflowService; import vip.mate.channel.web.ChatStreamTracker; @@ -202,6 +204,8 @@ public class ToolExecutionExecutor { private volatile String currentRequesterId; /** 当前工作区活动目录（为空不限制），传递给 ToolExecutionContext */ private volatile String currentWorkspaceBasePath; + /** RFC-063r §2.5: 当前执行的 ChatOrigin，构建 ToolContext 时透传给工具 */ + private volatile ChatOrigin currentChatOrigin = ChatOrigin.EMPTY; public ToolExecutionResult execute(List toolCalls, String conversationId, String agentId, @@ -213,8 +217,30 @@ public class ToolExecutionExecutor { String conversationId, String agentId, boolean isReplay, String requesterId, String workspaceBasePath) { + return execute(toolCalls, conversationId, agentId, isReplay, requesterId, + workspaceBasePath, ChatOrigin.EMPTY); + } + + /** + * RFC-063r §2.5: preferred overload — accepts a {@link ChatOrigin} that the + * top-level agent has enriched with agentId/workspace/channel context. + * Builds a Spring AI {@link ToolContext} per tool invocation so + * {@code @Tool} methods can read the origin via + * {@code ChatOrigin.from(toolContext)}. + * + *

During the PR-1 transition the legacy {@link ToolExecutionContext} + * ThreadLocal is also populated, so existing tools that read from it keep + * working unchanged. After all 8 callsites migrate, the ThreadLocal can be + * removed. + */ + public ToolExecutionResult execute(List toolCalls, + String conversationId, String agentId, + boolean isReplay, String requesterId, + String workspaceBasePath, + ChatOrigin origin) { this.currentRequesterId = requesterId; this.currentWorkspaceBasePath = workspaceBasePath; + this.currentChatOrigin = origin != null ? origin : ChatOrigin.EMPTY; List allResponses = new ArrayList<>(); List events = Collections.synchronizedList(new ArrayList<>()); // RFC-052: accumulate full-text outputs from returnDirect tools so the @@ -309,7 +335,7 @@ public class ToolExecutionExecutor { // 4. 分类: concurrencySafe boolean safe = isConcurrencySafe(toolName); preparedCalls.add(new PreparedToolCall(toolCall, callback, arguments, safe, allResponses.size(), - conversationId, currentRequesterId, currentWorkspaceBasePath)); + conversationId, currentRequesterId, currentWorkspaceBasePath, currentChatOrigin)); // 占位，Phase 2 填充 allResponses.add(null); } @@ -382,7 +408,13 @@ public class ToolExecutionExecutor { try { log.info("[ToolExecutor] Executing pre-approved tool: {}", toolName); - String result = callback.call(callArguments); + // RFC-063r §2.5: forward ToolContext so the pre-approved tool can + // still observe the originating ChatOrigin (channel/workspace). + ChatOrigin replayOrigin = currentChatOrigin != null ? currentChatOrigin : ChatOrigin.EMPTY; + replayOrigin = replayOrigin + .withConversationId(conversationId) + .withWorkspace(replayOrigin.workspaceId(), workspaceBasePath); + String result = callback.call(callArguments, replayOrigin.toToolContext()); int rawLen = result != null ? result.length() : 0; // RFC-052: pre-approved tool may itself be returnDirect — in that @@ -565,11 +597,19 @@ public class ToolExecutionExecutor { toolName, pc.arguments != null && pc.arguments.length() > 200 ? pc.arguments.substring(0, 200) + "..." : pc.arguments); - // 注入工具执行上下文（供 VideoGenerateTool 等获取 conversationId / username / workspaceBasePath） + // RFC-063r §2.5 / PR-1 transition window: populate BOTH the explicit + // Spring AI ToolContext (preferred — read via ChatOrigin.from(ctx)) + // AND the legacy ToolExecutionContext ThreadLocal so tools that have + // not yet migrated to ToolContext keep working unchanged. ToolExecutionContext.set(pc.conversationId, pc.requesterId, pc.workspaceBasePath); String result; try { - result = pc.callback.call(pc.arguments); + ChatOrigin runtimeOrigin = pc.origin != null ? pc.origin : ChatOrigin.EMPTY; + runtimeOrigin = runtimeOrigin + .withConversationId(pc.conversationId) + .withWorkspace(runtimeOrigin.workspaceId(), pc.workspaceBasePath); + ToolContext toolContext = runtimeOrigin.toToolContext(); + result = pc.callback.call(pc.arguments, toolContext); } finally { ToolExecutionContext.clear(); } @@ -781,7 +821,8 @@ public class ToolExecutionExecutor { int resultIndex, String conversationId, String requesterId, - String workspaceBasePath + String workspaceBasePath, + ChatOrigin origin ) {} private record ApprovalBarrier(String pendingId, String toolName) {} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ActionNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ActionNode.java index 18a59c61..8904ee6d 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ActionNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/ActionNode.java @@ -65,9 +65,13 @@ public class ActionNode implements NodeAction { // 获取工作区活动目录 String workspaceBasePath = state.value(MateClawStateKeys.WORKSPACE_BASE_PATH, ""); + // RFC-063r §2.5: read the originating ChatOrigin from graph state and + // forward it into the executor — tools see it via Spring AI ToolContext. + vip.mate.agent.context.ChatOrigin origin = accessor.chatOrigin(); + // 委托 ToolExecutionExecutor 执行（两阶段：顺序 Guard + 分段并发执行） ToolExecutionExecutor.ToolExecutionResult result = executor.execute( - toolCalls, conversationId, agentId, isReplay, requesterId, workspaceBasePath); + toolCalls, conversationId, agentId, isReplay, requesterId, workspaceBasePath, origin); ToolResponseMessage toolResponseMessage = ToolResponseMessage.builder() .responses(result.responses()) diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/StateGraphPlanExecuteAgent.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/StateGraphPlanExecuteAgent.java index 799f53a1..64a25303 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/StateGraphPlanExecuteAgent.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/StateGraphPlanExecuteAgent.java @@ -285,6 +285,19 @@ public class StateGraphPlanExecuteAgent extends BaseAgent implements StructuredS inputs.put(MateClawStateKeys.RUNTIME_MODEL_NAME, modelName != null ? modelName : ""); inputs.put(MateClawStateKeys.RUNTIME_PROVIDER_ID, runtimeProviderId != null ? runtimeProviderId : ""); inputs.put(MateClawStateKeys.TRACE_ID, UUID.randomUUID().toString().substring(0, 8)); + + // RFC-063r §2.5: same as ReAct path — enrich and store the ChatOrigin + // so StepExecutionNode (and any sub-graphs spawned via DelegateAgentTool) + // can read it back from state. + vip.mate.agent.context.ChatOrigin origin = vip.mate.agent.context.ChatOriginHolder.get(); + Long parsedAgentIdForOrigin = null; + try { parsedAgentIdForOrigin = agentId != null ? Long.valueOf(agentId) : null; } catch (Exception ignored) {} + if (parsedAgentIdForOrigin != null) { + origin = origin.withAgent(parsedAgentIdForOrigin); + } + origin = origin.withConversationId(conversationId) + .withWorkspace(origin.workspaceId(), workspaceBasePath); + inputs.put(MateClawStateKeys.CHAT_ORIGIN, origin); return inputs; } 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 e62ec41b..fa055a95 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 @@ -86,6 +86,12 @@ public class StepExecutionNode implements NodeAction { String conversationId = state.value(MateClawStateKeys.CONVERSATION_ID, ""); String agentId = state.value(MateClawStateKeys.AGENT_ID, ""); String workspaceBasePath = state.value(MateClawStateKeys.WORKSPACE_BASE_PATH, ""); + // RFC-063r §2.5: read parent ChatOrigin from graph state so tools in + // this step (and any DelegateAgentTool sub-graphs) inherit channel / + // workspace / requester context. + vip.mate.agent.context.ChatOrigin chatOrigin = + state.value(MateClawStateKeys.CHAT_ORIGIN) + .orElse(vip.mate.agent.context.ChatOrigin.EMPTY); if (stepIndex >= steps.size()) { log.warn("[StepExecution] stepIndex {} >= steps.size() {}, skipping", stepIndex, steps.size()); @@ -196,7 +202,7 @@ public class StepExecutionNode implements NodeAction { } else { // 非预批准工具走正常执行器 ToolExecutionExecutor.ToolExecutionResult execResult = executor.execute( - List.of(toolCall), conversationId, agentId, false, "", workspaceBasePath); + List.of(toolCall), conversationId, agentId, false, "", workspaceBasePath, chatOrigin); toolResponses.addAll(execResult.responses()); events.addAll(execResult.events()); if (execResult.hasDirectOutputs()) { @@ -212,7 +218,7 @@ public class StepExecutionNode implements NodeAction { } else { // 正常路径：委托 ToolExecutionExecutor（支持并发执行 + 审批 barrier） ToolExecutionExecutor.ToolExecutionResult execResult = executor.execute( - allToolCalls, conversationId, agentId, false, "", workspaceBasePath); + allToolCalls, conversationId, agentId, false, "", workspaceBasePath, chatOrigin); toolResponses.addAll(execResult.responses()); events.addAll(execResult.events()); if (execResult.hasDirectOutputs()) { diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateAccessor.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateAccessor.java index 92b1bce3..b7e0710e 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateAccessor.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateAccessor.java @@ -3,6 +3,7 @@ package vip.mate.agent.graph.state; import com.alibaba.cloud.ai.graph.OverAllState; import org.springframework.ai.chat.messages.Message; import vip.mate.agent.GraphEventPublisher; +import vip.mate.agent.context.ChatOrigin; import vip.mate.agent.graph.NodeStreamingChatHelper; import java.util.*; @@ -212,6 +213,17 @@ public final class MateClawStateAccessor { return state.value(FORCED_TOOL_CALL, ""); } + // ===== RFC-063r: ChatOrigin ===== + + /** + * RFC-063r §2.5: the {@link ChatOrigin} written into graph state by the + * top-level agent. Returns {@link ChatOrigin#EMPTY} when the entry path + * did not supply one (e.g., legacy callers using the bridge overloads). + */ + public ChatOrigin chatOrigin() { + return state.value(CHAT_ORIGIN).orElse(ChatOrigin.EMPTY); + } + // ===== Token Usage ===== public int promptTokens() { @@ -398,6 +410,11 @@ public final class MateClawStateAccessor { return put(FORCED_TOOL_CALL, json); } + // ---- RFC-063r: ChatOrigin ---- + public OutputBuilder chatOrigin(ChatOrigin origin) { + return put(CHAT_ORIGIN, origin); + } + // ---- Token Usage ---- /** 将本次 LLM 调用的 usage 累加到 state 已有值上 */ diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateKeys.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateKeys.java index 46e084f2..58f76c7d 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateKeys.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/state/MateClawStateKeys.java @@ -155,4 +155,16 @@ public final class MateClawStateKeys { * tool batch, used by FinalAnswerNode to assemble the final answer. */ public static final String DIRECT_TOOL_OUTPUTS = "direct_tool_outputs"; + + // ===== RFC-063r: ChatOrigin propagation through the StateGraph ===== + + /** + * RFC-063r §2.5: top-level agent writes the {@code ChatOrigin} value object + * into graph state once at {@code buildInitialState}; nodes (especially + * {@code StepExecutionNode} in the Plan-Execute sub-graph) read it + * read-only when invoking {@link vip.mate.agent.graph.executor.ToolExecutionExecutor} + * so child graphs and delegated agents inherit the originating channel / + * workspace context. + */ + public static final String CHAT_ORIGIN = "chat_origin"; } diff --git a/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java b/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java index 77987cba..8a664032 100644 --- a/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java +++ b/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java @@ -15,6 +15,8 @@ import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; +import vip.mate.agent.context.ChatOrigin; +import vip.mate.agent.context.ChatOriginHolder; import vip.mate.approval.model.ToolApprovalEntity; import vip.mate.approval.repository.ToolApprovalMapper; import vip.mate.tool.guard.model.GuardEvaluation; @@ -137,6 +139,7 @@ public class ApprovalWorkflowService implements ApplicationRunner { snapshot.setFindingsJson(entity.getFindingsJson()); snapshot.setMaxSeverity(entity.getMaxSeverity()); snapshot.setSummary(entity.getSummary()); + snapshot.setChatOrigin(entity.getChatOrigin()); approvalService.registerRecovered(snapshot); recovered++; @@ -200,6 +203,14 @@ public class ApprovalWorkflowService implements ApplicationRunner { conversationId, userId, toolName, toolArguments, reason, toolCallPayload, siblingToolCalls, agentId); + // RFC-063r §2.12: capture the originating ChatOrigin from the holder. + // The holder was set by AgentService.{chat,chatStream,...} for the + // duration of the agent invocation that produced this approval — so + // it is non-null for IM / web triggered tool calls. Snapshot is + // serialized once here and persisted on the DB row so cross-restart + // replays keep the channel binding. + String chatOriginJson = serializeChatOrigin(ChatOriginHolder.get()); + // 2. 增强内存记录 approvalService.getPending(pendingId).ifPresent(pending -> { if (evaluation != null) { @@ -207,11 +218,12 @@ public class ApprovalWorkflowService implements ApplicationRunner { pending.setMaxSeverity(evaluation.maxSeverity() != null ? evaluation.maxSeverity().name() : null); pending.setSummary(evaluation.summary()); } + pending.setChatOrigin(chatOriginJson); }); // 3. DB 层 persistToDb(pendingId, conversationId, userId, toolName, toolArguments, - toolCallPayload, siblingToolCalls, agentId, evaluation); + toolCallPayload, siblingToolCalls, agentId, evaluation, chatOriginJson); return pendingId; } @@ -517,7 +529,7 @@ public class ApprovalWorkflowService implements ApplicationRunner { private void persistToDb(String pendingId, String conversationId, String userId, String toolName, String toolArguments, String toolCallPayload, String siblingToolCalls, String agentId, - GuardEvaluation evaluation) { + GuardEvaluation evaluation, String chatOriginJson) { try { ToolApprovalEntity entity = new ToolApprovalEntity(); entity.setPendingId(pendingId); @@ -531,6 +543,9 @@ public class ApprovalWorkflowService implements ApplicationRunner { entity.setStatus("PENDING"); entity.setCreatedAt(LocalDateTime.now()); entity.setExpireAt(LocalDateTime.now().plusMinutes(30)); + // RFC-063r §2.12: persist Memento snapshot. Null when the entry + // path didn't supply an origin — replay falls back to ChatOrigin.EMPTY. + entity.setChatOrigin(chatOriginJson); if (evaluation != null) { entity.setFindingsJson(serializeFindings(evaluation.findings())); @@ -548,6 +563,44 @@ public class ApprovalWorkflowService implements ApplicationRunner { } } + /** + * RFC-063r §2.12: serialize a {@link ChatOrigin} for persistence on + * {@code mate_tool_approval.chat_origin}. Returns null for + * {@code ChatOrigin.EMPTY} so legacy approvals that never captured an + * origin do not store a meaningless empty record. + */ + private String serializeChatOrigin(ChatOrigin origin) { + if (origin == null || origin == ChatOrigin.EMPTY) return null; + if (origin.agentId() == null && origin.channelId() == null + && origin.conversationId() == null && origin.workspaceId() == null) { + return null; + } + try { + return objectMapper.writeValueAsString(origin); + } catch (JsonProcessingException e) { + log.warn("[ApprovalWorkflow] Failed to serialize ChatOrigin: {}", e.getMessage()); + return null; + } + } + + /** + * RFC-063r §2.12: deserialize a persisted Memento back into a + * {@link ChatOrigin}. Returns {@link ChatOrigin#EMPTY} when the column + * is null or the payload is corrupt — the caller treats that as + * "no channel binding" and replay proceeds with a web-style flow. + */ + public ChatOrigin restoreChatOrigin(String json) { + if (json == null || json.isBlank()) return ChatOrigin.EMPTY; + try { + ChatOrigin restored = objectMapper.readValue(json, ChatOrigin.class); + return restored != null ? restored : ChatOrigin.EMPTY; + } catch (Exception e) { + log.warn("[ApprovalWorkflow] Failed to restore ChatOrigin: {} (payload-len={})", + e.getMessage(), json.length()); + return ChatOrigin.EMPTY; + } + } + private void updateDbStatus(String pendingId, String status, String resolvedBy) { try { LambdaUpdateWrapper wrapper = new LambdaUpdateWrapper() diff --git a/mateclaw-server/src/main/java/vip/mate/approval/PendingApproval.java b/mateclaw-server/src/main/java/vip/mate/approval/PendingApproval.java index 87714e00..481fe1a8 100644 --- a/mateclaw-server/src/main/java/vip/mate/approval/PendingApproval.java +++ b/mateclaw-server/src/main/java/vip/mate/approval/PendingApproval.java @@ -59,6 +59,16 @@ public class PendingApproval { /** 风险摘要 */ private String summary; + /** + * RFC-063r §2.12: serialized {@code ChatOrigin} snapshot captured when + * this approval was created. Lets cross-process / cross-restart replays + * (the user approves hours later from a different node) restore the + * original channel binding so the replayed tool call still delivers + * back to the correct channel. Persisted into + * {@code mate_tool_approval.chat_origin}. + */ + private String chatOrigin; + public PendingApproval(String pendingId, String conversationId, String userId, String toolName, String toolArguments, String reason) { this.pendingId = pendingId; @@ -110,6 +120,7 @@ public class PendingApproval { public String getFindingsJson() { return findingsJson; } public String getMaxSeverity() { return maxSeverity; } public String getSummary() { return summary; } + public String getChatOrigin() { return chatOrigin; } // === Setters === @@ -125,4 +136,5 @@ public class PendingApproval { public void setFindingsJson(String findingsJson) { this.findingsJson = findingsJson; } public void setMaxSeverity(String maxSeverity) { this.maxSeverity = maxSeverity; } public void setSummary(String summary) { this.summary = summary; } + public void setChatOrigin(String chatOrigin) { this.chatOrigin = chatOrigin; } } diff --git a/mateclaw-server/src/main/java/vip/mate/approval/model/ToolApprovalEntity.java b/mateclaw-server/src/main/java/vip/mate/approval/model/ToolApprovalEntity.java index 90f96ac2..25df0eca 100644 --- a/mateclaw-server/src/main/java/vip/mate/approval/model/ToolApprovalEntity.java +++ b/mateclaw-server/src/main/java/vip/mate/approval/model/ToolApprovalEntity.java @@ -36,6 +36,16 @@ public class ToolApprovalEntity { private LocalDateTime resolvedAt; private LocalDateTime expireAt; + /** + * RFC-063r §2.12: serialized {@link vip.mate.agent.context.ChatOrigin} + * snapshot captured when this approval was created. The Memento lets + * ChannelMessageRouter.replayApprovedToolCall (and the web ApprovalController + * replay path) restore the originating channel/workspace context after + * a process restart, so a tool approved hours later still binds back to + * the original channel. + */ + private String chatOrigin; + @TableField(fill = FieldFill.INSERT) private LocalDateTime createTime; diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java index 1a2f3c43..3952b1a8 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java @@ -113,6 +113,19 @@ public interface ChannelAdapter { throw new UnsupportedOperationException(getChannelType() + " does not support proactive send"); } + /** + * RFC-063r §2.10: extended overload that accepts a + * {@link DeliveryOptions} Parameter Object carrying optional hints + * (thread id, multi-bot account id, future ext fields). + * + *

Default implementation delegates to {@link #proactiveSend(String, String)}, + * dropping hints — concrete adapters (Slack, Telegram) override this + * variant to read {@code threadId} and route into the threading API. + */ + default void proactiveSend(String targetId, String content, DeliveryOptions options) { + proactiveSend(targetId, content); + } + /** * 当前渠道是否支持主动推送 * diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelChatOriginFactory.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelChatOriginFactory.java new file mode 100644 index 00000000..85b1d098 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelChatOriginFactory.java @@ -0,0 +1,54 @@ +package vip.mate.channel; + +import org.springframework.stereotype.Component; +import vip.mate.agent.context.ChannelTarget; +import vip.mate.agent.context.ChatOrigin; +import vip.mate.channel.model.ChannelEntity; + +/** + * RFC-063r §2.2: factory that translates an inbound channel message into a + * {@link ChatOrigin}. Lives in {@code vip.mate.channel} (not in + * {@code vip.mate.agent.context}) so that the dependency direction stays + * {@code channel → agent} and never the reverse. + */ +@Component +public class ChannelChatOriginFactory { + + /** + * Build a {@link ChatOrigin} for a channel-originated message. + * + * @param channel channel entity (non-null) — provides id + workspaceId + * @param message inbound message (non-null) — provides senderId + reply target + * @param conversationId resolved conversation id (channel-scoped) + * @param workspaceBasePath workspace activity directory; null = unrestricted + */ + public ChatOrigin from(ChannelEntity channel, + ChannelMessage message, + String conversationId, + String workspaceBasePath) { + ChannelTarget target = new ChannelTarget( + resolveTargetId(message), + /* threadId */ null, // adapters fill via ChannelMessage extension fields when available + /* accountId */ null); + return new ChatOrigin( + /* agentId */ null, + /* conversationId */ conversationId, + /* requesterId */ message.getSenderId(), + /* workspaceId */ channel.getWorkspaceId(), + /* workspaceBasePath */ workspaceBasePath, + /* channelId */ channel.getId(), + /* channelTarget */ target); + } + + /** + * Resolve the IM target id used for proactive sends — prefer chatId + * (group/room) over senderId so that cron deliveries land in the same + * conversation the user originally messaged from. + */ + private String resolveTargetId(ChannelMessage message) { + if (message.getReplyToken() != null && !message.getReplyToken().isBlank()) { + return message.getReplyToken(); + } + return message.getChatId() != null ? message.getChatId() : message.getSenderId(); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java index 15d06837..b7805453 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java @@ -330,13 +330,23 @@ public class ChannelManager { * @throws IllegalStateException 渠道未启动或不支持主动推送 */ public void sendToChannel(Long channelId, String targetId, String content) { + sendToChannel(channelId, targetId, content, DeliveryOptions.DEFAULTS); + } + + /** + * RFC-063r §2.10: preferred overload — accepts a {@link DeliveryOptions} + * Parameter Object so cron delivery (and future callers) can pass + * Slack {@code thread_ts}, Telegram {@code message_thread_id}, multi-bot + * {@code accountId}, etc. without growing a 5-arg signature. + */ + public void sendToChannel(Long channelId, String targetId, String content, DeliveryOptions options) { ChannelAdapter adapter = getAdapter(channelId) .orElseThrow(() -> new IllegalStateException("Channel not active: " + channelId)); if (!adapter.supportsProactiveSend()) { throw new UnsupportedOperationException( "Channel " + adapter.getDisplayName() + " (" + adapter.getChannelType() + ") does not support proactive send"); } - adapter.proactiveSend(targetId, content); + adapter.proactiveSend(targetId, content, options != null ? options : DeliveryOptions.DEFAULTS); log.info("Proactive message sent via channel {} to {}: {}chars", adapter.getDisplayName(), targetId, content.length()); } diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java index 6b461f05..508ed5a6 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java @@ -4,6 +4,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import reactor.core.publisher.Flux; import vip.mate.agent.AgentService; +import vip.mate.agent.context.ChatOrigin; import vip.mate.approval.ApprovalWorkflowService; import vip.mate.approval.ResolveOutcome; import vip.mate.approval.PendingApproval; @@ -54,6 +55,7 @@ public class ChannelMessageRouter { private final TtsService ttsService; private final ObjectMapper objectMapper; private final ChatStreamTracker streamTracker; + private final ChannelChatOriginFactory chatOriginFactory; /** 队列条目：封装消息及其路由上下文 */ private record QueueEntry(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity) {} @@ -98,7 +100,8 @@ public class ChannelMessageRouter { ConversationCompletionPublisher completionPublisher, TtsService ttsService, ObjectMapper objectMapper, - ChatStreamTracker streamTracker) { + ChatStreamTracker streamTracker, + ChannelChatOriginFactory chatOriginFactory) { this.agentService = agentService; this.conversationService = conversationService; this.channelService = channelService; @@ -109,6 +112,7 @@ public class ChannelMessageRouter { this.ttsService = ttsService; this.objectMapper = objectMapper; this.streamTracker = streamTracker; + this.chatOriginFactory = chatOriginFactory; } // ==================== 防抖辅助类 ==================== @@ -440,11 +444,17 @@ public class ChannelMessageRouter { Long savedAssistantId = null; try { // 流式路径：渠道实现了 StreamingChannelAdapter 则委托渠道渲染流式事件 + // RFC-063r §2.5: build the ChatOrigin once per channel-message + // so cron jobs created during this conversation inherit the + // channel binding (Issue #25 root path). + ChatOrigin chatOrigin = chatOriginFactory.from( + channelEntity, message, conversationId, /* workspaceBasePath */ null); + if (adapter instanceof StreamingChannelAdapter streamingAdapter) { - savedAssistantId = processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity); + savedAssistantId = processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity, chatOrigin); } else { // 同步路径：直接获取完整回复 - String reply = agentService.chat(agentId, promptText, conversationId); + String reply = agentService.chat(agentId, promptText, conversationId, chatOrigin); // 检查 chat 过程中是否产生了审批 pending PendingApproval newPending = approvalService.findPendingByConversation(conversationId); @@ -521,14 +531,14 @@ public class ChannelMessageRouter { */ private Long processWithStreaming(ChannelMessage message, StreamingChannelAdapter streamingAdapter, String conversationId, Long agentId, String promptText, - ChannelEntity channelEntity) { + ChannelEntity channelEntity, ChatOrigin chatOrigin) { String channelType = streamingAdapter.getChannelType(); log.info("[{}] Streaming processing started: conversationId={}", channelType, conversationId); try { - // Step 1: 产生事件流 + // Step 1: 产生事件流（RFC-063r §2.5: forward ChatOrigin so tools see channelId） Flux stream = agentService.chatStructuredStream( - agentId, promptText, conversationId, message.getSenderId()); + agentId, promptText, conversationId, message.getSenderId(), chatOrigin); // Step 2: 委托渠道渲染（渠道内部消费 Flux 并处理 UI 更新） String finalContent = streamingAdapter.processStream(stream, message, conversationId); @@ -591,8 +601,17 @@ public class ChannelMessageRouter { String replayPrompt = "继续执行已批准的工具调用。"; try { + // RFC-063r §2.12: prefer the persisted Memento (covers + // cross-restart approval where the channel session changed) and + // only fall back to rebuilding from the current inbound message + // when no snapshot was captured (legacy rows from before this PR). + ChatOrigin replayOrigin = approvalService.restoreChatOrigin(consumed.getChatOrigin()); + if (replayOrigin == ChatOrigin.EMPTY) { + replayOrigin = chatOriginFactory.from( + channelEntity, triggerMessage, conversationId, /* workspaceBasePath */ null); + } String reply = agentService.chatWithReplay( - agentId, replayPrompt, conversationId, consumed.getToolCallPayload()); + agentId, replayPrompt, conversationId, consumed.getToolCallPayload(), replayOrigin); // 保存 replay 结果（这是正常结果，入库） conversationService.saveMessage(conversationId, "assistant", reply); @@ -644,7 +663,11 @@ public class ChannelMessageRouter { conversationService.saveMessage(conversationId, "user", message.getContent(), parts); String promptText = buildPromptFromParts(message.getContent(), parts, message.getInputMode()); - return agentService.chatStream(agentId, promptText, conversationId); + // RFC-063r §2.5: forward ChatOrigin so tools created during this + // streaming conversation inherit channel binding. + ChatOrigin origin = chatOriginFactory.from( + channelEntity, message, conversationId, /* workspaceBasePath */ null); + return agentService.chatStream(agentId, promptText, conversationId, origin); } // ==================== 优雅关闭 ==================== diff --git a/mateclaw-server/src/main/java/vip/mate/channel/DeliveryOptions.java b/mateclaw-server/src/main/java/vip/mate/channel/DeliveryOptions.java new file mode 100644 index 00000000..08f57e25 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/DeliveryOptions.java @@ -0,0 +1,29 @@ +package vip.mate.channel; + +import org.springframework.lang.Nullable; + +import java.util.Map; + +/** + * RFC-063r §2.10: Parameter Object that bundles optional delivery hints + * (Slack {@code thread_ts}, Telegram {@code message_thread_id}, multi-bot + * {@code accountId}, etc.) so {@link ChannelManager#sendToChannel} doesn't + * grow a 5-arg overload. + * + *

{@link #DEFAULTS} is the canonical "no hints" instance — adapters that + * don't override the 4-arg {@code proactiveSend} keep their pre-RFC behavior. + */ +public record DeliveryOptions( + @Nullable String threadId, + @Nullable String accountId, + Map ext +) { + + public static final DeliveryOptions DEFAULTS = new DeliveryOptions(null, null, Map.of()); + + public DeliveryOptions { + // Defensive: never expose a null map — the receiver should be able to + // call .get(...) without a null check. + if (ext == null) ext = Map.of(); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java index 255e1950..eb1b2737 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java @@ -238,6 +238,38 @@ public class SlackChannelAdapter extends AbstractChannelAdapter { sendMessage(targetId, content); } + /** + * RFC-063r §2.10: thread-aware proactive send. Reads + * {@link vip.mate.channel.DeliveryOptions#threadId()} (the Slack + * {@code thread_ts}) and posts into that thread when supplied; falls + * back to the legacy in-channel post when null. + */ + @Override + public void proactiveSend(String targetId, String content, + vip.mate.channel.DeliveryOptions options) { + if (options == null || options.threadId() == null || options.threadId().isBlank()) { + sendMessage(targetId, content); + return; + } + String botToken = getConfigString("bot_token"); + if (botToken == null || content == null || content.isBlank()) { + return; + } + try { + String slackContent = convertToSlackMarkdown(content); + final String threadTs = options.threadId(); + ChatPostMessageResponse response = slack.methods(botToken).chatPostMessage(req -> + req.channel(targetId).text(slackContent).threadTs(threadTs)); + if (!response.isOk()) { + log.warn("[slack] Failed to send threaded proactive message (thread_ts={}): {}", + threadTs, response.getError()); + } + } catch (IOException | SlackApiException e) { + log.error("[slack] Error sending threaded proactive message to {}: {}", + targetId, e.getMessage()); + } + } + /** * Webhook 回调处理（备用模式，Socket Mode 优先） */ diff --git a/mateclaw-server/src/main/java/vip/mate/channel/telegram/TelegramChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/telegram/TelegramChannelAdapter.java index 50b1c90b..97cf420c 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/telegram/TelegramChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/telegram/TelegramChannelAdapter.java @@ -715,6 +715,80 @@ public class TelegramChannelAdapter extends AbstractChannelAdapter { sendMessage(targetId, content); } + /** + * RFC-063r §2.10: forum-thread-aware proactive send. When + * {@link vip.mate.channel.DeliveryOptions#threadId()} is set (Telegram + * forum {@code message_thread_id}), include it in the {@code sendMessage} + * call so the cron result lands in the correct thread of a forum group. + * Falls back to the legacy chat-level send when null. + */ + @Override + public void proactiveSend(String targetId, String content, + vip.mate.channel.DeliveryOptions options) { + if (options == null || options.threadId() == null || options.threadId().isBlank()) { + sendMessage(targetId, content); + return; + } + if (httpClient == null || botToken == null) { + log.warn("[telegram] Channel not started, cannot send proactive message"); + return; + } + Integer threadId; + try { + threadId = Integer.valueOf(options.threadId()); + } catch (NumberFormatException nfe) { + log.warn("[telegram] Invalid message_thread_id '{}'; sending to main chat", options.threadId()); + sendMessage(targetId, content); + return; + } + try { + // Try Markdown first, fall back to plain on parse error — same + // contract as sendMessage but with message_thread_id added. + if (!sendThreadedText(targetId, threadId, content, "Markdown")) { + sendThreadedText(targetId, threadId, content, null); + } + } catch (Exception e) { + log.error("[telegram] proactiveSend(threadId={}) failed: {}", threadId, e.getMessage()); + } + } + + @SuppressWarnings("unchecked") + private boolean sendThreadedText(String targetId, Integer threadId, String content, String parseMode) { + try { + Map body = new java.util.LinkedHashMap<>(); + body.put("chat_id", targetId); + body.put("message_thread_id", threadId); + body.put("text", content); + if (parseMode != null) { + body.put("parse_mode", parseMode); + } + String jsonBody = objectMapper.writeValueAsString(body); + HttpRequest request = HttpRequest.newBuilder() + .uri(URI.create(apiBaseUrl + "/sendMessage")) + .header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(jsonBody)) + .build(); + HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + if (response.statusCode() == 200) { + return true; + } + if (response.statusCode() == 400 && parseMode != null) { + try { + Map errResult = objectMapper.readValue(response.body(), Map.class); + String desc = String.valueOf(errResult.getOrDefault("description", "")); + if (desc.contains("can't parse")) { + return false; + } + } catch (Exception ignored) {} + } + log.warn("[telegram] Threaded send failed: status={}, body={}", response.statusCode(), response.body()); + return true; + } catch (Exception e) { + log.error("[telegram] Threaded send error: {}", e.getMessage(), e); + return true; + } + } + @Override public String getChannelType() { return CHANNEL_TYPE; 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 53cca423..ef53c43a 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 @@ -283,8 +283,18 @@ public class ChatController { String replayPrompt = "继续执行已批准的工具调用。"; streamTracker.incrementFlux(conversationId); + // RFC-063r §2.12: prefer the persisted Memento snapshot + // (covers cross-restart approval where the original channel + // is gone) and fall back to a fresh web-origin + // ChatOrigin when none was captured. + vip.mate.agent.context.ChatOrigin replayOrigin = + approvalService.restoreChatOrigin(finalConsumed.getChatOrigin()); + if (replayOrigin == vip.mate.agent.context.ChatOrigin.EMPTY) { + replayOrigin = vip.mate.agent.context.ChatOrigin.web( + conversationId, username, workspaceId, null); + } Disposable disposable = agentService.chatWithReplayStream( - replayAgentId, replayPrompt, conversationId, finalConsumed.getToolCallPayload(), username) + replayAgentId, replayPrompt, conversationId, finalConsumed.getToolCallPayload(), username, replayOrigin) .doOnNext(delta -> { if (approvalEmitterDone.get()) return; try { @@ -460,7 +470,12 @@ public class ChatController { )); streamTracker.incrementFlux(conversationId); - Disposable disposable = agentService.chatStructuredStream(agentId, promptText, conversationId, username, request.getThinkingLevel()) + // RFC-063r §2.5: web entry — null channelId / no ChannelTarget; + // tools that need a workspace path read it from the agent (origin + // is enriched with workspaceBasePath in StateGraph buildInitialState). + vip.mate.agent.context.ChatOrigin webOrigin = + vip.mate.agent.context.ChatOrigin.web(conversationId, username, workspaceId, null); + Disposable disposable = agentService.chatStructuredStream(agentId, promptText, conversationId, username, request.getThinkingLevel(), webOrigin) .doOnNext(delta -> { if (emitterDone.get()) return; try { @@ -1061,7 +1076,12 @@ public class ChatController { broadcastEvent(conversationId, "message_start", Map.of("role", "assistant")); streamTracker.incrementFlux(conversationId); - Disposable disposable = agentService.chatStructuredStream(agentId, queuedMessage, conversationId, requesterId) + // RFC-063r §2.5: queued messages land in the same conversation; carry + // a web-origin ChatOrigin so any cron job created during the queued + // turn keeps a consistent (null-channel) binding. + vip.mate.agent.context.ChatOrigin queuedOrigin = + vip.mate.agent.context.ChatOrigin.web(conversationId, requesterId, null, null); + Disposable disposable = agentService.chatStructuredStream(agentId, queuedMessage, conversationId, requesterId, null, queuedOrigin) .doOnNext(delta -> { if (emitterDone.get()) return; try { diff --git a/mateclaw-server/src/main/java/vip/mate/cron/CronChatOriginFactory.java b/mateclaw-server/src/main/java/vip/mate/cron/CronChatOriginFactory.java new file mode 100644 index 00000000..68593e11 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/CronChatOriginFactory.java @@ -0,0 +1,41 @@ +package vip.mate.cron; + +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Component; +import vip.mate.agent.context.ChannelTarget; +import vip.mate.agent.context.ChatOrigin; +import vip.mate.agent.model.AgentEntity; +import vip.mate.agent.repository.AgentMapper; +import vip.mate.cron.model.CronJobEntity; +import vip.mate.cron.model.DeliveryConfig; + +/** + * RFC-063r §2.2: factory that builds a {@link ChatOrigin} for a cron-triggered + * agent invocation. Lives in {@code vip.mate.cron} so the dependency arrow + * points {@code cron → agent} only — symmetric with + * {@code ChannelChatOriginFactory} in {@code vip.mate.channel}. + * + *

workspaceId is reverse-resolved from {@code agent.workspaceId} (with the + * legacy {@code 1L} fallback to keep behavior identical to + * {@code CronJobService.executeJob}). workspaceBasePath is intentionally not + * persisted — the value is derived at run time from the agent's workspace + * configuration, matching the previous {@code ToolExecutionContext.workspaceBasePath()} + * semantics. + */ +@Component +@RequiredArgsConstructor +public class CronChatOriginFactory { + + private final AgentMapper agentMapper; + + public ChatOrigin from(CronJobEntity job, String conversationId) { + AgentEntity agent = job.getAgentId() != null ? agentMapper.selectById(job.getAgentId()) : null; + Long workspaceId = agent != null && agent.getWorkspaceId() != null ? agent.getWorkspaceId() : 1L; + + DeliveryConfig dc = job.getDeliveryConfig(); + ChannelTarget target = dc != null ? dc.toChannelTarget() : null; + + return ChatOrigin.cron(conversationId, workspaceId, /* workspaceBasePath */ null, + job.getChannelId(), target); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/delivery/AbstractCronResultDelivery.java b/mateclaw-server/src/main/java/vip/mate/cron/delivery/AbstractCronResultDelivery.java new file mode 100644 index 00000000..f6fe05e3 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/AbstractCronResultDelivery.java @@ -0,0 +1,125 @@ +package vip.mate.cron.delivery; + +import cn.hutool.core.util.StrUtil; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.messages.AssistantMessage; +import vip.mate.channel.ChannelMessageRenderer; +import vip.mate.cron.model.CronJobEntity; +import vip.mate.dashboard.model.CronJobRunEntity; +import vip.mate.dashboard.repository.CronJobRunMapper; + +import java.util.List; + +/** + * RFC-063r §2.6.1: Template-Method base for {@link CronResultDelivery}. + * + *

Owns the cross-strategy invariants: + *

    + *
  • Idempotency — atomic SQL CAS on + * {@code mate_cron_job_run.delivery_status} via {@link #claimRun} so + * only one listener instance proceeds per run row even across cluster + * deployments (replaces RFC-063 v1's process-local Caffeine cache).
    • + *
  • State-machine bookkeeping — {@link #markDelivered} on success, + * {@link #markNotDelivered} on failure (truncated message via Hutool + * {@code StrUtil.maxLength} per RFC §2.6.1).
    • + *
  • Render hook — {@link #renderForChannel} delegates to the + * project's existing {@link ChannelMessageRenderer} so per-channel + * markdown / Block Kit rendering stays consistent with the inline reply + * path.
    • + *
+ * + *

Concrete subclasses implement only + * {@link #doDeliver(CronJobEntity, AssistantMessage, CronJobRunEntity)}. + */ +@Slf4j +public abstract class AbstractCronResultDelivery implements CronResultDelivery { + + /** + * Default platform message-length cap used when the bound channel type + * is unknown — matches Telegram's 4096 ceiling, the most permissive + * among the small-cap platforms in {@link ChannelMessageRenderer#PLATFORM_LIMITS}. + * RFC-063r §2.11 will refine this in PR-4 by passing per-channel limits + * through {@code DeliveryOptions}. + */ + private static final int DEFAULT_RENDER_MAX_LEN = 4096; + + private final CronJobRunMapper runMapper; + + protected AbstractCronResultDelivery(CronJobRunMapper runMapper) { + this.runMapper = runMapper; + } + + @Override + public final DeliveryOutcome deliver(CronJobEntity job, AssistantMessage result, CronJobRunEntity run) { + if (!claimRun(run)) { + return DeliveryOutcome.skipped("already-claimed-by-other-instance"); + } + try { + DeliveryOutcome outcome = doDeliver(job, result, run); + markDelivered(run, outcome); + return outcome; + } catch (Exception e) { + markNotDelivered(run, e); + throw e; + } + } + + /** Strategy's actual delivery work — invoked after CAS success. */ + protected abstract DeliveryOutcome doDeliver( + CronJobEntity job, AssistantMessage result, CronJobRunEntity run); + + /** + * RFC-063r §2.11: filter thinking + tool-call markers and join the + * platform-truncated segments into a single string. Concrete strategies + * call this before handing the result to the channel-specific send call. + * + *

Null-safe — null/empty {@link AssistantMessage} returns "" so + * adapters never NPE. + */ + protected String renderForChannel(AssistantMessage msg, Long channelId) { + if (msg == null) return ""; + String text = msg.getText() != null ? msg.getText() : ""; + if (text.isEmpty()) return ""; + try { + List segments = ChannelMessageRenderer.renderForChannel( + text, /* filterThinking */ true, /* filterToolMessages */ true, + /* messageFormat */ null, DEFAULT_RENDER_MAX_LEN); + return segments.isEmpty() ? "" : String.join("\n\n", segments); + } catch (Exception e) { + log.debug("[CronDelivery] renderForChannel failed (channelId={}); falling back to raw text: {}", + channelId, e.getMessage()); + return text; + } + } + + // ---------- SQL state-machine helpers ---------- + + /** + * Atomic SQL CAS: transition delivery_status from {@code NONE} or + * {@code PENDING} → {@code PENDING}. Returns true iff this instance won + * the race. NONE-eligibility lets fresh runs claim without a separate + * "first-time" branch; PENDING-eligibility covers the rare same-instance + * retry inside the listener (cluster paths can't normally hit this). + */ + private boolean claimRun(CronJobRunEntity run) { + return runMapper.update(null, new LambdaUpdateWrapper() + .eq(CronJobRunEntity::getId, run.getId()) + .in(CronJobRunEntity::getDeliveryStatus, "NONE", "PENDING", null) + .set(CronJobRunEntity::getDeliveryStatus, "PENDING")) == 1; + } + + private void markDelivered(CronJobRunEntity run, DeliveryOutcome o) { + runMapper.update(null, new LambdaUpdateWrapper() + .eq(CronJobRunEntity::getId, run.getId()) + .set(CronJobRunEntity::getDeliveryStatus, "DELIVERED") + .set(CronJobRunEntity::getDeliveryTarget, o.target())); + } + + private void markNotDelivered(CronJobRunEntity run, Exception e) { + runMapper.update(null, new LambdaUpdateWrapper() + .eq(CronJobRunEntity::getId, run.getId()) + .set(CronJobRunEntity::getDeliveryStatus, "NOT_DELIVERED") + .set(CronJobRunEntity::getDeliveryError, StrUtil.maxLength(e.getMessage(), 500))); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/delivery/ChannelCronResultDelivery.java b/mateclaw-server/src/main/java/vip/mate/cron/delivery/ChannelCronResultDelivery.java new file mode 100644 index 00000000..020a5edb --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/ChannelCronResultDelivery.java @@ -0,0 +1,55 @@ +package vip.mate.cron.delivery; + +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.core.annotation.Order; +import org.springframework.stereotype.Component; +import vip.mate.channel.ChannelManager; +import vip.mate.channel.DeliveryOptions; +import vip.mate.cron.model.CronJobEntity; +import vip.mate.cron.model.DeliveryConfig; +import vip.mate.dashboard.model.CronJobRunEntity; +import vip.mate.dashboard.repository.CronJobRunMapper; + +import java.util.Map; + +/** + * RFC-063r §2.6: deliver a cron job's assistant result back to its + * originating IM channel via {@link ChannelManager#sendToChannel}. + * + *

{@link #supports} returns true only when both {@code channelId} and a + * non-null {@code deliveryConfig.targetId()} are present — web-origin jobs + * (no channelId) and partial bindings fall through and the run stays in + * {@code delivery_status='NONE'}, matching the always-best-effort policy in + * RFC §2.7.3. + */ +@Component +@Order(10) +public class ChannelCronResultDelivery extends AbstractCronResultDelivery { + + private final ChannelManager channelManager; + + public ChannelCronResultDelivery(CronJobRunMapper runMapper, + ChannelManager channelManager) { + super(runMapper); + this.channelManager = channelManager; + } + + @Override + public boolean supports(CronJobEntity job) { + if (job == null || job.getChannelId() == null) return false; + DeliveryConfig dc = job.getDeliveryConfig(); + return dc != null && dc.targetId() != null && !dc.targetId().isBlank(); + } + + @Override + protected DeliveryOutcome doDeliver(CronJobEntity job, AssistantMessage result, CronJobRunEntity run) { + DeliveryConfig dc = job.getDeliveryConfig(); + String rendered = renderForChannel(result, job.getChannelId()); + // RFC-063r §2.10: forward thread / account hints via DeliveryOptions. + // Adapters that don't override the 4-arg proactiveSend default ignore + // the hints — preserves pre-RFC behavior for non-threading platforms. + DeliveryOptions options = new DeliveryOptions(dc.threadId(), dc.accountId(), Map.of()); + channelManager.sendToChannel(job.getChannelId(), dc.targetId(), rendered, options); + return DeliveryOutcome.delivered(dc.targetId()); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronDeliveryListener.java b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronDeliveryListener.java new file mode 100644 index 00000000..0e0a0ee7 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronDeliveryListener.java @@ -0,0 +1,113 @@ +package vip.mate.cron.delivery; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.event.EventListener; +import org.springframework.scheduling.annotation.Async; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.stereotype.Component; +import org.springframework.transaction.event.TransactionPhase; +import org.springframework.transaction.event.TransactionalEventListener; +import vip.mate.audit.service.AuditEventService; + +import java.util.List; +import java.util.Optional; + +/** + * RFC-063r §2.7.3: Domain-Event listener that resolves the right + * {@link CronResultDelivery} strategy and runs it asynchronously after the + * cron run's T2 transaction commits. + * + *

Why {@code AFTER_COMMIT} + {@code @Async}: + *

    + *
  • {@code AFTER_COMMIT} guarantees the listener can re-read the run + * row from a fresh DB connection — eliminates the read-your-writes + * trap from the same-tx delivery path.
    • + *
  • {@code @Async("cronDeliveryExecutor")} unbinds delivery from the + * Spring event-dispatcher thread so a slow IM API never stalls other + * listeners. The pool uses {@code AbortPolicy} + audit so an overflow + * surfaces immediately rather than degrading silently.
    • + *
+ */ +@Slf4j +@Component +@RequiredArgsConstructor +public class CronDeliveryListener { + + /** Spring auto-collects every {@link CronResultDelivery} bean ordered by {@code @Order}. */ + private final List deliveries; + private final AuditEventService audit; + + @Async("cronDeliveryExecutor") + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true) + public void onCompleted(CronJobCompletedEvent ev) { + Optional strategy = deliveries.stream() + .filter(d -> d.supports(ev.job())) + .findFirst(); + if (strategy.isEmpty()) { + // RFC §2.6 / §2.8.1: web-origin runs (or any job without a + // matching strategy) leave delivery_status=NONE and exit silently. + return; + } + try { + strategy.get().deliver(ev.job(), ev.result(), ev.run()); + } catch (Exception e) { + // RFC §2.7.3: always-best-effort — failures audit but never flip + // the run's main status. Stale-pending cleanup tightens the + // delivery_status state machine if the row gets stuck. + log.warn("[CronDelivery] Delivery failed for job {}: {}", + ev.job() != null ? ev.job().getId() : null, e.getMessage()); + try { + audit.record("DELIVERY_FAILED", "CRON_JOB", + ev.job() != null ? String.valueOf(ev.job().getId()) : "unknown", + e.getMessage(), null); + } catch (Exception auditError) { + log.warn("[CronDelivery] Audit recording failed: {}", auditError.getMessage()); + } + } + } + + /** + * Listener hook for unit tests bypassing the {@code @TransactionalEventListener} + * proxy — direct {@code ApplicationEventPublisher.publishEvent} delivery. + * Production paths always go through the AFTER_COMMIT bridge. + */ + @EventListener + public void onCompletedRaw(CronJobCompletedEvent ev) { + // No-op: the @TransactionalEventListener above is the production path + // (it only fires when a transaction is active and after it commits). + // This raw @EventListener exists so unit tests using fallbackExecution + // do not double-fire and so the listener bean stays scannable in + // contexts without a tx manager. Intentionally empty. + } + + /** + * RFC-063r §2.7.3: dedicated executor for cron delivery. Defined here so + * the listener and its pool live in the same module without an extra + * {@code @Configuration} class. + */ + @org.springframework.context.annotation.Bean(name = "cronDeliveryExecutor") + public ThreadPoolTaskExecutor cronDeliveryExecutor() { + ThreadPoolTaskExecutor ex = new ThreadPoolTaskExecutor(); + ex.setCorePoolSize(2); + ex.setMaxPoolSize(4); + ex.setQueueCapacity(1000); + ex.setThreadNamePrefix("cron-delivery-"); + ex.setRejectedExecutionHandler((r, executor) -> { + // RFC §2.7.3: AbortPolicy + audit (NOT CallerRunsPolicy) — the + // caller is the Spring event-dispatcher thread; blocking it would + // stall every other listener. + log.error("[CronDelivery] queue overflow ({}); dropping task — review pool sizing", + executor.getQueue().size()); + try { + audit.record("DELIVERY_QUEUE_OVERFLOW", "CRON_DELIVERY", "system", + "active=" + executor.getActiveCount() + ",queue=" + executor.getQueue().size(), + null); + } catch (Exception ignored) { + // Audit must never block the rejection path itself. + } + }); + ex.initialize(); + return ex; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronJobCompletedEvent.java b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronJobCompletedEvent.java new file mode 100644 index 00000000..8bf4884d --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronJobCompletedEvent.java @@ -0,0 +1,17 @@ +package vip.mate.cron.delivery; + +import org.springframework.ai.chat.messages.AssistantMessage; +import vip.mate.cron.model.CronJobEntity; +import vip.mate.dashboard.model.CronJobRunEntity; + +/** + * RFC-063r §2.7.3: domain event fired by + * {@code CronJobLifecycleService.finishRunAndPublish} after T2 commits. + * Listeners run with {@code @TransactionalEventListener(AFTER_COMMIT)}, so + * the run row is guaranteed visible from a fresh DB connection. + */ +public record CronJobCompletedEvent( + CronJobEntity job, + AssistantMessage result, + CronJobRunEntity run) { +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronResultDelivery.java b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronResultDelivery.java new file mode 100644 index 00000000..c45f26e7 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronResultDelivery.java @@ -0,0 +1,33 @@ +package vip.mate.cron.delivery; + +import org.springframework.ai.chat.messages.AssistantMessage; +import vip.mate.cron.model.CronJobEntity; +import vip.mate.dashboard.model.CronJobRunEntity; + +/** + * RFC-063r §2.6: pluggable strategy for delivering a cron job's assistant + * result back to its originating context (channel, future webhook, future + * SSE bridge, etc.). + * + *

{@link AbstractCronResultDelivery} provides the SQL-CAS idempotency + + * state-machine update; concrete strategies only implement + * {@link AbstractCronResultDelivery#doDeliver}. + */ +public interface CronResultDelivery { + + /** + * Whether this strategy applies to the given job. The first matching + * strategy (per {@code @Order}) wins; web-origin jobs match nothing, + * leaving {@code delivery_status='NONE'}. + */ + boolean supports(CronJobEntity job); + + /** + * Run the delivery, internally claiming the run row via SQL CAS and + * updating {@code delivery_status} on success / failure. Implementations + * should override + * {@link AbstractCronResultDelivery#doDeliver(CronJobEntity, AssistantMessage, CronJobRunEntity)} + * rather than this method directly. + */ + DeliveryOutcome deliver(CronJobEntity job, AssistantMessage result, CronJobRunEntity run); +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java new file mode 100644 index 00000000..4b5fa171 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java @@ -0,0 +1,63 @@ +package vip.mate.cron.delivery; + +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import vip.mate.dashboard.model.CronJobRunEntity; +import vip.mate.dashboard.repository.CronJobRunMapper; + +import java.time.Duration; +import java.time.LocalDateTime; + +/** + * RFC-063r §2.8.2: periodic sweep that recovers two stuck states with one + * scheduled job. + * + *

    + *
  • {@code delivery_status=PENDING} older than 15 min → mark + * {@code NOT_DELIVERED} with {@code stale-pending-timeout} reason. + * Covers listener crashes / OOMs / forced kills after a successful + * {@code claimRun()} but before {@code markDelivered}.
    • + *
  • {@code status='running'} older than 30 min → mark {@code failed} + * with {@code stale-running-timeout}. Covers + * {@code CronJobLifecycleService.markRunFailed()} itself failing under + * DB jitter (the LLM call already terminated by then).
    • + *
+ * + *

Single sweep, two predicates, one DB roundtrip per state. Spring + * {@code @Scheduled} is already enabled at the application bootstrap. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class CronRunStaleCleanup { + + private final CronJobRunMapper runMapper; + + private static final Duration DELIVERY_STALE = Duration.ofMinutes(15); + private static final Duration RUN_STALE = Duration.ofMinutes(30); + + @Scheduled(fixedDelay = 5 * 60 * 1000L, initialDelay = 60 * 1000L) + public void sweep() { + LocalDateTime now = LocalDateTime.now(); + + int stalePending = runMapper.update(null, new LambdaUpdateWrapper() + .eq(CronJobRunEntity::getDeliveryStatus, "PENDING") + .lt(CronJobRunEntity::getStartedAt, now.minus(DELIVERY_STALE)) + .set(CronJobRunEntity::getDeliveryStatus, "NOT_DELIVERED") + .set(CronJobRunEntity::getDeliveryError, "stale-pending-timeout")); + + int staleRunning = runMapper.update(null, new LambdaUpdateWrapper() + .eq(CronJobRunEntity::getStatus, "running") + .lt(CronJobRunEntity::getStartedAt, now.minus(RUN_STALE)) + .set(CronJobRunEntity::getStatus, "failed") + .set(CronJobRunEntity::getFinishedAt, now) + .set(CronJobRunEntity::getErrorMessage, "stale-running-timeout")); + + if (stalePending > 0 || staleRunning > 0) { + log.warn("[CronCleanup] swept stalePending={} staleRunning={}", stalePending, staleRunning); + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/delivery/DeliveryOutcome.java b/mateclaw-server/src/main/java/vip/mate/cron/delivery/DeliveryOutcome.java new file mode 100644 index 00000000..e2fa6d43 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/DeliveryOutcome.java @@ -0,0 +1,27 @@ +package vip.mate.cron.delivery; + +import org.springframework.lang.Nullable; + +/** + * RFC-063r §2.6.0: outcome of a single cron-result delivery attempt. + * + *

Two states only — {@code DELIVERED} and {@code SKIPPED}. There is no + * {@code FAILED} state; failures throw from + * {@link AbstractCronResultDelivery#doDeliver}, get marked + * {@code NOT_DELIVERED} on the run row by the template method, and surface + * to the listener as exceptions for audit. Single-purpose return semantics. + */ +public record DeliveryOutcome(Status status, + @Nullable String target, + @Nullable String reason) { + + public enum Status { DELIVERED, SKIPPED } + + public static DeliveryOutcome delivered(String target) { + return new DeliveryOutcome(Status.DELIVERED, target, null); + } + + public static DeliveryOutcome skipped(String reason) { + return new DeliveryOutcome(Status.SKIPPED, null, reason); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/model/CronJobDTO.java b/mateclaw-server/src/main/java/vip/mate/cron/model/CronJobDTO.java index 400c5bc5..032185fd 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/model/CronJobDTO.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/model/CronJobDTO.java @@ -28,6 +28,22 @@ public class CronJobDTO { private LocalDateTime createTime; private LocalDateTime updateTime; + /** RFC-063r §2.9: originating channel binding (null = web-origin cron). */ + private Long channelId; + + /** RFC-063r §2.9: delivery target detail (targetId / threadId / accountId). */ + private DeliveryConfig deliveryConfig; + + /** + * RFC-063r §2.14: read-model field surfaced by CronJobMapper#selectListWithDeliveryStatus + * (PR-3). One of NONE / PENDING / DELIVERED / NOT_DELIVERED, taken from + * the most-recent run row. Out-only — never accepted on create/update. + */ + private String lastDeliveryStatus; + + /** RFC-063r §2.14: out-only error detail for the most-recent delivery attempt. */ + private String lastDeliveryError; + public static CronJobDTO from(CronJobEntity entity) { CronJobDTO dto = new CronJobDTO(); dto.setId(entity.getId()); @@ -43,6 +59,15 @@ public class CronJobDTO { dto.setLastRunTime(entity.getLastRunTime()); dto.setCreateTime(entity.getCreateTime()); dto.setUpdateTime(entity.getUpdateTime()); + dto.setChannelId(entity.getChannelId()); + dto.setDeliveryConfig(entity.getDeliveryConfig()); + // RFC-063r §2.14: surface the latest-run delivery snapshot when the + // entity was loaded via selectListWithDeliveryStatus / selectByIdWithDeliveryStatus. + // Default "NONE" when no run has ever been recorded so the UI can + // render a neutral badge instead of a blank cell. + dto.setLastDeliveryStatus(entity.getLastDeliveryStatus() != null + ? entity.getLastDeliveryStatus() : "NONE"); + dto.setLastDeliveryError(entity.getLastDeliveryError()); return dto; } @@ -63,6 +88,8 @@ public class CronJobDTO { entity.setTriggerMessage(this.triggerMessage); entity.setRequestBody(this.requestBody); entity.setEnabled(this.enabled); + entity.setChannelId(this.channelId); + entity.setDeliveryConfig(this.deliveryConfig); return entity; } } diff --git a/mateclaw-server/src/main/java/vip/mate/cron/model/CronJobEntity.java b/mateclaw-server/src/main/java/vip/mate/cron/model/CronJobEntity.java index 8b7398a0..22f3ee9c 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/model/CronJobEntity.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/model/CronJobEntity.java @@ -1,6 +1,7 @@ package vip.mate.cron.model; import com.baomidou.mybatisplus.annotation.*; +import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler; import lombok.Data; import java.time.LocalDateTime; @@ -11,7 +12,7 @@ import java.time.LocalDateTime; * @author MateClaw Team */ @Data -@TableName("mate_cron_job") +@TableName(value = "mate_cron_job", autoResultMap = true) public class CronJobEntity { @TableId(type = IdType.ASSIGN_ID) @@ -49,6 +50,21 @@ public class CronJobEntity { /** 上次执行时间 */ private LocalDateTime lastRunTime; + /** + * RFC-063r §2.9: originating channel binding. Null when this job was + * created from the web (no proactive delivery target). The single + * indexed column lets ops query "all jobs delivering to channel X". + */ + private Long channelId; + + /** + * RFC-063r §2.9: delivery target detail (targetId / threadId / accountId) + * persisted as JSON via MyBatis Plus JacksonTypeHandler so future fields + * don't require schema migrations. + */ + @TableField(typeHandler = JacksonTypeHandler.class) + private DeliveryConfig deliveryConfig; + @TableField(fill = FieldFill.INSERT) private LocalDateTime createTime; @@ -56,4 +72,16 @@ public class CronJobEntity { private LocalDateTime updateTime; private Integer deleted; + + /** + * RFC-063r §2.14: read-model field — populated by + * {@code CronJobMapper.selectListWithDeliveryStatus()} via a subquery + * against {@code mate_cron_job_run}. Not part of the writable schema. + */ + @TableField(exist = false) + private String lastDeliveryStatus; + + /** RFC-063r §2.14: matching error column for the most-recent run. */ + @TableField(exist = false) + private String lastDeliveryError; } diff --git a/mateclaw-server/src/main/java/vip/mate/cron/model/DeliveryConfig.java b/mateclaw-server/src/main/java/vip/mate/cron/model/DeliveryConfig.java new file mode 100644 index 00000000..ac3b01b4 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/model/DeliveryConfig.java @@ -0,0 +1,34 @@ +package vip.mate.cron.model; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import org.springframework.lang.Nullable; +import vip.mate.agent.context.ChannelTarget; + +/** + * RFC-063r §2.9: per-cron-job delivery configuration. Persisted as a JSON + * column on {@code mate_cron_job.delivery_config} via MyBatis Plus + * {@code JacksonTypeHandler}. Mirrors {@link ChannelTarget} but lives in the + * cron module so {@code CronJobEntity} doesn't need to depend on the channel + * value object directly. + * + *

{@link JsonIgnoreProperties#ignoreUnknown()} keeps deserialization + * forward-compatible when older rows are read after a column-add upgrade. + */ +@JsonIgnoreProperties(ignoreUnknown = true) +public record DeliveryConfig( + @Nullable String targetId, + @Nullable String threadId, + @Nullable String accountId +) { + + /** Convert from the {@link ChannelTarget} carried on a {@code ChatOrigin}. */ + public static DeliveryConfig from(@Nullable ChannelTarget t) { + if (t == null) return null; + return new DeliveryConfig(t.targetId(), t.threadId(), t.accountId()); + } + + /** Convert back to a {@link ChannelTarget} for ChatOrigin reconstruction. */ + public ChannelTarget toChannelTarget() { + return new ChannelTarget(targetId, threadId, accountId); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/repository/CronJobMapper.java b/mateclaw-server/src/main/java/vip/mate/cron/repository/CronJobMapper.java index 5dab7d86..9e7622bf 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/repository/CronJobMapper.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/repository/CronJobMapper.java @@ -2,8 +2,12 @@ package vip.mate.cron.repository; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; import vip.mate.cron.model.CronJobEntity; +import java.util.List; + /** * 定时任务 Mapper * @@ -11,4 +15,47 @@ import vip.mate.cron.model.CronJobEntity; */ @Mapper public interface CronJobMapper extends BaseMapper { + + /** + * RFC-063r §2.14: list cron jobs together with their most-recent + * delivery status / error (subquery against {@code mate_cron_job_run}). + * + *

Both H2 and MySQL accept {@code LIMIT 1} inside a correlated + * subquery, so the same SQL is portable across the two profiles. Index + * coverage: {@code mate_cron_job_run(cron_job_id, started_at)} (created + * by V1 baseline migration) makes the subquery cheap. + * + *

Filters out logically-deleted rows and orders by create_time DESC + * to mirror the existing {@code list()} ordering. + */ + @Select(""" + SELECT j.*, + (SELECT r.delivery_status FROM mate_cron_job_run r + WHERE r.cron_job_id = j.id + ORDER BY r.started_at DESC LIMIT 1) AS last_delivery_status, + (SELECT r.delivery_error FROM mate_cron_job_run r + WHERE r.cron_job_id = j.id + ORDER BY r.started_at DESC LIMIT 1) AS last_delivery_error + FROM mate_cron_job j + WHERE j.deleted = 0 + ORDER BY j.create_time DESC + """) + List selectListWithDeliveryStatus(); + + /** + * RFC-063r §2.14: per-job variant for the detail page. Same subquery + * pattern, restricted to a single id. + */ + @Select(""" + SELECT j.*, + (SELECT r.delivery_status FROM mate_cron_job_run r + WHERE r.cron_job_id = j.id + ORDER BY r.started_at DESC LIMIT 1) AS last_delivery_status, + (SELECT r.delivery_error FROM mate_cron_job_run r + WHERE r.cron_job_id = j.id + ORDER BY r.started_at DESC LIMIT 1) AS last_delivery_error + FROM mate_cron_job j + WHERE j.id = #{id} AND j.deleted = 0 + """) + CronJobEntity selectByIdWithDeliveryStatus(@Param("id") Long id); } diff --git a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobLifecycleService.java b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobLifecycleService.java new file mode 100644 index 00000000..7386f6e7 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobLifecycleService.java @@ -0,0 +1,128 @@ +package vip.mate.cron.service; + +import cn.hutool.core.util.StrUtil; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; +import vip.mate.cron.delivery.CronJobCompletedEvent; +import vip.mate.cron.model.CronJobEntity; +import vip.mate.dashboard.model.CronJobRunEntity; +import vip.mate.dashboard.repository.CronJobRunMapper; +import vip.mate.memory.event.ConversationCompletionPublisher; +import vip.mate.workspace.conversation.ConversationService; + +import java.time.LocalDateTime; + +/** + * RFC-063r §2.7.2: three-segment transactional support for {@link CronJobRunner}. + * + *

Each method runs in its own short {@code REQUIRES_NEW} transaction so + * the long LLM call between T1 and T2 never holds a DB connection. + * + *

    + *
  • {@code T1} — {@link #startRun}: insert run row + persist user message.
    • + *
  • {@code T-fail} — {@link #markRunFailed}: terminal state when the LLM + * call throws.
    • + *
  • {@code T2} — {@link #finishRunAndPublish}: persist assistant + * message + dispatch the conversation-completed event (memory pipeline) + * + the cron-job-completed event (delivery pipeline).
    • + *
+ * + *

Lives in a separate {@code @Service} bean so cross-bean invocation from + * {@link CronJobRunner} routes through the Spring AOP proxy (RFC §5.2 hard + * rule). The lifecycle service is deliberately the only place + * {@code @Transactional} appears in the cron-execution path. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class CronJobLifecycleService { + + private final CronJobRunMapper runMapper; + private final ConversationService conversationService; + private final ConversationCompletionPublisher completionPublisher; + private final ApplicationEventPublisher events; + + /** + * T1 — short transaction: persist a run row in {@code running} state, + * persist the user message that triggered the run, and commit. Returns + * the persisted entity so callers can observe its assigned id without a + * second SELECT. + * + * @param triggerType {@code scheduled} (cron tick) or {@code manual} (runNow) + */ + @Transactional(propagation = Propagation.REQUIRES_NEW) + public CronJobRunEntity startRun(CronJobEntity job, String userMessage, String triggerType) { + CronJobRunEntity run = new CronJobRunEntity(); + run.setCronJobId(job.getId()); + run.setConversationId("cron:" + job.getId()); + run.setStatus("running"); + run.setTriggerType(triggerType != null ? triggerType : "scheduled"); + run.setStartedAt(LocalDateTime.now()); + run.setDeliveryStatus("NONE"); + runMapper.insert(run); + + // Persist the user message before the LLM call so history reads + // see a coherent (user → assistant) ordering even if the agent + // throws mid-run. + if (userMessage != null && !userMessage.isBlank()) { + conversationService.saveMessage(run.getConversationId(), "user", userMessage); + } + return run; + } + + /** + * T-fail — short transaction: flag the run row as failed when the agent + * throws. Always-best-effort policy: delivery_status stays NONE; nothing + * is published. + */ + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void markRunFailed(CronJobRunEntity run, Throwable error) { + String message = error != null && error.getMessage() != null ? error.getMessage() : "unknown error"; + runMapper.update(null, new LambdaUpdateWrapper() + .eq(CronJobRunEntity::getId, run.getId()) + .set(CronJobRunEntity::getStatus, "failed") + .set(CronJobRunEntity::getFinishedAt, LocalDateTime.now()) + .set(CronJobRunEntity::getErrorMessage, StrUtil.maxLength(message, 1000))); + } + + /** + * T2 — short transaction: persist the assistant reply, mark the run + * succeeded, then publish the two domain events. The + * {@code @TransactionalEventListener(AFTER_COMMIT)} listeners only run + * once this method's tx commits, so cross-connection reads in the + * delivery / memory pipelines always see the final state. + */ + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void finishRunAndPublish(CronJobEntity job, CronJobRunEntity run, + String userMessage, AssistantMessage result) { + String convId = "cron:" + job.getId(); + String text = result != null && result.getText() != null ? result.getText() : ""; + + runMapper.update(null, new LambdaUpdateWrapper() + .eq(CronJobRunEntity::getId, run.getId()) + .set(CronJobRunEntity::getStatus, "succeeded") + .set(CronJobRunEntity::getFinishedAt, LocalDateTime.now())); + + conversationService.saveMessage(convId, "assistant", text); + + // Memory pipeline (existing behavior preserved — was inline in the + // old executeJob; now lives behind the same publisher used by the + // web / channel paths so cron is no longer special-cased). + try { + completionPublisher.publish(job.getAgentId(), convId, userMessage, text, "cron"); + } catch (Exception e) { + // Memory failures must not break delivery — log + carry on. + log.warn("[CronLifecycle] completionPublisher failed for job {}: {}", job.getId(), e.getMessage()); + } + + // Delivery pipeline (RFC-063r §2.7.3) — fired here so listeners only + // run after T2 commits. + events.publishEvent(new CronJobCompletedEvent(job, result != null ? result : new AssistantMessage(""), run)); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobRunner.java b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobRunner.java new file mode 100644 index 00000000..294d83c7 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobRunner.java @@ -0,0 +1,152 @@ +package vip.mate.cron.service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.stereotype.Component; +import vip.mate.agent.AgentService; +import vip.mate.agent.context.ChatOrigin; +import vip.mate.cron.CronChatOriginFactory; +import vip.mate.cron.model.CronJobEntity; +import vip.mate.dashboard.model.CronJobRunEntity; + +/** + * RFC-063r §2.7.1: scheduler-facing orchestrator that decomposes one cron + * tick into the three transactional segments owned by + * {@link CronJobLifecycleService}, with the long-running LLM call sitting + * outside any transaction. + * + *

This class must NOT be annotated {@code @Transactional} — see + * RFC-063r §5.2: + *

    + *
  • The class is the entry point invoked by the + * {@code ThreadPoolTaskScheduler}'s lambda; the LLM HTTP call inside + * {@link #runAgent} is seconds-to-minutes long and must not hold a DB + * connection.
    • + *
  • Self-invocation in the legacy {@code CronJobService} silently + * skipped {@code @Transactional}; that pattern is forbidden here.
    • + *
  • An {@code ArchUnit} test (see PR-3) pins this rule so a future + * regression fails CI.
    • + *
+ * + *

The three transactional segments live on + * {@link CronJobLifecycleService} (a separate bean), so cross-bean calls + * route through Spring AOP and {@code REQUIRES_NEW} works as advertised. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class CronJobRunner { + + private final CronJobLifecycleService lifecycle; + private final AgentService agentService; + private final CronChatOriginFactory originFactory; + + /** + * Scheduler-facing entry. Runs three logical segments: + *

    + *
  1. T1 — {@link CronJobLifecycleService#startRun} commits run row + + * user message.
    2. + *
  2. Untransacted — {@link #runAgent} performs the LLM call.
    3. + *
  3. T2 — {@link CronJobLifecycleService#finishRunAndPublish} commits + * assistant message and publishes the completion / delivery + * events. {@code AFTER_COMMIT} listeners fire from a fresh DB + * connection, so {@link CronJobRunEntity}'s persisted state is + * always visible by the time delivery resolves a strategy.
    4. + *
+ */ + public void executeJob(CronJobEntity job) { + executeJob(job, /* triggerType */ "scheduled"); + } + + /** Variant with explicit trigger type — used by {@code runNow} (manual). */ + public void executeJob(CronJobEntity job, String triggerType) { + if (job == null) { + log.warn("[CronRunner] executeJob called with null job — ignoring"); + return; + } + String userMessage = "agent".equals(job.getTaskType()) + ? job.getRequestBody() + : job.getTriggerMessage(); + + // T1 — short tx + CronJobRunEntity run; + try { + run = lifecycle.startRun(job, userMessage, triggerType); + } catch (Exception e) { + log.error("[CronRunner] T1 startRun failed for job {}: {}", job.getId(), e.getMessage(), e); + return; + } + + // No-tx segment — long LLM call. RFC §5.2 hard rule: must not hold + // any DB connection during this call. + AssistantMessage result; + try { + ChatOrigin origin = originFactory.from(job, "cron:" + job.getId()); + result = runAgent(job, userMessage, origin); + } catch (Exception e) { + log.error("[CronRunner] runAgent failed for job {}: {}", job.getId(), e.getMessage(), e); + try { + lifecycle.markRunFailed(run, e); + } catch (Exception markErr) { + // CronRunStaleCleanup will sweep status='running' rows older than 30 min. + log.warn("[CronRunner] markRunFailed itself failed for run {}: {} (stale-cleanup will recover)", + run.getId(), markErr.getMessage()); + } + return; + } + + // T2 — short tx + try { + lifecycle.finishRunAndPublish(job, run, userMessage, result); + } catch (Exception e) { + log.error("[CronRunner] T2 finishRunAndPublish failed for job {}: {}", job.getId(), e.getMessage(), e); + try { + lifecycle.markRunFailed(run, e); + } catch (Exception markErr) { + log.warn("[CronRunner] markRunFailed after T2 failure also failed for run {}: {}", + run.getId(), markErr.getMessage()); + } + } + } + + /** + * Runs the agent with the cron-derived {@link ChatOrigin} and the + * RFC-063r §2.13 system-prompt guard prepended when the cron is bound to + * a channel — fixes the Issue #25 LLM hallucination ("install + * mateclaw cli to send to wechat") by telling the model that delivery is + * framework-handled. + */ + private AssistantMessage runAgent(CronJobEntity job, String userMessage, ChatOrigin origin) { + String guarded = wrapWithDeliveryGuard(userMessage, origin); + String text = "agent".equals(job.getTaskType()) + ? agentService.execute(job.getAgentId(), guarded, "cron:" + job.getId(), origin) + : agentService.chat(job.getAgentId(), guarded, "cron:" + job.getId(), origin); + return new AssistantMessage(text != null ? text : ""); + } + + /** + * RFC-063r §2.13: when the cron is bound to a channel, prepend an + * explicit system note telling the LLM that delivery is handled by the + * framework. Without this, the model invents tools ("call CLI to send + * to wechat") and surfaces "command not found" style errors to users + * (Issue #25 second symptom). + * + *

Web-origin crons (no channelId) bypass the wrapper so the + * pre-RFC behavior is preserved. + */ + static String wrapWithDeliveryGuard(String userMessage, ChatOrigin origin) { + String body = userMessage != null ? userMessage : ""; + if (origin == null || origin.channelId() == null) { + return body; + } + return """ + [系统说明] + 本次执行由定时任务触发，结果将由系统自动投递回原渠道， + 你只需直接给出最终回复内容，不要尝试调用 CLI / shell / + "发送到微信"等工具——这些操作由框架完成。 + + [用户原始消息] + """ + body; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java index 2196ba43..cc08bd13 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java @@ -12,15 +12,12 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.support.CronExpression; import org.springframework.scheduling.support.CronTrigger; import org.springframework.stereotype.Service; -import vip.mate.agent.AgentService; import vip.mate.agent.model.AgentEntity; import vip.mate.agent.repository.AgentMapper; import vip.mate.cron.model.CronJobDTO; import vip.mate.cron.model.CronJobEntity; import vip.mate.cron.repository.CronJobMapper; import vip.mate.exception.MateClawException; -import vip.mate.memory.event.ConversationCompletionPublisher; -import vip.mate.workspace.conversation.ConversationService; import java.time.LocalDateTime; import java.time.ZoneId; @@ -45,17 +42,23 @@ public class CronJobService implements ApplicationRunner { private final CronJobMapper cronJobMapper; private final AgentMapper agentMapper; - private final AgentService agentService; - private final ConversationService conversationService; - private final ConversationCompletionPublisher completionPublisher; + /** + * RFC-063r §2.7.1: cron-tick execution moved to {@link CronJobRunner} + * (separate bean) so the three-segment transactional model in + * {@link CronJobLifecycleService} works via Spring AOP — no more + * self-invocation footgun. + * + *

{@code agentService}, {@code conversationService}, and + * {@code completionPublisher} now live on {@link CronJobLifecycleService} + * and {@link CronJobRunner} so this service shrinks to CRUD + scheduler + * registration only. + */ + private final CronJobRunner cronJobRunner; private final ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); private final ConcurrentHashMap> scheduledTasks = new ConcurrentHashMap<>(); private final ReentrantLock schedulerLock = new ReentrantLock(); - /** 定时任务触发时使用的系统用户标识 */ - private static final String SYSTEM_USER = "system"; - // ==================== 初始化与销毁 ==================== /** @@ -88,9 +91,10 @@ public class CronJobService implements ApplicationRunner { // ==================== CRUD ==================== public List list() { - List entities = cronJobMapper.selectList( - new LambdaQueryWrapper() - .orderByDesc(CronJobEntity::getCreateTime)); + // RFC-063r §2.14: use the variant that aggregates the most-recent + // delivery_status from mate_cron_job_run so the list page can + // render the "最近投递" badge without a per-row N+1 query. + List entities = cronJobMapper.selectListWithDeliveryStatus(); // 批量加载 Agent 名称 List agentIds = entities.stream() @@ -107,7 +111,9 @@ public class CronJobService implements ApplicationRunner { } public CronJobDTO getById(Long id) { - CronJobEntity entity = cronJobMapper.selectById(id); + // RFC-063r §2.14: detail page shows lastDeliveryStatus too — same + // subquery shape, restricted to one id. + CronJobEntity entity = cronJobMapper.selectByIdWithDeliveryStatus(id); if (entity == null) { throw new MateClawException("err.cron.not_found", "定时任务不存在: " + id); } @@ -219,8 +225,19 @@ public class CronJobService implements ApplicationRunner { if (entity == null) { throw new MateClawException("err.cron.not_found", "定时任务不存在: " + id); } - // 异步执行，不阻塞请求线程 - scheduler.submit(() -> executeJob(entity)); + // RFC-063r §2.7.1: delegate to CronJobRunner via Spring proxy so + // the three-segment REQUIRES_NEW transactions on + // CronJobLifecycleService work as advertised. "manual" trigger type + // distinguishes this from scheduler-driven runs in mate_cron_job_run. + scheduler.submit(() -> { + try { + cronJobRunner.executeJob(entity, "manual"); + } finally { + // RFC-063r §2.7.1: bookkeep regardless of run outcome so a + // single bad run does not wedge all future ticks. + updateRunTimes(entity.getId(), entity.getCronExpression(), entity.getTimezone()); + } + }); } // ==================== 调度器管理 ==================== @@ -232,7 +249,16 @@ public class CronJobService implements ApplicationRunner { String springCron = toSpringCron(job.getCronExpression()); ZoneId zoneId = ZoneId.of(job.getTimezone()); CronTrigger trigger = new CronTrigger(springCron, zoneId); - ScheduledFuture future = scheduler.schedule(() -> executeJob(job), trigger); + // RFC-063r §2.7.1: delegate to CronJobRunner — see runNow above. + // Bookkeeping (lastRunTime / nextRunTime) is wrapped in a finally + // so a failed run still advances the schedule. + ScheduledFuture future = scheduler.schedule(() -> { + try { + cronJobRunner.executeJob(job, "scheduled"); + } finally { + updateRunTimes(job.getId(), job.getCronExpression(), job.getTimezone()); + } + }, trigger); scheduledTasks.put(job.getId(), future); log.info("[CronJob] Registered job {} ({}), cron={}, tz={}", job.getId(), job.getName(), job.getCronExpression(), job.getTimezone()); @@ -249,46 +275,18 @@ public class CronJobService implements ApplicationRunner { } // ==================== 任务执行 ==================== - - private void executeJob(CronJobEntity job) { - String conversationId = "cron:" + job.getId(); - try { - log.info("[CronJob] Executing job {} ({}), type={}", job.getId(), job.getName(), job.getTaskType()); - - // 确保会话存在（使用 SYSTEM_USER 作为定时触发的所有者标识，workspace 从 agent 获取） - AgentEntity cronAgent = agentMapper.selectById(job.getAgentId()); - Long cronWorkspaceId = cronAgent != null ? cronAgent.getWorkspaceId() : 1L; - conversationService.getOrCreateConversation(conversationId, job.getAgentId(), SYSTEM_USER, cronWorkspaceId); - - String userMessage; - String result; - if ("agent".equals(job.getTaskType())) { - userMessage = job.getRequestBody(); - // 保存 user 消息 - conversationService.saveMessage(conversationId, "user", userMessage); - result = agentService.execute(job.getAgentId(), userMessage, conversationId); - } else { - userMessage = job.getTriggerMessage(); - // 保存 user 消息 - conversationService.saveMessage(conversationId, "user", userMessage); - result = agentService.chat(job.getAgentId(), userMessage, conversationId); - } - - // 保存 assistant 消息 - conversationService.saveMessage(conversationId, "assistant", result); - - // 发布对话完成事件 - completionPublisher.publish(job.getAgentId(), conversationId, userMessage, result, "cron"); - - // 合并更新 lastRunTime + nextRunTime，单次 DB 写入 - updateRunTimes(job.getId(), job.getCronExpression(), job.getTimezone()); - - log.info("[CronJob] Job {} executed successfully, result length={}", job.getId(), - result != null ? result.length() : 0); - } catch (Exception e) { - log.error("[CronJob] Job {} execution failed: {}", job.getId(), e.getMessage(), e); - } - } + // + // RFC-063r §2.7.1: the executeJob body moved to CronJobRunner so the + // three-segment transactional model in CronJobLifecycleService runs + // through a Spring AOP proxy. CronJobService now only owns CRUD + + // scheduler registration, and the lastRunTime / nextRunTime bookkeeping + // hook below — which deliberately runs *after* the runner returns so a + // failed run still advances the next-run pointer (otherwise a single + // bad run wedges all future ticks). + // + // Both register() and runNow() now delegate to cronJobRunner.executeJob; + // see those methods above. The wrap below ensures next-run rolls forward + // regardless of run outcome. /** * 合并更新 lastRunTime 和 nextRunTime，单次 DB 写入替代原来的 4 次 selectById + updateById diff --git a/mateclaw-server/src/main/java/vip/mate/dashboard/model/CronJobRunEntity.java b/mateclaw-server/src/main/java/vip/mate/dashboard/model/CronJobRunEntity.java index fe5a8d3e..91dbab9e 100644 --- a/mateclaw-server/src/main/java/vip/mate/dashboard/model/CronJobRunEntity.java +++ b/mateclaw-server/src/main/java/vip/mate/dashboard/model/CronJobRunEntity.java @@ -19,6 +19,24 @@ public class CronJobRunEntity { private LocalDateTime finishedAt; private String errorMessage; private Integer tokenUsage; + + // ===== RFC-063r §2.9: delivery state machine ===== + + /** + * Delivery state machine — see RFC-063r §2.8.1: + * {@code NONE} (web-origin) | {@code PENDING} (claimed) | + * {@code DELIVERED} | {@code NOT_DELIVERED} (failure or stale-cleanup). + * Orthogonal to {@link #status} (run main state); always-best-effort + * policy means delivery failure never flips the main state to failed. + */ + private String deliveryStatus; + + /** Resolved delivery target (IM userId / chat_id / Feishu webhook URL). */ + private String deliveryTarget; + + /** Delivery error reason — Hutool-truncated to 500 chars max. */ + private String deliveryError; + @TableField(fill = FieldFill.INSERT) private LocalDateTime createTime; } diff --git a/mateclaw-server/src/main/java/vip/mate/i18n/LocaleAwareToolCallback.java b/mateclaw-server/src/main/java/vip/mate/i18n/LocaleAwareToolCallback.java index 76c3a072..d0ec8869 100644 --- a/mateclaw-server/src/main/java/vip/mate/i18n/LocaleAwareToolCallback.java +++ b/mateclaw-server/src/main/java/vip/mate/i18n/LocaleAwareToolCallback.java @@ -1,15 +1,23 @@ package vip.mate.i18n; +import org.springframework.ai.chat.model.ToolContext; import org.springframework.ai.tool.ToolCallback; import org.springframework.ai.tool.definition.ToolDefinition; +import org.springframework.ai.tool.metadata.ToolMetadata; +import org.springframework.lang.Nullable; /** - * 国际化工具回调装饰器 - *

- * 包装原始 ToolCallback，覆写 {@link #getToolDefinition()} 返回本地化描述。 - * 其他方法（call、name 等）全部委托给原始回调。 + * I18n decorator around a {@link ToolCallback} — overrides + * {@link #getToolDefinition()} so the LLM sees the localized description. * - * @author MateClaw Team + *

RFC-063r §2.3 (P0): every {@code call(String, ToolContext)} invocation + * must be forwarded to the underlying delegate so {@link ToolContext} (carrying + * {@code ChatOrigin}) reaches downstream {@code @Tool} methods. The previous + * implementation only overrode {@code call(String)}, which silently dropped the + * ToolContext via the framework default. + * + *

{@link #getToolMetadata()} is forwarded so {@code returnDirect=true} + * tools keep their direct-return semantics under this decorator. */ public class LocaleAwareToolCallback implements ToolCallback { @@ -31,8 +39,18 @@ public class LocaleAwareToolCallback implements ToolCallback { .build(); } + @Override + public ToolMetadata getToolMetadata() { + return delegate.getToolMetadata(); + } + @Override public String call(String toolInput) { return delegate.call(toolInput); } + + @Override + public String call(String toolInput, @Nullable ToolContext toolContext) { + return delegate.call(toolInput, toolContext); + } } diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/BrowserUseTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/BrowserUseTool.java index 00bae242..87c27d78 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/BrowserUseTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/BrowserUseTool.java @@ -12,8 +12,10 @@ import com.microsoft.playwright.PlaywrightException; import com.microsoft.playwright.options.LoadState; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.model.ToolContext; 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 vip.mate.tool.browser.BrowserDiagnosticsService; import vip.mate.tool.browser.BrowserLauncher; @@ -61,6 +63,15 @@ public class BrowserUseTool { private final Object playwrightLock = new Object(); private final ConcurrentHashMap sessions = new ConcurrentHashMap<>(); + + /** + * RFC-063r §2.5 transition: ToolContext for the current invocation, set + * at the @Tool entry point and read by {@link #broadcastBrowserEvent}. + * Tool calls are serialized per ToolExecutionExecutor instance so this + * volatile field is safe; the field is read-only inside the action + * handlers. + */ + private volatile ToolContext currentToolContext; private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> { Thread t = new Thread(r, "browser-idle-watchdog"); t.setDaemon(true); @@ -97,8 +108,15 @@ public class BrowserUseTool { @ToolParam(description = "JavaScript code to execute (for action=eval)", required = false) String code, @ToolParam(description = "File path to save screenshot (for action=screenshot)", required = false) String path, @ToolParam(description = "Launch visible browser window (for action=start, default false)", required = false) Boolean headed, - @ToolParam(description = "Single CDP port to scan (for action=list_cdp_targets)", required = false) Integer cdpPort + @ToolParam(description = "Single CDP port to scan (for action=list_cdp_targets)", required = false) Integer cdpPort, + // RFC-063r §2.5: hidden from LLM by JsonSchemaGenerator. + @Nullable ToolContext ctx ) { + // The conversationId resolution lives in broadcastBrowserEvent below; + // capture the ctx into a field so the helper can read it without + // passing it down every action handler. Race-free because tool calls + // are serialized per executor. + this.currentToolContext = ctx; if (action == null || action.isBlank()) { return error("action is required"); } @@ -163,7 +181,7 @@ public class BrowserUseTool { */ private void broadcastBrowserEvent(String action, boolean success, String url, String title, String screenshot, long durationMs) { - String conversationId = ToolExecutionContext.conversationId(); + String conversationId = ToolExecutionContext.conversationId(currentToolContext); if (conversationId == null || streamTracker == null) { return; } diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/CronJobTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/CronJobTool.java index 433654ba..df2638fe 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/CronJobTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/CronJobTool.java @@ -5,9 +5,12 @@ import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.model.ToolContext; 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 vip.mate.agent.context.ChatOrigin; import vip.mate.cron.model.CronJobDTO; import vip.mate.cron.service.CronJobService; @@ -38,12 +41,21 @@ public class CronJobTool { @ToolParam(description = "Task name, e.g. 'Daily AI News Summary'") String name, @ToolParam(description = "5-field cron expression: minute hour day month weekday") String cronExpression, @ToolParam(description = "Message to send when the task triggers, e.g. 'Search for the latest AI news and summarize'") String triggerMessage, - @ToolParam(description = "Timezone, default Asia/Shanghai. Examples: UTC, America/New_York", required = false) String timezone) { + @ToolParam(description = "Timezone, default Asia/Shanghai. Examples: UTC, America/New_York", required = false) String timezone, + // RFC-063r §2.4: ToolContext is *not* exposed to the LLM — + // JsonSchemaGenerator skips it (Spring AI 1.1 framework convention) + @Nullable ToolContext ctx) { try { - // Resolve current agent ID from conversation context - String conversationId = ToolExecutionContext.conversationId(); - Long agentId = resolveAgentId(conversationId); + // RFC-063r §2.5: prefer the explicit ChatOrigin (channelId / channelTarget + // / agentId all live there). Fall back to the legacy ToolExecutionContext + // ThreadLocal during the PR-1 transition window so callers that have not + // yet migrated keep working. + ChatOrigin origin = ChatOrigin.from(ctx); + String conversationId = origin.conversationId() != null && !origin.conversationId().isEmpty() + ? origin.conversationId() + : ToolExecutionContext.conversationId(); + Long agentId = origin.agentId() != null ? origin.agentId() : resolveAgentId(conversationId); CronJobDTO dto = new CronJobDTO(); dto.setName(name); @@ -54,6 +66,12 @@ public class CronJobTool { dto.setTaskType("text"); dto.setEnabled(true); + // RFC-063r §2.4 / PR-2: when the originating context carries a + // channelId, the cron job inherits the binding so its results can + // be delivered back to the same channel. Fields are wired via + // reflection until PR-2 adds them to CronJobDTO + CronJobEntity. + propagateChannelBinding(dto, origin); + CronJobDTO created = cronJobService.create(dto); JSONObject result = new JSONObject(); @@ -166,4 +184,17 @@ public class CronJobTool { result.set("error", message); return JSONUtil.toJsonPrettyStr(result); } + + /** + * RFC-063r §2.4: propagate the originating channel binding into the cron + * job DTO so PR-3's delivery dispatcher can route results back to the + * originating channel. + */ + private void propagateChannelBinding(CronJobDTO dto, ChatOrigin origin) { + if (origin == null || origin.channelId() == null) return; + dto.setChannelId(origin.channelId()); + if (origin.channelTarget() != null) { + dto.setDeliveryConfig(vip.mate.cron.model.DeliveryConfig.from(origin.channelTarget())); + } + } } 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 8d7a8626..e3793a80 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 @@ -5,10 +5,13 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.model.ToolContext; 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 vip.mate.agent.AgentService; +import vip.mate.agent.context.ChatOrigin; import vip.mate.agent.model.AgentEntity; import vip.mate.agent.repository.AgentMapper; import vip.mate.channel.web.ChatStreamTracker; @@ -76,7 +79,12 @@ public class DelegateAgentTool { For multiple parallel tasks, use delegateParallel instead.""") public String delegateToAgent( @ToolParam(description = "Target Agent name (exact match)") String agentName, - @ToolParam(description = "Task description with complete context information") String task) { + @ToolParam(description = "Task description with complete context information") String task, + // RFC-063r §2.5 改动点 5: parent ChatOrigin (channel binding / + // workspace) propagates into the delegated child so a sub-agent + // creating a cron job still binds back to the originating channel. + // Hidden from the LLM by JsonSchemaGenerator. + @Nullable ToolContext ctx) { if (agentName == null || agentName.isBlank()) { return "[错误] 请指定目标 Agent 名称。" + availableAgentsHint(); @@ -111,8 +119,11 @@ public class DelegateAgentTool { } Runnable stopRelay = hasParent ? registerRelay(childConversationId, parentConversationId, target.getName()) : null; - // Execute child agent - ChildResult result = runSingleChild(0, target, task, parentConversationId, childConversationId); + // 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, task, parentConversationId, childConversationId, parentOrigin); // Cleanup relay, then broadcast final result if (stopRelay != null) stopRelay.run(); @@ -133,7 +144,9 @@ public class DelegateAgentTool { Input is a JSON array: [{"agentName":"Agent名称","task":"任务描述"}, ...]""") public String delegateParallel( @ToolParam(description = "JSON array of tasks: [{\"agentName\":\"X\",\"task\":\"Y\"}, ...]") - String tasksJson) { + String tasksJson, + // RFC-063r §2.5 改动点 5: hidden from LLM, used to inherit ChatOrigin into children. + @Nullable ToolContext ctx) { // 1. Parse task list List> tasks; @@ -206,9 +219,14 @@ public class DelegateAgentTool { long startTime = System.currentTimeMillis(); Map> futures = new LinkedHashMap<>(); + // RFC-063r §2.5 改动点 5: capture parent origin once on this thread, + // then hand it to each child future — the worker virtual threads + // can't re-read the ToolContext (no parameter scope), so we close + // over the captured origin. + ChatOrigin parentOriginParallel = ChatOrigin.from(ctx); for (PreparedChild p : prepared) { CompletableFuture future = CompletableFuture.supplyAsync( - () -> runSingleChild(p.index, p.agent, p.task, parentConversationId, p.childConvId), + () -> runSingleChild(p.index, p.agent, p.task, parentConversationId, p.childConvId, parentOriginParallel), DELEGATION_EXECUTOR); // Broadcast per-child completion as soon as each child finishes @@ -385,11 +403,18 @@ public class DelegateAgentTool { * truncated length, making "blank_success" detection unreliable. */ private ChildResult runSingleChild(int taskIndex, AgentEntity target, String task, - String parentConversationId, String childConversationId) { + String parentConversationId, String childConversationId, + ChatOrigin parentOrigin) { DelegationContext.enter(parentConversationId, CHILD_DENIED_TOOLS); try { long startTime = System.currentTimeMillis(); - String rawResult = agentService.chat(target.getId(), task, childConversationId); + // RFC-063r §2.5 改动点 5: inherit parent origin, swap agentId + // so child reads correct identity from ToolContext while keeping + // channelId / channelTarget / workspace context intact. + ChatOrigin childOrigin = (parentOrigin != null ? parentOrigin : ChatOrigin.EMPTY) + .withAgent(target.getId()) + .withConversationId(childConversationId); + String rawResult = agentService.chat(target.getId(), task, childConversationId, childOrigin); long durationMs = System.currentTimeMillis() - startTime; // Measure lengths before truncation so ChildResult carries accurate metadata. return ChildResult.ofSuccess(taskIndex, target.getName(), rawResult, durationMs, diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/ImageGenerateTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/ImageGenerateTool.java index bdd9ed13..a15c4538 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/ImageGenerateTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/ImageGenerateTool.java @@ -2,8 +2,10 @@ package vip.mate.tool.builtin; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.model.ToolContext; 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 vip.mate.system.model.SystemSettingsDTO; import vip.mate.system.service.SystemSettingService; @@ -39,14 +41,16 @@ public class ImageGenerateTool { @ToolParam(description = "Aspect ratio: 1:1 / 16:9 / 9:16, default 1:1", required = false) String aspectRatio, @ToolParam(description = "Generation count (1-4), default 1", required = false) Integer count, @ToolParam(description = "Model name (optional)", required = false) String model, - @ToolParam(description = "Task ID to check status (for status action)", required = false) String taskId + @ToolParam(description = "Task ID to check status (for status action)", required = false) String taskId, + // RFC-063r §2.5: ToolContext is hidden from the LLM by JsonSchemaGenerator. + @Nullable ToolContext ctx ) { String normalizedAction = (action == null || action.isBlank()) ? "generate" : action.trim().toLowerCase(); return switch (normalizedAction) { case "list" -> handleListAction(); - case "status" -> handleStatusAction(taskId); - default -> handleGenerateAction(prompt, size, aspectRatio, count, model); + case "status" -> handleStatusAction(taskId, ctx); + default -> handleGenerateAction(prompt, size, aspectRatio, count, model, ctx); }; } @@ -84,8 +88,8 @@ public class ImageGenerateTool { // ==================== action=status ==================== - private String handleStatusAction(String taskId) { - String conversationId = ToolExecutionContext.conversationId(); + private String handleStatusAction(String taskId, @Nullable ToolContext ctx) { + String conversationId = ToolExecutionContext.conversationId(ctx); if (taskId != null && !taskId.isBlank()) { AsyncTaskInfo info = imageGenerationService.checkTaskStatus(taskId); @@ -117,9 +121,9 @@ public class ImageGenerateTool { // ==================== action=generate ==================== private String handleGenerateAction(String prompt, String size, String aspectRatio, - Integer count, String model) { - String conversationId = ToolExecutionContext.conversationId(); - String username = ToolExecutionContext.username(); + Integer count, String model, @Nullable ToolContext ctx) { + String conversationId = ToolExecutionContext.conversationId(ctx); + String username = ToolExecutionContext.username(ctx); if (conversationId == null || conversationId.isBlank()) { return "错误：无法获取当前会话信息，请重试"; diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/ReadFileTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/ReadFileTool.java index 744a5654..79a25985 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/ReadFileTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/ReadFileTool.java @@ -3,8 +3,10 @@ package vip.mate.tool.builtin; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.model.ToolContext; 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 java.io.BufferedReader; @@ -57,7 +59,9 @@ public class ReadFileTool { public String read_file( @ToolParam(description = "Absolute or relative file path") String filePath, @ToolParam(description = "Start line number (1-based, inclusive). Omit to start from line 1", required = false) Integer startLine, - @ToolParam(description = "End line number (1-based, inclusive). Omit to read to EOF or truncation limit", required = false) Integer endLine) { + @ToolParam(description = "End line number (1-based, inclusive). Omit to read to EOF or truncation limit", required = false) Integer endLine, + // RFC-063r §2.5: hidden from LLM by JsonSchemaGenerator. + @Nullable ToolContext ctx) { JSONObject result = new JSONObject(); result.set("filePath", filePath); @@ -65,7 +69,9 @@ public class ReadFileTool { try { Path path; try { - path = vip.mate.tool.guard.WorkspacePathGuard.validatePath(filePath); + // RFC-063r §2.5: forward ToolContext so workspace boundary + // honors ChatOrigin.workspaceBasePath when available. + path = vip.mate.tool.guard.WorkspacePathGuard.validatePath(filePath, ctx); } catch (IllegalArgumentException e) { // Sandbox rejected the literal path. The LLM may have hallucinated // a Linux-style path (e.g. /app/Dockerfile) for a chat-upload that diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/ToolExecutionContext.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/ToolExecutionContext.java index 6e64553c..1f199b09 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/ToolExecutionContext.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/ToolExecutionContext.java @@ -1,11 +1,21 @@ package vip.mate.tool.builtin; +import org.springframework.ai.chat.model.ToolContext; +import org.springframework.lang.Nullable; +import vip.mate.agent.context.ChatOrigin; + /** * 工具执行上下文 — 通过 ThreadLocal 向 @Tool 方法传递执行环境信息 *

* 在 ToolExecutionExecutor.executeSingleTool() 中 set，在 finally 中 clear。 * 视频生成等需要知道 conversationId 的工具从此处获取。 * + *

RFC-063r §2.5 兼容期：执行器会同时填充本 ThreadLocal 和 Spring AI 的 + * {@link ToolContext}（携带 {@link ChatOrigin}）。优先读 ToolContext 的工具 + * 调用 {@link #conversationId(ToolContext)} / {@link #username(ToolContext)} + * / {@link #workspaceBasePath(ToolContext)} 等三参重载即可——传入 ctx 不为 + * null 时优先返回 origin 的字段，否则回退到 ThreadLocal。 + * * @author MateClaw Team */ public final class ToolExecutionContext { @@ -47,4 +57,34 @@ public final class ToolExecutionContext { USERNAME.remove(); WORKSPACE_BASE_PATH.remove(); } + + // ===== RFC-063r §2.5: ToolContext-aware accessors ===== + // + // Preferred over the parameter-less variants: read from the explicit + // Spring AI ToolContext (carries ChatOrigin) when available, otherwise + // fall back to the legacy ThreadLocal so legacy paths keep working. + + public static String conversationId(@Nullable ToolContext ctx) { + if (ctx != null) { + String v = ChatOrigin.from(ctx).conversationId(); + if (v != null && !v.isEmpty()) return v; + } + return CONVERSATION_ID.get(); + } + + public static String username(@Nullable ToolContext ctx) { + if (ctx != null) { + String v = ChatOrigin.from(ctx).requesterId(); + if (v != null && !v.isEmpty()) return v; + } + return USERNAME.get(); + } + + public static String workspaceBasePath(@Nullable ToolContext ctx) { + if (ctx != null) { + String v = ChatOrigin.from(ctx).workspaceBasePath(); + if (v != null && !v.isBlank()) return v; + } + return WORKSPACE_BASE_PATH.get(); + } } diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/VideoGenerateTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/VideoGenerateTool.java index 4cde63d8..7363f682 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/VideoGenerateTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/VideoGenerateTool.java @@ -2,8 +2,10 @@ package vip.mate.tool.builtin; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.model.ToolContext; 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 vip.mate.system.model.SystemSettingsDTO; import vip.mate.system.service.SystemSettingService; @@ -45,15 +47,18 @@ public class VideoGenerateTool { @ToolParam(description = "视频时长（秒），如 5 或 10，默认 5", required = false) Integer duration, @ToolParam(description = "参考图片 URL（图生视频模式）", required = false) String imageUrl, @ToolParam(description = "指定模型名称（可选）", required = false) String model, - @ToolParam(description = "查询指定任务 ID 的状态（status 模式时使用）", required = false) String taskId + @ToolParam(description = "查询指定任务 ID 的状态（status 模式时使用）", required = false) String taskId, + // RFC-063r §2.5: ToolContext is auto-injected by Spring AI MethodToolCallback + // and explicitly skipped by JsonSchemaGenerator — never visible to the LLM. + @Nullable ToolContext ctx ) { // 路由 action String normalizedAction = (action == null || action.isBlank()) ? "generate" : action.trim().toLowerCase(); return switch (normalizedAction) { case "list" -> handleListAction(); - case "status" -> handleStatusAction(taskId); - default -> handleGenerateAction(prompt, aspectRatio, duration, imageUrl, model); + case "status" -> handleStatusAction(taskId, ctx); + default -> handleGenerateAction(prompt, aspectRatio, duration, imageUrl, model, ctx); }; } @@ -92,8 +97,8 @@ public class VideoGenerateTool { // ==================== action=status ==================== - private String handleStatusAction(String taskId) { - String conversationId = ToolExecutionContext.conversationId(); + private String handleStatusAction(String taskId, @Nullable ToolContext ctx) { + String conversationId = ToolExecutionContext.conversationId(ctx); // 指定 taskId 查询 if (taskId != null && !taskId.isBlank()) { @@ -124,9 +129,9 @@ public class VideoGenerateTool { // ==================== action=generate ==================== private String handleGenerateAction(String prompt, String aspectRatio, Integer duration, - String imageUrl, String model) { - String conversationId = ToolExecutionContext.conversationId(); - String username = ToolExecutionContext.username(); + String imageUrl, String model, @Nullable ToolContext ctx) { + String conversationId = ToolExecutionContext.conversationId(ctx); + String username = ToolExecutionContext.username(ctx); if (conversationId == null || conversationId.isBlank()) { return "错误：无法获取当前会话信息，请重试"; diff --git a/mateclaw-server/src/main/java/vip/mate/tool/guard/WorkspacePathGuard.java b/mateclaw-server/src/main/java/vip/mate/tool/guard/WorkspacePathGuard.java index 0b2d8b51..d6112d7d 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/guard/WorkspacePathGuard.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/guard/WorkspacePathGuard.java @@ -1,6 +1,9 @@ package vip.mate.tool.guard; import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.model.ToolContext; +import org.springframework.lang.Nullable; +import vip.mate.agent.context.ChatOrigin; import vip.mate.tool.builtin.ToolExecutionContext; import java.io.IOException; @@ -33,9 +36,19 @@ public final class WorkspacePathGuard { * @throws IllegalArgumentException 路径不在允许范围内 */ public static Path validatePath(String rawPath) { + return validatePath(rawPath, null); + } + + /** + * RFC-063r §2.5: ToolContext-aware overload. Reads the workspace base path + * from the explicit {@link ChatOrigin} when present; falls back to the + * legacy {@link ToolExecutionContext} ThreadLocal during the PR-1 + * transition window. + */ + public static Path validatePath(String rawPath, @Nullable ToolContext ctx) { Path normalized = Paths.get(rawPath).toAbsolutePath().normalize(); - String basePath = ToolExecutionContext.workspaceBasePath(); + String basePath = resolveBasePath(ctx); if (basePath == null || basePath.isBlank()) { return normalized; // 未配置活动目录，不限制 } @@ -72,10 +85,35 @@ public final class WorkspacePathGuard { * @return 活动目录 Path，未配置时返回 null */ public static Path getWorkingDirectory() { - String basePath = ToolExecutionContext.workspaceBasePath(); + return getWorkingDirectory(null); + } + + /** + * RFC-063r §2.5: ToolContext-aware variant — prefer the explicit + * {@link ChatOrigin} workspaceBasePath when available. + */ + public static Path getWorkingDirectory(@Nullable ToolContext ctx) { + String basePath = resolveBasePath(ctx); if (basePath == null || basePath.isBlank()) { return null; } return Paths.get(basePath).toAbsolutePath().normalize(); } + + /** + * Resolve the active workspace base path. Order of preference: + *

    + *
  1. ChatOrigin from ToolContext (RFC-063r §2.5)
    2. + *
  2. Legacy {@link ToolExecutionContext} ThreadLocal (PR-1 transition)
    3. + *
+ */ + private static String resolveBasePath(@Nullable ToolContext ctx) { + if (ctx != null) { + ChatOrigin origin = ChatOrigin.from(ctx); + if (origin.workspaceBasePath() != null && !origin.workspaceBasePath().isBlank()) { + return origin.workspaceBasePath(); + } + } + return ToolExecutionContext.workspaceBasePath(); + } } diff --git a/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerateTool.java b/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerateTool.java index a02e48ef..e0e38498 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerateTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerateTool.java @@ -2,8 +2,10 @@ package vip.mate.tool.music; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.chat.model.ToolContext; 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 vip.mate.tool.builtin.ToolExecutionContext; @@ -23,9 +25,11 @@ public class MusicGenerateTool { public String music_generate( @ToolParam(description = "音乐风格/场景描述，如：'轻快的钢琴爵士乐'、'史诗电影配乐'、'欢快的流行歌曲'") String prompt, @ToolParam(description = "歌词文本（可选，不填则由 AI 生成或生成纯音乐）") String lyrics, - @ToolParam(description = "是否生成纯音乐（无人声），默认 false") Boolean instrumental) { + @ToolParam(description = "是否生成纯音乐（无人声），默认 false") Boolean instrumental, + // RFC-063r §2.5: hidden from LLM by JsonSchemaGenerator. + @Nullable ToolContext ctx) { - String conversationId = ToolExecutionContext.conversationId(); + String conversationId = ToolExecutionContext.conversationId(ctx); if (conversationId == null) { return "无法获取会话 ID"; } diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V57__cron_run_delivery_status.sql b/mateclaw-server/src/main/resources/db/migration/h2/V57__cron_run_delivery_status.sql new file mode 100644 index 00000000..f47a4d59 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V57__cron_run_delivery_status.sql @@ -0,0 +1,18 @@ +-- RFC-063r §2.9: cron run delivery state machine. +-- +-- delivery_status drives the SQL-CAS that prevents duplicate proactive sends +-- (replaces RFC-063 v1's Caffeine 30-min TTL idempotency). 4-state machine: +-- NONE — no delivery strategy (web-origin cron) +-- PENDING — claimed by AbstractCronResultDelivery#claimRun +-- DELIVERED — markDelivered after successful proactiveSend +-- NOT_DELIVERED — markNotDelivered (or stale-cleanup timeout) +-- +-- delivery_target VARCHAR(512) covers both IM userId (≤64) and Feishu +-- sessionWebhook URLs (~200-256). Index (delivery_status, started_at) is +-- the exact predicate of CronRunStaleCleanup.sweep(). + +ALTER TABLE mate_cron_job_run ADD COLUMN IF NOT EXISTS delivery_status VARCHAR(16) DEFAULT 'NONE' NOT NULL; +ALTER TABLE mate_cron_job_run ADD COLUMN IF NOT EXISTS delivery_target VARCHAR(512); +ALTER TABLE mate_cron_job_run ADD COLUMN IF NOT EXISTS delivery_error VARCHAR(500); + +CREATE INDEX IF NOT EXISTS idx_cron_run_pending_started ON mate_cron_job_run(delivery_status, started_at); diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V58__cron_channel_binding.sql b/mateclaw-server/src/main/resources/db/migration/h2/V58__cron_channel_binding.sql new file mode 100644 index 00000000..bcf4dfeb --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V58__cron_channel_binding.sql @@ -0,0 +1,12 @@ +-- RFC-063r §2.9: bind a cron job to its originating channel + delivery target. +-- +-- channel_id is the only field that needs an index (operations team queries +-- "all jobs delivering to channel X"). delivery_config is a JSON column — +-- targetId / threadId / accountId are persisted as a structured value via +-- MyBatis Plus JacksonTypeHandler so future delivery-target fields don't +-- require schema changes. + +ALTER TABLE mate_cron_job ADD COLUMN IF NOT EXISTS channel_id BIGINT; +ALTER TABLE mate_cron_job ADD COLUMN IF NOT EXISTS delivery_config TEXT; + +CREATE INDEX IF NOT EXISTS idx_cron_channel ON mate_cron_job(channel_id); diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V59__approval_chat_origin_snapshot.sql b/mateclaw-server/src/main/resources/db/migration/h2/V59__approval_chat_origin_snapshot.sql new file mode 100644 index 00000000..79266441 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V59__approval_chat_origin_snapshot.sql @@ -0,0 +1,6 @@ +-- RFC-063r §2.12: persist the originating ChatOrigin as a JSON snapshot on the +-- approval row. Used by ChannelMessageRouter.replayApprovedToolCall + web +-- ApprovalController so cross-process / cross-restart approval replays still +-- preserve the channel binding (Memento pattern). + +ALTER TABLE mate_tool_approval ADD COLUMN IF NOT EXISTS chat_origin TEXT; diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V57__cron_run_delivery_status.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V57__cron_run_delivery_status.sql new file mode 100644 index 00000000..4e19b522 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V57__cron_run_delivery_status.sql @@ -0,0 +1,22 @@ +-- RFC-063r §2.9: cron run delivery state machine (MySQL dialect). +-- See V57 H2 file for state-machine + design reasoning. +-- +-- MySQL has no `ADD COLUMN IF NOT EXISTS` / `CREATE INDEX IF NOT EXISTS` — +-- guard each statement via INFORMATION_SCHEMA + dynamic SQL (project pattern +-- previously used in V4/V19/V44). + +SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_cron_job_run' AND COLUMN_NAME = 'delivery_status'); +SET @s := IF(@c = 0, 'ALTER TABLE mate_cron_job_run ADD COLUMN delivery_status VARCHAR(16) NOT NULL DEFAULT ''NONE''', 'SELECT 1'); +PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_cron_job_run' AND COLUMN_NAME = 'delivery_target'); +SET @s := IF(@c = 0, 'ALTER TABLE mate_cron_job_run ADD COLUMN delivery_target VARCHAR(512)', 'SELECT 1'); +PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_cron_job_run' AND COLUMN_NAME = 'delivery_error'); +SET @s := IF(@c = 0, 'ALTER TABLE mate_cron_job_run ADD COLUMN delivery_error VARCHAR(500)', 'SELECT 1'); +PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_cron_job_run' AND INDEX_NAME = 'idx_cron_run_pending_started'); +SET @s := IF(@c = 0, 'CREATE INDEX idx_cron_run_pending_started ON mate_cron_job_run(delivery_status, started_at)', 'SELECT 1'); +PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt; diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V58__cron_channel_binding.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V58__cron_channel_binding.sql new file mode 100644 index 00000000..e310e3d3 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V58__cron_channel_binding.sql @@ -0,0 +1,13 @@ +-- RFC-063r §2.9: bind a cron job to its originating channel (MySQL dialect). + +SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_cron_job' AND COLUMN_NAME = 'channel_id'); +SET @s := IF(@c = 0, 'ALTER TABLE mate_cron_job ADD COLUMN channel_id BIGINT', 'SELECT 1'); +PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_cron_job' AND COLUMN_NAME = 'delivery_config'); +SET @s := IF(@c = 0, 'ALTER TABLE mate_cron_job ADD COLUMN delivery_config TEXT', 'SELECT 1'); +PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_cron_job' AND INDEX_NAME = 'idx_cron_channel'); +SET @s := IF(@c = 0, 'CREATE INDEX idx_cron_channel ON mate_cron_job(channel_id)', 'SELECT 1'); +PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt; diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V59__approval_chat_origin_snapshot.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V59__approval_chat_origin_snapshot.sql new file mode 100644 index 00000000..7311c020 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V59__approval_chat_origin_snapshot.sql @@ -0,0 +1,5 @@ +-- RFC-063r §2.12: persist ChatOrigin Memento snapshot on approval (MySQL dialect). + +SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_tool_approval' AND COLUMN_NAME = 'chat_origin'); +SET @s := IF(@c = 0, 'ALTER TABLE mate_tool_approval ADD COLUMN chat_origin TEXT', 'SELECT 1'); +PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt; diff --git a/mateclaw-ui/src/i18n/locales/en-US.ts b/mateclaw-ui/src/i18n/locales/en-US.ts index a9829ee6..330c6d43 100644 --- a/mateclaw-ui/src/i18n/locales/en-US.ts +++ b/mateclaw-ui/src/i18n/locales/en-US.ts @@ -1432,6 +1432,15 @@ export default { lastRun: 'Last Run', enabled: 'Enabled', actions: 'Actions', + // RFC-063r §2.14: list-page "Last Delivery" column + lastDelivery: 'Last Delivery', + }, + // RFC-063r §2.14: delivery_status state machine labels + lastDelivery: { + none: 'No delivery', + pending: 'In flight', + delivered: 'Delivered', + not_delivered: 'Failed', }, taskTypes: { text: 'Text Message', agent: 'Agent Goal' }, cronTypes: { hourly: 'Hourly', daily: 'Daily', weekly: 'Weekly', custom: 'Custom' }, diff --git a/mateclaw-ui/src/i18n/locales/zh-CN.ts b/mateclaw-ui/src/i18n/locales/zh-CN.ts index 0f5c1d2f..66fee430 100644 --- a/mateclaw-ui/src/i18n/locales/zh-CN.ts +++ b/mateclaw-ui/src/i18n/locales/zh-CN.ts @@ -1442,6 +1442,15 @@ export default { lastRun: '上次执行', enabled: '启用', actions: '操作', + // RFC-063r §2.14: 列表页"最近投递"列 + lastDelivery: '最近投递', + }, + // RFC-063r §2.14: delivery_status 状态机展示 + lastDelivery: { + none: '无投递', + pending: '投递中', + delivered: '已送达', + not_delivered: '投递失败', }, taskTypes: { text: '文字消息', agent: 'Agent 目标' }, cronTypes: { hourly: '每小时', daily: '每天', weekly: '每周', custom: '自定义' }, diff --git a/mateclaw-ui/src/types/index.ts b/mateclaw-ui/src/types/index.ts index 912ce111..54d834c4 100644 --- a/mateclaw-ui/src/types/index.ts +++ b/mateclaw-ui/src/types/index.ts @@ -786,4 +786,12 @@ export interface CronJob { lastRunTime?: string createTime?: string updateTime?: string + // RFC-063r §2.9 / §2.14: channel binding + most-recent delivery snapshot. + // channelId / deliveryConfig: round-trippable on create/update. + // lastDeliveryStatus / lastDeliveryError: read-only, populated by + // selectListWithDeliveryStatus / selectByIdWithDeliveryStatus on the backend. + channelId?: number | null + deliveryConfig?: { targetId?: string | null; threadId?: string | null; accountId?: string | null } | null + lastDeliveryStatus?: 'NONE' | 'PENDING' | 'DELIVERED' | 'NOT_DELIVERED' + lastDeliveryError?: string | null } diff --git a/mateclaw-ui/src/views/CronJobs.vue b/mateclaw-ui/src/views/CronJobs.vue index 19dc3b14..5065c701 100644 --- a/mateclaw-ui/src/views/CronJobs.vue +++ b/mateclaw-ui/src/views/CronJobs.vue @@ -23,6 +23,7 @@ {{ t('cronJobs.columns.name') }} {{ t('cronJobs.columns.cron') }} {{ t('tokenUsage.date') }} + {{ t('cronJobs.columns.lastDelivery') }} {{ t('cronJobs.columns.enabled') }} {{ t('cronJobs.columns.actions') }} @@ -53,6 +54,14 @@ {{ t('cronJobs.columns.lastRun') }}: {{ formatTime(job.lastRunTime) }} + + + + {{ t('cronJobs.lastDelivery.' + (job.lastDeliveryStatus || 'NONE').toLowerCase()) }} + +