mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(cron): post-deploy bug bundle — flakiness, scheduler, channel UI
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
This commit is contained in:
matevip
parent
4011050ceb
commit
b4697f2806
@ -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)
|
||||
|
||||
@ -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<AssistantMessage.ToolCall> 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.
|
||||
*
|
||||
* <p><b>Thread safety</b>: 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 <em>method-local</em> — 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<AssistantMessage.ToolCall> 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<ToolResponseMessage.ToolResponse> allResponses = new ArrayList<>();
|
||||
List<GraphEventPublisher.GraphEvent> 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<GraphEventPublisher.GraphEvent> 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: 并发执行 ====================
|
||||
|
||||
@ -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.
|
||||
*
|
||||
* <p><b>Critical</b>: 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.
|
||||
*
|
||||
* <p>Resolution order — both fields are stable identifiers across all
|
||||
* supported channels:
|
||||
* <ol>
|
||||
* <li>{@code chatId} — group / channel / room identifier; preferred so
|
||||
* cron messages land in the same conversation the user triggered
|
||||
* the cron from</li>
|
||||
* <li>{@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.</li>
|
||||
* </ol>
|
||||
*/
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
@ -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.
|
||||
*
|
||||
* <p>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<CronJobRunEntity>()
|
||||
.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;
|
||||
}
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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;
|
||||
|
||||
|
||||
@ -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<Long, ScheduledFuture<?>> 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.
|
||||
*
|
||||
* <p>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<Long> channelIds = entities.stream()
|
||||
.map(CronJobEntity::getChannelId)
|
||||
.filter(Objects::nonNull)
|
||||
.distinct()
|
||||
.collect(Collectors.toList());
|
||||
Map<Long, String> 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());
|
||||
|
||||
@ -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);
|
||||
|
||||
@ -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: {
|
||||
|
||||
@ -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: {
|
||||
|
||||
@ -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
|
||||
|
||||
@ -23,6 +23,7 @@
|
||||
<th>{{ t('cronJobs.columns.name') }}</th>
|
||||
<th>{{ t('cronJobs.columns.cron') }}</th>
|
||||
<th>{{ t('tokenUsage.date') }}</th>
|
||||
<th>{{ t('cronJobs.columns.channel') }}</th>
|
||||
<th>{{ t('cronJobs.columns.lastDelivery') }}</th>
|
||||
<th>{{ t('cronJobs.columns.enabled') }}</th>
|
||||
<th>{{ t('cronJobs.columns.actions') }}</th>
|
||||
@ -54,6 +55,15 @@
|
||||
<span v-if="job.lastRunTime" class="time-subtext" :title="`${t('cronJobs.columns.lastRun')}: ${formatTime(job.lastRunTime)}`">{{ t('cronJobs.columns.lastRun') }}: {{ formatTime(job.lastRunTime) }}</span>
|
||||
</div>
|
||||
</td>
|
||||
<td>
|
||||
<!-- Channel binding visibility — RFC-063r post-deploy fix.
|
||||
Cron created from web (no channelId) shows "—". -->
|
||||
<span v-if="job.channelId" class="channel-binding"
|
||||
:title="job.deliveryConfig?.targetId ? t('cronJobs.columns.targetId') + ': ' + job.deliveryConfig.targetId : ''">
|
||||
{{ job.channelName || ('#' + job.channelId) }}
|
||||
</span>
|
||||
<span v-else class="time-empty">—</span>
|
||||
</td>
|
||||
<td>
|
||||
<!-- RFC-063r §2.14: most-recent delivery status badge.
|
||||
hover surfaces the error detail when not delivered. -->
|
||||
@ -96,7 +106,7 @@
|
||||
</td>
|
||||
</tr>
|
||||
<tr v-if="store.jobs.length === 0">
|
||||
<td colspan="6" class="empty-row">
|
||||
<td colspan="7" class="empty-row">
|
||||
<div class="empty-state">
|
||||
<span class="empty-icon">⏱</span>
|
||||
<p>{{ t('cronJobs.noJobs') }}</p>
|
||||
@ -146,6 +156,26 @@
|
||||
<div class="detail-label">{{ t('cronJobs.columns.lastRun') }}</div>
|
||||
<div class="detail-value">{{ detailJob.lastRunTime ? formatTime(detailJob.lastRunTime) : '-' }}</div>
|
||||
</div>
|
||||
<!-- RFC-063r post-deploy: channel binding visibility in detail page -->
|
||||
<div class="detail-item" v-if="detailJob.channelId">
|
||||
<div class="detail-label">{{ t('cronJobs.columns.channel') }}</div>
|
||||
<div class="detail-value">{{ detailJob.channelName || ('#' + detailJob.channelId) }}</div>
|
||||
</div>
|
||||
<div class="detail-item" v-if="detailJob.deliveryConfig?.targetId">
|
||||
<div class="detail-label">{{ t('cronJobs.columns.targetId') }}</div>
|
||||
<div class="detail-value mono">{{ detailJob.deliveryConfig.targetId }}</div>
|
||||
</div>
|
||||
<div class="detail-item" v-if="detailJob.lastDeliveryStatus && detailJob.lastDeliveryStatus !== 'NONE'">
|
||||
<div class="detail-label">{{ t('cronJobs.columns.lastDelivery') }}</div>
|
||||
<div class="detail-value">
|
||||
<span class="delivery-badge" :class="'delivery-' + detailJob.lastDeliveryStatus.toLowerCase()">
|
||||
{{ t('cronJobs.lastDelivery.' + detailJob.lastDeliveryStatus.toLowerCase()) }}
|
||||
</span>
|
||||
<div v-if="detailJob.lastDeliveryError" class="detail-subvalue" style="color: rgb(239,68,68);">
|
||||
{{ detailJob.lastDeliveryError }}
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
<div class="detail-item detail-item-full" v-if="detailJob.taskType === 'text'">
|
||||
<div class="detail-label">{{ t('cronJobs.fields.triggerMessage') }}</div>
|
||||
<div class="detail-value detail-block">{{ detailJob.triggerMessage || '-' }}</div>
|
||||
@ -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; }
|
||||
|
||||
Loading…
Reference in New Issue
Block a user