feat(bidding): 投标产线接入流程裁剪——BID_FLOW_STAGES声明11阶段(不可裁4/警告3/自由裁4+deps+roles);bid_flow对账器全流程计划门禁(被裁维度不派/qc裁剪视为通过/prep/章节评审直通approved/整书评分直通passed/交付直通completed+驳回暂停);bid_common OPEN_TASK_STATES含waiting(防拆解子任务重复派发)+DIM_TO_STAGE映射;dispatch_analysis_dim拒派被裁维度

pipeline_bidding/bid_ability.py

@@ -11,7 +11,7 @@ import json
 import logging
 
 from pipeline_core import (
-    ToolDefinition, PipelineAbility, RoleSpec, register_ability,
+    ToolDefinition, PipelineAbility, RoleSpec, FlowStage, register_ability,
     SlashCommand, register_slash_command,
 )
 
@@ -19,6 +19,53 @@ logger = logging.getLogger("pipeline.bidding.ability")
 
 PIPELINE_ID = "bidding_general"
 
+# ══════════════════ 预制流程阶段声明（流程裁剪机制，pipeline_flow_plans 消费）══════════════════
+# trim: no=不可裁（流程不成立）/ warn=可裁但确认待办里给风险提示（质量门禁类）/ yes=自由裁
+# deps 只写「硬依赖」（裁了下游不成立）；QC 类门禁的放行语义在 bid_flow 里处理
+#（qc_analysis 被裁 → 章节依赖自动视为通过），不进 deps。
+# roles 用于：① 确认时收敛取消被裁阶段在办任务（仅当角色不属于任何保留阶段）；
+#            ② 对账器按阶段门禁派发。
+BID_FLOW_STAGES = [
+    FlowStage(key="analysis_scoring", label="评分项与得分规则解析",
+              description="从招标文件抽取评分项、得分规则、否决项——全流程交叉核对基准",
+              trim="no", roles=["agent.tender_analyst"]),
+    FlowStage(key="analysis_quals", label="资质清单解析",
+              description="抽取招标所需资质与要求（商务章与资料准备的上游）",
+              trim="warn", warn_note="裁掉资质解析后，商务章编写与资质文件准备失去依据，商务分可能失分",
+              roles=["agent.tender_analyst"], deps=["analysis_scoring"]),
+    FlowStage(key="analysis_reqs_outline", label="投标文件要求与章节骨架",
+              description="抽取投标文件格式/结构要求并生成章节骨架（评分项挂载）——所有章节编写的依据",
+              trim="no", roles=["agent.tender_analyst"], deps=["analysis_scoring"]),
+    FlowStage(key="analysis_cost_benefit", label="成本收益分析",
+              description="投标成本与收益测算（保证金/预算/付款条件），报价章的上游",
+              trim="yes", roles=["agent.tender_analyst"],
+              deps=["analysis_scoring", "analysis_reqs_outline"]),
+    FlowStage(key="qc_analysis", label="解析产出QC契合度审核",
+              description="QC 对照招标文件原文逐类审核解析产出（10分制，不过退回重做）",
+              trim="warn", warn_note="裁掉解析QC后，评分项/资质/骨架的抽取错误将不经核对直接进入编写阶段，可能整章返工",
+              roles=["agent.qc"], deps=["analysis_scoring", "analysis_reqs_outline"]),
+    FlowStage(key="prep", label="投标资料准备",
+              description="知识库资质匹配 + 缺失资质的人类文件清单（只阻塞商务章）",
+              trim="yes", roles=["agent.bid_prep"], deps=["analysis_quals"]),
+    FlowStage(key="chapter_write", label="分章节编写",
+              description="按章节骨架逐章编写（商务/技术分角色、并发派发）",
+              trim="no", roles=["agent.bid_writer", "agent.bid_biz_writer"],
+              deps=["analysis_reqs_outline"]),
+    FlowStage(key="chapter_review", label="章节评审",
+              description="每章对照评分项/投标文件要求评审打分，不达标退回重写（章节级质量循环）",
+              trim="warn", warn_note="裁掉章节评审后，章节写完即通过，无质量把关，整书评分（若保留）大概率反复打回",
+              roles=["agent.bid_reviewer"], deps=["chapter_write"]),
+    FlowStage(key="compose", label="标书合成",
+              description="全部章节通过后合成完整标书（md+docx）",
+              trim="no", roles=["agent.bid_compositor"], deps=["chapter_write"]),
+    FlowStage(key="whole_score", label="整书评分（模拟评标）",
+              description="按评分项+得分规则给整书打分，不达标改进意见落到章节退回重写",
+              trim="yes", roles=["agent.bid_scorer"], deps=["compose"]),
+    FlowStage(key="delivery_confirm", label="交付确认",
+              description="人工复核用印/密封/份数/递交后确认交付，项目完成",
+              trim="yes", roles=[], deps=["compose"]),
+]
+
 # ══════════════════ 工具定义（主 agent / 驾驶舱）══════════════════
 
 BID_TOOLS = [
@@ -117,6 +164,17 @@ BID_PROMPT = """你是「投标产线」的驾驶舱 agent，负责从招标文
 - 用户问「能得多少分 / 差在哪」→ bid_score_report + list_reviews，讲清失分项与改进意见落到了哪个章节。
 - 用户要补资质/案例 → add_kb_doc 录入公司投标知识库；知识库缺的用 request_human_docs 让人工提供。
 
+## 流程裁剪（2026-09-11 新增）
+用户提出「简化流程/跳过某阶段/只编写不评分」这类要求时：
+1. 先 show_flow_template 列出预制流程各阶段（含可裁性）；
+2. 把用户要求映射为 trim_keys（要裁掉的阶段 key），调 propose_flow_plan 生成裁剪草案——
+   合法性由系统硬校验（不可裁阶段/依赖断裂会被拒绝并说明原因，你把原因如实转告用户）；
+3. 草案生成后系统自动发「流程裁剪确认」待办：**只有用户在待办里确认才生效，你没有确认权**。
+   告知用户去待办确认，并如实转述 warn 级阶段的风险提示；
+4. 用户驳回 → get_flow_plan 读驳回意见 → 按意见修订 trim_keys 重新 propose（version 自动+1）；
+5. 裁剪只在项目启动阶段有效（已有任务产出结果后拒绝）；确认后不可改。
+裁剪被驳回期间对账器暂停派发——收到驳回后**主动**按意见修订重提，不要等用户催。
+
 ## 硬规则
 - 招标金额、投标时间、资质要求这些要素，只能来自招标文件原文或人工确认，**禁止编造**；
   抽取不到就如实说"文件未载明"，必要时 ask_user。
@@ -446,6 +504,7 @@ def register_bid_ability():
         system_prompt=BID_PROMPT,
         handlers=handlers,
         roles=BID_ROLES,
+        flow_stages=BID_FLOW_STAGES,
         menus=[
             {"label": "📄 招标文件", "icon": "", "url": "/pipeline-bidding/bid_tender_files/index.ui",
              "type": "popup", "width": "88%", "height": "82%", "require_project": True},

pipeline_bidding/bid_common.py

@@ -37,11 +37,14 @@ HT_TENDER_FILE = "tender_file_upload"
 HT_HUMAN_DOCS = "human_docs_request"
 HT_QC_ESCALATION = "qc_escalation"
 HT_DELIVERY_CONFIRM = "bid_delivery_confirm"
+HT_FLOW_CONFIRM = "flow_plan_confirm"    # 流程裁剪确认（引擎通用类型，pipeline_flow_plans 机制）
 
-BLOCKING_HT_TYPES = (HT_TENDER_FILE,)
+BLOCKING_HT_TYPES = (HT_TENDER_FILE, HT_FLOW_CONFIRM)
 # 2026-09-03 起只保留「等招标文件」为全局阻塞：
 # - HT_HUMAN_DOCS（等人类资质文件）降级为章节级依赖——只阻塞商务章（等资料准备），技术/报价章不受影响；
 # - HT_QC_ESCALATION（QC 轮次用尽）降级为章节级依赖——只阻塞依赖该产出类型的章节，其余章节照常流转。
+# 2026-09-11 增 HT_FLOW_CONFIRM：裁剪计划待用户确认期间全局阻塞（确认后才按计划派发，
+# 防「用户在确认、对账器按旧流程抢跑」）。驳回后的暂停由 bid_flow 查 pipeline_flow_plans 状态兜住。
 
 # ── QC 契合度审核类型（解析产出逐类审核）──
 QC_TYPE_SCORING = "scoring_items"          # 评分项 + 得分规则
@@ -60,8 +63,30 @@ QC_OUTPUT_TABLE = {
     QC_TYPE_COST: "bid_cost_benefit",
 }
 
+# ── 流程裁剪（pipeline_flow_plans 机制，2026-09-11）──
+# 裁剪计划阶段 key（bid_ability.BID_FLOW_STAGES 声明）→ 分析维度映射。
+# 对账器/PM 编排据此判定某维度是否执行、被裁维度的 QC 依赖是否视为通过。
+STAGE_ANALYSIS = "analysis_scoring"
+STAGE_QUALS = "analysis_quals"
+STAGE_REQS_OUTLINE = "analysis_reqs_outline"
+STAGE_COST_BENEFIT = "analysis_cost_benefit"
+STAGE_QC = "qc_analysis"
+STAGE_PREP = "prep"
+STAGE_CH_REVIEW = "chapter_review"
+STAGE_WHOLE_SCORE = "whole_score"
+STAGE_DELIVERY = "delivery_confirm"
+
+DIM_TO_STAGE = {
+    "scoring": STAGE_ANALYSIS,
+    "quals": STAGE_QUALS,
+    "reqs_outline": STAGE_REQS_OUTLINE,
+    "cost_benefit": STAGE_COST_BENEFIT,
+}
+
 # 任务未终结状态（判断「是否已有在办任务」用）
-OPEN_TASK_STATES = ("submitted", "running", "review", "qc_review")
+# waiting 计入在办（2026-09-11）：PM 复杂度拆解的子任务带 depends_on 时为 waiting，
+# 不算在办会导致对账器重复派发整维度任务/对半成品提前派 QC（09-04 编排堆积同款事故模式）。
+OPEN_TASK_STATES = ("submitted", "running", "review", "qc_review", "waiting")
 # failed 任务单独处理：角色 agent 执行失败（如机构未配 LLM）→ 对账器不重复造任务，抛人工介入
 
 # ── 门限默认值（可用 appbase params 表覆盖）──

pipeline_bidding/bid_flow.py

@@ -26,7 +26,9 @@ from .bid_common import (
     DOC_DRAFT, DOC_REVIEWING, DOC_PASSED, DOC_REJECTED, DOC_DELIVERED,
     HT_TENDER_FILE, HT_HUMAN_DOCS, HT_QC_ESCALATION, HT_DELIVERY_CONFIRM,
     QC_TYPE_SCORING, QC_TYPE_QUALS, QC_TYPE_REQS, QC_TYPE_OUTLINE, QC_TYPE_COST, QC_TYPES,
-    QC_OUTPUT_TABLE,
+    QC_OUTPUT_TABLE, DIM_TO_STAGE,
+    STAGE_ANALYSIS, STAGE_QUALS, STAGE_REQS_OUTLINE, STAGE_COST_BENEFIT,
+    STAGE_QC, STAGE_PREP, STAGE_CH_REVIEW, STAGE_WHOLE_SCORE, STAGE_DELIVERY,
 )
 
 logger = logging.getLogger("pipeline.bidding")
@@ -428,6 +430,32 @@ async def _auto_import_workspace_tender(sor, project_id):
     return ""
 
 
+async def _flow_stage_table(project_id):
+    """取项目裁剪计划的阶段启用表（流程裁剪机制，pipeline_flow_plans）。
+
+    返回 None = 无已确认计划（标准全流程，存量项目零影响）；
+    dict = {stage_key: bool}，False 即该阶段被裁（用户已在待办确认）。
+    引擎不可用时诚实降级 None（不阻塞流转），并记 warning。
+    """
+    try:
+        from pipeline_service.flow_plan_capability import plan_stage_enabled
+        return await plan_stage_enabled(project_id)
+    except Exception as e:
+        logger.warning("bid_flow: flow plan unavailable project=%s: %s",
+                       project_id, str(e)[:160])
+        return None
+
+
+async def _latest_flow_plan_status(project_id):
+    """项目最新流程计划状态（'' = 从未提案）。驳回后暂停派发用。"""
+    try:
+        from pipeline_service.flow_plan_capability import get_latest_plan
+        plan = await get_latest_plan(project_id)
+        return (plan or {}).get("status", "") or "", (plan or {}).get("reject_comment", "") or ""
+    except Exception:
+        return "", ""
+
+
 async def reconcile_project(sor, project_id, project_name=""):
     """对单个投标项目做一次状态对账。返回动作说明列表（幂等，可反复调用）。"""
     acts = []
@@ -482,7 +510,38 @@ async def reconcile_project(sor, project_id, project_name=""):
                 await sor.sqlExe("COMMIT", {})
                 return acts + ["created: 上传招标文件任务 %s" % hid]
     if await has_pending_blocking_human_task(sor, project_id):
-        return ["blocked: 有待办人类任务（等招标文件/等人类资料/等QC人工介入），本轮不派发"]
+        return ["blocked: 有待办人类任务（等招标文件/等流程裁剪确认），本轮不派发"]
+
+    # ── 0c. 流程裁剪门禁（2026-09-11）──
+    # 最新计划被驳回 → 暂停派发，等会话 agent 按驳回意见修订重提（用户已表达要改流程，
+    # 不得按旧流程抢跑）；pending_confirm 由 0b 的阻塞待办挡住；confirmed 计划在下方
+    # 各段按 stage_table 生效；从未提案（''）→ 标准全流程（存量项目零影响）。
+    stage_table = await _flow_stage_table(project_id)
+    _fp_status, _fp_comment = await _latest_flow_plan_status(project_id)
+    if _fp_status == "rejected" and stage_table is None:
+        return acts + ["blocked: 流程裁剪方案被用户驳回（意见：%s），等会话 agent 修订重提，本轮不派发"
+                       % (_fp_comment[:200] or "未填")]
+
+    def _stage_on(key):
+        """阶段是否执行：无确认计划（None）= 标准全流程都执行。"""
+        if stage_table is None:
+            return True
+        return bool(stage_table.get(key, True))
+
+    async def _qc_passed_eff():
+        """章节依赖判定用的 QC 通过集（裁剪语义，2026-09-11）：
+        - qc_analysis 被裁 → 全部类型视为通过（用户已确认放弃解析QC门禁）；
+        - 某分析维度被裁 → 该维度对应 QC 类型视为通过（不会再有产出与审核）；
+        - 其余按 bid_qc_reviews 最新一轮实际判定。
+        """
+        if not _stage_on(STAGE_QC):
+            return set(QC_TYPES)
+        passed = await _qc_passed_types(sor, project_id)
+        for dim, stage in DIM_TO_STAGE.items():
+            if not _stage_on(stage):
+                for t in DIM_QC_TYPES.get(dim, ()):
+                    passed.add(t)
+        return passed
 
     th = await get_thresholds(sor)
 
@@ -524,6 +583,12 @@ async def reconcile_project(sor, project_id, project_name=""):
             if await _reset_analysis_for_new_file(sor, project_id, _ft):
                 acts.append("recovered: 检测到新招标文件，已清空旧分析残留，重新四维度分析")
         missing = await _missing_dims(n_items, n_struct, n_quals, n_cost, chapters)
+        # 流程裁剪：被裁分析维度不派（用户已在确认待办里确认）
+        _trimmed_dims = [d for d, _ in missing if not _stage_on(DIM_TO_STAGE.get(d, ""))]
+        if _trimmed_dims:
+            missing = [(d, lb) for d, lb in missing
+                       if _stage_on(DIM_TO_STAGE.get(d, ""))]
+            acts.append("trimmed: 分析维度「%s」已被用户裁剪，跳过" % "、".join(_trimmed_dims))
         created_dims = []
         first_run_pending = []
         for dim, label in missing:
@@ -591,7 +656,7 @@ async def reconcile_project(sor, project_id, project_name=""):
         # 过滤塞进 else 分支、创建挪出分支外，导致编排任务在办时仍用未过滤列表
         # 每 15 秒重复建任务（实测 2 分钟堆 9 个）。
         if first_run_pending:
-            qc_passed = await _qc_passed_types(sor, project_id)
+            qc_passed = await _qc_passed_eff()
             dispatchable, waiting_up = [], []
             for d, lb in first_run_pending:
                 up_ok = all(t in qc_passed
@@ -637,7 +702,8 @@ async def reconcile_project(sor, project_id, project_name=""):
     # 2026-09-03 章节级门禁：本段只负责「推进 QC」（重做/审核/冒泡），不再提前 return
     # 阻塞全局流转。哪些章节可以开写，由 C 段按章节依赖矩阵（CH_SECTION_DEPS）逐章判定——
     # 某一维度卡住只阻塞依赖它的章节，其余章节照常流转。
-    qc_pending = await _qc_pending_types(sor, project_id)
+    # 2026-09-11 流程裁剪：qc_analysis 阶段被裁 → 整段跳过（章节依赖由 _qc_passed_eff 视为全过）。
+    qc_pending = [] if not _stage_on(STAGE_QC) else await _qc_pending_types(sor, project_id)
     over_types = set()
     if qc_pending:
         # 轮次用尽守卫：最新审核已达上限仍未通过且产出未更新 → 不再空转派审核任务，
@@ -671,6 +737,8 @@ async def reconcile_project(sor, project_id, project_name=""):
                 dim = QC_TYPE_TO_DIM.get(t, "scoring")
                 if dim in redo_dims:
                     continue
+                if not _stage_on(DIM_TO_STAGE.get(dim, "")):
+                    continue      # 流程裁剪：被裁维度不派重做
                 if await _open_dim_tasks(sor, project_id, dim) > 0:
                     continue
                 n_redo_done = await _dim_redo_done_count(sor, project_id, dim)
@@ -756,8 +824,9 @@ async def reconcile_project(sor, project_id, project_name=""):
     # ── B. 资料准备（知识库资质匹配 + 人类文件清单）──
     # 2026-09-03 章节级依赖：资料准备是商务章（business）的上游依赖，
     # 不再阻塞全局流转——技术/报价章在上游 QC 通过后立即开写。
-    prep_done = await has_done_task(sor, project_id, R_PREP)
-    if n_quals > 0 and not prep_done:
+    # 2026-09-11 流程裁剪：prep 被裁 → 不派；商务章 biz_ok 判定同步视为完成。
+    prep_done = (not _stage_on(STAGE_PREP)) or await has_done_task(sor, project_id, R_PREP)
+    if _stage_on(STAGE_PREP) and n_quals > 0 and not prep_done:
         if await count_open_tasks(sor, project_id, role=R_PREP) > 0:
             acts.append("waiting: 资料准备任务在办（只阻塞商务章）")
         else:
@@ -778,7 +847,7 @@ async def reconcile_project(sor, project_id, project_name=""):
     # （CH_SECTION_DEPS），依赖满足立即派发——某一维度卡住只阻塞依赖它的章节，
     # 其余章节照常流转（根治：此前五类 QC 全过才放行，cost_benefit 一个维度慢
     # 导致全部章节冻结）。商务章额外等资料准备完成。
-    qc_passed = await _qc_passed_types(sor, project_id)
+    qc_passed = await _qc_passed_eff()
     hd_pending = await find_human_task(sor, project_id, HT_HUMAN_DOCS, status="pending")
     biz_ok = (prep_done or n_quals == 0) and not hd_pending
     created_w = 0
@@ -822,8 +891,29 @@ async def reconcile_project(sor, project_id, project_name=""):
         await sor.sqlExe("COMMIT", {})
 
     # ── D. 章节评审（written 且无在办评审任务）──
+    # 2026-09-11 流程裁剪：chapter_review 被裁 → written 章节直接置 approved（用户已确认
+    # 放弃章节级质量把关，确认待办里有风险提示），不派评审任务。
+    if not _stage_on(STAGE_CH_REVIEW):
+        _auto_appr = [c for c in chapters if c.get("status") == CH_WRITTEN]
+        for c in _auto_appr:
+            await sor.sqlExe(
+                "UPDATE bid_chapters SET status=${st}$, updated_at=NOW() WHERE id=${i}$ "
+                "AND status=${old}$",
+                {"st": CH_APPROVED, "i": c["id"], "old": CH_WRITTEN})
+        if _auto_appr:
+            await sor.sqlExe("COMMIT", {})
+            acts.append("trimmed: chapter_review 已裁剪，%d 个章节 written→approved 直通"
+                        % len(_auto_appr))
+            # 状态变了，本轮重读章节供 E 段判定
+            chs2 = await sor.sqlExe(
+                "SELECT id, chapter_no, title, section, status, revise_count FROM bid_chapters "
+                "WHERE project_id=${p}$ ORDER BY order_no, chapter_no", {"p": project_id})
+            await sor.sqlExe("COMMIT", {})
+            chapters = rows_to_dicts(chs2, limit=500)
     created_r = 0
     for c in chapters:
+        if not _stage_on(STAGE_CH_REVIEW):
+            break
         if c.get("status") != CH_WRITTEN:
             continue
         if await count_open_tasks(sor, project_id, role=R_REVIEWER, chapter_id=c["id"]) > 0:
@@ -929,7 +1019,17 @@ async def reconcile_project(sor, project_id, project_name=""):
             return acts + ["created: 标书合成任务 %s" % tid]
 
         # ── F. 整书评分 ──
+        # 2026-09-11 流程裁剪：whole_score 被裁 → 合成后直接置 passed（用户已确认
+        # 放弃模拟评标门禁），进入交付环节。
         if doc.get("status") in (DOC_DRAFT, DOC_REVIEWING):
+            if not _stage_on(STAGE_WHOLE_SCORE):
+                await sor.sqlExe(
+                    "UPDATE bid_documents SET status=${st}$, updated_at=NOW() WHERE id=${d}$ "
+                    "AND status IN ('" + DOC_DRAFT + "','" + DOC_REVIEWING + "')",
+                    {"st": DOC_PASSED, "d": doc.get("id")})
+                await sor.sqlExe("COMMIT", {})
+                return acts + ["trimmed: whole_score 已裁剪，标书 v%s 直接置 passed"
+                               % doc.get("version")]
             if await count_open_tasks(sor, project_id, role=R_SCORER) > 0:
                 return acts + ["waiting: 整书评分任务在办"]
             tid = await create_role_task(
@@ -942,6 +1042,16 @@ async def reconcile_project(sor, project_id, project_name=""):
 
         # ── G. 通过 → 交付确认 → 项目完成 ──
         if doc.get("status") == DOC_PASSED:
+            # 2026-09-11 流程裁剪：delivery_confirm 被裁 → 跳过人工交付确认直接完结
+            if not _stage_on(STAGE_DELIVERY):
+                await sor.sqlExe(
+                    "UPDATE bid_documents SET status=${st}$, updated_at=NOW() WHERE id=${d}$",
+                    {"st": DOC_DELIVERED, "d": doc.get("id")})
+                await sor.sqlExe(
+                    "UPDATE sd_projects SET status='completed', updated_at=NOW() "
+                    "WHERE id=${p}$ AND status<>'completed'", {"p": project_id})
+                await sor.sqlExe("COMMIT", {})
+                return acts + ["trimmed: delivery_confirm 已裁剪，标书直接置 delivered，项目完成"]
             ht = await find_human_task(sor, project_id, HT_DELIVERY_CONFIRM)
             if not ht:
                 hid = await create_human_task(
@@ -1074,7 +1184,7 @@ def start_poller():
                     POLLER_STATE["last_error"] = ""
                     for pid, acts in (res or {}).items():
                         for a in acts:
-                            if a.startswith(("created", "escalated", "completed", "error")):
+                            if a.startswith(("created", "escalated", "completed", "error", "trimmed")):
                                 try:
                                     from appPublic.log import debug as _dbg
                                     _dbg("bid_flow %s: %s" % (pid, a))

pipeline_bidding/bid_orchestration_capability.py

@@ -19,13 +19,24 @@ MAX_PM_INSTRUCTIONS = 2000
 
 
 async def _progress(sor, project_id):
-    """各维度进度：产出计数 / 最新 QC / 在办任务 / 依赖门禁状态。"""
+    """各维度进度：产出计数 / 最新 QC / 在办任务 / 依赖门禁状态 / 裁剪状态。"""
     from .bid_flow import (ANALYSIS_DIMS, DIM_QC_TYPES, DIM_UPSTREAM,
                            QC_OUTPUT_TABLE, _qc_passed_types, _open_dim_tasks)
+    from .bid_common import DIM_TO_STAGE
+    # 流程裁剪计划（None = 无确认计划 = 标准全流程）
+    try:
+        from pipeline_service.flow_plan_capability import plan_stage_enabled
+        stage_table = await plan_stage_enabled(project_id)
+    except Exception:
+        stage_table = None
     passed = await _qc_passed_types(sor, project_id)
     rows = []
     for dim, label, qc_ts in ANALYSIS_DIMS:
         info = {"dim": dim, "label": label}
+        if stage_table is not None and not stage_table.get(DIM_TO_STAGE.get(dim, ""), True):
+            info["裁剪"] = "已被用户裁剪（不得派发）"
+            rows.append(info)
+            continue
         cnt = 0
         for t in qc_ts:
             tbl = QC_OUTPUT_TABLE[t]
@@ -85,6 +96,7 @@ async def dispatch_analysis_dim(dim, instructions="", project_id="",
     """
     from .bid_flow import (ANALYSIS_DIMS, DIM_QC_TYPES, DIM_UPSTREAM,
                            R_ANALYST, _qc_passed_types, _open_dim_tasks)
+    from .bid_common import DIM_TO_STAGE
     dims = {d: lb for d, lb, _ts in ANALYSIS_DIMS}
     if dim not in dims:
         return False, "未知维度 %s（可选：%s）" % (dim, "、".join(dims))
@@ -92,6 +104,15 @@ async def dispatch_analysis_dim(dim, instructions="", project_id="",
         project_id = await resolve_project_id(project_id)
     except ValueError as e:
         return False, str(e)[:200]
+    # 流程裁剪门禁（2026-09-11）：被裁维度拒绝派发（用户已确认裁剪，PM 不得绕过）
+    try:
+        from pipeline_service.flow_plan_capability import plan_stage_enabled
+        _st = await plan_stage_enabled(project_id)
+        if _st is not None and not _st.get(DIM_TO_STAGE.get(dim, ""), True):
+            return False, ("维度「%s」已被用户在流程裁剪中确认裁掉，不得派发。"
+                           "用 analysis_progress 查看当前裁剪后的维度清单。" % dims[dim])
+    except ImportError:
+        pass
     label = dims[dim]
     db, dbname = get_db()
     async with db.sqlorContext(dbname) as sor: