feat: 项目owner+人类任务清单+bug验收+流转门禁
- check_project_owner: owner=created_by 统一校验 - human_task_capability: 派发/处理/QC/列表/计数/bug验收(复用pipeline_human_tasks) - bug_flow: verified 分流(human报告→创建验收任务给报告人, agent→自动close) - qc_review_run: human_task_qc 分支(检查对象=pipeline_human_tasks 结果) - 流转门禁: agent/pm poller + start_next_iteration 检查未完成人类任务 - pipeline_human_tasks 加 project_id/iteration_id/bug_id/qc_status/qc_comment/title/description
This commit is contained in:
ymq
parent
c49bb1c756
commit
d417772f42
@ -27,7 +27,9 @@
|
||||
{"name": "iteration_id", "title": "迭代ID", "type": "str", "length": 32},
|
||||
{"name": "bug_id", "title": "关联BugID", "type": "str", "length": 32},
|
||||
{"name": "qc_status", "title": "QC状态", "type": "str", "length": 20, "default": "pending"},
|
||||
{"name": "qc_comment", "title": "QC意见", "type": "str", "length": 500}
|
||||
{"name": "qc_comment", "title": "QC意见", "type": "str", "length": 500},
|
||||
{"name": "title", "title": "任务标题", "type": "str", "length": 500},
|
||||
{"name": "description", "title": "任务描述", "type": "text"}
|
||||
],
|
||||
"indexes": [
|
||||
{"name": "idx_pht_task", "idxtype": "index", "idxfields": ["task_id"]},
|
||||
|
||||
@ -2023,6 +2023,16 @@ async def qc_review_run(project_id, agent_id=None, model_name=None):
|
||||
task_id = task.id
|
||||
title = getattr(task, "title", "") or ""
|
||||
task_role = _normalize_role(getattr(task, "role", "") or "")
|
||||
# 判断是否 human_task_qc（人类任务 QC，检查对象是 pipeline_human_tasks 而非交付件）
|
||||
task_kind = ""
|
||||
human_task_id = ""
|
||||
try:
|
||||
_tp = json.loads(getattr(task, "params", "") or "{}")
|
||||
if isinstance(_tp, dict):
|
||||
task_kind = _tp.get("task_kind", "") or ""
|
||||
human_task_id = _tp.get("human_task_id", "") or ""
|
||||
except Exception:
|
||||
pass
|
||||
# 机构 llm 检测：机构没配 llm → 冒泡问题暂停
|
||||
llm_missing, _names = await _check_org_llm(sor, org_id)
|
||||
if llm_missing:
|
||||
@ -2043,11 +2053,32 @@ async def qc_review_run(project_id, agent_id=None, model_name=None):
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id)
|
||||
if not deliverable_content:
|
||||
from .task_capability import qc_reject_task
|
||||
await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment="没有交付件")
|
||||
return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"}
|
||||
if task_kind == 'human_task_qc' and human_task_id:
|
||||
# 人类任务 QC：读 pipeline_human_tasks 的处理结果作为检查对象
|
||||
ht_recs = await sor.sqlExe(
|
||||
"SELECT title, description, result_data FROM pipeline_human_tasks WHERE id=${hid}$",
|
||||
{"hid": human_task_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not ht_recs:
|
||||
from .task_capability import qc_reject_task
|
||||
await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment="人类任务不存在")
|
||||
return {"status": "rejected", "task_id": task_id, "reason": "人类任务不存在"}
|
||||
ht_title = getattr(ht_recs[0], 'title', '') or ''
|
||||
ht_desc = getattr(ht_recs[0], 'description', '') or ''
|
||||
ht_result = getattr(ht_recs[0], 'result_data', '') or ''
|
||||
deliverable_content = f"人类任务标题：{ht_title}\n任务描述：{ht_desc}\n\n处理结果：\n{ht_result}"
|
||||
deliverable_type = "human_task"
|
||||
title = ht_title or title
|
||||
if not ht_result.strip():
|
||||
from .task_capability import qc_reject_task
|
||||
await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment="人类任务无处理结果")
|
||||
return {"status": "rejected", "task_id": task_id, "reason": "人类任务无处理结果"}
|
||||
else:
|
||||
deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id)
|
||||
if not deliverable_content:
|
||||
from .task_capability import qc_reject_task
|
||||
await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment="没有交付件")
|
||||
return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"}
|
||||
|
||||
content_preview = deliverable_content[:6000]
|
||||
role_skills = await _build_role_skills_block(sor, project_id, _normalize_role('qc'), org_id)
|
||||
@ -2107,12 +2138,22 @@ async def qc_review_run(project_id, agent_id=None, model_name=None):
|
||||
if decision['status'] == 'approved':
|
||||
from .task_capability import qc_approve_task
|
||||
await qc_approve_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment=decision.get('comment', ''))
|
||||
if task_kind == 'human_task_qc' and human_task_id:
|
||||
from .human_task_capability import qc_human_task
|
||||
await qc_human_task(human_task_id, True, decision.get('comment', ''), operator_id='agent.qc')
|
||||
logger.info(f"qc_review_run approve human_task: {human_task_id} qc passed")
|
||||
logger.info(f"qc_review_run approve: task={task_id} -> review")
|
||||
return {"status": "approved", "task_id": task_id, "comment": decision.get('comment', '')}
|
||||
else:
|
||||
comment = decision.get('comment', '')
|
||||
from .task_capability import qc_reject_task
|
||||
ok, msg = await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment=comment)
|
||||
if task_kind == 'human_task_qc' and human_task_id:
|
||||
# 人类任务 QC 不通过：标记退回重做（qc_status=rejected + status=pending），不冒泡给 agent 角色
|
||||
from .human_task_capability import qc_human_task
|
||||
await qc_human_task(human_task_id, False, comment, operator_id='agent.qc')
|
||||
logger.info(f"qc_review_run reject human_task: {human_task_id} 退回重做")
|
||||
return {"status": "rejected", "task_id": task_id, "comment": comment}
|
||||
if not ok and "failed" in msg:
|
||||
# 重复退回达上限 → 已转 failed，不再建 qc_reject 问题（failed poller 会报 fault_report 给人工）
|
||||
logger.info(f"qc_review_run reject->failed: task={task_id} {msg}")
|
||||
|
||||
@ -78,18 +78,60 @@ async def _has_active_task(sor, project_id, iteration_name, task_kind):
|
||||
return c > 0
|
||||
|
||||
|
||||
async def _advance_bug_states(sor):
|
||||
"""一轮 bug 循环：确定性流转（verified→close / fixed→复测 / confirmed→修复）。"""
|
||||
result = {"closed": 0, "fix_tasks": 0, "retest_tasks": 0}
|
||||
async def _create_bug_acceptance_task(sor, bug_id, iteration_id, reporter_id):
|
||||
"""为 human 报告的 verified bug 创建验收任务（幂等：已有 pending 验收任务则不重复建）。"""
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT i.project_id, i.iteration_name, b.title FROM sd_bugs b "
|
||||
"JOIN sd_iterations i ON b.iteration_id=i.id WHERE b.id=${bid}$",
|
||||
{"bid": bug_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not recs:
|
||||
return None
|
||||
pid = getattr(recs[0], 'project_id', '') or ''
|
||||
title = getattr(recs[0], 'title', '') or ''
|
||||
# 幂等：已有 pending 验收任务则不重复建
|
||||
ex = await sor.sqlExe(
|
||||
"SELECT COUNT(*) AS c FROM pipeline_human_tasks "
|
||||
"WHERE bug_id=${bid}$ AND task_type='bug_acceptance' AND status='pending'",
|
||||
{"bid": bug_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if ex and getattr(ex[0], 'c', 0) > 0:
|
||||
return None
|
||||
hid = getID()
|
||||
await sor.C('pipeline_human_tasks', {
|
||||
'id': hid, 'task_id': '', 'step_name': '', 'version': 1,
|
||||
'task_type': 'bug_acceptance',
|
||||
'assignee_role': '', 'assignee_id': reporter_id or '',
|
||||
'form_schema': None, 'result_data': None,
|
||||
'status': 'pending',
|
||||
'project_id': pid, 'iteration_id': iteration_id, 'bug_id': bug_id,
|
||||
'qc_status': 'pending',
|
||||
'title': f"Bug 验收：{title}",
|
||||
'description': "您报告的 Bug 已修复并测试通过，请验收（通过则关闭，不通过则重新打开）。",
|
||||
})
|
||||
logger.info(f"bug_flow: 创建 bug 验收任务 {hid} (bug={bug_id}, reporter={reporter_id})")
|
||||
return hid
|
||||
|
||||
# 1) verified → closed（自动关闭，无需 LLM）
|
||||
|
||||
async def _advance_bug_states(sor):
|
||||
"""一轮 bug 循环：确定性流转（verified→close/验收 / fixed→复测 / confirmed→修复）。"""
|
||||
result = {"closed": 0, "accept_tasks": 0, "fix_tasks": 0, "retest_tasks": 0}
|
||||
|
||||
# 1) verified → 分流：agent 报告自动 close；human 报告创建 bug 验收任务（等报告人验收）
|
||||
verified = await sor.sqlExe(
|
||||
"SELECT id, iteration_id FROM sd_bugs WHERE status='verified' LIMIT 100", {})
|
||||
"SELECT id, iteration_id, reporter_type, reporter_id FROM sd_bugs "
|
||||
"WHERE status='verified' LIMIT 100", {})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
for r in (verified or []):
|
||||
d = _rec_to_dict(r)
|
||||
bid, iid = d.get('id', ''), d.get('iteration_id', '')
|
||||
if bid and iid:
|
||||
if not bid or not iid:
|
||||
continue
|
||||
if (d.get('reporter_type', '') or '') == 'human':
|
||||
# human 报告 → 创建验收任务（幂等），bug 停在 verified 等报告人验收
|
||||
await _create_bug_acceptance_task(sor, bid, iid, d.get('reporter_id', '') or '')
|
||||
result["accept_tasks"] += 1
|
||||
else:
|
||||
ok, _ = await bug_capability.close_bug(bid, iid, who='agent.bug_flow')
|
||||
if ok:
|
||||
result["closed"] += 1
|
||||
|
||||
412
pipeline_service/human_task_capability.py
Normal file
412
pipeline_service/human_task_capability.py
Normal file
@ -0,0 +1,412 @@
|
||||
"""SDLC 项目级人类任务清单 capability。
|
||||
|
||||
人类待办三类来源（统一视图，见 list_my_human_todos / count_my_human_tasks）：
|
||||
- general 显式人类任务 → 落 pipeline_human_tasks（owner/PM 派发）
|
||||
- bug_acceptance bug 验收任务 → 落 pipeline_human_tasks（bug_flow 在 verified 时创建）
|
||||
- question 冒泡问题 → 不落此表，union 查 pipeline_agent_questions
|
||||
|
||||
处理权限：处理者须同机构（org_id 匹配）+ 匹配 assignee_role 或 assignee_id。
|
||||
|
||||
状态：status(pending/done/rejected) + qc_status(pending/passed/rejected，仅 general 走 QC)。
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
|
||||
from sqlor.dbpools import DBPools
|
||||
from appPublic.uniqueID import getID
|
||||
|
||||
DBNAME = "pipeline"
|
||||
logger = logging.getLogger("pipeline.human_task_capability")
|
||||
|
||||
S_PENDING = "pending"
|
||||
S_DONE = "done"
|
||||
S_REJECTED = "rejected"
|
||||
|
||||
QC_PENDING = "pending"
|
||||
QC_PASSED = "passed"
|
||||
QC_REJECTED = "rejected"
|
||||
|
||||
T_GENERAL = "general"
|
||||
T_BUG_ACCEPT = "bug_acceptance"
|
||||
|
||||
|
||||
def _get_db():
|
||||
db = DBPools()
|
||||
if not db.databases:
|
||||
from appPublic.jsonConfig import getConfig
|
||||
config = getConfig()
|
||||
if config.databases:
|
||||
db.databases = config.databases
|
||||
return db, DBNAME
|
||||
|
||||
|
||||
def _rec_to_dict(rec):
|
||||
if rec is None:
|
||||
return {}
|
||||
if isinstance(rec, dict):
|
||||
return dict(rec)
|
||||
try:
|
||||
return dict(rec)
|
||||
except (TypeError, ValueError):
|
||||
return {}
|
||||
|
||||
|
||||
async def _get_user_org(sor, user_id):
|
||||
"""查用户 org_id。"""
|
||||
if not user_id:
|
||||
return ""
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT orgid FROM users WHERE id=${u}$ LIMIT 1", {"u": user_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
return getattr(recs[0], 'orgid', '') if recs else ''
|
||||
|
||||
|
||||
async def _get_user_roles(sor, user_id):
|
||||
"""查用户 RBAC 角色名列表（{orgtypeid}.{name} 格式）。"""
|
||||
roles = []
|
||||
if not user_id:
|
||||
return roles
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT r.orgtypeid, r.name FROM userrole ur JOIN role r ON ur.roleid=r.id "
|
||||
"WHERE ur.userid=${u}$", {"u": user_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
for r in (recs or []):
|
||||
o = getattr(r, 'orgtypeid', '') or ''
|
||||
n = getattr(r, 'name', '') or ''
|
||||
if o and n:
|
||||
roles.append(f"{o}.{n}")
|
||||
return roles
|
||||
|
||||
|
||||
async def create_human_task(project_id, title, description="", task_type=T_GENERAL,
|
||||
assignee_role=None, assignee_id=None,
|
||||
iteration_id=None, bug_id=None, created_by=None):
|
||||
"""派发项目级人类任务。assignee_role 或 assignee_id 至少指定一个。
|
||||
|
||||
返回 (True, human_task_id) 或 (False, 错误信息)。
|
||||
"""
|
||||
if not project_id:
|
||||
return False, "缺少 project_id"
|
||||
if not title or not title.strip():
|
||||
return False, "缺少标题"
|
||||
if not assignee_role and not assignee_id:
|
||||
return False, "须指定 assignee_role 或 assignee_id"
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
# 项目存在性 + 迭代（未显式传时取当前迭代）
|
||||
proj = await sor.sqlExe(
|
||||
"SELECT id, org_id FROM sd_projects WHERE id=${pid}$", {"pid": project_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not proj:
|
||||
return False, "项目不存在"
|
||||
if not iteration_id:
|
||||
it = await sor.sqlExe(
|
||||
"SELECT id FROM sd_iterations WHERE project_id=${pid}$ "
|
||||
"AND status='in_progress' ORDER BY seq_no DESC LIMIT 1",
|
||||
{"pid": project_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
iteration_id = getattr(it[0], 'id', '') if it else ''
|
||||
|
||||
hid = getID()
|
||||
await sor.C('pipeline_human_tasks', {
|
||||
'id': hid,
|
||||
'task_id': '', # 项目级人类任务不绑定引擎任务
|
||||
'step_name': '', # 同上
|
||||
'version': 1,
|
||||
'task_type': task_type or T_GENERAL,
|
||||
'assignee_role': assignee_role or '',
|
||||
'assignee_id': assignee_id or '',
|
||||
'form_schema': None,
|
||||
'result_data': None,
|
||||
'status': S_PENDING,
|
||||
'submitted_by': created_by or '',
|
||||
'project_id': project_id,
|
||||
'iteration_id': iteration_id or '',
|
||||
'bug_id': bug_id or '',
|
||||
'qc_status': QC_PENDING,
|
||||
'title': title.strip(),
|
||||
'description': description or '',
|
||||
})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
logger.info("create_human_task: %s project=%s type=%s title=%s",
|
||||
hid, project_id, task_type, title.strip())
|
||||
return True, hid
|
||||
|
||||
|
||||
async def complete_human_task(human_task_id, result_data, operator_id=None):
|
||||
"""处理者提交完成（done）。校验同机构 + 匹配 assignee。
|
||||
|
||||
general 任务 done 后创建 qc_review 的 QC 任务（复用 qc poller → qc_review_run）。
|
||||
bug_acceptance 任务不走这里（走 bug_accept）。
|
||||
"""
|
||||
if not human_task_id:
|
||||
return False, "缺少 human_task_id"
|
||||
if not operator_id:
|
||||
return False, "未登录"
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT * FROM pipeline_human_tasks WHERE id=${hid}$", {"hid": human_task_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not recs:
|
||||
return False, "人类任务不存在"
|
||||
ht = _rec_to_dict(recs[0])
|
||||
if ht.get('status') != S_PENDING:
|
||||
return False, f"任务不在待处理状态 (当前: {ht.get('status')})"
|
||||
|
||||
# 同机构校验
|
||||
pid = ht.get('project_id', '')
|
||||
proj = await sor.sqlExe(
|
||||
"SELECT org_id FROM sd_projects WHERE id=${pid}$", {"pid": pid})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
proj_org = getattr(proj[0], 'org_id', '') if proj else ''
|
||||
user_org = await _get_user_org(sor, operator_id)
|
||||
if proj_org and user_org != proj_org:
|
||||
return False, "仅同机构用户可处理该任务"
|
||||
|
||||
# 匹配 assignee：assignee_id 指定则须等于本人；assignee_role 指定则本人角色须含之
|
||||
assignee_id = ht.get('assignee_id', '') or ''
|
||||
assignee_role = ht.get('assignee_role', '') or ''
|
||||
if assignee_id:
|
||||
if operator_id != assignee_id:
|
||||
return False, "该任务未指派给你"
|
||||
elif assignee_role:
|
||||
roles = await _get_user_roles(sor, operator_id)
|
||||
if assignee_role not in roles:
|
||||
return False, "该任务未指派给你的角色"
|
||||
|
||||
task_type = ht.get('task_type', '') or T_GENERAL
|
||||
result_json = json.dumps(result_data, ensure_ascii=False, default=str) \
|
||||
if isinstance(result_data, (dict, list)) else str(result_data or '')
|
||||
|
||||
if task_type == T_BUG_ACCEPT:
|
||||
return False, "bug 验收任务请走 bug_accept 接口"
|
||||
|
||||
# general：done + 触发 QC（创建 qc_review 任务）
|
||||
await sor.sqlExe(
|
||||
"UPDATE pipeline_human_tasks SET status='done', result_data=${rd}$, "
|
||||
"submitted_by=${oid}$, submitted_at=NOW() WHERE id=${hid}$",
|
||||
{"rd": result_json, "oid": operator_id, "hid": human_task_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
|
||||
# 创建 QC 任务（复用 qc poller → qc_review_run）
|
||||
qid = getID()
|
||||
qparams = {
|
||||
"task_kind": "human_task_qc",
|
||||
"human_task_id": human_task_id,
|
||||
"title": ht.get('title', '') or '',
|
||||
"description": (ht.get('description', '') or '')[:2000],
|
||||
"iteration_id": ht.get('iteration_id', '') or '',
|
||||
}
|
||||
# 补 title（pipeline_human_tasks 无 title 列，用 result 里带或 form_schema 里带）
|
||||
await sor.C('pipeline_tasks', {
|
||||
'id': qid, 'tenant_id': pid, 'pipeline_id': 'role_task',
|
||||
'owner_id': 'human_task_qc',
|
||||
'title': f"人类任务 QC 检查（{human_task_id[:8]}）",
|
||||
'state': 'qc_review', 'role': 'agent.qc',
|
||||
'params': json.dumps(qparams, ensure_ascii=False),
|
||||
})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
return True, human_task_id
|
||||
|
||||
|
||||
async def qc_human_task(human_task_id, passed, comment=None, operator_id=None):
|
||||
"""QC 检查人类任务（general）。passed → qc_status=passed；否则 qc_status=rejected + status=pending 退回。"""
|
||||
if not human_task_id:
|
||||
return False, "缺少 human_task_id"
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT status FROM pipeline_human_tasks WHERE id=${hid}$", {"hid": human_task_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not recs:
|
||||
return False, "人类任务不存在"
|
||||
if passed:
|
||||
await sor.sqlExe(
|
||||
"UPDATE pipeline_human_tasks SET qc_status='passed', qc_comment=${c}$ "
|
||||
"WHERE id=${hid}$", {"c": comment or '', "hid": human_task_id})
|
||||
else:
|
||||
# 退回重做：status 回 pending，qc_status=rejected
|
||||
await sor.sqlExe(
|
||||
"UPDATE pipeline_human_tasks SET status='pending', qc_status='rejected', "
|
||||
"qc_comment=${c}$ WHERE id=${hid}$", {"c": comment or '', "hid": human_task_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
return True, human_task_id
|
||||
|
||||
|
||||
async def list_project_human_tasks(project_id, status=None, assignee_id=None):
|
||||
"""列出项目人类任务（显式任务 + bug 验收，不含冒泡问题）。"""
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
conds = ["project_id=${pid}$"]
|
||||
params = {"pid": project_id}
|
||||
if status:
|
||||
conds.append("status=${st}$")
|
||||
params["st"] = status
|
||||
if assignee_id:
|
||||
conds.append("assignee_id=${aid}$")
|
||||
params["aid"] = assignee_id
|
||||
where = " AND ".join(conds)
|
||||
recs = await sor.sqlExe(
|
||||
f"SELECT * FROM pipeline_human_tasks WHERE {where} ORDER BY created_at DESC",
|
||||
params)
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
return [_rec_to_dict(r) for r in (recs or [])]
|
||||
|
||||
|
||||
async def has_blocking_human_task(sor, project_id):
|
||||
"""当前迭代是否有未完成的人类任务（流转门禁用）。
|
||||
|
||||
general：status=pending 或 status=done 但 qc_status!=passed → 阻塞
|
||||
bug_acceptance：status=pending → 阻塞（done/rejected = 已验收出结果，不阻塞）
|
||||
"""
|
||||
if not project_id:
|
||||
return False
|
||||
it = await sor.sqlExe(
|
||||
"SELECT id FROM sd_iterations WHERE project_id=${pid}$ AND status='in_progress' "
|
||||
"ORDER BY seq_no DESC LIMIT 1", {"pid": project_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not it:
|
||||
return False
|
||||
iid = getattr(it[0], 'id', '') or ''
|
||||
if not iid:
|
||||
return False
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT COUNT(*) AS c FROM pipeline_human_tasks "
|
||||
"WHERE iteration_id=${iid}$ AND ("
|
||||
" (task_type='general' AND (status='pending' OR (status='done' AND qc_status != 'passed'))) "
|
||||
" OR (task_type='bug_acceptance' AND status='pending')"
|
||||
")",
|
||||
{"iid": iid})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
c = getattr(recs[0], 'c', 0) if recs else 0
|
||||
return int(c) > 0
|
||||
|
||||
|
||||
async def count_my_human_tasks(user_id, project_id=None):
|
||||
"""角标 X = pending 问题(agentid=我) + 人类任务(assignee_id=我) + 人类任务(assignee_role∈我角色)。"""
|
||||
if not user_id:
|
||||
return 0
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
# ① pending 问题 current_handler_agentid = 我
|
||||
q_cond = "status='pending' AND current_handler_agentid=${u}$"
|
||||
q_params = {"u": user_id}
|
||||
if project_id:
|
||||
q_cond += " AND tenant_id=${pid}$"
|
||||
q_params["pid"] = project_id
|
||||
q = await sor.sqlExe(
|
||||
f"SELECT COUNT(*) AS c FROM pipeline_agent_questions WHERE {q_cond}", q_params)
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
|
||||
# ② + ③ 人类任务（assignee_id=我 或 assignee_role∈我角色）
|
||||
roles = await _get_user_roles(sor, user_id)
|
||||
ht_cond = "status='pending' AND (assignee_id=${u}$"
|
||||
ht_params = {"u": user_id}
|
||||
if roles:
|
||||
ph = ",".join(f"'{r}'" for r in roles)
|
||||
ht_cond += f" OR assignee_role IN ({ph})"
|
||||
ht_cond += ")"
|
||||
if project_id:
|
||||
ht_cond += " AND project_id=${pid}$"
|
||||
ht_params["pid"] = project_id
|
||||
ht = await sor.sqlExe(
|
||||
f"SELECT COUNT(*) AS c FROM pipeline_human_tasks WHERE {ht_cond}", ht_params)
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
|
||||
n_q = getattr(q[0], 'c', 0) if q else 0
|
||||
n_ht = getattr(ht[0], 'c', 0) if ht else 0
|
||||
return int(n_q) + int(n_ht)
|
||||
|
||||
|
||||
async def list_my_human_todos(user_id, limit=100):
|
||||
"""跨项目「我的待办」列表：union 人类任务 + 冒泡问题，按创建时间倒序。"""
|
||||
if not user_id:
|
||||
return []
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
roles = await _get_user_roles(sor, user_id)
|
||||
role_ph = ",".join(f"'{r}'" for r in roles) if roles else "''"
|
||||
|
||||
# 人类任务：assignee_id=我 或 assignee_role∈我角色，pending
|
||||
ht = await sor.sqlExe(
|
||||
f"SELECT id, project_id, iteration_id, bug_id, task_type, assignee_role, "
|
||||
f"assignee_id, status, qc_status, created_at, submitted_by, result_data "
|
||||
f"FROM pipeline_human_tasks "
|
||||
f"WHERE status='pending' AND (assignee_id=${{u}}$ OR assignee_role IN ({role_ph})) "
|
||||
f"ORDER BY created_at DESC LIMIT {limit}", {"u": user_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
|
||||
# 冒泡问题：current_handler_agentid=我 或 current_handler_role∈我角色，pending
|
||||
qs = await sor.sqlExe(
|
||||
f"SELECT id, tenant_id AS project_id, task_id, from_role, question, "
|
||||
f"problem_type, current_handler_role, current_handler_agentid, created_at "
|
||||
f"FROM pipeline_agent_questions "
|
||||
f"WHERE status='pending' AND (current_handler_agentid=${{u}}$ "
|
||||
f"OR current_handler_role IN ({role_ph})) "
|
||||
f"ORDER BY created_at DESC LIMIT {limit}", {"u": user_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
|
||||
todos = []
|
||||
for r in (ht or []):
|
||||
d = _rec_to_dict(r)
|
||||
d['source'] = 'human_task'
|
||||
todos.append(d)
|
||||
for r in (qs or []):
|
||||
d = _rec_to_dict(r)
|
||||
d['source'] = 'question'
|
||||
todos.append(d)
|
||||
todos.sort(key=lambda x: str(x.get('created_at') or ''), reverse=True)
|
||||
return todos[:limit]
|
||||
|
||||
|
||||
async def bug_accept(bug_id, accept, user_id, comment=None):
|
||||
"""bug 验收：accept=True → bug closed + 验收任务 done；accept=False → bug reopen + 任务 rejected。
|
||||
|
||||
校验 user_id == bug.reporter_id。返回 (True, 消息) 或 (False, 错误)。
|
||||
"""
|
||||
if not bug_id:
|
||||
return False, "缺少 bug_id"
|
||||
if not user_id:
|
||||
return False, "未登录"
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
brecs = await sor.sqlExe(
|
||||
"SELECT reporter_id, iteration_id, status FROM sd_bugs WHERE id=${bid}$",
|
||||
{"bid": bug_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not brecs:
|
||||
return False, "Bug 不存在"
|
||||
reporter_id = getattr(brecs[0], 'reporter_id', '') or ''
|
||||
iteration_id = getattr(brecs[0], 'iteration_id', '') or ''
|
||||
if reporter_id and user_id != reporter_id:
|
||||
return False, "仅 Bug 报告人可验收"
|
||||
|
||||
if accept:
|
||||
from . import bug_capability
|
||||
ok, msg = await bug_capability.close_bug(bug_id, iteration_id, who='human')
|
||||
if not ok:
|
||||
return False, f"关闭失败: {msg}"
|
||||
# 验收任务 → done
|
||||
await sor.sqlExe(
|
||||
"UPDATE pipeline_human_tasks SET status='done', result_data=${c}$, "
|
||||
"submitted_by=${u}$, submitted_at=NOW() "
|
||||
"WHERE bug_id=${bid}$ AND task_type='bug_acceptance' AND status='pending'",
|
||||
{"c": json.dumps({"accept": True, "comment": comment or ''}, ensure_ascii=False),
|
||||
"u": user_id, "bid": bug_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
return True, "验收通过，Bug 已关闭"
|
||||
else:
|
||||
from . import bug_capability
|
||||
ok, msg = await bug_capability.reopen_bug(bug_id, iteration_id, who='human')
|
||||
if not ok:
|
||||
return False, f"重开失败: {msg}"
|
||||
await sor.sqlExe(
|
||||
"UPDATE pipeline_human_tasks SET status='rejected', result_data=${c}$, "
|
||||
"submitted_by=${u}$, submitted_at=NOW() "
|
||||
"WHERE bug_id=${bid}$ AND task_type='bug_acceptance' AND status='pending'",
|
||||
{"c": json.dumps({"accept": False, "comment": comment or ''}, ensure_ascii=False),
|
||||
"u": user_id, "bid": bug_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
return True, "验收不通过，Bug 已重新打开"
|
||||
@ -509,6 +509,20 @@ def load_pipeline_service():
|
||||
env.approval_reject = approval_reject
|
||||
env.human_task_list = human_list
|
||||
|
||||
# SDLC 项目级人类任务清单 + owner 校验
|
||||
from .human_task_capability import (
|
||||
create_human_task, complete_human_task, qc_human_task,
|
||||
list_project_human_tasks, count_my_human_tasks, list_my_human_todos, bug_accept)
|
||||
from .project_capability import check_project_owner
|
||||
env.create_human_task = create_human_task
|
||||
env.complete_human_task = complete_human_task
|
||||
env.qc_human_task = qc_human_task
|
||||
env.list_project_human_tasks = list_project_human_tasks
|
||||
env.count_my_human_tasks = count_my_human_tasks
|
||||
env.list_my_human_todos = list_my_human_todos
|
||||
env.bug_accept = bug_accept
|
||||
env.check_project_owner = check_project_owner
|
||||
|
||||
# Register default handler
|
||||
register_default_handler()
|
||||
|
||||
@ -647,6 +661,13 @@ def load_pipeline_service():
|
||||
"WHERE t.state='submitted' AND t.pipeline_id='role_task' "
|
||||
"AND t.claimed_by IS NULL "
|
||||
"AND NOT EXISTS (SELECT 1 FROM sd_projects p WHERE p.id=t.tenant_id AND p.status='paused') "
|
||||
"AND NOT EXISTS ("
|
||||
" SELECT 1 FROM pipeline_human_tasks ht WHERE ht.iteration_id = ("
|
||||
" SELECT i.id FROM sd_iterations i WHERE i.project_id=t.tenant_id "
|
||||
" AND i.status='in_progress' ORDER BY i.seq_no DESC LIMIT 1) "
|
||||
" AND ((ht.task_type='general' AND (ht.status='pending' OR (ht.status='done' AND ht.qc_status != 'passed'))) "
|
||||
" OR (ht.task_type='bug_acceptance' AND ht.status='pending'))"
|
||||
") "
|
||||
"ORDER BY t.created_at ASC LIMIT 200",
|
||||
{})
|
||||
|
||||
@ -749,6 +770,13 @@ def load_pipeline_service():
|
||||
"SELECT t.id, t.tenant_id, t.role FROM pipeline_tasks t "
|
||||
"WHERE t.state='review' AND t.claimed_by IS NULL "
|
||||
"AND NOT EXISTS (SELECT 1 FROM sd_projects p WHERE p.id=t.tenant_id AND p.status='paused') "
|
||||
"AND NOT EXISTS ("
|
||||
" SELECT 1 FROM pipeline_human_tasks ht WHERE ht.iteration_id = ("
|
||||
" SELECT i.id FROM sd_iterations i WHERE i.project_id=t.tenant_id "
|
||||
" AND i.status='in_progress' ORDER BY i.seq_no DESC LIMIT 1) "
|
||||
" AND ((ht.task_type='general' AND (ht.status='pending' OR (ht.status='done' AND ht.qc_status != 'passed'))) "
|
||||
" OR (ht.task_type='bug_acceptance' AND ht.status='pending'))"
|
||||
") "
|
||||
"ORDER BY t.created_at ASC LIMIT 200",
|
||||
{})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
|
||||
@ -228,6 +228,10 @@ async def start_next_iteration(project_id, who=None, agent_id=None, confirm=Fals
|
||||
return False, "缺少 project_id"
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
# 人类任务门禁：当前迭代有未完成人类任务时，阻止切迭代（须先完成 + QC 合格）
|
||||
from .human_task_capability import has_blocking_human_task
|
||||
if await has_blocking_human_task(sor, project_id):
|
||||
return False, "当前迭代存在未完成的人类任务，须完成并通过 QC 后才能切换迭代"
|
||||
cur = await get_current_iteration(sor, project_id)
|
||||
# 二次确认：当前迭代有活跃任务时，须 confirm=True 才作废（否则返回确认提示）
|
||||
if cur and not confirm:
|
||||
|
||||
@ -104,6 +104,30 @@ async def create_project(name, project_type="web_app", description="",
|
||||
return True, pid
|
||||
|
||||
|
||||
async def check_project_owner(project_id, user_id):
|
||||
"""校验 user_id 是否为项目 owner（sd_projects.created_by）。
|
||||
|
||||
独立开 context（供 pipeline-task 等 dspy 直接调用）。
|
||||
返回 (True, '') 或 (False, 错误信息)。
|
||||
"""
|
||||
if not project_id:
|
||||
return False, "缺少 project_id"
|
||||
if not user_id:
|
||||
return False, "未登录"
|
||||
db, dbname = _get_db()
|
||||
async with db.sqlorContext(dbname) as sor:
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT created_by, name FROM sd_projects WHERE id=${pid}$",
|
||||
{"pid": project_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not recs:
|
||||
return False, "项目不存在"
|
||||
owner = getattr(recs[0], 'created_by', '') or ''
|
||||
if owner == user_id:
|
||||
return True, ''
|
||||
return False, "仅项目 owner 可执行此操作"
|
||||
|
||||
|
||||
async def start_project(project_id, who=None, agent_id=None):
|
||||
"""启动项目：draft → active（也兼容 archived → active 重新激活）。"""
|
||||
ok, msg = await _transition(project_id, S_DRAFT, S_ACTIVE, 'start',
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user