feat(iteration): 迭代状态机改造——显式「当前迭代」+ 序号 seq_no + 启动下一迭代，废除 created_at 推断
背景：用户明确迭代状态机语义——当前迭代唯一(状态 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
This commit is contained in:
ymq
parent
eb910bc7b8
commit
d77910ea2e
@ -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 互引，第二遍解析）
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user