"""任务能力 — pipeline_tasks 的状态机语义化迁移（通用、产线无关）。 定位：storage.py 是数据 CRUD 层（create_task/list_tasks/get_task/update_task_state 低层改状态）， 本模块是状态机语义化操作层 —— 每个迁移 CAS 原子（防并发/防越权覆盖）+ 租户隔离 + 审计。 任务的状态机/流转规则在 task skill 里（LLM 读 skill 判断合法性）；本模块只固化操作原语。 角色规范：role 参数用 agent.{role}（无前缀自动补 agent.）；人角色用 {orgtype}.{role}。 """ import json import logging from sqlor.dbpools import DBPools from appPublic.uniqueID import getID from .audit import record_audit DBNAME = "pipeline" logger = logging.getLogger("pipeline.task_capability") # 任务状态（SDLC 默认，状态机语义见 task skill） S_SUBMITTED = "submitted" # 待认领 S_RUNNING = "running" # 角色 agent 执行中 S_QC_REVIEW = "qc_review" # 已交付，待 QC 合规/质量检查 S_REVIEW = "review" # QC 通过，待 PM 审核验收 S_APPROVED = "approved" # 审核通过 S_COMPLETED = "completed" # 全流程完成 S_WAITING = "waiting" # 挂起等回答 S_FAILED = "failed" # 失败 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 _normalize_role(role): """角色规范：agent 角色补 agent. 前缀；人角色 {orgtype}.{role} 保留原样。""" role = (role or "").strip() if not role: return "" if "." in role: return role return f"agent.{role}" async def _transition(task_id, tenant_id, from_state, to_state, action, who=None, agent_id=None, detail=None): """CAS 状态迁移（清 claimed_by）+ 审计。返回 (ok, message)。""" if not task_id or not tenant_id: return False, "缺少 task_id 或 tenant_id" db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: await sor.sqlExe( "UPDATE pipeline_tasks SET state=${to}$, claimed_by=NULL, updated_at=NOW() " "WHERE id=${tid}$ AND tenant_id=${tn}$ AND state=${from}$", {"to": to_state, "tid": task_id, "tn": tenant_id, "from": from_state}) recs = await sor.R('pipeline_tasks', {'id': task_id}) await sor.sqlExe("COMMIT", {}) if not recs: return False, "任务不存在" cur = getattr(recs[0], 'state', '') if cur != to_state: return False, f"状态迁移失败(CAS): 期望 from={from_state} 实际 state={cur}" await record_audit(tenant_id, 'pipeline_tasks', task_id, action, from_state=from_state, to_state=to_state, who=who, agent_id=agent_id, detail=detail, sor=sor) return True, to_state async def claim_task(tenant_id, role, agent_id, from_state=S_SUBMITTED, to_state=S_RUNNING): """认领任务：CAS claimed_by IS NULL 保证并发认领原子；设置 claimed_by=agent_id。 返回 (ok, task_id_or_message)。 """ role_norm = _normalize_role(role) # 审计记录用规范名；查询用原样(兼容存量裸名数据) db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: claim_token = agent_id or getID() recs = await sor.sqlExe( "SELECT id FROM pipeline_tasks " "WHERE tenant_id=${tn}$ AND state=${st}$ AND role=${role}$ " "AND (claimed_by IS NULL OR claimed_by='') LIMIT 1", {"tn": tenant_id, "st": from_state, "role": role}) await sor.sqlExe("COMMIT", {}) if not recs: return False, f"没有可认领的 {role} 任务(state={from_state})" task_id = getattr(recs[0], 'id', '') # CAS：claimed_by IS NULL 保证原子（并发 poller 不会双重认领） await sor.sqlExe( "UPDATE pipeline_tasks SET state=${to}$, claimed_by=${cb}$, updated_at=NOW() " "WHERE id=${tid}$ AND tenant_id=${tn}$ AND state=${from}$ AND claimed_by IS NULL", {"to": to_state, "cb": claim_token, "tid": task_id, "tn": tenant_id, "from": from_state}) chk = await sor.R('pipeline_tasks', {'id': task_id}) await sor.sqlExe("COMMIT", {}) if not chk or getattr(chk[0], 'state', '') != to_state: return False, "认领竞争失败(已被他人认领)" await record_audit(tenant_id, 'pipeline_tasks', task_id, 'claim', from_state=from_state, to_state=to_state, who=role_norm, agent_id=agent_id, sor=sor) return True, task_id async def _task_skip_generic_qc(task_id): """任务是否声明豁免通用 QC 门禁（params.skip_generic_qc，产线自带质量门禁时用）。""" if not task_id: return False db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: recs = await sor.sqlExe( "SELECT params FROM pipeline_tasks WHERE id=${tid}$", {"tid": task_id}) await sor.sqlExe("COMMIT", {}) if not recs: return False try: p = json.loads(getattr(recs[0], 'params', '') or '{}') except Exception: return False return isinstance(p, dict) and bool(p.get('skip_generic_qc')) async def submit_task(task_id, tenant_id, who=None, agent_id=None): """提交产出：running → qc_review（清 claimed_by，先过 QC 合规/质量门禁）。 声明式豁免（2026-09-03）：params.skip_generic_qc=true 的任务直接 running → approved， 跳过通用 QC/PM 门禁——该任务的质量判定由产线自身流转引擎负责 （如投标产线：章节评审打分 + 解析产出契合度审核），通用门禁不接管。 """ if await _task_skip_generic_qc(task_id): return await _transition(task_id, tenant_id, S_RUNNING, S_APPROVED, 'submit', who=who, agent_id=agent_id, detail='skip_generic_qc：产线自带质量门禁，豁免通用QC/PM门禁') return await _transition(task_id, tenant_id, S_RUNNING, S_QC_REVIEW, 'submit', who=who, agent_id=agent_id) async def qc_approve_task(task_id, tenant_id, who=None, agent_id=None, comment=None): """QC 通过：qc_review → review（进入 PM 审核验收）。""" return await _transition(task_id, tenant_id, S_QC_REVIEW, S_REVIEW, 'qc_approve', who=who, agent_id=agent_id, detail=comment) async def _reject_with_retry_limit(task_id, tenant_id, from_state, action, who=None, agent_id=None, comment=None): """退回重做 + 重试上限：retry_count+1，超 task_max_retry 则转 failed（打破死循环）。 未超限：from_state → submitted（正常退回重做）。 超限：from_state → failed（释放并发名额，交 failed poller 报故障给人工）。 返回 (ok, message)。 """ if not task_id or not tenant_id: return False, "缺少 task_id 或 tenant_id" db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: # 重复计数 +1（CAS：仅当任务仍在 from_state） await sor.sqlExe( "UPDATE pipeline_tasks SET retry_count=retry_count+1 " "WHERE id=${tid}$ AND tenant_id=${tn}$ AND state=${from}$", {"tid": task_id, "tn": tenant_id, "from": from_state}) recs = await sor.sqlExe( "SELECT retry_count FROM pipeline_tasks WHERE id=${tid}$", {"tid": task_id}) await sor.sqlExe("COMMIT", {}) rc = int(getattr(recs[0], 'retry_count', 0) or 0) if recs else 0 from .workspace import get_max_task_retry max_retry = await get_max_task_retry(sor) if rc >= max_retry: err = f"{action} 退回 {rc} 次已达上限({max_retry})：{comment or '交付件不合规'}" await sor.sqlExe( "UPDATE pipeline_tasks SET state='failed', claimed_by=NULL, " "last_error=${err}$, updated_at=NOW() " "WHERE id=${tid}$ AND state=${from}$", {"tid": task_id, "from": from_state, "err": err}) await record_audit(tenant_id, 'pipeline_tasks', task_id, action, from_state=from_state, to_state=S_FAILED, who=who, agent_id=agent_id, detail=err, sor=sor) logger.info("reject retry limit hit: task=%s rc=%s -> failed", task_id, rc) return False, f"退回 {rc} 次已达上限({max_retry})，任务已标记 failed，等待人工介入" # 未超限：正常退回 submitted return await _transition(task_id, tenant_id, from_state, S_SUBMITTED, action, who=who, agent_id=agent_id, detail=comment) async def qc_reject_task(task_id, tenant_id, who=None, agent_id=None, comment=None): """QC 退回：qc_review → submitted（重复计数+1，超上限转 failed 打破死循环）。""" return await _reject_with_retry_limit(task_id, tenant_id, S_QC_REVIEW, 'qc_reject', who=who, agent_id=agent_id, comment=comment) async def qc_exempt_task(task_id, tenant_id, who=None, agent_id=None, comment=None): """豁免放行：qc_review → approved（任务声明 skip_generic_qc 时，通用门禁的兜底放行）。 用于存量已卡在 qc_review 的豁免任务：认领后不再做交付件检查，直接放行。 """ return await _transition(task_id, tenant_id, S_QC_REVIEW, S_APPROVED, 'qc_exempt', who=who, agent_id=agent_id, detail=comment) async def approve_task(task_id, tenant_id, who=None, agent_id=None, comment=None): """审核通过：review → approved。""" return await _transition(task_id, tenant_id, S_REVIEW, S_APPROVED, 'approve', who=who, agent_id=agent_id, detail=comment) async def reject_task(task_id, tenant_id, who=None, agent_id=None, comment=None): """审核退回：review → submitted（重复计数+1，超上限转 failed 打破死循环）。""" return await _reject_with_retry_limit(task_id, tenant_id, S_REVIEW, 'reject', who=who, agent_id=agent_id, comment=comment) async def complete_task(task_id, tenant_id, who=None, agent_id=None): """标记完成：approved → completed（也兼容 review → completed）。""" ok, msg = await _transition(task_id, tenant_id, S_APPROVED, S_COMPLETED, 'complete', who=who, agent_id=agent_id) if ok: return ok, msg return await _transition(task_id, tenant_id, S_REVIEW, S_COMPLETED, 'complete', who=who, agent_id=agent_id) async def mark_failed(task_id, tenant_id, who=None, agent_id=None, error=None): """标记失败（任意状态 → failed，清 claimed_by）。""" db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: recs = await sor.R('pipeline_tasks', {'id': task_id}) if not recs: return False, "任务不存在" from_state = getattr(recs[0], 'state', '') await sor.sqlExe( "UPDATE pipeline_tasks SET state='failed', last_error=${e}$, " "claimed_by=NULL, updated_at=NOW() WHERE id=${tid}$ AND tenant_id=${tn}$", {"e": (error or '')[:4000], "tid": task_id, "tn": tenant_id}) await sor.sqlExe("COMMIT", {}) await record_audit(tenant_id, 'pipeline_tasks', task_id, 'fail', from_state=from_state, to_state=S_FAILED, who=who, agent_id=agent_id, detail=error, sor=sor) return True, S_FAILED async def cancel_task(task_id, tenant_id, who=None, agent_id=None, comment=None): """取消任务（任意非终态 → cancelled，清 claimed_by）。 重做/作废场景必须先取消旧任务，避免两个相同任务并存；正在执行的 agent 会在 下一轮心跳检测到 cancelled 后自行中止。终态(completed/cancelled/failed/approved) 不重复取消。CAS：仅当任务仍处原非终态时取消，防并发覆盖。 """ if not task_id or not tenant_id: return False, "缺少 task_id 或 tenant_id" db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: recs = await sor.R('pipeline_tasks', {'id': task_id}) if not recs: return False, "任务不存在" from_state = getattr(recs[0], 'state', '') if from_state in (S_COMPLETED, 'cancelled', S_FAILED, S_APPROVED): return False, f"任务已是终态 {from_state}，无需取消" await sor.sqlExe( "UPDATE pipeline_tasks SET state='cancelled', claimed_by=NULL, " "updated_at=NOW() WHERE id=${tid}$ AND tenant_id=${tn}$ AND state=${from}$", {"tid": task_id, "tn": tenant_id, "from": from_state}) chk = await sor.R('pipeline_tasks', {'id': task_id}) await sor.sqlExe("COMMIT", {}) if not chk or getattr(chk[0], 'state', '') != 'cancelled': return False, "取消失败(CAS): 任务状态已变化" await record_audit(tenant_id, 'pipeline_tasks', task_id, 'cancel', from_state=from_state, to_state='cancelled', who=who, agent_id=agent_id, detail=comment, sor=sor) return True, 'cancelled' async def retry_task(task_id, tenant_id, who=None, agent_id=None): """重试：failed → submitted（retry_count+1，清 claimed_by）。""" db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: await sor.sqlExe( "UPDATE pipeline_tasks SET retry_count=retry_count+1, state='submitted', " "claimed_by=NULL, last_error=NULL, updated_at=NOW() " "WHERE id=${tid}$ AND tenant_id=${tn}$ AND state='failed'", {"tid": task_id, "tn": tenant_id}) chk = await sor.R('pipeline_tasks', {'id': task_id}) await sor.sqlExe("COMMIT", {}) if not chk or getattr(chk[0], 'state', '') != S_SUBMITTED: return False, "重试失败(任务不在 failed 状态)" await record_audit(tenant_id, 'pipeline_tasks', task_id, 'retry', from_state=S_FAILED, to_state=S_SUBMITTED, who=who, agent_id=agent_id, sor=sor) return True, S_SUBMITTED async def reset_task_retry(task_id, tenant_id, who=None, agent_id=None): """人工处理后恢复：retry_count 归零，waiting/failed/submitted → submitted 重新执行。 任务重复超限后任务链暂停、故障抛给人工；人工处理完毕调用本函数把该任务重复数清零并重新执行。 state IN 含 submitted（2026-09-16 实测 bug）：标准解链三步 resolve_problem (resume_task=True) → reset_task_retry → resume_project 中，resolve 已把 waiting→submitted，本函数旧 CAS 只认 waiting/failed → UPDATE 0 行；而成功 判定只看「终态==submitted」→ 空转也报 True，retry_count 从未清零、陈旧 last_error 残留，任务带着满重试计数重新认领，一次退回即再触顶 fault。 running 态仍排除（防中途拽回 submitted 打断在跑协程）。 """ db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: recs = await sor.R('pipeline_tasks', {'id': task_id}) if not recs: return False, "任务不存在" from_state = getattr(recs[0], 'state', '') if from_state not in ('waiting', 'failed', 'submitted'): return False, f"恢复失败(任务状态 {from_state} 不在 waiting/failed/submitted)" # 2026-09-18 M4a 事故配套：人工恢复时基础设施回置预算（params.infra_requeue） # 一并清零——否则任务带着满 infra 计数重新认领，一次 LLM 瞬时故障即触顶。 _new_params = getattr(recs[0], 'params', '') or '' try: _p = json.loads(_new_params or '{}') if isinstance(_p, dict) and _p.get('infra_requeue'): _p['infra_requeue'] = 0 _new_params = json.dumps(_p, ensure_ascii=False) except Exception: pass await sor.sqlExe( "UPDATE pipeline_tasks SET retry_count=0, state='submitted', " "claimed_by=NULL, last_error=NULL, params=${p}$, updated_at=NOW() " "WHERE id=${tid}$ AND tenant_id=${tn}$ AND state IN ('waiting','failed','submitted')", {"p": _new_params, "tid": task_id, "tn": tenant_id}) chk = await sor.R('pipeline_tasks', {'id': task_id}) await sor.sqlExe("COMMIT", {}) # 成功判定=UPDATE 真实生效（state 与 retry_count 双核对），不只看终态—— # 终态可能本来就是 submitted（空转假成功，见 docstring 实测事故）。 if (not chk or getattr(chk[0], 'state', '') != S_SUBMITTED or int(getattr(chk[0], 'retry_count', -1) or -1) != 0): return False, "恢复失败(计数未清零或状态竞态，请重试)" await record_audit(tenant_id, 'pipeline_tasks', task_id, 'reset_retry', from_state=from_state, to_state=S_SUBMITTED, who=who, agent_id=agent_id, sor=sor) return True, S_SUBMITTED async def suspend_task(task_id, tenant_id, who=None, agent_id=None): """挂起等回答：running → waiting（清 claimed_by）。""" return await _transition(task_id, tenant_id, S_RUNNING, S_WAITING, 'suspend', who=who, agent_id=agent_id) async def revive_task(task_id, tenant_id, who=None, agent_id=None): """恢复认领：waiting → submitted（清 claimed_by，重新认领）。""" return await _transition(task_id, tenant_id, S_WAITING, S_SUBMITTED, 'revive', who=who, agent_id=agent_id) async def set_task_state(task_id, tenant_id, from_state, to_state, who=None, agent_id=None, detail=None): """通用 CAS 状态迁移兜底（跨产线自定义状态机用）。""" return await _transition(task_id, tenant_id, from_state, to_state, 'set_state', who=who, agent_id=agent_id, detail=detail)