diff --git a/pipeline_service/bug_capability.py b/pipeline_service/bug_capability.py
new file mode 100644
index 0000000..e9c67ff
--- /dev/null
+++ b/pipeline_service/bug_capability.py
@@ -0,0 +1,220 @@
+"""Bug 能力 — sd_bugs 的状态机语义化迁移（通用、产线无关）。
+
+定位：Bug CRUD 走 xls2ui 生成的端点；本模块只做「生命周期状态流转」——
+每个迁移 CAS 原子（防并发/越权）+ 租户隔离 + 审计。
+
+Bug 的状态机/流转规则在 bug skill 里（LLM 读 skill 判断合法性）；本模块只固化操作原语。
+
+角色规范：role 参数用 agent.{role}（无前缀自动补 agent.）；人角色用 {orgtype}.{role}。
+
+scope 约定：sd_bugs 无 project_id 列，用 iteration_id 做范围校验（CAS 的 WHERE 里带 iteration_id）。
+"""
+
+import logging
+from datetime import datetime
+from sqlor.dbpools import DBPools
+from appPublic.uniqueID import getID
+from .audit import record_audit
+
+DBNAME = "pipeline"
+logger = logging.getLogger("pipeline.bug_capability")
+
+TABLE = "sd_bugs"
+
+# Bug 状态（SDLC 默认，状态机语义见 bug skill）
+S_OPEN = "open"                # 新建
+S_CONFIRMED = "confirmed"      # 已确认
+S_FIXING = "fixing"            # 修复中
+S_FIXED = "fixed"              # 已修复
+S_VERIFIED = "verified"        # 已验证
+S_CLOSED = "closed"            # 已关闭
+S_REJECTED = "rejected"        # 已驳回
+
+
+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(bug_id, iteration_id, from_states, to_state, action,
+                      extra_updates=None, who=None, agent_id=None, detail=None):
+    """CAS 状态迁移（支持多 from 状态 + 额外字段更新）+ 审计。返回 (ok, message)。"""
+    if not bug_id or not iteration_id:
+        return False, "缺少 bug_id 或 iteration_id"
+    if isinstance(from_states, str):
+        from_states = [from_states]
+    from_in = ",".join(f"${{s{i}}}$" for i in range(len(from_states)))
+    params = {"to": to_state, "bid": bug_id, "iid": iteration_id}
+    for i, s in enumerate(from_states):
+        params[f"s{i}"] = s
+
+    extra_sets = ""
+    for col, val in (extra_updates or {}).items():
+        key = col
+        params[key] = val
+        extra_sets += f", {col}=${{{key}}}$"
+
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        await sor.sqlExe(
+            f"UPDATE {TABLE} SET status=${{to}}$, updated_at=NOW(){extra_sets} "
+            f"WHERE id=${{bid}}$ AND iteration_id=${{iid}}$ AND status IN ({from_in})",
+            params)
+        recs = await sor.R(TABLE, {'id': bug_id})
+        await sor.sqlExe("COMMIT", {})
+        if not recs:
+            return False, "Bug 不存在"
+        cur = getattr(recs[0], 'status', '')
+        if cur != to_state:
+            return False, f"状态迁移失败(CAS): 期望 from={from_states} 实际 status={cur}"
+        await record_audit(iteration_id, TABLE, bug_id, action,
+                           from_state="|".join(from_states), to_state=to_state,
+                           who=who, agent_id=agent_id, detail=detail, sor=sor)
+        return True, to_state
+
+
+async def report_bug(iteration_id, title, description="", severity="major",
+                     priority="P1", case_id="", step_name="", reporter_type="agent",
+                     reporter_id="", assignee_id="", who=None, agent_id=None):
+    """上报 Bug：新建记录，status=open。返回 (ok, bug_id_or_message)。"""
+    if not iteration_id:
+        return False, "缺少 iteration_id"
+    if not title or not title.strip():
+        return False, "缺少 Bug 标题"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        bid = getID()
+        await sor.C(TABLE, {
+            'id': bid,
+            'iteration_id': iteration_id,
+            'case_id': case_id or '',
+            'step_name': step_name or '',
+            'title': title.strip(),
+            'description': description or '',
+            'severity': severity or 'major',
+            'priority': priority or 'P1',
+            'status': S_OPEN,
+            'reporter_type': reporter_type or 'agent',
+            'reporter_id': reporter_id or '',
+            'assignee_id': assignee_id or '',
+        })
+        await record_audit(iteration_id, TABLE, bid, 'report',
+                           to_state=S_OPEN, who=_normalize_role(who),
+                           agent_id=agent_id, sor=sor)
+        logger.info("report_bug: %s iteration=%s", bid, iteration_id)
+        return True, bid
+
+
+async def confirm_bug(bug_id, iteration_id, who=None, agent_id=None):
+    """确认 Bug：open → confirmed。"""
+    return await _transition(bug_id, iteration_id, [S_OPEN], S_CONFIRMED, 'confirm',
+                             who=who, agent_id=agent_id)
+
+
+async def start_fix(bug_id, iteration_id, assignee_id="", who=None, agent_id=None):
+    """开始修复：confirmed → fixing（记录处理人）。"""
+    return await _transition(bug_id, iteration_id, [S_CONFIRMED], S_FIXING, 'start_fix',
+                             extra_updates={"assignee_id": assignee_id or ''},
+                             who=who, agent_id=agent_id)
+
+
+async def fix_bug(bug_id, iteration_id, fix_description="", fix_commit="",
+                  who=None, agent_id=None):
+    """修复完成：fixing → fixed（附修复说明 + 提交）。"""
+    return await _transition(bug_id, iteration_id, [S_FIXING], S_FIXED, 'fix',
+                             extra_updates={"fix_description": fix_description or '',
+                                            "fix_commit": fix_commit or ''},
+                             who=who, agent_id=agent_id, detail=fix_description)
+
+
+async def verify_bug(bug_id, iteration_id, verified_by="", who=None, agent_id=None):
+    """验证修复：fixed → verified（记录验证人）。"""
+    return await _transition(bug_id, iteration_id, [S_FIXED], S_VERIFIED, 'verify',
+                             extra_updates={"verified_by": verified_by or ''},
+                             who=who, agent_id=agent_id)
+
+
+async def close_bug(bug_id, iteration_id, who=None, agent_id=None):
+    """关闭 Bug：verified → closed（记录关闭时间）。"""
+    return await _transition(bug_id, iteration_id, [S_VERIFIED], S_CLOSED, 'close',
+                             extra_updates={"closed_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')},
+                             who=who, agent_id=agent_id)
+
+
+async def reject_bug(bug_id, iteration_id, who=None, agent_id=None, comment=None):
+    """驳回 Bug：open/confirmed → rejected（附意见）。"""
+    if not comment or not comment.strip():
+        return False, "驳回必须附意见"
+    return await _transition(bug_id, iteration_id, [S_OPEN, S_CONFIRMED], S_REJECTED, 'reject',
+                             who=who, agent_id=agent_id, detail=comment)
+
+
+async def reopen_bug(bug_id, iteration_id, who=None, agent_id=None, comment=None):
+    """重新打开：closed/rejected → open。"""
+    return await _transition(bug_id, iteration_id, [S_CLOSED, S_REJECTED], S_OPEN, 'reopen',
+                             who=who, agent_id=agent_id, detail=comment)
+
+
+async def set_bug_state(bug_id, iteration_id, from_state, to_state,
+                        who=None, agent_id=None, detail=None):
+    """通用 CAS 状态迁移兜底（跨产线自定义状态机用）。"""
+    return await _transition(bug_id, iteration_id, [from_state], to_state, 'set_state',
+                             who=who, agent_id=agent_id, detail=detail)
+
+
+async def list_bugs(iteration_id, status=None, severity=None, limit=100) -> list:
+    """列出 Bug（可按状态/严重度过滤）。"""
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        conditions = ["iteration_id=${iid}$"]
+        params = {"iid": iteration_id}
+        if status:
+            conditions.append("status=${status}$")
+            params["status"] = status
+        if severity:
+            conditions.append("severity=${sev}$")
+            params["sev"] = severity
+        where = " AND ".join(conditions)
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError):
+            limit = 100
+        sql = (f"SELECT * FROM {TABLE} WHERE {where} "
+               f"ORDER BY created_at DESC LIMIT {limit}")
+        recs = await sor.sqlExe(sql, params)
+        # 释放 SELECT 元数据锁
+        await sor.sqlExe("COMMIT", {})
+        result = []
+        for rec in (recs or []):
+            result.append(_rec_to_dict(rec))
+        return result
+
+
+def _rec_to_dict(rec):
+    """把 sqlor 记录对象转成 dict（sqlor 行是 DictObject，必须 dict(rec) 取列）。"""
+    if isinstance(rec, dict):
+        return dict(rec)
+    try:
+        return dict(rec)
+    except (TypeError, ValueError):
+        if hasattr(rec, 'to_dict'):
+            try:
+                return rec.to_dict()
+            except Exception:
+                pass
+        return {}
diff --git a/pipeline_service/deliverable_capability.py b/pipeline_service/deliverable_capability.py
new file mode 100644
index 0000000..fb78fed
--- /dev/null
+++ b/pipeline_service/deliverable_capability.py
@@ -0,0 +1,182 @@
+"""交付件能力 — pipeline_deliverables 的评审状态机语义化迁移（通用、产线无关）。
+
+定位：交付件 CRUD 走 xls2ui 生成的端点；本模块只做「评审状态流转」——
+每个迁移 CAS 原子（防并发/越权）+ 租户隔离 + 审计。
+
+交付件的评审状态机/流转规则在 deliverable skill 里（LLM 读 skill 判断合法性）；本模块只固化操作原语。
+
+角色规范：role 参数用 agent.{role}（无前缀自动补 agent.）；人角色用 {orgtype}.{role}。
+
+scope 约定：pipeline_deliverables 用 project_id 做范围校验（CAS 的 WHERE 里带 project_id）。
+
+注意：评审状态列名是 review_status（非 status）。
+"""
+
+import logging
+from sqlor.dbpools import DBPools
+from appPublic.uniqueID import getID
+from .audit import record_audit
+
+DBNAME = "pipeline"
+logger = logging.getLogger("pipeline.deliverable_capability")
+
+TABLE = "pipeline_deliverables"
+STATUS_COL = "review_status"
+
+# 交付件评审状态（SDLC 默认，状态机语义见 deliverable skill）
+S_PENDING = "pending"      # 待评审
+S_APPROVED = "approved"    # 评审通过
+S_REJECTED = "rejected"    # 评审驳回
+
+
+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(deliverable_id, project_id, from_state, to_state, action,
+                      who=None, agent_id=None, detail=None):
+    """CAS 评审状态迁移 + 审计。返回 (ok, message)。"""
+    if not deliverable_id or not project_id:
+        return False, "缺少 deliverable_id 或 project_id"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        await sor.sqlExe(
+            f"UPDATE {TABLE} SET {STATUS_COL}=${{to}}$, reviewed_by=${{who}}$ "
+            "WHERE id=${did}$ AND project_id=${pid}$ AND "
+            f"{STATUS_COL}=${{from}}$",
+            {"to": to_state, "who": _normalize_role(who) or '',
+             "did": deliverable_id, "pid": project_id, "from": from_state})
+        recs = await sor.R(TABLE, {'id': deliverable_id})
+        await sor.sqlExe("COMMIT", {})
+        if not recs:
+            return False, "交付件不存在"
+        cur = getattr(recs[0], STATUS_COL, '')
+        if cur != to_state:
+            return False, f"状态迁移失败(CAS): 期望 from={from_state} 实际 {STATUS_COL}={cur}"
+        await record_audit(project_id, TABLE, deliverable_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 submit_deliverable(project_id, title, deliverable_type="code",
+                             task_id="", content="", repo_name="", target_path="",
+                             file_path="", quality_score=80, created_by="",
+                             who=None, agent_id=None):
+    """提交交付件：新建记录，review_status=pending。返回 (ok, deliverable_id_or_message)。"""
+    if not project_id:
+        return False, "缺少 project_id"
+    if not title or not title.strip():
+        return False, "缺少交付件标题"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        did = getID()
+        try:
+            quality_score = int(quality_score)
+        except (TypeError, ValueError):
+            quality_score = 80
+        await sor.C(TABLE, {
+            'id': did,
+            'project_id': project_id,
+            'task_id': task_id or '',
+            'title': title.strip(),
+            'deliverable_type': deliverable_type or 'code',
+            'repo_name': repo_name or '',
+            'target_path': target_path or '',
+            'file_path': file_path or '',
+            'content': content or '',
+            'quality_score': quality_score,
+            STATUS_COL: S_PENDING,
+            'created_by': created_by or '',
+        })
+        await record_audit(project_id, TABLE, did, 'submit',
+                           to_state=S_PENDING, who=_normalize_role(who),
+                           agent_id=agent_id, sor=sor)
+        logger.info("submit_deliverable: %s project=%s", did, project_id)
+        return True, did
+
+
+async def approve_deliverable(deliverable_id, project_id, who=None, agent_id=None, comment=None):
+    """评审通过：pending → approved。"""
+    return await _transition(deliverable_id, project_id, S_PENDING, S_APPROVED, 'approve',
+                             who=who, agent_id=agent_id, detail=comment)
+
+
+async def reject_deliverable(deliverable_id, project_id, who=None, agent_id=None, comment=None):
+    """评审驳回：pending → rejected（意见必填）。"""
+    if not comment or not comment.strip():
+        return False, "驳回必须附评审意见"
+    return await _transition(deliverable_id, project_id, S_PENDING, S_REJECTED, 'reject',
+                             who=who, agent_id=agent_id, detail=comment)
+
+
+async def reopen_deliverable(deliverable_id, project_id, who=None, agent_id=None, comment=None):
+    """重新提交评审：rejected → pending（修改后重新提交）。"""
+    return await _transition(deliverable_id, project_id, S_REJECTED, S_PENDING, 'reopen',
+                             who=who, agent_id=agent_id, detail=comment)
+
+
+async def set_deliverable_state(deliverable_id, project_id, from_state, to_state,
+                                who=None, agent_id=None, detail=None):
+    """通用 CAS 状态迁移兜底（跨产线自定义状态机用）。"""
+    return await _transition(deliverable_id, project_id, from_state, to_state, 'set_state',
+                             who=who, agent_id=agent_id, detail=detail)
+
+
+async def list_deliverables(project_id, task_id=None, review_status=None, limit=50) -> list:
+    """列出交付件（可按任务/评审状态过滤）。"""
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        conditions = ["project_id=${pid}$"]
+        params = {"pid": project_id}
+        if task_id:
+            conditions.append("task_id=${tid}$")
+            params["tid"] = task_id
+        if review_status:
+            conditions.append(f"{STATUS_COL}=${{rs}}$")
+            params["rs"] = review_status
+        where = " AND ".join(conditions)
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError):
+            limit = 50
+        sql = (f"SELECT * FROM {TABLE} WHERE {where} "
+               f"ORDER BY created_at DESC LIMIT {limit}")
+        recs = await sor.sqlExe(sql, params)
+        # 释放 SELECT 元数据锁
+        await sor.sqlExe("COMMIT", {})
+        result = []
+        for rec in (recs or []):
+            result.append(_rec_to_dict(rec))
+        return result
+
+
+def _rec_to_dict(rec):
+    """把 sqlor 记录对象转成 dict（sqlor 行是 DictObject，必须 dict(rec) 取列）。"""
+    if isinstance(rec, dict):
+        return dict(rec)
+    try:
+        return dict(rec)
+    except (TypeError, ValueError):
+        if hasattr(rec, 'to_dict'):
+            try:
+                return rec.to_dict()
+            except Exception:
+                pass
+        return {}
diff --git a/pipeline_service/deploy_capability.py b/pipeline_service/deploy_capability.py
new file mode 100644
index 0000000..5f34c09
--- /dev/null
+++ b/pipeline_service/deploy_capability.py
@@ -0,0 +1,199 @@
+"""部署环境能力 — sd_deploy_envs 的状态机语义化迁移（通用、产线无关）。
+
+定位：环境 CRUD 走 xls2ui 生成的端点；本模块只做「验证状态流转」——
+每个迁移 CAS 原子（防并发/越权）+ 租户隔离 + 审计。
+
+环境的状态机/流转规则在 deploy-env skill 里（LLM 读 skill 判断合法性）；本模块只固化操作原语。
+发布本体（部署动作）复用 task 状态机（deploy 角色）；本模块管「环境」这个可追踪实体。
+
+角色规范：role 参数用 agent.{role}（无前缀自动补 agent.）；人角色用 {orgtype}.{role}。
+
+scope 约定：sd_deploy_envs 用 project_id 做范围校验（CAS 的 WHERE 里带 project_id）。
+"""
+
+import logging
+from datetime import datetime
+from sqlor.dbpools import DBPools
+from appPublic.uniqueID import getID
+from .audit import record_audit
+
+DBNAME = "pipeline"
+logger = logging.getLogger("pipeline.deploy_capability")
+
+TABLE = "sd_deploy_envs"
+
+# 部署环境状态（SDLC 默认，状态机语义见 deploy-env skill）
+S_CONFIGURED = "configured"    # 已配置
+S_VERIFIED = "verified"        # 已验证
+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(env_id, project_id, from_states, to_state, action,
+                      extra_updates=None, who=None, agent_id=None, detail=None):
+    """CAS 状态迁移（支持多 from 状态 + 额外字段更新）+ 审计。返回 (ok, message)。"""
+    if not env_id or not project_id:
+        return False, "缺少 env_id 或 project_id"
+    if isinstance(from_states, str):
+        from_states = [from_states]
+    from_in = ",".join(f"${{s{i}}}$" for i in range(len(from_states)))
+    params = {"to": to_state, "eid": env_id, "pid": project_id}
+    for i, s in enumerate(from_states):
+        params[f"s{i}"] = s
+
+    extra_sets = ""
+    for col, val in (extra_updates or {}).items():
+        key = col
+        params[key] = val
+        extra_sets += f", {col}=${{{key}}}$"
+
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        await sor.sqlExe(
+            f"UPDATE {TABLE} SET status=${{to}}$, updated_at=NOW(){extra_sets} "
+            f"WHERE id=${{eid}}$ AND project_id=${{pid}}$ AND status IN ({from_in})",
+            params)
+        recs = await sor.R(TABLE, {'id': env_id})
+        await sor.sqlExe("COMMIT", {})
+        if not recs:
+            return False, "环境不存在"
+        cur = getattr(recs[0], 'status', '')
+        if cur != to_state:
+            return False, f"状态迁移失败(CAS): 期望 from={from_states} 实际 status={cur}"
+        await record_audit(project_id, TABLE, env_id, action,
+                           from_state="|".join(from_states), to_state=to_state,
+                           who=who, agent_id=agent_id, detail=detail, sor=sor)
+        return True, to_state
+
+
+async def configure_env(project_id, env_type, host, user, deploy_path,
+                        port=22, ssh_key_path="", sudo_enabled="", python_path="",
+                        db_host="", db_port=3306, db_name="", db_user="", db_password="",
+                        who=None, agent_id=None):
+    """配置环境：新建记录，status=configured。返回 (ok, env_id_or_message)。"""
+    if not project_id:
+        return False, "缺少 project_id"
+    if not env_type or not env_type.strip():
+        return False, "缺少 env_type"
+    if not host or not host.strip():
+        return False, "缺少 host"
+    if not user or not user.strip():
+        return False, "缺少 user"
+    if not deploy_path or not deploy_path.strip():
+        return False, "缺少 deploy_path"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        eid = getID()
+        try:
+            port = int(port)
+        except (TypeError, ValueError):
+            port = 22
+        try:
+            db_port = int(db_port)
+        except (TypeError, ValueError):
+            db_port = 3306
+        await sor.C(TABLE, {
+            'id': eid,
+            'project_id': project_id,
+            'env_type': env_type.strip(),
+            'host': host.strip(),
+            'port': port,
+            'user': user.strip(),
+            'ssh_key_path': ssh_key_path or '',
+            'sudo_enabled': sudo_enabled or '',
+            'deploy_path': deploy_path.strip(),
+            'python_path': python_path or '',
+            'db_host': db_host or '',
+            'db_port': db_port,
+            'db_name': db_name or '',
+            'db_user': db_user or '',
+            'db_password': db_password or '',
+            'status': S_CONFIGURED,
+        })
+        await record_audit(project_id, TABLE, eid, 'configure',
+                           to_state=S_CONFIGURED, who=_normalize_role(who),
+                           agent_id=agent_id, sor=sor)
+        logger.info("configure_env: %s project=%s type=%s", eid, project_id, env_type.strip())
+        return True, eid
+
+
+async def verify_env(env_id, project_id, who=None, agent_id=None):
+    """验证环境通过：configured/failed → verified（记录验证时间）。"""
+    return await _transition(env_id, project_id, [S_CONFIGURED, S_FAILED], S_VERIFIED, 'verify',
+                             extra_updates={"verified_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')},
+                             who=who, agent_id=agent_id)
+
+
+async def fail_env(env_id, project_id, reason="", who=None, agent_id=None):
+    """验证环境失败：configured/verified → failed（附失败原因）。"""
+    return await _transition(env_id, project_id, [S_CONFIGURED, S_VERIFIED], S_FAILED, 'fail',
+                             who=who, agent_id=agent_id, detail=reason)
+
+
+async def set_env_state(env_id, project_id, from_state, to_state,
+                        who=None, agent_id=None, detail=None):
+    """通用 CAS 状态迁移兜底（跨产线自定义状态机用）。"""
+    return await _transition(env_id, project_id, [from_state], to_state, 'set_state',
+                             who=who, agent_id=agent_id, detail=detail)
+
+
+async def list_envs(project_id, env_type=None, status=None, limit=50) -> list:
+    """列出环境（可按类型/状态过滤）。"""
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        conditions = ["project_id=${pid}$"]
+        params = {"pid": project_id}
+        if env_type:
+            conditions.append("env_type=${et}$")
+            params["et"] = env_type
+        if status:
+            conditions.append("status=${status}$")
+            params["status"] = status
+        where = " AND ".join(conditions)
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError):
+            limit = 50
+        sql = (f"SELECT * FROM {TABLE} WHERE {where} "
+               f"ORDER BY created_at ASC LIMIT {limit}")
+        recs = await sor.sqlExe(sql, params)
+        # 释放 SELECT 元数据锁
+        await sor.sqlExe("COMMIT", {})
+        result = []
+        for rec in (recs or []):
+            result.append(_rec_to_dict(rec))
+        return result
+
+
+def _rec_to_dict(rec):
+    """把 sqlor 记录对象转成 dict（sqlor 行是 DictObject，必须 dict(rec) 取列）。"""
+    if isinstance(rec, dict):
+        return dict(rec)
+    try:
+        return dict(rec)
+    except (TypeError, ValueError):
+        if hasattr(rec, 'to_dict'):
+            try:
+                return rec.to_dict()
+            except Exception:
+                pass
+        return {}
diff --git a/pipeline_service/iteration_capability.py b/pipeline_service/iteration_capability.py
new file mode 100644
index 0000000..3e594bd
--- /dev/null
+++ b/pipeline_service/iteration_capability.py
@@ -0,0 +1,172 @@
+"""迭代能力 — sd_iterations 的状态机语义化迁移（通用、产线无关）。
+
+定位：迭代 CRUD 走 xls2ui 生成的端点；本模块只做「生命周期状态流转」——
+每个迁移 CAS 原子（防并发/越权）+ 租户隔离 + 审计。
+
+迭代的生命周期/流转规则在 iteration skill 里（LLM 读 skill 判断合法性）；本模块只固化操作原语。
+
+角色规范：role 参数用 agent.{role}（无前缀自动补 agent.）；人角色用 {orgtype}.{role}。
+
+scope 约定：sd_iterations 用 project_id 做范围校验（CAS 的 WHERE 里带 project_id），
+与 pipeline_tasks 用 tenant_id 不同——sd_* 业务表用自己的归属列。
+"""
+
+import logging
+from sqlor.dbpools import DBPools
+from appPublic.uniqueID import getID
+from .audit import record_audit
+
+DBNAME = "pipeline"
+logger = logging.getLogger("pipeline.iteration_capability")
+
+TABLE = "sd_iterations"
+
+# 迭代状态（SDLC 默认，状态机语义见 iteration skill）
+S_PLANNING = "planning"          # 规划中
+S_IN_PROGRESS = "in_progress"    # 进行中
+S_COMPLETED = "completed"        # 已完成
+S_CANCELLED = "cancelled"        # 已取消
+
+
+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(iteration_id, project_id, from_state, to_state, action,
+                      who=None, agent_id=None, detail=None):
+    """CAS 状态迁移 + 审计。返回 (ok, message)。"""
+    if not iteration_id or not project_id:
+        return False, "缺少 iteration_id 或 project_id"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        await sor.sqlExe(
+            f"UPDATE {TABLE} SET status=${{to}}$, updated_at=NOW() "
+            "WHERE id=${iid}$ AND project_id=${pid}$ AND status=${from}$",
+            {"to": to_state, "iid": iteration_id, "pid": project_id, "from": from_state})
+        recs = await sor.R(TABLE, {'id': iteration_id})
+        await sor.sqlExe("COMMIT", {})
+        if not recs:
+            return False, "迭代不存在"
+        cur = getattr(recs[0], 'status', '')
+        if cur != to_state:
+            return False, f"状态迁移失败(CAS): 期望 from={from_state} 实际 status={cur}"
+        await record_audit(project_id, TABLE, iteration_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 create_iteration(project_id, iteration_name, iteration_type="new_feature",
+                           scope="", priority=5, created_by="", who=None, agent_id=None):
+    """创建迭代：新建记录，status=planning。返回 (ok, iteration_id_or_message)。"""
+    if not project_id:
+        return False, "缺少 project_id"
+    if not iteration_name or not iteration_name.strip():
+        return False, "缺少 iteration_name"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        iid = getID()
+        try:
+            priority = int(priority)
+        except (TypeError, ValueError):
+            priority = 5
+        await sor.C(TABLE, {
+            'id': iid,
+            'project_id': project_id,
+            'iteration_name': iteration_name.strip(),
+            'iteration_type': iteration_type or 'new_feature',
+            'scope': scope or '',
+            'status': S_PLANNING,
+            'priority': priority,
+            '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)
+        return True, iid
+
+
+async def start_iteration(iteration_id, project_id, who=None, agent_id=None):
+    """开始迭代：planning → in_progress。"""
+    return await _transition(iteration_id, project_id, S_PLANNING, S_IN_PROGRESS, 'start',
+                             who=who, agent_id=agent_id)
+
+
+async def complete_iteration(iteration_id, project_id, who=None, agent_id=None):
+    """完成迭代：in_progress → completed。"""
+    return await _transition(iteration_id, project_id, S_IN_PROGRESS, S_COMPLETED, 'complete',
+                             who=who, agent_id=agent_id)
+
+
+async def cancel_iteration(iteration_id, project_id, who=None, agent_id=None, comment=None):
+    """取消迭代：planning/in_progress → cancelled。"""
+    ok, msg = await _transition(iteration_id, project_id, S_PLANNING, S_CANCELLED, 'cancel',
+                                who=who, agent_id=agent_id, detail=comment)
+    if ok:
+        return ok, msg
+    return await _transition(iteration_id, project_id, S_IN_PROGRESS, S_CANCELLED, 'cancel',
+                             who=who, agent_id=agent_id, detail=comment)
+
+
+async def set_iteration_state(iteration_id, project_id, from_state, to_state,
+                              who=None, agent_id=None, detail=None):
+    """通用 CAS 状态迁移兜底（跨产线自定义状态机用）。"""
+    return await _transition(iteration_id, project_id, from_state, to_state, 'set_state',
+                             who=who, agent_id=agent_id, detail=detail)
+
+
+async def list_iterations(project_id, status=None, limit=50) -> list:
+    """列出迭代（可按状态过滤）。"""
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        conditions = ["project_id=${pid}$"]
+        params = {"pid": project_id}
+        if status:
+            conditions.append("status=${status}$")
+            params["status"] = status
+        where = " AND ".join(conditions)
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError):
+            limit = 50
+        sql = (f"SELECT * FROM {TABLE} WHERE {where} "
+               f"ORDER BY created_at ASC LIMIT {limit}")
+        recs = await sor.sqlExe(sql, params)
+        # 释放 SELECT 元数据锁
+        await sor.sqlExe("COMMIT", {})
+        result = []
+        for rec in (recs or []):
+            result.append(_rec_to_dict(rec))
+        return result
+
+
+def _rec_to_dict(rec):
+    """把 sqlor 记录对象转成 dict（sqlor 行是 DictObject，必须 dict(rec) 取列）。"""
+    if isinstance(rec, dict):
+        return dict(rec)
+    try:
+        return dict(rec)
+    except (TypeError, ValueError):
+        if hasattr(rec, 'to_dict'):
+            try:
+                return rec.to_dict()
+            except Exception:
+                pass
+        return {}
diff --git a/pipeline_service/project_capability.py b/pipeline_service/project_capability.py
new file mode 100644
index 0000000..409cecf
--- /dev/null
+++ b/pipeline_service/project_capability.py
@@ -0,0 +1,178 @@
+"""项目能力 — sd_projects 的状态机语义化迁移（通用、产线无关）。
+
+定位：项目 CRUD 走 xls2ui 生成的端点；本模块只做「生命周期状态流转」——
+每个迁移 CAS 原子（防并发/越权）+ 租户隔离 + 审计。
+
+项目的生命周期/流转规则在 project skill 里（LLM 读 skill 判断合法性）；本模块只固化操作原语。
+
+角色规范：role 参数用 agent.{role}（无前缀自动补 agent.）；人角色用 {orgtype}.{role}。
+
+scope 约定：sd_projects 是顶层容器，用 id 唯一标识（CAS WHERE 带 id 即可），
+org_id 用于 list 过滤（租户隔离）；audit 的 tenant_id 用 project_id（项目即最外层容器）。
+"""
+
+import logging
+from sqlor.dbpools import DBPools
+from appPublic.uniqueID import getID
+from .audit import record_audit
+
+DBNAME = "pipeline"
+logger = logging.getLogger("pipeline.project_capability")
+
+TABLE = "sd_projects"
+
+# 项目状态（SDLC 默认，状态机语义见 project skill）
+S_DRAFT = "draft"            # 草稿（已创建，未启动）
+S_ACTIVE = "active"          # 进行中
+S_COMPLETED = "completed"    # 已完成
+S_ARCHIVED = "archived"      # 已归档
+
+
+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(project_id, from_state, to_state, action,
+                      who=None, agent_id=None, detail=None):
+    """CAS 状态迁移 + 审计。返回 (ok, message)。"""
+    if not project_id:
+        return False, "缺少 project_id"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        await sor.sqlExe(
+            f"UPDATE {TABLE} SET status=${{to}}$, updated_at=NOW() "
+            "WHERE id=${pid}$ AND status=${from}$",
+            {"to": to_state, "pid": project_id, "from": from_state})
+        recs = await sor.R(TABLE, {'id': project_id})
+        await sor.sqlExe("COMMIT", {})
+        if not recs:
+            return False, "项目不存在"
+        cur = getattr(recs[0], 'status', '')
+        if cur != to_state:
+            return False, f"状态迁移失败(CAS): 期望 from={from_state} 实际 status={cur}"
+        await record_audit(project_id, TABLE, project_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 create_project(name, project_type="web_app", description="",
+                         org_id="0", created_by="", who=None, agent_id=None):
+    """创建项目：新建记录，status=draft。返回 (ok, project_id_or_message)。"""
+    if not name or not name.strip():
+        return False, "缺少项目名称"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        pid = getID()
+        await sor.C(TABLE, {
+            'id': pid,
+            'name': name.strip(),
+            'description': description or '',
+            'project_type': project_type or 'web_app',
+            'status': S_DRAFT,
+            'org_id': org_id or '0',
+            'created_by': created_by or '',
+        })
+        await record_audit(pid, TABLE, pid, 'create',
+                           to_state=S_DRAFT, who=_normalize_role(who),
+                           agent_id=agent_id, sor=sor)
+        logger.info("create_project: %s name=%s", pid, name.strip())
+        return True, pid
+
+
+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',
+                                who=who, agent_id=agent_id)
+    if ok:
+        return ok, msg
+    return await _transition(project_id, S_ARCHIVED, S_ACTIVE, 'start',
+                             who=who, agent_id=agent_id)
+
+
+async def complete_project(project_id, who=None, agent_id=None):
+    """完成项目：active → completed。"""
+    return await _transition(project_id, S_ACTIVE, S_COMPLETED, 'complete',
+                             who=who, agent_id=agent_id)
+
+
+async def archive_project(project_id, who=None, agent_id=None):
+    """归档项目：completed → archived（也兼容 active → archived 强制归档）。"""
+    ok, msg = await _transition(project_id, S_COMPLETED, S_ARCHIVED, 'archive',
+                                who=who, agent_id=agent_id)
+    if ok:
+        return ok, msg
+    return await _transition(project_id, S_ACTIVE, S_ARCHIVED, 'archive',
+                             who=who, agent_id=agent_id)
+
+
+async def reopen_project(project_id, who=None, agent_id=None):
+    """重新打开：archived → active。"""
+    return await _transition(project_id, S_ARCHIVED, S_ACTIVE, 'reopen',
+                             who=who, agent_id=agent_id)
+
+
+async def set_project_state(project_id, from_state, to_state,
+                            who=None, agent_id=None, detail=None):
+    """通用 CAS 状态迁移兜底（跨产线自定义状态机用）。"""
+    return await _transition(project_id, from_state, to_state, 'set_state',
+                             who=who, agent_id=agent_id, detail=detail)
+
+
+async def list_projects(org_id=None, status=None, limit=50) -> list:
+    """列出项目（可按组织/状态过滤；org_id 空则列全部）。"""
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        conditions = ["1=1"]
+        params = {}
+        if org_id:
+            conditions.append("org_id=${org}$")
+            params["org"] = org_id
+        if status:
+            conditions.append("status=${status}$")
+            params["status"] = status
+        where = " AND ".join(conditions)
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError):
+            limit = 50
+        sql = (f"SELECT * FROM {TABLE} WHERE {where} "
+               f"ORDER BY created_at DESC LIMIT {limit}")
+        recs = await sor.sqlExe(sql, params)
+        # 释放 SELECT 元数据锁
+        await sor.sqlExe("COMMIT", {})
+        result = []
+        for rec in (recs or []):
+            result.append(_rec_to_dict(rec))
+        return result
+
+
+def _rec_to_dict(rec):
+    """把 sqlor 记录对象转成 dict（sqlor 行是 DictObject，必须 dict(rec) 取列）。"""
+    if isinstance(rec, dict):
+        return dict(rec)
+    try:
+        return dict(rec)
+    except (TypeError, ValueError):
+        if hasattr(rec, 'to_dict'):
+            try:
+                return rec.to_dict()
+            except Exception:
+                pass
+        return {}
diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py
index 67f0771..8f50889 100644
--- a/pipeline_service/sdlc_ability.py
+++ b/pipeline_service/sdlc_ability.py
@@ -151,6 +151,49 @@ SDL_TOOLS = [
         parameters={"feature_id": "功能ID"},
         category="feature",
     ),
+    # ── 项目生命周期 ──
+    ToolDefinition(name="list_projects", description="列出项目列表(含状态)", parameters={"status": "按状态筛选(可选)"}, category="project"),
+    ToolDefinition(name="start_project", description="启动项目(draft/archived→active)", parameters={"project_id": "项目ID"}, category="project"),
+    ToolDefinition(name="complete_project", description="完成项目(active→completed)", parameters={"project_id": "项目ID"}, category="project"),
+    ToolDefinition(name="archive_project", description="归档项目(completed→archived)", parameters={"project_id": "项目ID"}, category="project"),
+    ToolDefinition(name="reopen_project", description="重新打开归档项目(archived→active)", parameters={"project_id": "项目ID"}, category="project"),
+    # ── 迭代 ──
+    ToolDefinition(name="list_iterations", description="查看当前项目迭代列表", parameters={"status": "按状态筛选(可选)"}, category="iteration"),
+    ToolDefinition(name="create_iteration", description="创建迭代(planning)", parameters={"iteration_name": "迭代名称", "iteration_type": "迭代类型(可选)", "scope": "迭代范围(可选)"}, category="iteration"),
+    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="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"),
+    ToolDefinition(name="reject_deliverable", description="评审驳回交付件(pending→rejected，附意见)", parameters={"deliverable_id": "交付件ID", "comment": "评审意见(必填)"}, category="deliverable"),
+    # ── 测试计划 ──
+    ToolDefinition(name="list_plans", description="查看迭代测试计划列表", parameters={"iteration_id": "迭代ID(可选)", "status": "按状态筛选(可选)"}, category="test-plan"),
+    ToolDefinition(name="create_test_plan", description="创建测试计划(draft)", parameters={"iteration_id": "迭代ID", "plan_name": "方案名称", "plan_type": "方案类型(可选)", "scope": "测试范围(可选)"}, category="test-plan"),
+    ToolDefinition(name="approve_plan", description="审批测试计划(draft→approved)", parameters={"plan_id": "计划ID"}, category="test-plan"),
+    ToolDefinition(name="start_plan", description="开始执行测试计划(approved→executing)", parameters={"plan_id": "计划ID"}, category="test-plan"),
+    ToolDefinition(name="complete_plan", description="完成测试计划(executing→completed)", parameters={"plan_id": "计划ID"}, category="test-plan"),
+    # ── 测试用例 ──
+    ToolDefinition(name="list_cases", description="查看测试计划用例列表", parameters={"plan_id": "计划ID", "status": "按状态筛选(可选)"}, category="test-case"),
+    ToolDefinition(name="create_case", description="创建测试用例(pending)", parameters={"plan_id": "计划ID", "case_name": "用例名称", "case_type": "用例类型(可选)", "steps": "测试步骤(可选)", "expected_result": "预期结果(可选)"}, category="test-case"),
+    ToolDefinition(name="pass_case", description="用例通过(pending/fail→pass)", parameters={"case_id": "用例ID", "actual_result": "实际结果(可选)"}, category="test-case"),
+    ToolDefinition(name="fail_case", description="用例失败(pending/pass→fail，应联动建Bug)", parameters={"case_id": "用例ID", "actual_result": "实际结果(必填)"}, category="test-case"),
+    ToolDefinition(name="skip_case", description="用例跳过(pending→skipped，注明原因)", parameters={"case_id": "用例ID", "actual_result": "跳过原因(可选)"}, category="test-case"),
+    ToolDefinition(name="block_case", description="用例阻塞(pending→blocked，注明原因)", parameters={"case_id": "用例ID", "actual_result": "阻塞原因(可选)"}, category="test-case"),
+    # ── Bug ──
+    ToolDefinition(name="list_bugs", description="查看迭代Bug列表", parameters={"iteration_id": "迭代ID(可选)", "status": "按状态筛选(可选)"}, category="bug"),
+    ToolDefinition(name="report_bug", description="上报Bug(open)", parameters={"title": "Bug标题", "description": "描述", "severity": "严重度(可选)", "priority": "优先级(可选)", "iteration_id": "迭代ID(可选)", "case_id": "关联用例ID(可选)"}, category="bug"),
+    ToolDefinition(name="confirm_bug", description="确认Bug(open→confirmed)", parameters={"bug_id": "BugID"}, category="bug"),
+    ToolDefinition(name="start_fix", description="开始修复Bug(confirmed→fixing)", parameters={"bug_id": "BugID", "assignee_id": "处理人(可选)"}, category="bug"),
+    ToolDefinition(name="fix_bug", description="修复完成(fixing→fixed，附修复说明)", parameters={"bug_id": "BugID", "fix_description": "修复说明", "fix_commit": "修复Commit(可选)"}, category="bug"),
+    ToolDefinition(name="verify_bug", description="验证修复(fixed→verified)", parameters={"bug_id": "BugID"}, category="bug"),
+    ToolDefinition(name="close_bug", description="关闭Bug(verified→closed)", parameters={"bug_id": "BugID"}, category="bug"),
+    ToolDefinition(name="reject_bug", description="驳回Bug(open/confirmed→rejected，附意见)", parameters={"bug_id": "BugID", "comment": "驳回意见(必填)"}, category="bug"),
+    ToolDefinition(name="reopen_bug", description="重新打开Bug(closed/rejected→open)", parameters={"bug_id": "BugID"}, category="bug"),
+    # ── 部署环境 ──
+    ToolDefinition(name="list_envs", description="查看项目部署环境列表", parameters={"env_type": "环境类型(可选)", "status": "按状态筛选(可选)"}, category="deploy"),
+    ToolDefinition(name="configure_env", description="配置部署环境(configured)", parameters={"env_type": "环境类型(test/staging/production)", "host": "SSH主机", "user": "SSH用户", "deploy_path": "部署目录", "port": "SSH端口(可选)"}, category="deploy"),
+    ToolDefinition(name="verify_env", description="验证环境通过(configured→verified)", parameters={"env_id": "环境ID"}, category="deploy"),
 ]
 
 
@@ -166,7 +209,14 @@ SDL_PROMPT = """你是「开发产线」的驾驶舱 agent，负责软件项目
 - 仓库管理 → add_repo / list_repos / clone_repo。
 - 发现卡点（任务卡死、审核超时、失败、僵尸 claimed_by）时，主动定位根因并推动修复。
 - 发现「待回答问题」> 0 或角色提问时，加载 team-communication 技能按冒泡链处理（list_questions 列问题 → 能答则 answer_question → 不能答 escalate_question 沿冒泡路径转下一个处理方）。
-- 功能/需求管理 → 用户提新需求时 propose_feature；查功能清单 list_features；需求评审/验收时 approve_feature / reject_feature / verify_feature（状态机规范见 feature 技能）。"""
+- 功能/需求管理 → 用户提新需求时 propose_feature；查功能清单 list_features；需求评审/验收时 approve_feature / reject_feature / verify_feature（状态机规范见 feature 技能）。
+- 项目管理 → list_projects 列项目；启动/完成/归档/重开项目用 start_project / complete_project / archive_project / reopen_project（状态机规范见 project 技能）。
+- 迭代管理 → create_iteration 建迭代、list_iterations 列迭代；开始/完成/取消迭代用 start_iteration / complete_iteration / cancel_iteration（规范见 iteration 技能）。
+- 交付件评审 → submit_deliverable 提交、list_deliverables 列交付件；评审用 approve_deliverable / reject_deliverable（规范见 deliverable 技能）。
+- 测试计划 → create_test_plan 建计划、list_plans 列计划；审批/执行/完成用 approve_plan / start_plan / complete_plan（规范见 test-plan 技能）。
+- 测试用例 → create_case 建用例、list_cases 列用例；执行结果用 pass_case / fail_case / skip_case / block_case（失败用例应联动 report_bug，规范见 test-case 技能）。
+- Bug 管理 → report_bug 上报、list_bugs 列 Bug；确认/修复/验证/关闭/驳回/重开用 confirm_bug / start_fix / fix_bug / verify_bug / close_bug / reject_bug / reopen_bug（规范见 bug 技能）。
+- 部署环境 → configure_env 配环境、list_envs 列环境；验证用 verify_env（规范见 deploy-env 技能）。"""
 
 
 # ── 产线角色集（可插拔，替代硬编码 ROLE_SPECIFICS/ROLE_ALIASES/ROLE_CHAIN） ──
@@ -728,6 +778,600 @@ async def _h_verify_feature(sor, p, ctx):
     return f"OK: {msg}" if ok else f"ERROR: {msg}"
 
 
+# ── 通用 ID 前缀兜底 + scope 反查 ──
+
+async def _resolve_id(sor, table, fid):
+    """通用 ID 前缀兜底：精确匹配失败且入参≥6位时，用 LIKE 前缀解析回完整 ID。"""
+    if not fid:
+        return ""
+    recs = await sor.sqlExe(
+        f"SELECT id FROM {table} WHERE id=${{fid}}$", {"fid": fid})
+    if recs:
+        return getattr(recs[0], "id", "")
+    if len(fid) >= 6:
+        recs = await sor.sqlExe(
+            f"SELECT id FROM {table} WHERE id LIKE ${{prefix}}$ LIMIT 1",
+            {"prefix": fid + "%"})
+        if recs:
+            return getattr(recs[0], "id", "")
+    return ""
+
+
+async def _get_scope(sor, table, id_field, id_val, scope_field):
+    """反查实体的 scope 列值（如 iteration_id/plan_id）。返回 scope_val 或空。"""
+    recs = await sor.sqlExe(
+        f"SELECT {scope_field} FROM {table} WHERE {id_field}=${{id}}$",
+        {"id": id_val})
+    return getattr(recs[0], scope_field, "") if recs else ""
+
+
+async def _resolve_iteration_id(sor, pid, iteration_id):
+    """解析迭代 ID：显式传入则用，否则取项目第一个迭代。返回 (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 ASC LIMIT 1",
+        {"pid": pid})
+    if recs:
+        return getattr(recs[0], "id", ""), ""
+    return "", "项目无迭代，请先 create_iteration"
+
+
+async def _resolve_plan_id(sor, pid, plan_id):
+    """解析计划 ID：显式传入则用，否则取项目第一个计划（经迭代）。返回 (plan_id, error)。"""
+    if plan_id:
+        full = await _resolve_id(sor, "sd_test_plans", plan_id)
+        if full:
+            return full, ""
+        return "", f"计划不存在: {plan_id}"
+    iid, err = await _resolve_iteration_id(sor, pid, "")
+    if err:
+        return "", err
+    plans = await sor.sqlExe(
+        "SELECT id FROM sd_test_plans WHERE iteration_id=${iid}$ ORDER BY created_at ASC LIMIT 1",
+        {"iid": iid})
+    if not plans:
+        return "", "项目无测试计划，请先 create_test_plan"
+    return getattr(plans[0], "id", ""), ""
+
+
+# ── 项目生命周期 handler ──
+
+async def _h_list_projects(sor, p, ctx):
+    from .project_capability import list_projects
+    # 按当前用户 org 过滤（租户隔离），org_id 空则列全部
+    org_id = ""
+    uid = ctx.get("user_id", "")
+    if uid:
+        _u = await sor.sqlExe("SELECT orgid FROM users WHERE id=${u}$ LIMIT 1", {"u": uid})
+        if _u:
+            org_id = getattr(_u[0], "orgid", "") or ""
+    status = (p.get("status", "") or "").strip()
+    lst = await list_projects(org_id=org_id or None, status=status or None)
+    if not lst:
+        return "暂无项目"
+    lines = []
+    for r in lst:
+        lines.append(f"- [{r.get('status', '?')}] {r.get('name', '')} (id={r.get('id', '')})")
+    return "\n".join(lines)
+
+
+async def _h_start_project(sor, p, ctx):
+    pid = (p.get("project_id", "") or "").strip()
+    if not pid:
+        return "需要项目ID"
+    pid = await _resolve_id(sor, "sd_projects", pid) or pid
+    from .project_capability import start_project
+    ok, msg = await start_project(pid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_complete_project(sor, p, ctx):
+    pid = (p.get("project_id", "") or "").strip()
+    if not pid:
+        return "需要项目ID"
+    pid = await _resolve_id(sor, "sd_projects", pid) or pid
+    from .project_capability import complete_project
+    ok, msg = await complete_project(pid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_archive_project(sor, p, ctx):
+    pid = (p.get("project_id", "") or "").strip()
+    if not pid:
+        return "需要项目ID"
+    pid = await _resolve_id(sor, "sd_projects", pid) or pid
+    from .project_capability import archive_project
+    ok, msg = await archive_project(pid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_reopen_project(sor, p, ctx):
+    pid = (p.get("project_id", "") or "").strip()
+    if not pid:
+        return "需要项目ID"
+    pid = await _resolve_id(sor, "sd_projects", pid) or pid
+    from .project_capability import reopen_project
+    ok, msg = await reopen_project(pid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+# ── 迭代 handler ──
+
+async def _h_list_iterations(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    from .iteration_capability import list_iterations
+    status = (p.get("status", "") or "").strip()
+    lst = await list_iterations(pid, status=status or None)
+    if not lst:
+        return "暂无迭代"
+    lines = []
+    for r in lst:
+        lines.append(f"- [{r.get('status', '?')}] {r.get('iteration_name', '')} (id={r.get('id', '')})")
+    return "\n".join(lines)
+
+
+async def _h_create_iteration(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    name = (p.get("iteration_name", "") or "").strip()
+    if not name:
+        return "需要迭代名称"
+    from .iteration_capability import create_iteration
+    ok, iid = await create_iteration(
+        project_id=pid,
+        iteration_name=name,
+        iteration_type=p.get("iteration_type", "new_feature") or "new_feature",
+        scope=p.get("scope", "") or "",
+        created_by=ctx.get("user_id", "") or "",
+        who="agent.main_agent",
+    )
+    if not ok:
+        return f"ERROR: {iid}"
+    return f"OK: 已创建迭代「{name}」（id={iid}，状态 planning）"
+
+
+async def _h_start_iteration(sor, p, ctx):
+    iid = (p.get("iteration_id", "") or "").strip()
+    if not iid:
+        return "需要迭代ID"
+    pid = ctx.get("project_id", "")
+    iid = await _resolve_id(sor, "sd_iterations", iid) or iid
+    from .iteration_capability import start_iteration
+    ok, msg = await start_iteration(iid, pid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_complete_iteration(sor, p, ctx):
+    iid = (p.get("iteration_id", "") or "").strip()
+    if not iid:
+        return "需要迭代ID"
+    pid = ctx.get("project_id", "")
+    iid = await _resolve_id(sor, "sd_iterations", iid) or iid
+    from .iteration_capability import complete_iteration
+    ok, msg = await complete_iteration(iid, pid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_cancel_iteration(sor, p, ctx):
+    iid = (p.get("iteration_id", "") or "").strip()
+    if not iid:
+        return "需要迭代ID"
+    pid = ctx.get("project_id", "")
+    iid = await _resolve_id(sor, "sd_iterations", iid) or iid
+    from .iteration_capability import cancel_iteration
+    ok, msg = await cancel_iteration(iid, pid, who="agent.main_agent",
+                                     comment=p.get("comment", "") or "")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+# ── 交付件评审 handler ──
+
+async def _h_submit_deliverable(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    title = (p.get("title", "") or "").strip()
+    if not title:
+        return "需要交付件标题"
+    from .deliverable_capability import submit_deliverable
+    ok, did = await submit_deliverable(
+        project_id=pid,
+        title=title,
+        deliverable_type=p.get("deliverable_type", "code") or "code",
+        task_id=p.get("task_id", "") or "",
+        content=p.get("content", "") or "",
+        created_by=ctx.get("user_id", "") or "",
+        who="agent.main_agent",
+    )
+    if not ok:
+        return f"ERROR: {did}"
+    return f"OK: 已提交交付件「{title}」（id={did}，状态 pending）"
+
+
+async def _h_approve_deliverable(sor, p, ctx):
+    did = (p.get("deliverable_id", "") or "").strip()
+    if not did:
+        return "需要交付件ID"
+    pid = ctx.get("project_id", "")
+    did = await _resolve_id(sor, "pipeline_deliverables", did) or did
+    from .deliverable_capability import approve_deliverable
+    ok, msg = await approve_deliverable(did, pid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_reject_deliverable(sor, p, ctx):
+    did = (p.get("deliverable_id", "") or "").strip()
+    if not did:
+        return "需要交付件ID"
+    pid = ctx.get("project_id", "")
+    did = await _resolve_id(sor, "pipeline_deliverables", did) or did
+    from .deliverable_capability import reject_deliverable
+    ok, msg = await reject_deliverable(did, pid, who="agent.main_agent",
+                                       comment=p.get("comment", "") or "")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+# ── 测试计划 handler ──
+
+async def _h_list_plans(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    from .test_plan_capability import list_plans
+    iid, err = await _resolve_iteration_id(sor, pid, (p.get("iteration_id", "") or "").strip())
+    if err:
+        return err
+    status = (p.get("status", "") or "").strip()
+    lst = await list_plans(iid, status=status or None)
+    if not lst:
+        return "暂无测试计划"
+    lines = []
+    for r in lst:
+        lines.append(f"- [{r.get('status', '?')}] {r.get('plan_name', '')} (id={r.get('id', '')})")
+    return "\n".join(lines)
+
+
+async def _h_create_test_plan(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    name = (p.get("plan_name", "") or "").strip()
+    if not name:
+        return "需要方案名称"
+    iid, err = await _resolve_iteration_id(sor, pid, (p.get("iteration_id", "") or "").strip())
+    if err:
+        return err
+    from .test_plan_capability import create_test_plan
+    ok, plan_id = await create_test_plan(
+        iteration_id=iid,
+        plan_name=name,
+        plan_type=p.get("plan_type", "functional") or "functional",
+        scope=p.get("scope", "") or "",
+        created_by=ctx.get("user_id", "") or "",
+        who="agent.main_agent",
+    )
+    if not ok:
+        return f"ERROR: {plan_id}"
+    return f"OK: 已创建测试计划「{name}」（id={plan_id}，状态 draft）"
+
+
+async def _h_approve_plan(sor, p, ctx):
+    plan_id = (p.get("plan_id", "") or "").strip()
+    if not plan_id:
+        return "需要计划ID"
+    plan_id = await _resolve_id(sor, "sd_test_plans", plan_id) or plan_id
+    iid = await _get_scope(sor, "sd_test_plans", "id", plan_id, "iteration_id")
+    from .test_plan_capability import approve_plan
+    ok, msg = await approve_plan(plan_id, iid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_start_plan(sor, p, ctx):
+    plan_id = (p.get("plan_id", "") or "").strip()
+    if not plan_id:
+        return "需要计划ID"
+    plan_id = await _resolve_id(sor, "sd_test_plans", plan_id) or plan_id
+    iid = await _get_scope(sor, "sd_test_plans", "id", plan_id, "iteration_id")
+    from .test_plan_capability import start_plan
+    ok, msg = await start_plan(plan_id, iid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_complete_plan(sor, p, ctx):
+    plan_id = (p.get("plan_id", "") or "").strip()
+    if not plan_id:
+        return "需要计划ID"
+    plan_id = await _resolve_id(sor, "sd_test_plans", plan_id) or plan_id
+    iid = await _get_scope(sor, "sd_test_plans", "id", plan_id, "iteration_id")
+    from .test_plan_capability import complete_plan
+    ok, msg = await complete_plan(plan_id, iid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+# ── 测试用例 handler ──
+
+async def _h_list_cases(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    from .test_case_capability import list_cases
+    plan_id, err = await _resolve_plan_id(sor, pid, (p.get("plan_id", "") or "").strip())
+    if err:
+        return err
+    status = (p.get("status", "") or "").strip()
+    lst = await list_cases(plan_id, status=status or None)
+    if not lst:
+        return "暂无测试用例"
+    lines = []
+    for r in lst:
+        lines.append(f"- [{r.get('status', '?')}] {r.get('case_name', '')} (id={r.get('id', '')})")
+    return "\n".join(lines)
+
+
+async def _h_create_case(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    name = (p.get("case_name", "") or "").strip()
+    if not name:
+        return "需要用例名称"
+    plan_id, err = await _resolve_plan_id(sor, pid, (p.get("plan_id", "") or "").strip())
+    if err:
+        return err
+    from .test_case_capability import create_case
+    ok, cid = await create_case(
+        plan_id=plan_id,
+        case_name=name,
+        case_type=p.get("case_type", "functional") or "functional",
+        steps=p.get("steps", "") or "",
+        expected_result=p.get("expected_result", "") or "",
+        who="agent.main_agent",
+    )
+    if not ok:
+        return f"ERROR: {cid}"
+    return f"OK: 已创建测试用例「{name}」（id={cid}，状态 pending）"
+
+
+async def _h_pass_case(sor, p, ctx):
+    cid = (p.get("case_id", "") or "").strip()
+    if not cid:
+        return "需要用例ID"
+    cid = await _resolve_id(sor, "sd_test_cases", cid) or cid
+    plan_id = await _get_scope(sor, "sd_test_cases", "id", cid, "plan_id")
+    from .test_case_capability import pass_case
+    ok, msg = await pass_case(cid, plan_id, actual_result=p.get("actual_result", "") or "",
+                              who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_fail_case(sor, p, ctx):
+    cid = (p.get("case_id", "") or "").strip()
+    if not cid:
+        return "需要用例ID"
+    cid = await _resolve_id(sor, "sd_test_cases", cid) or cid
+    plan_id = await _get_scope(sor, "sd_test_cases", "id", cid, "plan_id")
+    from .test_case_capability import fail_case
+    ok, msg = await fail_case(cid, plan_id, actual_result=p.get("actual_result", "") or "",
+                              who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_skip_case(sor, p, ctx):
+    cid = (p.get("case_id", "") or "").strip()
+    if not cid:
+        return "需要用例ID"
+    cid = await _resolve_id(sor, "sd_test_cases", cid) or cid
+    plan_id = await _get_scope(sor, "sd_test_cases", "id", cid, "plan_id")
+    from .test_case_capability import skip_case
+    ok, msg = await skip_case(cid, plan_id, actual_result=p.get("actual_result", "") or "",
+                              who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_block_case(sor, p, ctx):
+    cid = (p.get("case_id", "") or "").strip()
+    if not cid:
+        return "需要用例ID"
+    cid = await _resolve_id(sor, "sd_test_cases", cid) or cid
+    plan_id = await _get_scope(sor, "sd_test_cases", "id", cid, "plan_id")
+    from .test_case_capability import block_case
+    ok, msg = await block_case(cid, plan_id, actual_result=p.get("actual_result", "") or "",
+                               who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+# ── Bug handler ──
+
+async def _h_list_bugs(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    from .bug_capability import list_bugs
+    iid, err = await _resolve_iteration_id(sor, pid, (p.get("iteration_id", "") or "").strip())
+    if err:
+        return err
+    status = (p.get("status", "") or "").strip()
+    lst = await list_bugs(iid, status=status or None)
+    if not lst:
+        return "暂无Bug"
+    lines = []
+    for r in lst:
+        lines.append(f"- [{r.get('status', '?')}][{r.get('severity', '') or ''}] "
+                     f"{r.get('title', '')} (id={r.get('id', '')})")
+    return "\n".join(lines)
+
+
+async def _h_report_bug(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    title = (p.get("title", "") or "").strip()
+    if not title:
+        return "需要Bug标题"
+    iid, err = await _resolve_iteration_id(sor, pid, (p.get("iteration_id", "") or "").strip())
+    if err:
+        return err
+    from .bug_capability import report_bug
+    ok, bid = await report_bug(
+        iteration_id=iid,
+        title=title,
+        description=p.get("description", "") or "",
+        severity=p.get("severity", "major") or "major",
+        priority=p.get("priority", "P1") or "P1",
+        case_id=p.get("case_id", "") or "",
+        reporter_type="agent",
+        reporter_id=ctx.get("user_id", "") or "",
+        who="agent.main_agent",
+    )
+    if not ok:
+        return f"ERROR: {bid}"
+    return f"OK: 已上报Bug「{title}」（id={bid}，状态 open）"
+
+
+async def _h_confirm_bug(sor, p, ctx):
+    bid = (p.get("bug_id", "") or "").strip()
+    if not bid:
+        return "需要BugID"
+    bid = await _resolve_id(sor, "sd_bugs", bid) or bid
+    iid = await _get_scope(sor, "sd_bugs", "id", bid, "iteration_id")
+    from .bug_capability import confirm_bug
+    ok, msg = await confirm_bug(bid, iid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_start_fix(sor, p, ctx):
+    bid = (p.get("bug_id", "") or "").strip()
+    if not bid:
+        return "需要BugID"
+    bid = await _resolve_id(sor, "sd_bugs", bid) or bid
+    iid = await _get_scope(sor, "sd_bugs", "id", bid, "iteration_id")
+    from .bug_capability import start_fix
+    ok, msg = await start_fix(bid, iid, assignee_id=p.get("assignee_id", "") or "",
+                              who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_fix_bug(sor, p, ctx):
+    bid = (p.get("bug_id", "") or "").strip()
+    if not bid:
+        return "需要BugID"
+    bid = await _resolve_id(sor, "sd_bugs", bid) or bid
+    iid = await _get_scope(sor, "sd_bugs", "id", bid, "iteration_id")
+    from .bug_capability import fix_bug
+    ok, msg = await fix_bug(bid, iid, fix_description=p.get("fix_description", "") or "",
+                            fix_commit=p.get("fix_commit", "") or "",
+                            who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_verify_bug(sor, p, ctx):
+    bid = (p.get("bug_id", "") or "").strip()
+    if not bid:
+        return "需要BugID"
+    bid = await _resolve_id(sor, "sd_bugs", bid) or bid
+    iid = await _get_scope(sor, "sd_bugs", "id", bid, "iteration_id")
+    from .bug_capability import verify_bug
+    ok, msg = await verify_bug(bid, iid, verified_by=ctx.get("user_id", "") or "",
+                               who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_close_bug(sor, p, ctx):
+    bid = (p.get("bug_id", "") or "").strip()
+    if not bid:
+        return "需要BugID"
+    bid = await _resolve_id(sor, "sd_bugs", bid) or bid
+    iid = await _get_scope(sor, "sd_bugs", "id", bid, "iteration_id")
+    from .bug_capability import close_bug
+    ok, msg = await close_bug(bid, iid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_reject_bug(sor, p, ctx):
+    bid = (p.get("bug_id", "") or "").strip()
+    if not bid:
+        return "需要BugID"
+    bid = await _resolve_id(sor, "sd_bugs", bid) or bid
+    iid = await _get_scope(sor, "sd_bugs", "id", bid, "iteration_id")
+    from .bug_capability import reject_bug
+    ok, msg = await reject_bug(bid, iid, who="agent.main_agent",
+                               comment=p.get("comment", "") or "")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+async def _h_reopen_bug(sor, p, ctx):
+    bid = (p.get("bug_id", "") or "").strip()
+    if not bid:
+        return "需要BugID"
+    bid = await _resolve_id(sor, "sd_bugs", bid) or bid
+    iid = await _get_scope(sor, "sd_bugs", "id", bid, "iteration_id")
+    from .bug_capability import reopen_bug
+    ok, msg = await reopen_bug(bid, iid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
+# ── 部署环境 handler ──
+
+async def _h_list_envs(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    from .deploy_capability import list_envs
+    env_type = (p.get("env_type", "") or "").strip()
+    status = (p.get("status", "") or "").strip()
+    lst = await list_envs(pid, env_type=env_type or None, status=status or None)
+    if not lst:
+        return "暂无部署环境"
+    lines = []
+    for r in lst:
+        lines.append(f"- [{r.get('env_type', '?')}][{r.get('status', '?')}] "
+                     f"{r.get('host', '')} ({r.get('user', '')}@{r.get('deploy_path', '')}) id={r.get('id', '')}")
+    return "\n".join(lines)
+
+
+async def _h_configure_env(sor, p, ctx):
+    pid = ctx.get("project_id", "")
+    if not pid:
+        return "请先切换到项目"
+    env_type = (p.get("env_type", "") or "").strip()
+    host = (p.get("host", "") or "").strip()
+    user = (p.get("user", "") or "").strip()
+    deploy_path = (p.get("deploy_path", "") or "").strip()
+    if not env_type or not host or not user or not deploy_path:
+        return "需要 env_type/host/user/deploy_path"
+    from .deploy_capability import configure_env
+    ok, eid = await configure_env(
+        project_id=pid,
+        env_type=env_type,
+        host=host,
+        user=user,
+        deploy_path=deploy_path,
+        port=p.get("port", 22),
+        who="agent.main_agent",
+    )
+    if not ok:
+        return f"ERROR: {eid}"
+    return f"OK: 已配置环境「{env_type}」（id={eid}，状态 configured）"
+
+
+async def _h_verify_env(sor, p, ctx):
+    eid = (p.get("env_id", "") or "").strip()
+    if not eid:
+        return "需要环境ID"
+    pid = ctx.get("project_id", "")
+    eid = await _resolve_id(sor, "sd_deploy_envs", eid) or eid
+    from .deploy_capability import verify_env
+    ok, msg = await verify_env(eid, pid, who="agent.main_agent")
+    return f"OK: {msg}" if ok else f"ERROR: {msg}"
+
+
 # ── 注册能力包 ──
 
 SDL_HANDLERS = {
@@ -751,6 +1395,49 @@ SDL_HANDLERS = {
     "approve_feature": _h_approve_feature,
     "reject_feature": _h_reject_feature,
     "verify_feature": _h_verify_feature,
+    # ── 项目生命周期 ──
+    "list_projects": _h_list_projects,
+    "start_project": _h_start_project,
+    "complete_project": _h_complete_project,
+    "archive_project": _h_archive_project,
+    "reopen_project": _h_reopen_project,
+    # ── 迭代 ──
+    "list_iterations": _h_list_iterations,
+    "create_iteration": _h_create_iteration,
+    "start_iteration": _h_start_iteration,
+    "complete_iteration": _h_complete_iteration,
+    "cancel_iteration": _h_cancel_iteration,
+    # ── 交付件评审 ──
+    "submit_deliverable": _h_submit_deliverable,
+    "approve_deliverable": _h_approve_deliverable,
+    "reject_deliverable": _h_reject_deliverable,
+    # ── 测试计划 ──
+    "list_plans": _h_list_plans,
+    "create_test_plan": _h_create_test_plan,
+    "approve_plan": _h_approve_plan,
+    "start_plan": _h_start_plan,
+    "complete_plan": _h_complete_plan,
+    # ── 测试用例 ──
+    "list_cases": _h_list_cases,
+    "create_case": _h_create_case,
+    "pass_case": _h_pass_case,
+    "fail_case": _h_fail_case,
+    "skip_case": _h_skip_case,
+    "block_case": _h_block_case,
+    # ── Bug ──
+    "list_bugs": _h_list_bugs,
+    "report_bug": _h_report_bug,
+    "confirm_bug": _h_confirm_bug,
+    "start_fix": _h_start_fix,
+    "fix_bug": _h_fix_bug,
+    "verify_bug": _h_verify_bug,
+    "close_bug": _h_close_bug,
+    "reject_bug": _h_reject_bug,
+    "reopen_bug": _h_reopen_bug,
+    # ── 部署环境 ──
+    "list_envs": _h_list_envs,
+    "configure_env": _h_configure_env,
+    "verify_env": _h_verify_env,
 }
 
 
diff --git a/pipeline_service/test_case_capability.py b/pipeline_service/test_case_capability.py
new file mode 100644
index 0000000..067180d
--- /dev/null
+++ b/pipeline_service/test_case_capability.py
@@ -0,0 +1,199 @@
+"""测试用例能力 — sd_test_cases 的执行状态机语义化迁移（通用、产线无关）。
+
+定位：用例 CRUD 走 xls2ui 生成的端点；本模块只做「执行结果状态流转」——
+每个迁移 CAS 原子（防并发/越权）+ 租户隔离 + 审计。
+
+用例的执行状态机/流转规则在 test-case skill 里（LLM 读 skill 判断合法性）；本模块只固化操作原语。
+失败用例联动建 Bug（case_id → sd_bugs.case_id）是规范，见 test-case skill。
+
+角色规范：role 参数用 agent.{role}（无前缀自动补 agent.）；人角色用 {orgtype}.{role}。
+
+scope 约定：sd_test_cases 无 project_id/iteration_id 列，用 plan_id 做范围校验（CAS 的 WHERE 里带 plan_id）。
+"""
+
+import logging
+from sqlor.dbpools import DBPools
+from appPublic.uniqueID import getID
+from .audit import record_audit
+
+DBNAME = "pipeline"
+logger = logging.getLogger("pipeline.test_case_capability")
+
+TABLE = "sd_test_cases"
+
+# 测试用例执行状态（SDLC 默认，状态机语义见 test-case skill）
+S_PENDING = "pending"      # 待执行
+S_PASS = "pass"            # 通过
+S_FAIL = "fail"            # 失败
+S_BLOCKED = "blocked"      # 阻塞
+S_SKIPPED = "skipped"      # 跳过
+
+
+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 _execute_transition(case_id, plan_id, from_states, to_state, action,
+                              actual_result="", executed_by="", duration_ms=None,
+                              who=None, agent_id=None):
+    """执行结果 CAS 迁移 + 审计（记录执行人/实际结果/执行时间）。返回 (ok, message)。"""
+    if not case_id or not plan_id:
+        return False, "缺少 case_id 或 plan_id"
+    if isinstance(from_states, str):
+        from_states = [from_states]
+    from_in = ",".join(f"${{s{i}}}$" for i in range(len(from_states)))
+    params = {"to": to_state, "cid": case_id, "pid": plan_id,
+              "ar": actual_result or '', "eb": executed_by or ''}
+    for i, s in enumerate(from_states):
+        params[f"s{i}"] = s
+    if duration_ms is not None:
+        try:
+            params["dur"] = int(duration_ms)
+            dur_set = ", duration_ms=${dur}$"
+        except (TypeError, ValueError):
+            dur_set = ""
+    else:
+        dur_set = ""
+
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        await sor.sqlExe(
+            f"UPDATE {TABLE} SET status=${{to}}$, actual_result=${{ar}}$, "
+            f"executed_by=${{eb}}$, executed_at=NOW(){dur_set} "
+            f"WHERE id=${{cid}}$ AND plan_id=${{pid}}$ AND status IN ({from_in})",
+            params)
+        recs = await sor.R(TABLE, {'id': case_id})
+        await sor.sqlExe("COMMIT", {})
+        if not recs:
+            return False, "用例不存在"
+        cur = getattr(recs[0], 'status', '')
+        if cur != to_state:
+            return False, f"状态迁移失败(CAS): 期望 from={from_states} 实际 status={cur}"
+        await record_audit(plan_id, TABLE, case_id, action,
+                           from_state="|".join(from_states), to_state=to_state,
+                           who=who, agent_id=agent_id, detail=actual_result, sor=sor)
+        return True, to_state
+
+
+async def create_case(plan_id, case_name, case_type="functional", priority="P2",
+                      precondition="", steps="", expected_result="",
+                      created_by="", who=None, agent_id=None):
+    """创建用例：新建记录，status=pending。返回 (ok, case_id_or_message)。"""
+    if not plan_id:
+        return False, "缺少 plan_id"
+    if not case_name or not case_name.strip():
+        return False, "缺少 case_name"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        cid = getID()
+        await sor.C(TABLE, {
+            'id': cid,
+            'plan_id': plan_id,
+            'case_name': case_name.strip(),
+            'case_type': case_type or 'functional',
+            'priority': priority or 'P2',
+            'precondition': precondition or '',
+            'steps': steps or '',
+            'expected_result': expected_result or '',
+            'status': S_PENDING,
+        })
+        await record_audit(plan_id, TABLE, cid, 'create',
+                           to_state=S_PENDING, who=_normalize_role(who),
+                           agent_id=agent_id, sor=sor)
+        logger.info("create_case: %s plan=%s", cid, plan_id)
+        return True, cid
+
+
+async def pass_case(case_id, plan_id, actual_result="", executed_by="",
+                    duration_ms=None, who=None, agent_id=None):
+    """用例通过：pending/fail → pass。"""
+    return await _execute_transition(case_id, plan_id, [S_PENDING, S_FAIL], S_PASS, 'pass',
+                                     actual_result=actual_result, executed_by=executed_by,
+                                     duration_ms=duration_ms, who=who, agent_id=agent_id)
+
+
+async def fail_case(case_id, plan_id, actual_result="", executed_by="",
+                    duration_ms=None, who=None, agent_id=None):
+    """用例失败：pending/pass → fail（失败应联动建 Bug，规范见 test-case skill）。"""
+    return await _execute_transition(case_id, plan_id, [S_PENDING, S_PASS], S_FAIL, 'fail',
+                                     actual_result=actual_result, executed_by=executed_by,
+                                     duration_ms=duration_ms, who=who, agent_id=agent_id)
+
+
+async def block_case(case_id, plan_id, actual_result="", executed_by="",
+                     who=None, agent_id=None):
+    """用例阻塞：pending → blocked（因环境/依赖无法执行）。"""
+    return await _execute_transition(case_id, plan_id, [S_PENDING], S_BLOCKED, 'block',
+                                     actual_result=actual_result, executed_by=executed_by,
+                                     who=who, agent_id=agent_id)
+
+
+async def skip_case(case_id, plan_id, actual_result="", executed_by="",
+                    who=None, agent_id=None):
+    """用例跳过：pending → skipped（有意不测，须注明原因）。"""
+    return await _execute_transition(case_id, plan_id, [S_PENDING], S_SKIPPED, 'skip',
+                                     actual_result=actual_result, executed_by=executed_by,
+                                     who=who, agent_id=agent_id)
+
+
+async def set_case_state(case_id, plan_id, from_state, to_state,
+                         who=None, agent_id=None, detail=None):
+    """通用 CAS 状态迁移兜底（跨产线自定义状态机用）。"""
+    return await _execute_transition(case_id, plan_id, [from_state], to_state, 'set_state',
+                                     actual_result=detail or '', who=who, agent_id=agent_id)
+
+
+async def list_cases(plan_id, status=None, limit=100) -> list:
+    """列出用例（可按状态过滤）。"""
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        conditions = ["plan_id=${pid}$"]
+        params = {"pid": plan_id}
+        if status:
+            conditions.append("status=${status}$")
+            params["status"] = status
+        where = " AND ".join(conditions)
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError):
+            limit = 100
+        sql = (f"SELECT * FROM {TABLE} WHERE {where} "
+               f"ORDER BY created_at ASC LIMIT {limit}")
+        recs = await sor.sqlExe(sql, params)
+        # 释放 SELECT 元数据锁
+        await sor.sqlExe("COMMIT", {})
+        result = []
+        for rec in (recs or []):
+            result.append(_rec_to_dict(rec))
+        return result
+
+
+def _rec_to_dict(rec):
+    """把 sqlor 记录对象转成 dict（sqlor 行是 DictObject，必须 dict(rec) 取列）。"""
+    if isinstance(rec, dict):
+        return dict(rec)
+    try:
+        return dict(rec)
+    except (TypeError, ValueError):
+        if hasattr(rec, 'to_dict'):
+            try:
+                return rec.to_dict()
+            except Exception:
+                pass
+        return {}
diff --git a/pipeline_service/test_plan_capability.py b/pipeline_service/test_plan_capability.py
new file mode 100644
index 0000000..3a94096
--- /dev/null
+++ b/pipeline_service/test_plan_capability.py
@@ -0,0 +1,166 @@
+"""测试计划能力 — sd_test_plans 的状态机语义化迁移（通用、产线无关）。
+
+定位：测试计划 CRUD 走 xls2ui 生成的端点；本模块只做「生命周期状态流转」——
+每个迁移 CAS 原子（防并发/越权）+ 租户隔离 + 审计。
+
+测试计划的生命周期/流转规则在 test-plan skill 里（LLM 读 skill 判断合法性）；本模块只固化操作原语。
+
+角色规范：role 参数用 agent.{role}（无前缀自动补 agent.）；人角色用 {orgtype}.{role}。
+
+scope 约定：sd_test_plans 无 project_id 列，用 iteration_id 做范围校验（CAS 的 WHERE 里带 iteration_id）。
+"""
+
+import logging
+from sqlor.dbpools import DBPools
+from appPublic.uniqueID import getID
+from .audit import record_audit
+
+DBNAME = "pipeline"
+logger = logging.getLogger("pipeline.test_plan_capability")
+
+TABLE = "sd_test_plans"
+
+# 测试计划状态（SDLC 默认，状态机语义见 test-plan skill）
+S_DRAFT = "draft"            # 草稿
+S_APPROVED = "approved"      # 已审批
+S_EXECUTING = "executing"    # 执行中
+S_COMPLETED = "completed"    # 已完成
+
+
+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(plan_id, iteration_id, from_state, to_state, action,
+                      who=None, agent_id=None, detail=None):
+    """CAS 状态迁移 + 审计。返回 (ok, message)。"""
+    if not plan_id or not iteration_id:
+        return False, "缺少 plan_id 或 iteration_id"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        await sor.sqlExe(
+            f"UPDATE {TABLE} SET status=${{to}}$, updated_at=NOW() "
+            "WHERE id=${pid}$ AND iteration_id=${iid}$ AND status=${from}$",
+            {"to": to_state, "pid": plan_id, "iid": iteration_id, "from": from_state})
+        recs = await sor.R(TABLE, {'id': plan_id})
+        await sor.sqlExe("COMMIT", {})
+        if not recs:
+            return False, "测试计划不存在"
+        cur = getattr(recs[0], 'status', '')
+        if cur != to_state:
+            return False, f"状态迁移失败(CAS): 期望 from={from_state} 实际 status={cur}"
+        await record_audit(iteration_id, TABLE, plan_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 create_test_plan(iteration_id, plan_name, plan_type="functional",
+                           scope="", environment="", entry_criteria="", exit_criteria="",
+                           created_by="", who=None, agent_id=None):
+    """创建测试计划：新建记录，status=draft。返回 (ok, plan_id_or_message)。"""
+    if not iteration_id:
+        return False, "缺少 iteration_id"
+    if not plan_name or not plan_name.strip():
+        return False, "缺少 plan_name"
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        pid = getID()
+        await sor.C(TABLE, {
+            'id': pid,
+            'iteration_id': iteration_id,
+            'plan_name': plan_name.strip(),
+            'plan_type': plan_type or 'functional',
+            'scope': scope or '',
+            'environment': environment or '',
+            'entry_criteria': entry_criteria or '',
+            'exit_criteria': exit_criteria or '',
+            'status': S_DRAFT,
+            'created_by': created_by or '',
+        })
+        await record_audit(iteration_id, TABLE, pid, 'create',
+                           to_state=S_DRAFT, who=_normalize_role(who),
+                           agent_id=agent_id, sor=sor)
+        logger.info("create_test_plan: %s iteration=%s", pid, iteration_id)
+        return True, pid
+
+
+async def approve_plan(plan_id, iteration_id, who=None, agent_id=None):
+    """审批通过：draft → approved。"""
+    return await _transition(plan_id, iteration_id, S_DRAFT, S_APPROVED, 'approve',
+                             who=who, agent_id=agent_id)
+
+
+async def start_plan(plan_id, iteration_id, who=None, agent_id=None):
+    """开始执行：approved → executing。"""
+    return await _transition(plan_id, iteration_id, S_APPROVED, S_EXECUTING, 'start',
+                             who=who, agent_id=agent_id)
+
+
+async def complete_plan(plan_id, iteration_id, who=None, agent_id=None):
+    """完成测试：executing → completed。"""
+    return await _transition(plan_id, iteration_id, S_EXECUTING, S_COMPLETED, 'complete',
+                             who=who, agent_id=agent_id)
+
+
+async def set_plan_state(plan_id, iteration_id, from_state, to_state,
+                         who=None, agent_id=None, detail=None):
+    """通用 CAS 状态迁移兜底（跨产线自定义状态机用）。"""
+    return await _transition(plan_id, iteration_id, from_state, to_state, 'set_state',
+                             who=who, agent_id=agent_id, detail=detail)
+
+
+async def list_plans(iteration_id, status=None, limit=50) -> list:
+    """列出测试计划（可按状态过滤）。"""
+    db, dbname = _get_db()
+    async with db.sqlorContext(dbname) as sor:
+        conditions = ["iteration_id=${iid}$"]
+        params = {"iid": iteration_id}
+        if status:
+            conditions.append("status=${status}$")
+            params["status"] = status
+        where = " AND ".join(conditions)
+        try:
+            limit = int(limit)
+        except (TypeError, ValueError):
+            limit = 50
+        sql = (f"SELECT * FROM {TABLE} WHERE {where} "
+               f"ORDER BY created_at ASC LIMIT {limit}")
+        recs = await sor.sqlExe(sql, params)
+        # 释放 SELECT 元数据锁
+        await sor.sqlExe("COMMIT", {})
+        result = []
+        for rec in (recs or []):
+            result.append(_rec_to_dict(rec))
+        return result
+
+
+def _rec_to_dict(rec):
+    """把 sqlor 记录对象转成 dict（sqlor 行是 DictObject，必须 dict(rec) 取列）。"""
+    if isinstance(rec, dict):
+        return dict(rec)
+    try:
+        return dict(rec)
+    except (TypeError, ValueError):
+        if hasattr(rec, 'to_dict'):
+            try:
+                return rec.to_dict()
+            except Exception:
+                pass
+        return {}