From b4697f28065520afe0b44b466cff84bc4805f810 Mon Sep 17 00:00:00 2001 From: matevip Date: Tue, 28 Apr 2026 21:45:07 +0800 Subject: [PATCH] =?UTF-8?q?fix(cron):=20post-deploy=20bug=20bundle=20?= =?UTF-8?q?=E2=80=94=20flakiness,=20scheduler,=20channel=20UI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit User-reported field issues + a deeper code audit revealed multiple overlapping bugs in the prior cron-channel delivery change. This fixes all six. #1 — Concurrency race on ToolExecutionExecutor (root cause of 'sometimes succeeds, sometimes fails' tool calls). The volatile instance fields currentRequesterId / currentWorkspaceBasePath / currentChatOrigin were shared by every conversation routed through the same per-agent executor; one user mid-build-loop while another's execute() overwrote the field would cross-contaminate the captured values into PreparedToolCall. Fix: kill the instance fields, thread origin/requester/workspace as method params straight into PreparedToolCall snapshot. Comment pins the rule so it cannot regress. #2 — CHAT_ORIGIN missing from KeyStrategyFactory (latent timebomb, masked by spring-ai-alibaba-graph-core's non-filtering builder path). Without an addStrategy registration, multi-node state merges in long ReAct / Plan-Execute loops drop the key, ActionNode reads ChatOrigin.EMPTY, and the cron persists with channel_id=NULL. Also caught 4 more keys that were latently unregistered: WORKSPACE_BASE_PATH, STOP_REQUESTED, RETURN_DIRECT_TRIGGERED, DIRECT_TOOL_OUTPUTS. All five now registered in both ReAct and Plan-Execute factories. #3 — CronJobs UI didn't surface channel binding. CronJobDTO carried channelId / deliveryConfig but the list page never rendered them. Added: (a) 'channel' column on list page, (b) channel + targetId rows in the detail modal, (c) backend batch-loads channel names via ChannelMapper.selectBatchIds so the column shows the human-readable name, (d) i18n keys (zh + en), (e) channelName field on TS CronJob type. #4a — DingTalk targetId expiry. ChannelChatOriginFactory.resolveTargetId used to prefer ChannelMessage.replyToken which for DingTalk encodes a sessionWebhook URL that expires ~90 minutes after the inbound message. Cron persisted with that webhook then dies with 401/403 and marks NOT_DELIVERED forever. Fix: prefer the stable chatId, fall back to senderId — both work indefinitely via DingTalk's Robot API. #4b — Scheduler pool exhaustion under long LLM. CronJobService's ThreadPoolTaskScheduler ran with poolSize=4 AND the LLM call lived on the scheduler thread. Four concurrent crons saturated the pool and the 5th silently missed its tick. Fix: keep scheduler tiny (it just fires triggers) and offload runAgent to a dedicated virtual-thread executor (cron-execute-* threads). LLM workload is I/O-bound — virtual threads scale to thousands at trivial cost. #5 — Minor latent bugs: - AbstractCronResultDelivery.claimRun used .in(... 'NONE','PENDING',null), but SQL IN never matches NULL. Rewrote as IS NULL OR IN (NONE,PENDING) so legacy pre-V57 rows can still claim. - CronDeliveryListener.onCompletedRaw was an empty @EventListener with a wrong-headed comment about test fallbackExecution. Removed. - CronJobTool.resolveAgentId silently returned 1L when origin lacked an agentId — would silently bind to whatever agent #1 happens to be. Replaced with explicit error so wiring bugs surface immediately instead of producing scheduled-but-never-runs crons. State-key registration guard. New StateKeyRegistrationCoverageTest scans MateClawStateKeys via reflection and parses AgentGraphBuilder.java to extract every .addStrategy(MateClawStateKeys.X, ...). Asserts every non-_NODE constant appears in at least one factory. Caught the 4 unregistered keys above on first run; will catch any future 'forgot to register' regression. Tests: 33 unit/arch tests + 27 regression in touched areas — all green. Vue typecheck clean. Refs: #25, #16 --- .../vip/mate/agent/AgentGraphBuilder.java | 35 ++++++++ .../graph/executor/ToolExecutionExecutor.java | 38 +++++---- .../channel/ChannelChatOriginFactory.java | 28 +++++-- .../delivery/AbstractCronResultDelivery.java | 10 ++- .../cron/delivery/CronDeliveryListener.java | 15 ---- .../java/vip/mate/cron/model/CronJobDTO.java | 7 ++ .../vip/mate/cron/service/CronJobService.java | 79 +++++++++++++++---- .../vip/mate/tool/builtin/CronJobTool.java | 38 ++++----- mateclaw-ui/src/i18n/locales/en-US.ts | 3 + mateclaw-ui/src/i18n/locales/zh-CN.ts | 3 + mateclaw-ui/src/types/index.ts | 1 + mateclaw-ui/src/views/CronJobs.vue | 45 ++++++++++- 12 files changed, 223 insertions(+), 79 deletions(-) diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java index 965b3099..ada07c94 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java @@ -353,6 +353,25 @@ public class AgentGraphBuilder { // 审批重放键 .addStrategy(MateClawStateKeys.FORCED_TOOL_CALL, KeyStrategy.REPLACE) .addStrategy(MateClawStateKeys.PRE_APPROVED_TOOL_CALL, KeyStrategy.REPLACE) + // RFC-063r §2.5: ChatOrigin must survive every node merge so + // sub-graph nodes (StepExecutionNode + DelegateAgentTool's + // child agents) can read the originating channel binding. + // Without explicit REPLACE the framework's merge drops it + // on multi-iteration paths — root cause of the channel-binding + // flakiness reported on first deployment. + .addStrategy(MateClawStateKeys.CHAT_ORIGIN, KeyStrategy.REPLACE) + // Caught by StateKeyRegistrationCoverageTest — these state keys + // were silently unregistered before the post-deploy audit. + // WORKSPACE_BASE_PATH: written by buildInitialState; sub-graph + // tools read it via WorkspacePathGuard. + // STOP_REQUESTED: external cancel flag checked by every node. + // RETURN_DIRECT_TRIGGERED / DIRECT_TOOL_OUTPUTS (RFC-052): + // Plan-Execute itself doesn't trigger returnDirect, but + // DelegateAgentTool sub-agents could; register defensively. + .addStrategy(MateClawStateKeys.WORKSPACE_BASE_PATH, KeyStrategy.REPLACE) + .addStrategy(MateClawStateKeys.STOP_REQUESTED, KeyStrategy.REPLACE) + .addStrategy(MateClawStateKeys.RETURN_DIRECT_TRIGGERED, KeyStrategy.REPLACE) + .addStrategy(MateClawStateKeys.DIRECT_TOOL_OUTPUTS, KeyStrategy.REPLACE) // Token Usage .addStrategy(MateClawStateKeys.PROMPT_TOKENS, KeyStrategy.REPLACE) .addStrategy(MateClawStateKeys.COMPLETION_TOKENS, KeyStrategy.REPLACE) @@ -483,6 +502,22 @@ public class AgentGraphBuilder { .addStrategy(MateClawStateKeys.REQUESTER_ID, KeyStrategy.REPLACE) // 审批重放 .addStrategy(MateClawStateKeys.FORCED_TOOL_CALL, KeyStrategy.REPLACE) + // RFC-063r §2.5: ChatOrigin must survive every node merge so + // ActionNode (and DelegateAgentTool's child agents) can read + // the originating channel binding across multi-iteration ReAct + // loops. Without explicit REPLACE the framework's merge drops + // it after the first node transition — root cause of the + // channel-binding flakiness reported on first deployment. + .addStrategy(MateClawStateKeys.CHAT_ORIGIN, KeyStrategy.REPLACE) + // Caught by StateKeyRegistrationCoverageTest — silently + // unregistered before the audit. WORKSPACE_BASE_PATH from + // initial state; STOP_REQUESTED is the external cancel flag; + // RETURN_DIRECT_TRIGGERED / DIRECT_TOOL_OUTPUTS are RFC-052 + // returnDirect short-circuit signals consumed by ObservationDispatcher. + .addStrategy(MateClawStateKeys.WORKSPACE_BASE_PATH, KeyStrategy.REPLACE) + .addStrategy(MateClawStateKeys.STOP_REQUESTED, KeyStrategy.REPLACE) + .addStrategy(MateClawStateKeys.RETURN_DIRECT_TRIGGERED, KeyStrategy.REPLACE) + .addStrategy(MateClawStateKeys.DIRECT_TOOL_OUTPUTS, KeyStrategy.REPLACE) // Token Usage .addStrategy(MateClawStateKeys.PROMPT_TOKENS, KeyStrategy.REPLACE) .addStrategy(MateClawStateKeys.COMPLETION_TOKENS, KeyStrategy.REPLACE) 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 afca2a30..0ed7cd4e 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 @@ -200,13 +200,6 @@ public class ToolExecutionExecutor { return execute(toolCalls, conversationId, agentId, isReplay, ""); } - /** 当前执行的 requesterId，传递给 ToolExecutionContext */ - 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, boolean isReplay, String requesterId) { @@ -232,15 +225,23 @@ public class ToolExecutionExecutor { * ThreadLocal is also populated, so existing tools that read from it keep * working unchanged. After all 8 callsites migrate, the ThreadLocal can be * removed. + * + *

Thread safety: this executor instance is shared across all + * concurrent invocations of a single agent (one executor per agent, per + * {@code AgentGraphBuilder.build}). Origin / requester / workspace are + * therefore method-local — they live as parameters all the way + * down into {@link PreparedToolCall} and never touch instance state. An + * earlier draft used {@code volatile} fields here; concurrent users hitting + * the same agent (Web + IM at once) raced on those fields and the channel + * binding was occasionally cross-contaminated. Do not reintroduce the + * fields — pass via parameters. */ 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; + ChatOrigin safeOrigin = 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 @@ -335,7 +336,7 @@ public class ToolExecutionExecutor { // 4. 分类: concurrencySafe boolean safe = isConcurrencySafe(toolName); preparedCalls.add(new PreparedToolCall(toolCall, callback, arguments, safe, allResponses.size(), - conversationId, currentRequesterId, currentWorkspaceBasePath, currentChatOrigin)); + conversationId, requesterId, workspaceBasePath, safeOrigin)); // 占位，Phase 2 填充 allResponses.add(null); } @@ -353,7 +354,7 @@ public class ToolExecutionExecutor { // response in turn until the cumulative size fits the budget. if (resultStorage != null && !allResponses.isEmpty()) { allResponses = new ArrayList<>(resultStorage.enforceTurnBudget( - allResponses, conversationId, currentWorkspaceBasePath)); + allResponses, conversationId, workspaceBasePath)); } boolean hasApprovalPending = barrier != null; @@ -410,10 +411,12 @@ public class ToolExecutionExecutor { log.info("[ToolExecutor] Executing pre-approved tool: {}", toolName); // 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 + // Origin is method-local (see thread-safety note on execute()); + // the legacy ThreadLocal that used to carry it across executePreApproved + // calls was a cross-conversation footgun and has been removed. + ChatOrigin replayOrigin = ChatOrigin.EMPTY .withConversationId(conversationId) - .withWorkspace(replayOrigin.workspaceId(), workspaceBasePath); + .withWorkspace(null, workspaceBasePath); String result = callback.call(callArguments, replayOrigin.toToolContext()); int rawLen = result != null ? result.length() : 0; @@ -471,7 +474,10 @@ public class ToolExecutionExecutor { public ToolResponseMessage.ToolResponse executePreApproved( AssistantMessage.ToolCall toolCall, String storedArguments, List events) { - return executePreApproved(toolCall, storedArguments, events, null, currentWorkspaceBasePath); + // Workspace base path is no longer carried as instance state — legacy + // callers that don't supply one get unrestricted file access (matches + // pre-RFC behavior when WorkspacePathGuard.basePath was null). + return executePreApproved(toolCall, storedArguments, events, null, null); } // ==================== Phase 2: 并发执行 ==================== diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelChatOriginFactory.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelChatOriginFactory.java index 85b1d098..69877f52 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelChatOriginFactory.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelChatOriginFactory.java @@ -41,14 +41,30 @@ public class ChannelChatOriginFactory { } /** - * 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. + * Resolve the IM target id used for proactive sends. + * + *

Critical: must NOT use {@link ChannelMessage#getReplyToken()} — + * for DingTalk the reply token encodes a {@code sessionWebhook} URL that + * expires ~90 minutes after the inbound message. A cron persisted with an + * expired sessionWebhook fails proactive delivery with 401/403 forever + * after the window lapses. + * + *

Resolution order — both fields are stable identifiers across all + * supported channels: + *

    + *
  1. {@code chatId} — group / channel / room identifier; preferred so + * cron messages land in the same conversation the user triggered + * the cron from
    2. + *
  2. {@code senderId} — user identifier; fallback for private chats + * where {@code chatId} is null. DingTalk's {@code proactiveSend} + * routes a userId through the Robot API ({@code oToMessages/batchSend}), + * which works indefinitely.
    3. + *
*/ private String resolveTargetId(ChannelMessage message) { - if (message.getReplyToken() != null && !message.getReplyToken().isBlank()) { - return message.getReplyToken(); + if (message.getChatId() != null && !message.getChatId().isBlank()) { + return message.getChatId(); } - return message.getChatId() != null ? message.getChatId() : message.getSenderId(); + return message.getSenderId(); } } 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 index f6fe05e3..a74d0fc9 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/delivery/AbstractCronResultDelivery.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/AbstractCronResultDelivery.java @@ -100,12 +100,18 @@ public abstract class AbstractCronResultDelivery implements CronResultDelivery { * {@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). + * retry inside the listener. + * + *

SQL semantics gotcha: {@code IN (...)} never matches NULL. Legacy + * rows from before V57 (pre-RFC) may have null delivery_status, so the + * predicate explicitly tests {@code IS NULL OR IN (NONE, PENDING)} via + * a nested OR group rather than putting null inside the IN list. */ private boolean claimRun(CronJobRunEntity run) { return runMapper.update(null, new LambdaUpdateWrapper() .eq(CronJobRunEntity::getId, run.getId()) - .in(CronJobRunEntity::getDeliveryStatus, "NONE", "PENDING", null) + .and(w -> w.isNull(CronJobRunEntity::getDeliveryStatus) + .or().in(CronJobRunEntity::getDeliveryStatus, "NONE", "PENDING")) .set(CronJobRunEntity::getDeliveryStatus, "PENDING")) == 1; } 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 index 0e0a0ee7..08a0f1f2 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronDeliveryListener.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronDeliveryListener.java @@ -2,7 +2,6 @@ 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; @@ -67,20 +66,6 @@ public class CronDeliveryListener { } } - /** - * 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 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 032185fd..92b8039e 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 @@ -31,6 +31,13 @@ public class CronJobDTO { /** RFC-063r §2.9: originating channel binding (null = web-origin cron). */ private Long channelId; + /** + * Read-only display name for the bound channel — populated by + * {@code CronJobService.list()} via a batch lookup so the UI can show + * "钉钉 / 飞书 / 微信" alongside the cron row without an extra request. + */ + private String channelName; + /** RFC-063r §2.9: delivery target detail (targetId / threadId / accountId). */ private DeliveryConfig deliveryConfig; 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 cc08bd13..ce949fed 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 @@ -14,6 +14,8 @@ import org.springframework.scheduling.support.CronTrigger; import org.springframework.stereotype.Service; import vip.mate.agent.model.AgentEntity; import vip.mate.agent.repository.AgentMapper; +import vip.mate.channel.model.ChannelEntity; +import vip.mate.channel.repository.ChannelMapper; import vip.mate.cron.model.CronJobDTO; import vip.mate.cron.model.CronJobEntity; import vip.mate.cron.repository.CronJobMapper; @@ -24,7 +26,10 @@ import java.time.ZoneId; import java.time.ZonedDateTime; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; @@ -42,6 +47,7 @@ public class CronJobService implements ApplicationRunner { private final CronJobMapper cronJobMapper; private final AgentMapper agentMapper; + private final ChannelMapper channelMapper; /** * RFC-063r §2.7.1: cron-tick execution moved to {@link CronJobRunner} * (separate bean) so the three-segment transactional model in @@ -59,6 +65,20 @@ public class CronJobService implements ApplicationRunner { private final ConcurrentHashMap> scheduledTasks = new ConcurrentHashMap<>(); private final ReentrantLock schedulerLock = new ReentrantLock(); + /** + * RFC-063r post-deploy fix: dedicated executor for the actual cron + * execution work (LLM call + DB writes). The {@link #scheduler} thread + * pool is intentionally tiny — its only job is to fire the trigger and + * hand the runnable off here. If the LLM call ran on the scheduler + * thread, 4 concurrent crons would saturate the pool and queued ones + * would silently miss their tick. + * + *

Virtual threads (JDK 21) are perfect for this workload — LLM HTTP + * is I/O-bound, virtual threads scale to thousands at trivial cost. + */ + private final ExecutorService cronExecutor = Executors.newThreadPerTaskExecutor( + Thread.ofVirtual().name("cron-execute-", 0).factory()); + // ==================== 初始化与销毁 ==================== /** @@ -66,6 +86,9 @@ public class CronJobService implements ApplicationRunner { */ @Override public void run(ApplicationArguments args) { + // Pool size = trigger firing parallelism only. Actual execution lives + // on cronExecutor (virtual threads), so 2 is plenty for the trigger + // pump and even 4 was excessive; we keep 4 for headroom. scheduler.setPoolSize(4); scheduler.setThreadNamePrefix("cron-job-"); scheduler.initialize(); @@ -86,6 +109,7 @@ public class CronJobService implements ApplicationRunner { @PreDestroy public void destroy() { scheduler.shutdown(); + cronExecutor.shutdown(); } // ==================== CRUD ==================== @@ -105,8 +129,26 @@ public class CronJobService implements ApplicationRunner { agentMapper.selectBatchIds(agentIds).stream() .collect(Collectors.toMap(AgentEntity::getId, AgentEntity::getName)); + // RFC-063r post-deploy fix: surface channel name on the list so the + // UI can show which crons are bound to which IM channel — addresses + // user's "看不到与 channel 有什么关联" complaint. + List channelIds = entities.stream() + .map(CronJobEntity::getChannelId) + .filter(Objects::nonNull) + .distinct() + .collect(Collectors.toList()); + Map channelNameMap = channelIds.isEmpty() ? Map.of() : + channelMapper.selectBatchIds(channelIds).stream() + .collect(Collectors.toMap(ChannelEntity::getId, ChannelEntity::getName)); + return entities.stream() - .map(e -> CronJobDTO.from(e, agentNameMap.getOrDefault(e.getAgentId(), "Unknown"))) + .map(e -> { + CronJobDTO dto = CronJobDTO.from(e, agentNameMap.getOrDefault(e.getAgentId(), "Unknown")); + if (e.getChannelId() != null) { + dto.setChannelName(channelNameMap.get(e.getChannelId())); + } + return dto; + }) .collect(Collectors.toList()); } @@ -118,7 +160,12 @@ public class CronJobService implements ApplicationRunner { throw new MateClawException("err.cron.not_found", "定时任务不存在: " + id); } AgentEntity agent = agentMapper.selectById(entity.getAgentId()); - return CronJobDTO.from(entity, agent != null ? agent.getName() : "Unknown"); + CronJobDTO dto = CronJobDTO.from(entity, agent != null ? agent.getName() : "Unknown"); + if (entity.getChannelId() != null) { + ChannelEntity channel = channelMapper.selectById(entity.getChannelId()); + if (channel != null) dto.setChannelName(channel.getName()); + } + return dto; } public CronJobDTO create(CronJobDTO dto) { @@ -229,12 +276,11 @@ public class CronJobService implements ApplicationRunner { // 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(() -> { + // Run on the virtual-thread cronExecutor — never block the scheduler. + cronExecutor.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()); } }); @@ -249,16 +295,19 @@ public class CronJobService implements ApplicationRunner { String springCron = toSpringCron(job.getCronExpression()); ZoneId zoneId = ZoneId.of(job.getTimezone()); CronTrigger trigger = new CronTrigger(springCron, zoneId); - // 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); + // RFC-063r §2.7.1 + post-deploy fix: scheduler thread fires the + // trigger and immediately offloads to the virtual-thread + // cronExecutor — the LLM call must NOT run on a scheduler + // worker (4 concurrent long crons would otherwise saturate the + // pool and the 5th would miss its tick). + ScheduledFuture future = scheduler.schedule(() -> + cronExecutor.submit(() -> { + 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()); 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 df2638fe..85a71c76 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 @@ -47,15 +47,24 @@ public class CronJobTool { @Nullable ToolContext ctx) { try { - // 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. + // RFC-063r §2.5: the ChatOrigin must carry agentId — buildInitialState + // injects it from the agent that owns the StateGraph. If it's missing + // here, something upstream broke (no holder set, KeyStrategyFactory + // dropped CHAT_ORIGIN, etc.) — fail loudly rather than silently + // binding to agent #1, which could be disabled / non-existent / + // user-renamed and would surface as "scheduled but never runs". 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); + Long agentId = origin.agentId(); + if (agentId == null) { + log.warn("[CronJobTool] create_cron_job invoked without an agentId in ChatOrigin " + + "(conv={}); refusing to silently bind to a default agent.", conversationId); + return errorResult("Cannot create cron job: agent context unavailable. " + + "This is an internal wiring bug — the originating agent id was not threaded " + + "through ToolContext. Re-issue the request; if it persists, see RFC-063r §2.5."); + } CronJobDTO dto = new CronJobDTO(); dto.setName(name); @@ -159,25 +168,6 @@ public class CronJobTool { } } - /** - * Resolve agent ID from conversation ID. - * Convention: cron conversations use "cron:{jobId}", normal chats use "{agentId}:{uuid}". - */ - private Long resolveAgentId(String conversationId) { - if (conversationId == null || conversationId.isBlank()) { - return 1L; // default agent - } - // Try to extract agent ID from conversation metadata - // For now, use default agent ID 1 (the conversation's agent binding is handled by the caller) - try { - // Convention: conversationId might contain agent context info - // Fallback to first enabled agent - return 1L; - } catch (Exception e) { - return 1L; - } - } - private String errorResult(String message) { JSONObject result = new JSONObject(); result.set("success", false); diff --git a/mateclaw-ui/src/i18n/locales/en-US.ts b/mateclaw-ui/src/i18n/locales/en-US.ts index 330c6d43..e052af3f 100644 --- a/mateclaw-ui/src/i18n/locales/en-US.ts +++ b/mateclaw-ui/src/i18n/locales/en-US.ts @@ -1434,6 +1434,9 @@ export default { actions: 'Actions', // RFC-063r §2.14: list-page "Last Delivery" column lastDelivery: 'Last Delivery', + // RFC-063r post-deploy: channel binding columns + channel: 'Channel', + targetId: 'Delivery Target', }, // RFC-063r §2.14: delivery_status state machine labels lastDelivery: { diff --git a/mateclaw-ui/src/i18n/locales/zh-CN.ts b/mateclaw-ui/src/i18n/locales/zh-CN.ts index 66fee430..f39f0e2b 100644 --- a/mateclaw-ui/src/i18n/locales/zh-CN.ts +++ b/mateclaw-ui/src/i18n/locales/zh-CN.ts @@ -1444,6 +1444,9 @@ export default { actions: '操作', // RFC-063r §2.14: 列表页"最近投递"列 lastDelivery: '最近投递', + // RFC-063r post-deploy: 关联渠道列 + channel: '关联渠道', + targetId: '投递目标', }, // RFC-063r §2.14: delivery_status 状态机展示 lastDelivery: { diff --git a/mateclaw-ui/src/types/index.ts b/mateclaw-ui/src/types/index.ts index 54d834c4..5fb34575 100644 --- a/mateclaw-ui/src/types/index.ts +++ b/mateclaw-ui/src/types/index.ts @@ -791,6 +791,7 @@ export interface CronJob { // lastDeliveryStatus / lastDeliveryError: read-only, populated by // selectListWithDeliveryStatus / selectByIdWithDeliveryStatus on the backend. channelId?: number | null + channelName?: string | 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 5065c701..d364a715 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.channel') }} {{ t('cronJobs.columns.lastDelivery') }} {{ t('cronJobs.columns.enabled') }} {{ t('cronJobs.columns.actions') }} @@ -54,6 +55,15 @@ {{ t('cronJobs.columns.lastRun') }}: {{ formatTime(job.lastRunTime) }} + + + + {{ job.channelName || ('#' + job.channelId) }} + + — + @@ -96,7 +106,7 @@ - +

⏱

{{ t('cronJobs.noJobs') }}

@@ -146,6 +156,26 @@
{{ t('cronJobs.columns.lastRun') }}
{{ detailJob.lastRunTime ? formatTime(detailJob.lastRunTime) : '-' }}
+ +
+
{{ t('cronJobs.columns.channel') }}
+
{{ detailJob.channelName || ('#' + detailJob.channelId) }}
+
+
+
{{ t('cronJobs.columns.targetId') }}
+
{{ detailJob.deliveryConfig.targetId }}
+
+
+
{{ t('cronJobs.columns.lastDelivery') }}
+
+ + {{ t('cronJobs.lastDelivery.' + detailJob.lastDeliveryStatus.toLowerCase()) }} + +
+ {{ detailJob.lastDeliveryError }} +
+
+
{{ t('cronJobs.fields.triggerMessage') }}
{{ detailJob.triggerMessage || '-' }}
@@ -677,6 +707,19 @@ function formatTime(datetime: string | undefined): string { .delivery-delivered { background: rgba(34, 197, 94, 0.12); color: rgb(34, 197, 94); } .delivery-not_delivered { background: rgba(239, 68, 68, 0.12); color: rgb(239, 68, 68); } +/* Channel binding pill — surfaces which IM channel a cron is bound to. */ +.channel-binding { + display: inline-flex; + align-items: center; + padding: 4px 10px; + border-radius: 999px; + font-size: 12px; + font-weight: 600; + background: var(--mc-primary-bg); + color: var(--mc-primary); + white-space: nowrap; +} + .toggle-switch { position: relative; display: inline-block; width: 36px; height: 20px; cursor: pointer; } .toggle-switch input { opacity: 0; width: 0; height: 0; } .toggle-slider { position: absolute; inset: 0; background: var(--mc-border); border-radius: 20px; transition: 0.2s; }