tools: {
  queryOrder: queryOrderTool,
  initiateRefund: initiateRefundTool,
  queryShipping: queryShippingTool,
},

import { Queue, Worker, Job } from 'bullmq';
import { Redis } from 'ioredis';
 
const connection = new Redis({ host: 'localhost', port: 6379 });
 
// 定义队列
const skillQueue = new Queue('skill-execution', { connection });
 
// 添加任务（生产者）
await skillQueue.add('execute-skill', {
  skillId: 'query-order',
  tenantId: 'tenant-a',
  input: { orderId: '#12345' },
});
 
// Worker 消费（消费者）
const worker = new Worker(
  'skill-execution',
  async (job: Job) => {
    // job.data 是添加时传入的数据
    const { skillId, tenantId, input } = job.data;
    // 执行 skill，返回结果
    const result = await executeSkill(skillId, tenantId, input);
    return result; // 返回值会存入 Job 的 returnvalue 字段
  },
  {
    connection,
    concurrency: 10, // 同时处理 10 个 Job
  }
);
 
// 监听 Worker 事件
worker.on('completed', (job, result) => {
  console.log(`Job ${job.id} 完成`, result);
});
 
worker.on('failed', (job, err) => {
  console.error(`Job ${job.id} 失败`, err.message);
});

// ❌ 直接执行退款，重试时会重复退款
async function executeRefund(orderId: string, amount: number) {
  await paymentService.refund(orderId, amount);
}
 
// ✅ 用 idempotencyKey 去重：支付服务层面保证相同 key 只退一次
async function executeRefund(orderId: string, amount: number, idempotencyKey: string) {
  await paymentService.refund(orderId, amount, {
    idempotencyKey, // 传给支付网关（Stripe、支付宝等均支持幂等 Key）
  });
}

interface WebhookSkillConfig {
  url: string;
  method: 'GET' | 'POST';
  // 请求头（通常包含认证信息，加密存储）
  headers: Record<string, string>;
  // 超时时间（毫秒）
  timeout: number;
  // 超时时是否重试一次
  retryOnTimeout: boolean;
  // 签名密钥：平台发出请求时会附上 HMAC-SHA256 签名
  // 企业服务端验证签名，防止伪造请求
  webhookSecret?: string;
}

interface CodeSkillConfig {
  // 用户上传的 JavaScript 代码（ES2020，不支持 import/require）
  code: string;
  // 入口函数名，平台会调用 module[entrypoint](input, context)
  entrypoint: string;
  // 允许调用的外部 URL 白名单（沙箱内网络请求受限）
  allowedUrls?: string[];
  // 内存上限（MB，默认 128）
  memoryLimitMb: number;
  // CPU 时间上限（毫秒，默认 5000）
  cpuTimeLimitMs: number;
}

interface SubAgentSkillConfig {
  // Mastra Agent 的端点（内部服务或外部 A2A 协议端点）
  agentEndpoint: string;
  // 通信协议：内部直接调用或通过 A2A 协议
  protocol: 'internal' | 'a2a';
  // 超时（复杂任务可能需要更长时间）
  timeout: number;
}

import { z } from 'zod';
 
// Skill 执行时的上下文信息
interface SkillContext {
  tenantId: string;
  sessionId: string;
  userId?: string;  // 终端用户 ID（如果已知）
  // 用于在 Skill 内部调用其他平台能力（比如查询租户配置）
  platformClient: PlatformClient;
}
 
// 内置 Skill 统一格式
interface BuiltinSkill {
  id: string;
  name: string;
  // description 会作为工具描述传给 LLM，直接影响 LLM 的选择决策
  description: string;
  category: SkillCategory;
  inputSchema: z.ZodSchema;
  outputSchema: z.ZodSchema;
  execute: (input: unknown, context: SkillContext) => Promise<unknown>;
  // 每分钟调用次数上限（平台级别，跨所有租户）
  rateLimit?: { rpm: number };
  // 超时（毫秒）
  timeout: number;
  // 结果是否可缓存（用于 LLM Gateway 的语义缓存）
  isCacheable: boolean;
}
 
type SkillCategory =
  | 'search'       // 信息检索
  | 'communication' // 邮件、通知
  | 'data'         // 数据查询
  | 'generation';  // 内容生成

// 查询天气的 Skill
 
// ❌ 模糊，LLM 可能在用户问"现在外面适合跑步吗"时选不到这个工具
description: '查询天气信息'
 
// ✅ 具体，包含适用场景和不适用场景
description: `查询指定城市的实时天气和未来 7 天预报。
适用：用户询问某地当前天气、温度、降雨概率、未来天气趋势。
不适用：历史天气查询（数据只保留最近 30 天）、室内环境温湿度。
输入城市名支持中文（"北京"）或英文（"Beijing"），不支持模糊地名（"北方"）。`
 
// 搜索网页的 Skill
 
// ❌ 过于宽泛，会和知识库检索混淆
description: '搜索信息'
 
// ✅ 明确区分和其他检索类 Skill 的边界
description: `通过 Brave Search 搜索互联网上的公开信息。
适用：查询新闻事件、公司公开信息、产品评测、技术文档、当前热点。
不适用：查询用户自己的数据（订单、账单、个人信息）——这类数据用专用的数据查询工具。
不适用：检索企业内部知识库——使用 search-knowledge-base 工具。
返回结果包含页面摘要和来源 URL，不保证实时性（结果可能有 24 小时延迟）。`

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import { z } from 'zod';
 
const server = new McpServer({
  name: 'tenant-b-tools',       // Server 名称，用于日志标识
  version: '1.0.0',
});
 
// 注册账单查询工具
server.tool(
  'get-user-billing',                    // 工具 ID
  '查询指定用户当月账单明细，包含各功能模块的用量和费用', // description（影响 LLM 选择）
  {
    // 输入 schema，用 zod 定义
    userId: z.string().describe('用户 ID'),
    month: z.string().optional().describe('账单月份，格式 YYYY-MM，默认当月'),
  },
  async ({ userId, month }) => {
    // 调用租户 B 自己的 Billing API
    const targetMonth = month ?? getCurrentMonth();
    const bill = await fetchBillingFromInternalAPI(userId, targetMonth);
 
    // 返回格式固定：content 数组，每项是 { type, text/data }
    return {
      content: [
        {
          type: 'text' as const,
          text: JSON.stringify({
            userId,
            month: targetMonth,
            totalAmount: bill.total,
            breakdown: bill.items,
          }),
        },
      ],
    };
  }
);
 
// 注册工单创建工具
server.tool(
  'create-support-ticket',
  '为用户创建技术支持工单，返回工单编号',
  {
    userId: z.string(),
    title: z.string().describe('工单标题'),
    description: z.string().describe('问题描述'),
    priority: z.enum(['low', 'medium', 'high']).default('medium'),
  },
  async ({ userId, title, description, priority }) => {
    const ticket = await createTicketInJira({ userId, title, description, priority });
    return {
      content: [{ type: 'text' as const, text: `工单已创建，编号：${ticket.id}` }],
    };
  }
);
 
// 启动（stdio transport 适合本地进程调用）
const transport = new StdioServerTransport();
await server.connect(transport);

interface MCPSkillConfig {
  // HTTP transport 端点（生产环境）
  serverUrl?: string;
  // stdio 命令（本地开发）
  stdioCommand?: string;
  stdioArgs?: string[];
  // 认证（Bearer Token）
  authToken?: string;
  // 允许使用哪些工具（不填则允许全部）
  allowedTools?: string[];
}

import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
 
async function callMCPSkill(config: MCPSkillConfig, toolName: string, args: unknown) {
  const client = new Client({ name: 'agentflow', version: '1.0.0' });
 
  // 连接 MCP Server
  const transport = new StreamableHTTPClientTransport(new URL(config.serverUrl!));
  await client.connect(transport);
 
  // 调用工具
  const result = await client.callTool({ name: toolName, arguments: args as Record<string, unknown> });
 
  await client.close();
  return result;
}

// ❌ 极度危险，绝对不能这样做
eval(userCode);
 
// ❌ vm 模块不提供真正的隔离
const vm = require('vm');
vm.runInNewContext(userCode, { input });

// 通过 constructor chain 逃逸，访问 Node.js 全局
({}).__proto__.constructor.constructor('return process')().env
// 在 vm 沙箱里，这行代码可以读取 process.env，获取所有环境变量（含 API Key）

import ivm from 'isolated-vm';
 
// 创建 Isolate，限制堆内存 128MB
// memoryLimit 触发后 Isolate 会抛出 RangeError，而不是让整个 Node 进程 OOM
const isolate = new ivm.Isolate({ memoryLimit: 128 });
 
// 创建执行上下文（每次执行创建新的 Context，隔离全局状态）
const context = await isolate.createContext();
const jail = context.global;
 
// 安全注入：让沙箱内代码能调用外部函数
// 注意：注入的函数以 Reference 形式传递，而不是直接暴露 JS 对象
// 这确保了沙箱内代码只能调用我们明确允许的函数
await jail.set(
  'fetchAllowed',
  new ivm.Reference(async (url: string, options?: string) => {
    // 在沙箱外执行，享有完整的 Node.js 权限
    // 但我们在这里做白名单检查，限制可访问的 URL
    if (!isUrlWhitelisted(url)) {
      throw new Error(`URL 不在白名单内: ${url}`);
    }
    const parsedOptions = options ? JSON.parse(options) : {};
    const response = await fetch(url, parsedOptions);
    return response.text();
  })
);
 
// 注入只读的输入数据（通过 JSON 序列化，防止对象引用传递）
await jail.set('__input__', new ivm.ExternalCopy(input).copyInto());
 
// 编译用户代码（编译阶段会做语法检查）
const script = await isolate.compileScript(`
  (async function() {
    // 用户代码在这里执行
    // 只能访问我们注入的 __input__ 和 fetchAllowed
    ${userCode}
  })()
`);
 
// 执行，设置 CPU 时间上限（timeout 单位毫秒）
// timeout 超出后抛出 Error: Script execution timed out
const resultRef = await script.run(context, { timeout: 5000 });
 
// 从沙箱内取出结果（同样通过序列化，防止沙箱对象泄漏到外部）
const result = await resultRef?.copy();
 
// 执行完毕后立即释放 Isolate（回收 V8 堆内存）
isolate.dispose();

-- 启用 pgvector 扩展
CREATE EXTENSION IF NOT EXISTS vector;
 
CREATE TABLE skills (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  -- NULL 表示平台内置 Skill，否则是租户自定义 Skill
  tenant_id UUID REFERENCES tenants(id),
  name VARCHAR(255) NOT NULL,
  description TEXT NOT NULL,
  -- 'builtin' | 'webhook' | 'code' | 'mcp' | 'subagent'
  type VARCHAR(50) NOT NULL,
  -- 类型相关配置（webhook URL、code 内容、mcp endpoint 等）
  config JSONB NOT NULL,
  -- 输入/输出 schema（JSON Schema 格式，用于 LLM 了解参数结构）
  input_schema JSONB NOT NULL,
  output_schema JSONB NOT NULL,
  -- description 的向量表示，维度 1024（voyage-3，全书默认 embedding 模型）
  -- 切到 OpenAI text-embedding-3-small 时改为 vector(1536) 并重建索引
  embedding vector(1024),
  is_active BOOLEAN DEFAULT true,
  rate_limit_rpm INTEGER,
  timeout_ms INTEGER DEFAULT 30000,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);
 
-- HNSW 索引，余弦相似度。比 IVFFlat 的优点：不需要预先训练，增量插入不需要重建索引
CREATE INDEX idx_skills_embedding ON skills
  USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);
 
-- 按租户查询的常规索引
CREATE INDEX idx_skills_tenant ON skills (tenant_id, is_active);

async function findRelevantSkills(
  userMessage: string,
  tenantId: string,
  topK: number = 10
): Promise<Skill[]> {
  // 第一步：向量化用户消息
  // 用和 embedding 列相同的模型（voyage-3），否则向量空间不对齐
  const embedding = await embedText(userMessage);
 
  // 第二步：pgvector 最近邻搜索（召回阶段）
  // 返回平台内置（tenant_id IS NULL）+ 该租户自定义的 Skill
  // <=> 是余弦距离算子，值越小越相似
  const candidates = await db.query<Skill & { distance: number }>(
    `SELECT
       id, name, description, type, config, input_schema, output_schema,
       embedding <=> $1 AS distance
     FROM skills
     WHERE (tenant_id IS NULL OR tenant_id = $2)
       AND is_active = true
     ORDER BY distance
     LIMIT $3`,
    [pgvector.toSql(embedding), tenantId, topK * 2]  // 多取一倍，留给精排过滤
  );
 
  if (candidates.rows.length === 0) return [];
 
  // 第三步：LLM 精排（从召回结果中选最相关的 topK 个）
  // 用小模型（claude-haiku / gpt-4o-mini），减少延迟和成本
  return rerankWithLLM(candidates.rows, userMessage, topK);
}
 
async function rerankWithLLM(
  candidates: Array<Skill & { distance: number }>,
  userMessage: string,
  topK: number
): Promise<Skill[]> {
  // 构造精排 prompt：让 LLM 从候选 Skill 中选出最合适的几个
  const candidateList = candidates
    .map((s, i) => `${i + 1}. [${s.id}] ${s.name}: ${s.description}`)
    .join('\n');
 
  const response = await llmClient.generate({
    model: 'claude-3-haiku-20240307',
    messages: [
      {
        role: 'user',
        content: `用户消息："${userMessage}"
 
以下是候选 Skill 列表：
${candidateList}
 
请选出最适合处理该用户消息的 ${topK} 个 Skill，按相关性从高到低排序。
只返回 Skill 编号，用逗号分隔，例如：3,1,7`,
      },
    ],
  });
 
  // 解析 LLM 返回的编号，映射回 Skill 列表
  const selectedIndices = response.content
    .split(',')
    .map((s) => parseInt(s.trim(), 10) - 1)
    .filter((i) => i >= 0 && i < candidates.length);
 
  return selectedIndices.map((i) => candidates[i]);
}

async function registerSkill(skill: CreateSkillInput): Promise<Skill> {
  // 计算 description 的向量（用于后续语义检索）
  const embedding = await embedText(skill.description);
 
  const result = await db.query(
    `INSERT INTO skills (tenant_id, name, description, type, config,
                         input_schema, output_schema, embedding,
                         rate_limit_rpm, timeout_ms)
     VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
     RETURNING *`,
    [
      skill.tenantId ?? null,
      skill.name,
      skill.description,
      skill.type,
      JSON.stringify(skill.config),
      JSON.stringify(skill.inputSchema),
      JSON.stringify(skill.outputSchema),
      pgvector.toSql(embedding),
      skill.rateLimitRpm ?? null,
      skill.timeoutMs ?? 30000,
    ]
  );
 
  return result.rows[0];
}