pipeline_core/wwwroot/api/skill_proposals.dspy

# skill_proposals.dspy - 技能提议管理（agent自省提议 + 人工创建，生命周期：pending→testing→approved→published）
# 发布动作：把提议的 SKILL.md 写入 skills/global 并增量同步到所有机构，然后状态置 published

import json
import os

uid = await get_user()
if not uid:
    uid = 'user-01'
org_id = await get_userorgid() or '0'

dbname = get_module_dbname('pipeline_core')
action = (params_kw or {}).get('action', 'list')


def _rec_to_dict(r):
    out = {}
    for k in ('id', 'name', 'description', 'content', 'source', 'status',
              'org_id', 'pipeline_id', 'feedback', 'created_by'):
        out[k] = getattr(r, k, '') or ''
    out['created_at'] = str(getattr(r, 'created_at', '') or '')
    out['updated_at'] = str(getattr(r, 'updated_at', '') or '')
    return out


async with DBPools().sqlorContext(dbname) as sor:

    if action == 'list':
        status = (params_kw or {}).get('status', '')
        where = "1=1"
        p = {}
        if status:
            where = "status=${status}$"
            p["status"] = status
        recs = await sor.sqlExe(
            "SELECT * FROM skill_proposals WHERE " + where +
            " ORDER BY created_at DESC LIMIT 200", p)
        proposals = [_rec_to_dict(r) for r in (recs or [])]
        return json.dumps({"success": True, "proposals": proposals},
                          ensure_ascii=False, default=str)

    elif action == 'create':
        name = (params_kw or {}).get('name', '').strip()
        description = (params_kw or {}).get('description', '').strip()
        content = (params_kw or {}).get('content', '').strip()
        if not name or not content:
            return json.dumps({"success": False, "error": "需要 name 和 content"},
                              ensure_ascii=False)
        from appPublic.uniqueID import getID
        await sor.C('skill_proposals', {
            'id': getID(),
            'name': name,
            'description': description,
            'content': content,
            'source': 'manual',
            'status': 'pending',
            'org_id': org_id,
            'pipeline_id': '',
            'created_by': uid,
        })
        return json.dumps({"success": True, "name": name}, ensure_ascii=False)

    elif action == 'update_status':
        qid = (params_kw or {}).get('id', '').strip()
        status = (params_kw or {}).get('status', '').strip()
        feedback = (params_kw or {}).get('feedback', '').strip()
        if not qid or not status:
            return json.dumps({"success": False, "error": "需要 id 和 status"},
                              ensure_ascii=False)
        if status not in ('pending', 'testing', 'approved', 'published', 'rejected'):
            return json.dumps({"success": False, "error": f"非法状态: {status}"},
                              ensure_ascii=False)
        recs = await sor.R('skill_proposals', {'id': qid})
        if not recs:
            return json.dumps({"success": False, "error": "提议不存在"}, ensure_ascii=False)
        await sor.U('skill_proposals', {'id': qid, 'status': status, 'feedback': feedback or ''})
        return json.dumps({"success": True, "id": qid, "status": status}, ensure_ascii=False)

    elif action == 'publish':
        # 发布：读提议 → 写 SKILL.md 到 skills/global → 增量同步到所有机构 → 状态 published
        qid = (params_kw or {}).get('id', '').strip()
        if not qid:
            return json.dumps({"success": False, "error": "需要提议ID"}, ensure_ascii=False)
        recs = await sor.R('skill_proposals', {'id': qid})
        if not recs:
            return json.dumps({"success": False, "error": "提议不存在"}, ensure_ascii=False)
        r = recs[0]
        name = getattr(r, 'name', '') or ''
        content = getattr(r, 'content', '') or ''
        if not name or not content:
            return json.dumps({"success": False, "error": "提议缺少 name 或 content"},
                              ensure_ascii=False)

        # 写 SKILL.md 到技能树。落点按提案 org_id 分流（机构技能层架构）：
        # - org_id='0'（ocai 组织/平台缺省）→ skills/global/{name}/（平台通用规范，所有组织缺省共享）
        # - org_id≠'0'（其他组织）→ skills/orgs/{org_id}/{name}/（组织自有开发规范，
        #   经 get_merged 的 org 缺省继承 + 组织覆盖机制，该组织项目角色可见且覆盖 ocai 同名）
        from pipeline_core.skill_pack import get_skills_base
        proposal_org = (getattr(r, 'org_id', '') or '').strip() or '0'
        base = get_skills_base()
        if proposal_org and proposal_org != '0':
            skill_dir = os.path.join(base, "orgs", proposal_org, name)
        else:
            skill_dir = os.path.join(base, "global", name)
        os.makedirs(skill_dir, exist_ok=True)
        with open(os.path.join(skill_dir, "SKILL.md"), "w", encoding="utf-8") as f:
            f.write(content)

        await sor.U('skill_proposals', {'id': qid, 'status': 'published'})
        return json.dumps({"success": True, "name": name},
                          ensure_ascii=False)

    else:
        return json.dumps({"success": False, "error": "Unknown action: " + str(action)},
                          ensure_ascii=False)