fix(upload): get_project_dir_by_id+copy_uploads_to_project；agent工作目录对空workspace_dir项目兜底解析(与落盘同一函数)

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)
 
 

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
 

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