From 5a7104aa5d9550927ce3852c17f1ff6b73421784 Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Wed, 5 Aug 2026 17:56:22 +0800
Subject: [PATCH] =?UTF-8?q?feat(v3.2.1):=20shell=5Fexec=20+=20skill=5Fimpo?=
 =?UTF-8?q?rt=5Fgit=20=E2=80=94=20=E4=B8=BBagent=20devops=E8=83=BD?=
 =?UTF-8?q?=E5=8A=9B?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- shell_exec: 安全子进程执行（沙箱在 /d/pipeline/workspaces 下，60s超时）
- skill_import_git: git clone + skills目录扫描，返回技能列表
- 注册到 env，cockpit_chat 可通过 dspy 直接调用
---
 pipeline_service/init.py | 76 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 76 insertions(+)

diff --git a/pipeline_service/init.py b/pipeline_service/init.py
index 4ba6d15..147affc 100644
--- a/pipeline_service/init.py
+++ b/pipeline_service/init.py
@@ -9,6 +9,7 @@
 
 import json
 import asyncio
+import os
 from ahserver.serverenv import ServerEnv
 from appPublic.uniqueID import getID
 from appPublic.log import debug
@@ -371,6 +372,77 @@ def pipeline_unregister_step_type(step_type):
     return json.dumps({"success": True, "step_type": step_type}, ensure_ascii=False)
 
 
+# ── Git / Shell 操作（v3.2.1: 主agent devops 能力）──
+
+_SHELL_BASE_DIR = '/d/pipeline/workspaces'
+
+async def shell_exec(command: str, workdir: str = None, timeout: int = 60):
+    """安全外壳执行。限制在 workspace 目录下，超时自动终止。
+    
+    Returns: {"rc": exit_code, "stdout": "...", "stderr": "..."}
+    """
+    cwd = workdir or _SHELL_BASE_DIR
+    cwd = os.path.abspath(cwd)
+    if not cwd.startswith(os.path.abspath(_SHELL_BASE_DIR)):
+        return {"rc": -1, "stdout": "", "stderr": f"安全限制：工作目录必须在 {_SHELL_BASE_DIR} 下"}
+    try:
+        proc = await asyncio.create_subprocess_shell(
+            command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+            cwd=cwd, executable='/bin/bash')
+        try:
+            stdout, stderr = await asyncio.wait_for(
+                proc.communicate(), timeout=timeout)
+        except asyncio.TimeoutError:
+            proc.kill()
+            await proc.wait()
+            return {"rc": -1, "stdout": "", "stderr": f"命令超时（{timeout}s）"}
+        return {"rc": proc.returncode or 0, "stdout": stdout.decode('utf-8', 'replace')[-8000:],
+                "stderr": stderr.decode('utf-8', 'replace')[-4000:]}
+    except Exception as e:
+        return {"rc": -1, "stdout": "", "stderr": str(e)[:500]}
+
+
+async def skill_import_git(repo_url: str, skills_dir: str = 'skills'):
+    """克隆 git 仓库，扫描指定目录下所有子目录（每个=一个企业技能）。
+    
+    Returns: {"success": bool, "repo":..., "target":..., "skills": [{"name","path","description"}]}
+    """
+    from appPublic.uniqueID import getID
+    repo_name = repo_url.rstrip('/').split('/')[-1].replace('.git', '') or f"repo_{getID()[:8]}"
+    target = os.path.join(_SHELL_BASE_DIR, repo_name)
+    
+    if not os.path.isdir(target):
+        r = await shell_exec(f'git clone {repo_url} {target}', workdir=_SHELL_BASE_DIR)
+        if r['rc'] != 0:
+            return {"success": False, "error": f"clone 失败: {r['stderr'][:500]}"}
+    
+    full_skills = os.path.join(target, skills_dir)
+    if not os.path.isdir(full_skills):
+        entries = ', '.join(os.listdir(target)[:10]) if os.path.isdir(target) else '(dir not found)'
+        return {"success": False, "error": f"skills目录不存在: {full_skills}，仓库内容: {entries}"}
+    
+    found = []
+    for entry in sorted(os.listdir(full_skills)):
+        entry_path = os.path.join(full_skills, entry)
+        if os.path.isdir(entry_path):
+            desc = ''
+            skill_file = os.path.join(entry_path, 'SKILL.md')
+            if os.path.isfile(skill_file):
+                try:
+                    with open(skill_file) as f:
+                        content = f.read(500)
+                    for line in content.split('\n'):
+                        if line.startswith('description:'):
+                            desc = line.split(':', 1)[1].strip()
+                            break
+                except:
+                    pass
+            found.append({"name": entry, "path": entry_path, "description": desc})
+    
+    return {"success": True, "repo": repo_url, "target": target,
+            "skills_dir": skills_dir, "skills": found}
+
+
 def load_pipeline_service():
     """注册所有函数到 ServerEnv。任何宿主应用调用此函数即可使用产线引擎。"""
     env = ServerEnv()
@@ -428,6 +500,10 @@ def load_pipeline_service():
     env.question_list = list_questions
     env.question_qna = get_task_qna
 
+    # DevOps: git/shell operations (v3.2.1)
+    env.shell_exec = shell_exec
+    env.skill_import_git = skill_import_git
+
     # Background poller: auto-dispatch submitted role_tasks to role agents
 
     async def _role_poller(app):