UHD_Media_Tool/apps/server/src/modules/schedules/agent-checks.ts

import {
  AgentCheckConfig,
  type AgentCheckConfigInput,
  CAPABILITY_CATALOG,
  type KnownCapabilityId,
} from "@umt/contracts";
import { sql } from "@umt/db";
import { recoveryGrants } from "../agent/investigate.ts";
import { INVESTIGATE_CAPABILITIES, PLAN_ONLY_KINDS, RECOVER_CAPABILITIES } from "../agent/runner.ts";
import { startRun } from "../agent/service.ts";
import type { ServerContext } from "../types.ts";
import type { JobRunContext, JobRunResult, ScheduledJobDef } from "./registry.ts";

// Agent 巡检(kind = agent_check):到点先按业务表判断触发条件,满足才渲染目标模板、入队 runAgent
// (trigger=schedule,能力白名单 + 预算;同一时间槽 dedupe_key 相同只建一个 run)。
// allowRecover 走与 autoInvestigate 相同的授权路径:有覆盖恢复动作的有效持续授权才用 execute 模式,
// 否则降级只读;实际每次调用仍由 authorize 按授权判定。run 结束后调度器回填结果(见 scheduler.harvest)。

export const HOUR = 3600_000;

/** 内置巡检模板(默认关闭) */
export const BUILTIN_AGENT_CHECKS: {
  key: string;
  name: string;
  description: string;
  everyMs: number;
  config: AgentCheckConfigInput;
}[] = [
  {
    key: "check.pending",
    name: "待处理巡检",
    description: "检查待处理事项并给出建议（只读）",
    everyMs: 6 * HOUR,
    config: {
      goalTemplate:
        "巡检待处理({{now}})：有 {{openDecisions}} 条未关闭的待处理事项。逐条查看证据，按紧急程度排序，" +
        "给出处理建议和理由。只读，不做任何变更。",
      capabilityAllowlist: [...INVESTIGATE_CAPABILITIES],
      budget: { maxToolCalls: 20, maxDurationSec: 300 },
      triggerCondition: "open_decisions",
    },
  },
  {
    key: "check.downloads",
    name: "下载诊断",
    description: "检查无进展或失败的下载并诊断；开启恢复后在授权内重试或重新校验",
    everyMs: 2 * HOUR,
    config: {
      goalTemplate:
        "巡检下载({{now}})：有 {{stalledDownloads}} 个下载任务无进展或失败。用只读能力查明原因（站点、做种、磁盘、任务错误），" +
        "给出诊断与建议；若本次允许恢复，只在授权内做重试或强制重新校验，每个任务最多一次，不换源、不删除。",
      capabilityAllowlist: [...INVESTIGATE_CAPABILITIES],
      budget: { maxToolCalls: 24, maxDurationSec: 420 },
      triggerCondition: "stalled_downloads",
      allowRecover: false,
    },
  },
  {
    key: "check.weekly",
    name: "订阅与站点周报",
    description: "检查订阅缺集与站点健康，汇总写入待处理（只读）",
    everyMs: 7 * 24 * HOUR,
    config: {
      goalTemplate:
        "生成本周巡检周报({{date}})：追踪中的订阅 {{activeSubscriptions}} 个。列出缺集的订阅及可能原因、" +
        "健康异常的站点与 RSS，给出本周建议。只读。最后用不超过 300 字的中文汇总。",
      capabilityAllowlist: [...INVESTIGATE_CAPABILITIES],
      budget: { maxToolCalls: 30, maxDurationSec: 600 },
      triggerCondition: "subscriptions_active",
      reportAsDecision: true,
    },
  },
];

interface Facts {
  openDecisions: number;
  stalledDownloads: number;
  activeSubscriptions: number;
}

async function facts(ctx: ServerContext): Promise<Facts> {
  const r = await sql<Facts>`
    select
      (select count(*)::int from decision where status in ('open', 'investigating', 'awaiting_user')) as "openDecisions",
      (select count(*)::int from download_task where removed_at is null and status in ('stalled', 'failed')) as "stalledDownloads",
      (select count(*)::int from subscription where status = 'active') as "activeSubscriptions"`.execute(ctx.db);
  return r.rows[0] ?? { openDecisions: 0, stalledDownloads: 0, activeSubscriptions: 0 };
}

export function renderGoal(template: string, now: Date, f: Facts): string {
  const vars: Record<string, string> = {
    now: now.toISOString(),
    date: now.toISOString().slice(0, 10),
    openDecisions: String(f.openDecisions),
    stalledDownloads: String(f.stalledDownloads),
    activeSubscriptions: String(f.activeSubscriptions),
  };
  return template.replace(/\{\{\s*(\w+)\s*\}\}/g, (m, k: string) => vars[k] ?? m);
}

const isPlanOnly = (id: string) => {
  const d = CAPABILITY_CATALOG[id as KnownCapabilityId];
  return !!d && PLAN_ONLY_KINDS.includes(d.kind);
};

/** agent_check 的执行体(内置与自定义共用) */
export async function runAgentCheck(ctx: ServerContext, c: JobRunContext): Promise<JobRunResult> {
  const parsed = AgentCheckConfig.safeParse(c.config);
  if (!parsed.success) throw new Error(`巡检参数无效：${parsed.error.issues[0]?.message ?? "未知"}`);
  const cfg = parsed.data;
  const f = await facts(ctx);
  const unmet =
    (cfg.triggerCondition === "open_decisions" && f.openDecisions === 0) ||
    (cfg.triggerCondition === "stalled_downloads" && f.stalledDownloads === 0) ||
    (cfg.triggerCondition === "subscriptions_active" && f.activeSubscriptions === 0);
  // 手动执行不看触发条件
  if (unmet && c.trigger === "schedule") return { status: "skipped", summary: "触发条件不满足，未唤醒智能体" };

  let mode: "execute" | "plan_only" = "plan_only";
  let allowlist = cfg.capabilityAllowlist.filter(isPlanOnly);
  let note = "";
  if (cfg.allowRecover) {
    const grants = await recoveryGrants(ctx, ["task.retry", "download.recheck"]);
    if (grants.length > 0) {
      mode = "execute";
      allowlist = [...new Set([...(allowlist.length ? allowlist : INVESTIGATE_CAPABILITIES), ...RECOVER_CAPABILITIES])];
    } else note = "（没有覆盖恢复动作的有效授权，改为只读）";
  }
  const goal = renderGoal(cfg.goalTemplate, c.now, f);
  const dedupeKey = c.trigger === "manual" ? `schedule:${c.jobId}:manual:${c.runId}` : `schedule:${c.jobId}:${c.slot}`;
  const { run, created } = await startRun(ctx, {
    goal,
    mode,
    capabilityAllowlist: allowlist,
    trigger: "schedule",
    createdBy: `schedule:${c.jobId}`,
    dedupeKey,
    budget: cfg.budget,
  });
  return {
    summary: `${created ? "已入队" : "本时段已有"} 智能体运行(${mode === "execute" ? "授权内恢复" : "只读"})${note}`,
    agentRunId: run.id,
  };
}

/** 内置巡检的注册定义(默认关闭) */
export function builtinAgentCheckDefs(ctx: ServerContext): ScheduledJobDef[] {
  return BUILTIN_AGENT_CHECKS.map((t) => ({
    key: t.key,
    kind: "agent_check" as const,
    defaults: {
      name: t.name,
      description: t.description,
      group: "agent" as const,
      schedule: { type: "interval" as const, everyMs: t.everyMs },
      enabled: false,
      config: AgentCheckConfig.parse(t.config) as unknown as Record<string, unknown>,
      firstRun: "after_interval" as const,
    },
    run: (c) => runAgentCheck(ctx, c),
  }));
}