memind_adm/server/skills/index.mjs

/**
 * SkillService — registry, flag check, and invocation.
 *
 * Supported executors:
 *   goosed       — delegates to a goosed agent session (one-shot)
 *   direct_llm   — (future) calls LLM API directly
 *   webhook      — (future) calls user-published HTTP endpoint
 */

import { resolveSkillFlag } from './flag-resolver.mjs';
import { invokeGoosedSkill } from './executors/goosed.mjs';

export function createSkillService(pool) {
  /** List all active skills (for marketplace browsing) */
  async function listSkills({ includeDisabled = false } = {}) {
    const where = includeDisabled ? '' : 'WHERE s.enabled = 1 AND s.status = "active"';
    const [rows] = await pool.query(
      `SELECT s.id, s.slug, s.name, s.type, s.status, s.enabled,
              s.rollout_pct, s.input_schema, s.pricing, s.meta,
              s.publisher_id
       FROM adm_skills s
       ${where}
       ORDER BY s.type, s.name`,
    );
    return rows.map(parseSkillRow);
  }

  /** Get a single skill by slug */
  async function getSkill(slug) {
    const [rows] = await pool.query(
      'SELECT * FROM adm_skills WHERE slug = ? LIMIT 1',
      [slug],
    );
    if (!rows.length) return null;
    return parseSkillRow(rows[0]);
  }

  /**
   * Invoke a skill.
   * context: { userId, planName, callerType }
   * Returns { output } or throws.
   */
  async function invokeSkill(slug, input, context = {}) {
    const skill = await getSkill(slug);
    if (!skill) {
      const err = new Error(`Skill not found: ${slug}`);
      err.code = 'skill_not_found';
      throw err;
    }

    // 1. Global on/off
    if (!skill.enabled) {
      const err = new Error('该 skill 当前不可用');
      err.code = 'skill_disabled';
      err.skill = slug;
      err.fallback_skills = skill.meta?.fallback_skills ?? [];
      throw err;
    }

    // 2. Feature flag chain
    const flag = await resolveSkillFlag(pool, skill.id, context);
    if (!flag.allowed) {
      const err = new Error('该 skill 当前不可用');
      err.code = 'skill_disabled';
      err.reason = flag.reason;
      err.skill = slug;
      err.fallback_skills = skill.meta?.fallback_skills ?? [];
      throw err;
    }

    // 3. Record invocation start
    const invocationId = await recordInvocation(skill.id, input, context, 'running');

    try {
      const output = await dispatch(skill, input);
      await updateInvocation(invocationId, 'success', { latency: output._latency_ms });
      return { invocation_id: invocationId, output: output.content ?? output };
    } catch (err) {
      await updateInvocation(invocationId, 'failed', { error: err.message });
      throw err;
    }
  }

  /** Admin: update skill enabled flag */
  async function setSkillEnabled(slug, enabled, note = '') {
    await pool.query(
      'UPDATE adm_skills SET enabled = ?, updated_at = NOW() WHERE slug = ?',
      [enabled ? 1 : 0, slug],
    );
    if (note) {
      await pool.query(
        `INSERT INTO adm_skill_flags (skill_id, scope_type, scope_id, enabled, note)
         SELECT id, 'global', NULL, ?, ?
         FROM adm_skills WHERE slug = ?
         ON DUPLICATE KEY UPDATE enabled = VALUES(enabled), note = VALUES(note)`,
        [enabled ? 1 : 0, note, slug],
      );
    }
  }

  /** Admin: upsert a flag override */
  async function setSkillFlag(slug, { scopeType, scopeId, enabled, note = '' }) {
    await pool.query(
      `INSERT INTO adm_skill_flags (skill_id, scope_type, scope_id, enabled, note)
       SELECT id, ?, ?, ?, ?
       FROM adm_skills WHERE slug = ?
       ON DUPLICATE KEY UPDATE enabled = VALUES(enabled), note = VALUES(note)`,
      [scopeType, scopeId ?? null, enabled ? 1 : 0, note, slug],
    );
  }

  return { listSkills, getSkill, invokeSkill, setSkillEnabled, setSkillFlag };

  // ── internals ──────────────────────────────────────────────────────────────

  async function dispatch(skill, input) {
    const executor = skill.runtime?.executor ?? 'goosed';
    const t0 = Date.now();

    let result;
    switch (executor) {
      case 'goosed':
        result = await invokeGoosedSkill(skill, input);
        break;
      default:
        throw new Error(`Unsupported executor: ${executor}`);
    }

    result._latency_ms = Date.now() - t0;
    return result;
  }

  async function recordInvocation(skillId, input, context, status) {
    const safeInput = sanitizeInput(input);
    const [res] = await pool.query(
      `INSERT INTO adm_skill_invocations
         (skill_id, caller_id, caller_type, input_snapshot, status, created_at)
       VALUES (?, ?, ?, ?, ?, NOW())`,
      [
        skillId,
        context.userId ?? null,
        context.callerType ?? 'unknown',
        JSON.stringify(safeInput),
        status,
      ],
    );
    return res.insertId;
  }

  async function updateInvocation(id, status, { latency, error } = {}) {
    await pool.query(
      `UPDATE adm_skill_invocations
       SET status = ?, latency_ms = ?, error_message = ?, updated_at = NOW()
       WHERE id = ?`,
      [status, latency ?? null, error ?? null, id],
    );
  }
}

function parseSkillRow(row) {
  return {
    ...row,
    input_schema: tryJson(row.input_schema),
    pricing: tryJson(row.pricing),
    meta: tryJson(row.meta),
    runtime: tryJson(row.runtime),
    enabled: Boolean(row.enabled),
  };
}

function tryJson(v) {
  if (!v) return null;
  if (typeof v === 'object') return v;
  try { return JSON.parse(v); } catch { return null; }
}

function sanitizeInput(input) {
  // Strip potentially sensitive fields before logging
  const { api_key, secret, password, token, ...safe } = input ?? {};
  return safe;
}