diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 55fc089..a8ddadd 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -1112,6 +1112,15 @@ async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True, async def _get_workspace_dir(sor, project_id): recs = await sor.sqlExe("SELECT workspace_dir FROM sd_projects WHERE id=${pid}$", {"pid": project_id}) ws = getattr(recs[0], 'workspace_dir', '') if recs else '' + if ws: + return _resolve_workspace(ws) + # 存量项目无 workspace_dir:兜底解析项目根目录(与上传落盘同一函数, + # 保证「上传放的位置 = agent 读的位置」) + from .workspace import get_project_dir_by_id + _pdir, _ = await get_project_dir_by_id(sor, project_id) + if _pdir: + os.makedirs(_pdir, exist_ok=True) + return _pdir return _resolve_workspace(ws) diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index 0146cc4..f6fc3b6 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/pipeline_service/agent_loop_v2.py @@ -373,6 +373,13 @@ class AgentExecutor: if ws: self.workspace_dir = ws self.pipeline_id = getattr(recs[0], 'pipeline_id', '') or '' + if not ws: + # 存量项目无 workspace_dir:兜底解析项目根目录(与上传落盘同一函数, + # 保证「上传放的位置 = agent 读的位置」) + from .workspace import get_project_dir_by_id + _pdir, _ = await get_project_dir_by_id(sor, self.project_id) + if _pdir: + self.workspace_dir = _pdir except Exception: pass diff --git a/pipeline_service/workspace.py b/pipeline_service/workspace.py index f82be1e..a9edf73 100644 --- a/pipeline_service/workspace.py +++ b/pipeline_service/workspace.py @@ -334,6 +334,67 @@ async def get_project_apps_modules(sor, uid, session_id=''): return sorted(set(apps)), sorted(modules) +async def get_project_dir_by_id(sor, project_id): + """项目根目录(按项目 id 直接解析,不依赖会话上下文)。 + + 解析顺序与 agent 工作目录完全一致,保证「上传放的位置 = agent 读的位置」: + 1. sd_projects.workspace_dir 非空(新项目)→ 直接用({space}/projects/{slug}) + 2. 为空(存量项目)→ 兜底旧平铺结构 {base}/{org}/{space}/{项目名} + """ + if not project_id: + return '', '' + recs = await sor.sqlExe( + "SELECT name, org_id, pipeline_id, workspace_dir FROM sd_projects WHERE id=${p}$ LIMIT 1", + {"p": project_id}) + if not recs: + return '', '' + r = recs[0] + workspace_base = await get_workspace_base(sor) + ws = (getattr(r, 'workspace_dir', '') or '').strip() + if ws.startswith('/'): + return ws, workspace_base + name = (getattr(r, 'name', '') or '').strip() + if not name: + return '', workspace_base + org_id = getattr(r, 'org_id', '0') or '0' + space = getattr(r, 'pipeline_id', '') or GENERAL_SPACE + # 不传 project_id:查询语义下禁用重名后缀逻辑,返回的就是平铺目录本身 + return build_workspace_path(workspace_base, org_id, space, name), workspace_base + + +def copy_uploads_to_project(project_dir, uploads): + """把已落盘的上传文件复制进项目根目录(会话上传 → 项目内可被 agent 找到)。 + + uploads: [(src_abs_path, filename)];同名文件自动加 _1/_2 后缀防覆盖。 + 返回 [(保存后文件名, 绝对路径)],与入参顺序对应。 + """ + import shutil + saved = [] + if not project_dir: + return saved + try: + os.makedirs(project_dir, exist_ok=True) + except Exception: + return saved + for src, name in uploads: + name = os.path.basename((name or '').strip()) + if not name or not os.path.isfile(src): + continue + target = os.path.join(project_dir, name) + if os.path.exists(target): + stem, ext = os.path.splitext(name) + i = 1 + while os.path.exists(os.path.join(project_dir, f"{stem}_{i}{ext}")): + i += 1 + target = os.path.join(project_dir, f"{stem}_{i}{ext}") + try: + shutil.copyfile(src, target) + saved.append((os.path.basename(target), target)) + except Exception as e: + logger.warning(f"copy upload failed: {src} -> {target} err={e}") + return saved + + def resolve_workspace_path(project_dir, space_dir, node_id): """把工作空间节点 ID 解析成绝对路径。 @@ -514,6 +575,8 @@ def load_workspace(): g.get_space_dir = get_space_dir g.get_project_dir = get_project_dir g.get_project_dir_pl = get_project_dir_pl + g.get_project_dir_by_id = get_project_dir_by_id + g.copy_uploads_to_project = copy_uploads_to_project g.get_project_apps_modules = get_project_apps_modules g.get_session_project_id = get_session_project_id g.get_session_context = get_session_context