tasks = $tasks; $this->chat = $chat; $this->runDao = $runDao; } /** * 执行单次任务运行(队列 Job 唯一入口) * * @param int $runId 运行记录 ID(必须处于 running 态,否则直接忽略——状态即锁,天然防重复投递) * @throws AdminException 运行记录或任务不存在 */ public function run(int $runId): void { $run = $this->tasks->getRun($runId); if ((string)$run['status'] !== self::STATUS_RUNNING) { return; } $task = $this->tasks->getTask((int)$run['task_id'], 0, false); $startedAt = time(); // 数据库 CAS 领取:仅 running 态可置 started_at,并发重复投递只有一个赢家 $claimed = $this->runDao->claimRunning($runId, $startedAt); if (!$claimed) { return; } $adminId = (int)$task['admin_id']; $instruction = (string)$task['instruction']; // 独立审计会话:每次运行新建,标题带任务名与日期,工具调用全过程落会话消息可回放 $conversationId = app()->make(AiAgentServices::class)->createConversation( 0, $adminId, '【自动任务】' . (string)$task['name'] . ' ' . date('m-d H:i', $startedAt) ); $this->tasks->updateRun($runId, ['conversation_id' => $conversationId]); $timeout = max(60, (int)$task['timeout']); $autonomy = (int)$task['autonomy'] === AgentTaskServices::AUTONOMY_AUTO ? AgentTaskServices::AUTONOMY_AUTO : AgentTaskServices::AUTONOMY_READONLY; // getTask() 返回原始库记录,JSON 字段是字符串,必须先解码(强转 (array) 会得到单元素字符串数组) $serverIds = $this->decodeArray($task['server_ids'] ?? ''); $skillKeys = $this->decodeArray($task['skill_keys'] ?? ''); $notifyChannels = $this->decodeArray($task['notify_channels'] ?? ''); $runtimeOptions = [ 'unattended' => true, 'autonomy' => $autonomy, 'max_steps' => min(30, max(3, (int)$task['max_steps'] ?: self::DEFAULT_MAX_STEPS)), 'deadline' => $startedAt + $timeout, // 任务登记的通知邮箱:无人值守邮件发送仅允许发往此处与系统白名单,防止业务数据外发 'notify_email' => (string)$task['notify_email'], ]; $summary = ''; $error = ''; $usage = []; $toolCallCount = 0; $status = self::STATUS_SUCCESS; // 交付目标显式注入:任务指令可能只写「发送到指定邮箱」而未带具体地址, // 无人值守下模型无法追问,必须把任务配置的收件目标透传给模型,防止编造收件人 $instruction .= "\n\n[运行时配置] 收件邮箱(交付结果必须发送到此地址):" . ((string)$task['notify_email'] !== '' ? $task['notify_email'] : '未配置'); try { $result = $this->chat->chat( $adminId, $conversationId, $instruction, (int)$task['model_id'], $serverIds, $skillKeys, $runtimeOptions ); $summary = trim((string)($result['content'] ?? '')); $usage = (array)($result['usage'] ?? []); $toolCalls = (array)($result['tool_calls'] ?? []); $toolCallCount = count($toolCalls); // 软超时判定:循环因 deadline 提前收尾时,产出标注 timeout 而非 failed(结论仍有效) if (time() >= $startedAt + $timeout) { $status = self::STATUS_TIMEOUT; } // 交付校验:模型未在循环内成功调用交付工具时,按任务通知配置兜底投递 if (!$this->hasDelivered($toolCalls)) { $delivered = $this->deliverFallback( (string)$task['name'], $summary !== '' ? $summary : '(本次运行未产出正文)', $status, $notifyChannels, (string)$task['notify_email'] ); $error = $delivered ? '' : '模型未调用交付工具且兜底投递未完成(请检查通知渠道配置)'; } } catch (\Throwable $e) { $status = self::STATUS_FAILED; $error = mb_substr($e->getMessage(), 0, 900); Log::error('[AgentTask] 任务运行失败: ' . $e->getMessage(), [ 'run_id' => $runId, 'task_id' => (int)$task['id'], 'conversation_id' => $conversationId, ]); } // 失败/超时时按通知配置发送失败提醒(与结果交付共用通道,独立 try-catch 不影响终态落库) if (in_array($status, [self::STATUS_FAILED, self::STATUS_TIMEOUT], true)) { try { $this->deliverFallback( (string)$task['name'], $summary !== '' ? $summary : ('运行' . ($status === self::STATUS_FAILED ? '失败' : '超时') . ':' . $error), $status, $notifyChannels, (string)$task['notify_email'] ); } catch (\Throwable $e) { Log::error('[AgentTask] 失败提醒发送异常: ' . $e->getMessage(), ['run_id' => $runId]); } } $this->tasks->updateRun($runId, [ 'status' => $status, 'summary' => mb_substr($summary, 0, 60000), 'error' => mb_substr($error, 0, 1000), 'tool_calls' => $toolCallCount, 'usage' => json_encode($usage, JSON_UNESCAPED_UNICODE), 'finished_at' => time(), ]); } /** * JSON 字符串/数组安全解码为数组(任务原始记录中的 JSON 字段专用) * @param mixed $value JSON 字符串或已是数组 * @return array */ private function decodeArray(mixed $value): array { if (is_array($value)) { return $value; } $decoded = json_decode((string)$value, true); return is_array($decoded) ? array_values($decoded) : []; } /** * 校验工具轨迹中是否存在成功交付(交付类工具无错误结果即视为已交付) * @param array $toolCalls 工具调用轨迹(chat() 返回的 tool_calls:name/arguments/result) * @return bool */ private function hasDelivered(array $toolCalls): bool { foreach ($toolCalls as $call) { $name = (string)($call['name'] ?? ''); if (in_array($name, McpToolService::DELIVERY_CONFIRM_TOOLS, true) && is_array($call['result'] ?? null) && empty($call['result']['error'])) { return true; } } return false; } /** * 兜底投递:按任务通知配置将文本直接投递到邮件/企业微信机器人 * * 不经 Notice 模板管道(AI 产出即成品正文,模板渲染无意义且依赖后台预配模板), * 直连渠道发送能力;单渠道失败互不影响。 * * @param string $taskName 任务名(用于标题) * @param string $content 正文 * @param string $status 运行状态(用于标题后缀) * @param array $channels 通知渠道列表(is_email/is_ent_wechat) * @param string $email 收件邮箱 * @return bool 任一渠道投递成功即 true */ private function deliverFallback(string $taskName, string $content, string $status, array $channels, string $email): bool { $statusText = match ($status) { self::STATUS_FAILED => '失败', self::STATUS_TIMEOUT => '超时(部分结论)', default => '完成', }; $subject = "【AI 任务】{$taskName} 运行{$statusText}"; $delivered = false; if (in_array('is_email', $channels, true) && filter_var($email, FILTER_VALIDATE_EMAIL)) { try { $delivered = app()->make(EmailChannel::class)->sendEmail($email, $subject, $content) || $delivered; } catch (\Throwable $e) { Log::error('[AgentTask] 兜底邮件投递失败: ' . $e->getMessage(), ['task' => $taskName, 'to' => $email]); } } if (in_array('is_ent_wechat', $channels, true)) { try { $delivered = $this->sendEntWechat($subject, $content) || $delivered; } catch (\Throwable $e) { Log::error('[AgentTask] 兜底企业微信投递失败: ' . $e->getMessage(), ['task' => $taskName]); } } return $delivered; } /** * 企业微信群机器人推送(Webhook 取系统配置 aiagent_qywechat_webhook,与 MCP 发送工具同源) * @param string $title 消息标题 * @param string $content 消息正文 * @return bool */ private function sendEntWechat(string $title, string $content): bool { $webhook = trim((string)sys_config('aiagent_qywechat_webhook', '')); if ($webhook === '') { Log::error('[AgentTask] 企业微信机器人 Webhook 未配置,跳过兜底投递'); return false; } $markdown = "## {$title}\n" . $content; Http::postRequest($webhook, json_encode([ 'msgtype' => 'markdown', 'markdown' => ['content' => $markdown], ], JSON_UNESCAPED_UNICODE)); return true; } }