From 82b094442482808f34afe6de824543aea14a763a Mon Sep 17 00:00:00 2001 From: ymq Date: Sat, 15 Aug 2026 10:00:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=85=A8=E5=B1=80=E5=B9=B6=E5=8F=91age?= =?UTF-8?q?nt=E6=95=B0=E9=99=90=E5=88=B6(max=5Fconcurrent=5Fagents=20param?= =?UTF-8?q?s=E8=A1=A8)+=E6=9C=BA=E6=9E=84=E5=B7=A5=E4=BD=9C=E7=9B=AE?= =?UTF-8?q?=E5=BD=95workspace=5Fbase=E5=8A=A8=E6=80=81=E5=8C=96(=5Fis=5Fsa?= =?UTF-8?q?fe=5Fworkdir=E8=AF=BBparams=E8=A1=A8)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_service/agent_loop.py | 47 ++++++++++++++++++++++++++++++---- pipeline_service/init.py | 18 +++++++++++-- pipeline_service/workspace.py | 30 ++++++++++++++++------ 3 files changed, 80 insertions(+), 15 deletions(-) diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index ac9e085..41549a1 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -120,7 +120,7 @@ def _resolve_workspace(workspace_dir): # ── Agent 工具函数 ── def _is_safe_workdir(workdir): - """检查目录是否在允许范围内。""" + """检查目录是否在允许范围内（同步版，仅用模块级白名单兜底）。""" wd = os.path.abspath(workdir) for allowed in _ALLOWED_WORKDIRS: awd = os.path.abspath(os.path.expanduser(allowed)) @@ -129,10 +129,47 @@ def _is_safe_workdir(workdir): return False +_allowed_workdirs_cache = None +_allowed_workdirs_cache_time = 0 + + +async def _get_allowed_workdirs(): + """动态读 workspace_base 参数（params 表），合并到允许目录列表（60s 缓存）。""" + global _allowed_workdirs_cache, _allowed_workdirs_cache_time + import time as _t + now = _t.time() + if _allowed_workdirs_cache is not None and (now - _allowed_workdirs_cache_time) < 60: + return _allowed_workdirs_cache + allowed = list(_ALLOWED_WORKDIRS) + try: + db = _get_db() + async with db.sqlorContext("pipeline") as sor: + from .workspace import get_workspace_base + base = await get_workspace_base(sor) + if base and base not in allowed: + allowed.append(base) + except Exception: + pass + _allowed_workdirs_cache = allowed + _allowed_workdirs_cache_time = now + return allowed + + +async def _is_safe_workdir_async(workdir): + """检查目录是否在允许范围内（含 params 表动态 workspace_base）。""" + wd = os.path.abspath(workdir) + allowed = await _get_allowed_workdirs() + for a in allowed: + awd = os.path.abspath(os.path.expanduser(a)) + if wd.startswith(awd): + return True + return False + + async def _run_shell(command, workdir, timeout=120): """安全执行 shell 命令。返回 {"rc": int, "stdout": str, "stderr": str}""" cwd = os.path.abspath(workdir) if workdir else os.path.expanduser('~/pipeline_ws') - if not _is_safe_workdir(cwd): + if not await _is_safe_workdir_async(cwd): return {"rc": -1, "stdout": "", "stderr": f"安全限制：目录 {cwd} 不在允许范围"} if not os.path.isdir(cwd): return {"rc": -1, "stdout": "", "stderr": f"目录不存在: {cwd}"} @@ -590,7 +627,7 @@ async def _exec_agent_tool(tool, params, workspace_dir): path = p.get('path', '') if not path: return 'FAIL: 需要文件路径' full = os.path.join(workspace_dir, path) - if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围' + if not await _is_safe_workdir_async(full): return 'FAIL: 路径不在允许范围' if not os.path.isfile(full): return f'FAIL: 文件不存在 {path}' with open(full, encoding='utf-8') as f: return f.read()[:30000] @@ -599,7 +636,7 @@ async def _exec_agent_tool(tool, params, workspace_dir): content = p.get('content', '') if not path: return 'FAIL: 需要文件路径' full = os.path.join(workspace_dir, path) - if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围' + if not await _is_safe_workdir_async(full): return 'FAIL: 路径不在允许范围' os.makedirs(os.path.dirname(full), exist_ok=True) with open(full, 'w', encoding='utf-8') as f: f.write(content) @@ -607,7 +644,7 @@ async def _exec_agent_tool(tool, params, workspace_dir): elif tool == 'list_files': path = p.get('path', '') or '.' full = os.path.join(workspace_dir, path) - if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围' + if not await _is_safe_workdir_async(full): return 'FAIL: 路径不在允许范围' if not os.path.isdir(full): return f'FAIL: 目录不存在 {path}' items = os.listdir(full)[:50] lines = [] diff --git a/pipeline_service/init.py b/pipeline_service/init.py index bc450c3..fea45a2 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -555,12 +555,26 @@ def load_pipeline_service(): "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL, updated_at=NOW() " "WHERE state='running' AND pipeline_id='role_task' " "AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {}) + + # 全局并发 agent 数限制（max_concurrent_agents，appbase params 表，默认 3）。 + # 以 DB state='running' 计数为准（权威），超出上限时本轮回合不再派发。 + from .workspace import get_max_concurrent_agents + max_n = await get_max_concurrent_agents(sor) + running_rows = await sor.sqlExe( + "SELECT COUNT(*) as c FROM pipeline_tasks " + "WHERE state='running' AND pipeline_id='role_task'", {}) + cur = getattr(running_rows[0], 'c', 0) if running_rows else 0 + avail = max(0, max_n - cur) + if avail <= 0: + # 达到并发上限，本轮不派发（退出 context 后末尾 sleep 再 poll） + continue + recs = await sor.sqlExe( "SELECT id, tenant_id, role FROM pipeline_tasks " "WHERE state='submitted' AND pipeline_id='role_task' " - "AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 5", + "AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 20", {}) - for rec in (recs or []): + for rec in (recs or [])[:avail]: tid = getattr(rec, 'id', '') pid = getattr(rec, 'tenant_id', '') r = getattr(rec, 'role', '') diff --git a/pipeline_service/workspace.py b/pipeline_service/workspace.py index 8276036..7456a95 100644 --- a/pipeline_service/workspace.py +++ b/pipeline_service/workspace.py @@ -33,17 +33,31 @@ def _get_file_icon(ext): return "📁" -async def get_workspace_base(sor): - """读 workspace_base 参数（appbase params 表，表不存在时兜底 WORKSPACE_BASE）。""" - workspace_base = WORKSPACE_BASE +async def get_param(sor, name, default=""): + """读 appbase params 表配置（params_name → params_value），表不存在/无值时兑底 default。""" try: - _wbr = await sor.sqlExe( - "SELECT params_value FROM params WHERE params_name='workspace_base' LIMIT 1", {}) - if _wbr and getattr(_wbr[0], 'params_value', ''): - workspace_base = getattr(_wbr[0], 'params_value', '') + recs = await sor.sqlExe( + "SELECT params_value FROM params WHERE params_name=${n}$ LIMIT 1", {"n": name}) + if recs and getattr(recs[0], 'params_value', '') not in (None, ''): + return getattr(recs[0], 'params_value', '') except Exception: pass - return workspace_base + return default + + +async def get_workspace_base(sor): + """读 workspace_base 参数（appbase params 表，表不存在时兜底 WORKSPACE_BASE）。""" + return await get_param(sor, 'workspace_base', WORKSPACE_BASE) + + +async def get_max_concurrent_agents(sor): + """读全局并发 agent 数限制（appbase params 表 max_concurrent_agents，默认 3）。""" + val = await get_param(sor, 'max_concurrent_agents', '3') + try: + n = int(float(str(val))) + return max(1, n) + except (ValueError, TypeError): + return 3 async def get_workspace_dir(sor, uid):