From d77910ea2e4b6b7897127f0e2842e8b51fdb15e0 Mon Sep 17 00:00:00 2001 From: ymq Date: Wed, 19 Aug 2026 17:06:31 +0800 Subject: [PATCH] =?UTF-8?q?feat(iteration):=20=E8=BF=AD=E4=BB=A3=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E6=9C=BA=E6=94=B9=E9=80=A0=E2=80=94=E2=80=94=E6=98=BE?= =?UTF-8?q?=E5=BC=8F=E3=80=8C=E5=BD=93=E5=89=8D=E8=BF=AD=E4=BB=A3=E3=80=8D?= =?UTF-8?q?+=20=E5=BA=8F=E5=8F=B7=20seq=5Fno=20+=20=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E4=B8=8B=E4=B8=80=E8=BF=AD=E4=BB=A3=EF=BC=8C=E5=BA=9F=E9=99=A4?= =?UTF-8?q?=20created=5Fat=20=E6=8E=A8=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 背景：用户明确迭代状态机语义——当前迭代唯一(状态 in_progress)、按序号 1,2,3 推进、 需用户指令启动下一迭代、启动下一迭代时若当前未结束则强制完成。 1. iteration_capability.py: - create_iteration 自动分配 seq_no = 项目内 max(seq_no)+1 - 新增 get_current_iteration(sor, pid)：当前迭代 = status='in_progress' 唯一迭代 - 新增 start_next_iteration(pid)：强制完成当前迭代 + 作废其活跃任务 + 启动下一个 planning 迭代 - list_iterations 按 seq_no 排序 2. sdlc_ability.py: 4 处 created_at 推断(默认迭代)收口到 get_current_iteration； 新增 start_next_iteration 工具；list_iterations 显示 seq_no + 当前迭代标记 3. agent_loop.py: PM 派发任务默认迭代收口到 get_current_iteration 4. agent_loop_v2.py: create_project 初始迭代 status 从非法的 'active' 改为 'in_progress' + seq_no=1 --- pipeline_service/agent_loop.py | 11 ++- pipeline_service/agent_loop_v2.py | 3 +- pipeline_service/iteration_capability.py | 102 ++++++++++++++++++++++- pipeline_service/sdlc_ability.py | 50 ++++++----- 4 files changed, 138 insertions(+), 28 deletions(-) diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index af6792a..05fdcb2 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -1230,13 +1230,12 @@ async def _pm_create_tasks(sor, project_id, params, parent_task_id=None): # 默认父任务：本次审核的里程碑任务（子任务挂它名下，任务树据此分层） default_parent = (params.get('parent_id') or parent_task_id or '').strip() - # 当前迭代（该 project 最新创建的迭代），作为任务默认归属 + # 当前迭代（status='in_progress' 的唯一迭代），作为任务默认归属 iteration_name = '' - irecs = await sor.sqlExe( - "SELECT iteration_name FROM sd_iterations WHERE project_id=${pid}$ ORDER BY created_at DESC LIMIT 1", - {"pid": project_id}) - if irecs: - iteration_name = getattr(irecs[0], 'iteration_name', '') or '' + from .iteration_capability import get_current_iteration + cur = await get_current_iteration(sor, project_id) + if cur: + iteration_name = cur.get('iteration_name', '') or '' created = [] # [(task_dict, tid, title, role)] key_map = {} # key → 真实任务ID（同批 depends_on 用 key 互引，第二遍解析） diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index 32ccce9..5f1c665 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/pipeline_service/agent_loop_v2.py @@ -868,7 +868,8 @@ class AgentExecutor: await sor.C("sd_iterations", { "id": getID(), "project_id": pid_val, "iteration_name": f"{name}-初始迭代", - "iteration_type": "default", "status": "active", "priority": 1, + "iteration_type": "default", "status": "in_progress", "priority": 1, + "seq_no": 1, }) self.project_id = pid_val await self._persist_project(sor, pid_val) diff --git a/pipeline_service/iteration_capability.py b/pipeline_service/iteration_capability.py index 3e594bd..c9ebff2 100644 --- a/pipeline_service/iteration_capability.py +++ b/pipeline_service/iteration_capability.py @@ -81,6 +81,12 @@ async def create_iteration(project_id, iteration_name, iteration_type="new_featu return False, "缺少 iteration_name" db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: + # 迭代序号 seq_no = 该项目 max(seq_no) + 1（1,2,3... 顺序编号） + r = await sor.sqlExe( + f"SELECT COALESCE(MAX(seq_no),0) AS m FROM {TABLE} WHERE project_id=${{pid}}$", + {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + seq_no = int(getattr(r[0], 'm', 0) or 0) + 1 iid = getID() try: priority = int(priority) @@ -94,12 +100,13 @@ async def create_iteration(project_id, iteration_name, iteration_type="new_featu 'scope': scope or '', 'status': S_PLANNING, 'priority': priority, + 'seq_no': seq_no, 'created_by': created_by or '', }) await record_audit(project_id, TABLE, iid, 'create', to_state=S_PLANNING, who=_normalize_role(who), agent_id=agent_id, sor=sor) - logger.info("create_iteration: %s project=%s", iid, project_id) + logger.info("create_iteration: %s project=%s seq_no=%s", iid, project_id, seq_no) return True, iid @@ -147,7 +154,7 @@ async def list_iterations(project_id, status=None, limit=50) -> list: except (TypeError, ValueError): limit = 50 sql = (f"SELECT * FROM {TABLE} WHERE {where} " - f"ORDER BY created_at ASC LIMIT {limit}") + f"ORDER BY seq_no ASC LIMIT {limit}") recs = await sor.sqlExe(sql, params) # 释放 SELECT 元数据锁 await sor.sqlExe("COMMIT", {}) @@ -157,6 +164,97 @@ async def list_iterations(project_id, status=None, limit=50) -> list: return result +async def get_current_iteration(sor, project_id): + """当前迭代 = status='in_progress' 的唯一迭代（seq_no 最小）。无则返回 None。 + + 取代旧的「ORDER BY created_at DESC LIMIT 1」推断：当前迭代是显式状态， + 不再靠时间戳猜。传入 sor（调用点已在 sqlorContext 内，避免嵌套开连接）。 + """ + recs = await sor.sqlExe( + f"SELECT * FROM {TABLE} WHERE project_id=${{pid}}$ AND status='in_progress' " + "ORDER BY seq_no ASC LIMIT 1", + {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + if recs: + return _rec_to_dict(recs[0]) + return None + + +async def _cancel_active_tasks(sor, project_id, iteration_name): + """作废某迭代内所有活跃任务（非终态 → cancelled + 清 claimed_by）。 + + 任务经 params.iteration_id（存迭代名）归属迭代；只作废该迭代的活跃任务， + 已终态(completed/approved/cancelled/failed)的不动。返回作废数量。 + """ + if not iteration_name: + return 0 + recs = await sor.sqlExe( + "SELECT id FROM pipeline_tasks WHERE tenant_id=${pid}$ " + "AND state NOT IN ('completed','approved','cancelled','failed') " + "AND JSON_UNQUOTE(JSON_EXTRACT(params,'$.iteration_id'))=${nm}$", + {"pid": project_id, "nm": iteration_name}) + await sor.sqlExe("COMMIT", {}) + ids = [getattr(r, 'id', '') for r in (recs or []) if getattr(r, 'id', '')] + if ids: + in_clause = ",".join(["'" + x.replace("'", "''") + "'" for x in ids]) + await sor.sqlExe( + f"UPDATE pipeline_tasks SET state='cancelled', claimed_by=NULL, updated_at=NOW() " + f"WHERE id IN ({in_clause})", {}) + return len(ids) + + +async def start_next_iteration(project_id, who=None, agent_id=None): + """启动下一个迭代（用户指令推进）： + + 1. 若存在当前迭代(in_progress)，先强制完成它 + 作废其活跃任务。 + 2. 找下一个 planning 迭代（seq_no 最小且 > 已结束迭代的最大 seq_no），start 它。 + 返回 (ok, message)。 + """ + if not project_id: + return False, "缺少 project_id" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + # 1. 强制完成当前迭代 + 作废其活跃任务 + cur = await get_current_iteration(sor, project_id) + if cur: + cur_id = cur.get('id', '') + cur_name = cur.get('iteration_name', '') or '' + if cur_name: + await _cancel_active_tasks(sor, project_id, cur_name) + await sor.sqlExe( + f"UPDATE {TABLE} SET status='completed', completed_at=NOW(), updated_at=NOW() " + f"WHERE id=${{iid}}$ AND project_id=${{pid}}$ AND status='in_progress'", + {"iid": cur_id, "pid": project_id}) + await record_audit(project_id, TABLE, cur_id, 'complete', + from_state=S_IN_PROGRESS, to_state=S_COMPLETED, + who=_normalize_role(who), agent_id=agent_id, + detail='启动下一迭代时强制完成', sor=sor) + logger.info("start_next_iteration: 强制完成当前迭代 %s (%s)", cur_id, cur_name) + # 2. 找下一个 planning 迭代（seq_no > 已结束迭代的最大 seq_no） + nxt = await sor.sqlExe( + f"SELECT * FROM {TABLE} WHERE project_id=${{pid}}$ AND status='planning' " + f"AND seq_no > (SELECT COALESCE(MAX(seq_no),0) FROM {TABLE} " + f"WHERE project_id=${{pid}}$ AND status IN ('completed','cancelled')) " + "ORDER BY seq_no ASC LIMIT 1", + {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + if not nxt: + return False, "没有可启动的下一个迭代（请先 create_iteration）" + nxt_id = getattr(nxt[0], 'id', '') + nxt_name = getattr(nxt[0], 'iteration_name', '') + nxt_seq = getattr(nxt[0], 'seq_no', '') + # 3. start 它（planning → in_progress） + await sor.sqlExe( + f"UPDATE {TABLE} SET status='in_progress', started_at=NOW(), updated_at=NOW() " + f"WHERE id=${{iid}}$ AND project_id=${{pid}}$ AND status='planning'", + {"iid": nxt_id, "pid": project_id}) + await record_audit(project_id, TABLE, nxt_id, 'start', + from_state=S_PLANNING, to_state=S_IN_PROGRESS, + who=_normalize_role(who), agent_id=agent_id, sor=sor) + logger.info("start_next_iteration: 启动 %s (seq_no=%s)", nxt_name, nxt_seq) + return True, f"已启动迭代「{nxt_name}」(seq_no={nxt_seq})" + + def _rec_to_dict(rec): """把 sqlor 记录对象转成 dict（sqlor 行是 DictObject，必须 dict(rec) 取列）。""" if isinstance(rec, dict): diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py index 68c455e..669dd88 100644 --- a/pipeline_service/sdlc_ability.py +++ b/pipeline_service/sdlc_ability.py @@ -174,6 +174,7 @@ SDL_TOOLS = [ ToolDefinition(name="start_iteration", description="开始迭代(planning→in_progress)", parameters={"iteration_id": "迭代ID"}, category="iteration"), ToolDefinition(name="complete_iteration", description="完成迭代(in_progress→completed)", parameters={"iteration_id": "迭代ID"}, category="iteration"), ToolDefinition(name="cancel_iteration", description="取消迭代(planning/in_progress→cancelled)", parameters={"iteration_id": "迭代ID", "comment": "取消原因(可选)"}, category="iteration"), + ToolDefinition(name="start_next_iteration", description="启动下一个迭代(强制完成当前迭代+作废其活跃任务，再启动编号最小的下一个planning迭代)", parameters={}, category="iteration"), # ── 交付件评审 ── ToolDefinition(name="submit_deliverable", description="提交交付件(pending)", parameters={"title": "交付件标题", "deliverable_type": "类型(code/doc/config等)", "content": "交付件内容", "task_id": "关联任务ID(可选)"}, category="deliverable"), ToolDefinition(name="approve_deliverable", description="评审通过交付件(pending→approved)", parameters={"deliverable_id": "交付件ID"}, category="deliverable"), @@ -387,11 +388,11 @@ async def _h_create_task(sor, p, ctx): if irecs and getattr(irecs[0], 'iteration_name', ''): iteration_name = getattr(irecs[0], 'iteration_name', '') else: - irecs = await sor.sqlExe( - "SELECT iteration_name FROM sd_iterations WHERE project_id=${pid}$ ORDER BY created_at DESC LIMIT 1", - {"pid": pid}) - if irecs: - iteration_name = getattr(irecs[0], 'iteration_name', '') or '' + # 默认当前迭代 = status='in_progress' 的唯一迭代（不再靠 created_at DESC 推断） + from .iteration_capability import get_current_iteration + cur = await get_current_iteration(sor, pid) + if cur: + iteration_name = cur.get('iteration_name', '') or '' params = {"description": desc} if iteration_name: @@ -784,13 +785,13 @@ async def _h_add_bug(sor, p, ctx): iteration_id = p.get("iteration_id", "") if not iteration_id: - its = await sor.sqlExe( - "SELECT id FROM sd_iterations WHERE project_id=${pid}$ ORDER BY created_at ASC LIMIT 1", - {"pid": pid}) - if its: - iteration_id = getattr(its[0], "id", "") + # 默认当前迭代（status='in_progress'），不再用 created_at ASC（最旧）——Bug 与任务归同一迭代 + from .iteration_capability import get_current_iteration + cur = await get_current_iteration(sor, pid) + if cur: + iteration_id = cur.get('id', '') if not iteration_id: - return "ERROR: 项目无迭代，无法提交Bug" + return "ERROR: 项目无当前迭代，无法提交Bug" await sor.C("sd_bugs", { "id": getID(), "title": title, @@ -923,18 +924,17 @@ async def _get_scope(sor, table, id_field, id_val, scope_field): async def _resolve_iteration_id(sor, pid, iteration_id): - """解析迭代 ID：显式传入则用，否则取项目第一个迭代。返回 (iteration_id, error)。""" + """解析迭代 ID：显式传入则用，否则取当前迭代（status='in_progress'）。返回 (iteration_id, error)。""" if iteration_id: full = await _resolve_id(sor, "sd_iterations", iteration_id) if full: return full, "" return "", f"迭代不存在: {iteration_id}" - recs = await sor.sqlExe( - "SELECT id FROM sd_iterations WHERE project_id=${pid}$ ORDER BY created_at DESC LIMIT 1", - {"pid": pid}) - if recs: - return getattr(recs[0], "id", ""), "" - return "", "项目无迭代，请先 create_iteration" + from .iteration_capability import get_current_iteration + cur = await get_current_iteration(sor, pid) + if cur: + return cur.get('id', ''), "" + return "", "项目无当前迭代，请先 create_iteration 并 start" async def _resolve_plan_id(sor, pid, plan_id): @@ -1072,10 +1072,21 @@ async def _h_list_iterations(sor, p, ctx): return "暂无迭代" lines = [] for r in lst: - lines.append(f"- [{r.get('status', '?')}] {r.get('iteration_name', '')} (id={r.get('id', '')})") + mark = "▶" if r.get('status', '') == 'in_progress' else " " + lines.append(f"- {mark}[{r.get('status', '?')}] #{r.get('seq_no', '?')} {r.get('iteration_name', '')} (id={r.get('id', '')})") return "\n".join(lines) +async def _h_start_next_iteration(sor, p, ctx): + pid = ctx.get("project_id", "") + if not pid: + return "请先切换到项目" + from .iteration_capability import start_next_iteration + ok, msg = await start_next_iteration(pid, who="agent.main_agent", + agent_id=ctx.get("user_id", "") or "") + return f"OK: {msg}" if ok else f"ERROR: {msg}" + + async def _h_create_iteration(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: @@ -1572,6 +1583,7 @@ SDL_HANDLERS = { "start_iteration": _h_start_iteration, "complete_iteration": _h_complete_iteration, "cancel_iteration": _h_cancel_iteration, + "start_next_iteration": _h_start_next_iteration, # ── 交付件评审 ── "submit_deliverable": _h_submit_deliverable, "approve_deliverable": _h_approve_deliverable,