feat(bidding): 批2纯技术方案流程bid_tech_proposal收口——bid_tech_capability(六类评估项抽取/补录/查询+模版三级检索+骨架提案用户确认门禁+确认落库)/bid_flow 0d多流程分支(LLM自动判流+_reconcile_tech_proposal T1-T7对账:解析→QC→模版确认→分章编写→章节评审→合成→交付)/BID_FLOW_TEMPLATES两模板注册/bid_qc tech_items审核维度+start_qc tech分支(查bid_tech_items)/5个tech工具schema注册/tech_template_confirm.dspy用户确认端点(agent无确认权)+load_path/码表补tech_items+cost_benefit/bug A修复:_resolve_file_path补workspace_base候选+import_tender_file落库前归一绝对路径(上传落点口径不一致致正文永读空,首医实测)/llm_json透传project_id记账

models/bid_tech_items.json

@@ -0,0 +1,30 @@
+{
+    "summary": [
+        {
+            "name": "bid_tech_items",
+            "title": "技术方案评估项(纯技术方案流程tech_analysis维度产出;kind=item单项/summary=结论;category六类:function_list功能清单/function_point功能点/hardware硬件配置/tech_arch技术架构/func_arch功能架构/deploy_arch部署架构;写入方bid_tech_capability.py;单表模式同bid_cost_benefit)",
+            "primary": ["id"],
+            "catelog": "entity"
+        }
+    ],
+    "fields": [
+        {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"},
+        {"name": "project_id", "title": "项目ID", "type": "str", "length": 32, "nullable": "no"},
+        {"name": "kind", "title": "行类型(item=单项/summary=结论)", "type": "str", "length": 16, "nullable": "no", "default": "item"},
+        {"name": "category", "title": "评估类别(function_list/function_point/hardware/tech_arch/func_arch/deploy_arch)", "type": "str", "length": 32, "nullable": "no", "default": "function_list"},
+        {"name": "item_name", "title": "条目名称", "type": "str", "length": 250, "nullable": "no"},
+        {"name": "requirement", "title": "需求书原文要求", "type": "text", "nullable": "yes"},
+        {"name": "response_plan", "title": "方案响应要点(技术方案怎么满足)", "type": "text", "nullable": "yes"},
+        {"name": "qty_note", "title": "数量/规模说明(功能点数/台数/容量等)", "type": "str", "length": 200, "nullable": "yes"},
+        {"name": "mandatory", "title": "是否硬性要求(1=需求书明确必须)", "type": "str", "length": 2, "nullable": "yes", "default": "0"},
+        {"name": "confidence", "title": "置信度(high/medium/low)", "type": "str", "length": 16, "nullable": "yes", "default": "medium"},
+        {"name": "source_ref", "title": "需求书出处(章节/条款号)", "type": "str", "length": 480, "nullable": "yes"},
+        {"name": "order_no", "title": "显示排序", "type": "int", "nullable": "yes", "default": "0"},
+        {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"},
+        {"name": "updated_at", "title": "更新时间(写入方显式NOW(),新机无ON UPDATE,平台约定)", "type": "timestamp", "nullable": "no"}
+    ],
+    "indexes": [
+        {"name": "idx_project", "idxtype": "index", "idxfields": ["project_id"]},
+        {"name": "idx_proj_cat", "idxtype": "index", "idxfields": ["project_id", "category"]}
+    ]
+}

pipeline_bidding/bid_ability.py

@@ -11,7 +11,7 @@ import json
 import logging
 
 from pipeline_core import (
-    ToolDefinition, PipelineAbility, RoleSpec, FlowStage, register_ability,
+    ToolDefinition, PipelineAbility, RoleSpec, FlowStage, FlowTemplate, register_ability,
     SlashCommand, register_slash_command,
 )
 
@@ -66,6 +66,55 @@ BID_FLOW_STAGES = [
               trim="yes", roles=[], deps=["compose"]),
 ]
 
+# ══════════════════ 纯技术方案流程（bid_tech_proposal，2026-09-14 批2）══════════════════
+# 用户 2026-09-11 定义：输入=技术需求书（非招标文件），分析集中在技术层面六类评估项
+# （功能清单/功能点/硬件配置/技术架构/功能架构/部署架构），方案书根据预定模版
+# （无则 RAG/网络检索），模版骨架**由用户确认后**按章节分段编写和 QC 评审，
+# 最终合成产出技术方案书。QC/评审**以用户提供的需求书为唯一基准**（对标此文件）。
+TECH_PROPOSAL_STAGES = [
+    FlowStage(key="tech_analysis", label="技术需求解析（六类评估项）",
+              description="从技术需求书抽取六类评估项：功能清单/功能点/硬件配置/技术架构/功能架构/部署架构，落库 bid_tech_items",
+              trim="no", roles=["agent.tender_analyst"]),
+    FlowStage(key="tech_qc", label="技术评估项QC契合度审核",
+              description="QC 对照技术需求书原文逐类审核评估项（10分制，不过退回重做）",
+              trim="warn", warn_note="裁掉技术QC后，评估项抽取错误将不经核对直接进入骨架拟定与编写阶段，可能整章偏离需求",
+              roles=["agent.qc"], deps=["tech_analysis"]),
+    FlowStage(key="template_confirm", label="方案模版确定+用户确认骨架",
+              description="预定模版（知识库）→ 无则 RAG 检索 → 无则网络搜索，拟定章节骨架（覆盖六类评估项），发待办由用户确认后才开写",
+              trim="no", roles=["agent.pm"], deps=["tech_analysis"]),
+    FlowStage(key="tech_chapter_write", label="技术方案分章编写",
+              description="按用户确认的骨架逐章编写（一章一任务并发），内容对标需求书评估项写到操作层",
+              trim="no", roles=["agent.bid_writer"], deps=["template_confirm"]),
+    FlowStage(key="tech_chapter_qc", label="章节QC评审（对标需求书）",
+              description="每章对照技术需求书评估项覆盖度评审打分，不达标退回重写（章节级质量循环）",
+              trim="warn", warn_note="裁掉章节评审后，章节写完即通过，无需求覆盖度把关，方案书可能漏答需求条目",
+              roles=["agent.bid_reviewer"], deps=["tech_chapter_write"]),
+    FlowStage(key="tech_compose", label="合成技术方案书",
+              description="全部章节通过后合成完整技术方案书（md+docx）",
+              trim="no", roles=["agent.bid_compositor"], deps=["tech_chapter_write"]),
+    FlowStage(key="tech_delivery_confirm", label="交付确认",
+              description="人工核对技术方案书后确认交付，项目完成",
+              trim="yes", roles=[], deps=["tech_compose"]),
+]
+
+# 多流程模板声明（≥2 条时文件导入触发 LLM 自动判流 + 用户确认，2026-09-14 批2）
+BID_FLOW_TEMPLATES = [
+    FlowTemplate(key="bid_standard", name="标准投标流程",
+                 description="收到招标文件 → 四维度解析 → QC → 资料准备 → 分章编写 → 章节评审 → 合成标书 → 整书评分 → 交付",
+                 detect_criteria=(
+                     "输入文件是【招标文件/招标公告/投标邀请书】：含评标办法、评分标准、分值分配、"
+                     "投标人资格要求、投标保证金、投标文件格式/密封/份数要求、开标时间等招投标程序性内容，"
+                     "最终产出应是一份响应评分标准的【投标文件（标书）】。"),
+                 stages=BID_FLOW_STAGES),
+    FlowTemplate(key="bid_tech_proposal", name="纯技术方案流程",
+                 description="技术需求书 → 六类技术评估项解析 → QC → 模版骨架用户确认 → 分章编写 → 章节QC对标需求书 → 合成技术方案书",
+                 detect_criteria=(
+                     "输入文件是【技术需求书/需求规格说明书/产品规格书/技术方案要求】：主体内容是功能需求、"
+                     "功能清单、功能点、性能指标、硬件配置、技术架构/功能架构/部署架构要求、接口/安全/验收标准等，"
+                     "**没有**评标办法/评分标准/投标程序等招投标内容，最终产出应是一份【技术方案书】而非标书。"),
+                 stages=TECH_PROPOSAL_STAGES),
+]
+
 # ══════════════════ 工具定义（主 agent / 驾驶舱）══════════════════
 
 BID_TOOLS = [
@@ -164,6 +213,16 @@ BID_PROMPT = """你是「投标产线」的驾驶舱 agent，负责从招标文
 - 用户问「能得多少分 / 差在哪」→ bid_score_report + list_reviews，讲清失分项与改进意见落到了哪个章节。
 - 用户要补资质/案例 → add_kb_doc 录入公司投标知识库；知识库缺的用 request_human_docs 让人工提供。
 
+## 多流程模板（2026-09-15 新增，纯技术方案流程）
+本产线有两条预制流程（show_flow_template 可列出）：
+- `bid_standard` 标准投标流程：输入=招标文件，产出=标书（默认）；
+- `bid_tech_proposal` 纯技术方案流程：输入=**技术需求书**（无评标办法/投标程序），产出=技术方案书——
+  六类技术评估项解析（功能清单/功能点/硬件配置/技术架构/功能架构/部署架构）→ QC 对标需求书
+  → 模版骨架**用户确认**后才分章编写 → 合成技术方案书。
+文件导入后系统会 LLM 自动判流并发「流程裁剪确认」待办（正文含识别结论与判据）——用户确认走哪条流程后才派发；
+判流失败会发人工待办，此时**不要替用户选流程**，引导用户在会话里说明文件性质，用
+propose_flow_plan(flow_key=...) 重新提案。用户明说「只要技术方案/这是需求书不是招标文件」时同样走 propose_flow_plan 指定。
+
 ## 流程裁剪（2026-09-11 新增）
 用户提出「简化流程/跳过某阶段/只编写不评分」这类要求时：
 1. 先 show_flow_template 列出预制流程各阶段（含可裁性）；
@@ -505,6 +564,7 @@ def register_bid_ability():
         handlers=handlers,
         roles=BID_ROLES,
         flow_stages=BID_FLOW_STAGES,
+        flow_templates=BID_FLOW_TEMPLATES,
         menus=[
             {"label": "📄 招标文件", "icon": "", "url": "/pipeline-bidding/bid_tender_files/index.ui",
              "type": "popup", "width": "88%", "height": "82%", "require_project": True},

pipeline_bidding/bid_analysis_capability.py

@@ -187,7 +187,8 @@ async def read_tender_file(file_id, offset=0, length=20000):
         txt = d.get("content_text") or ""
         if not txt.strip() and d.get("file_path"):
             pdir = await _project_dir_of(sor, d.get("project_id", ""))
-            real = _resolve_file_path(pdir, d["file_path"])
+            real = _resolve_file_path(pdir, d["file_path"],
+                                      ws_base=await _workspace_base_of(sor))
             txt = _extract_file_text(real) if real else ""
             if txt:
                 await sor.sqlExe(
@@ -256,10 +257,13 @@ async def _project_dir_of(sor, project_id):
         return ""
 
 
-def _resolve_file_path(project_dir, file_path):
+def _resolve_file_path(project_dir, file_path, ws_base=""):
     """把存储的 file_path 解析成磁盘绝对路径。
     - 绝对路径且存在：原样返回；
-    - 相对/裸文件名：依次尝试 项目根目录、项目根目录下的 docs/、进程 cwd，存在即返回；
+    - 相对/裸文件名：依次尝试 项目根目录、项目根目录下的 docs/、
+      工作空间根（ws_base，2026-09-15 补——上传落点相对工作空间根存路径，
+      如 "0/第五章采购需求.docx"，旧候选全解析不到 → 正文永远读空）、进程 cwd，
+      存在即返回；
     - 都找不到返回 ''。
     """
     fp = str(file_path or "").strip()
@@ -271,6 +275,8 @@ def _resolve_file_path(project_dir, file_path):
     if project_dir:
         cands.append(os.path.join(project_dir, fp))
         cands.append(os.path.join(project_dir, "docs", fp))
+    if ws_base:
+        cands.append(os.path.join(ws_base, fp))
     cands.append(fp)
     for c in cands:
         if os.path.isfile(c):
@@ -278,6 +284,16 @@ def _resolve_file_path(project_dir, file_path):
     return ""
 
 
+async def _workspace_base_of(sor):
+    """工作空间根目录（params.workspace_base，引擎统一口径）。失败返回 ''。"""
+    try:
+        from pipeline_service.workspace import get_workspace_base
+        return await get_workspace_base(sor) or ""
+    except Exception as e:
+        logger.warning("workspace base resolve failed: %s", str(e)[:120])
+        return ""
+
+
 async def _best_tender_record(sor, project_id):
     """选最优招标文件记录（修复 2026-09-01「解析读旧截断稿」根因）。
 
@@ -297,12 +313,13 @@ async def _best_tender_record(sor, project_id):
         return {}
     rows = rows_to_dicts(recs, limit=50)
     pdir = await _project_dir_of(sor, project_id)
+    ws_base = await _workspace_base_of(sor)
     best, best_score, best_real = {}, -1, ""
     for r in rows:
         score = len((r.get("content_text") or "").strip())
         real = ""
         if (r.get("file_path") or "").strip():
-            real = _resolve_file_path(pdir, r["file_path"])
+            real = _resolve_file_path(pdir, r["file_path"], ws_base=ws_base)
             if real:
                 try:
                     score = max(score, os.path.getsize(real))
@@ -587,7 +604,8 @@ async def _load_tender_text(sor, project_id, file_id=""):
     txt = d.get("content_text") or ""
     if not txt.strip() and d.get("file_path"):
         pdir = await _project_dir_of(sor, project_id)
-        real = d.get("_resolved_path") or _resolve_file_path(pdir, d["file_path"])
+        real = d.get("_resolved_path") or _resolve_file_path(
+            pdir, d["file_path"], ws_base=await _workspace_base_of(sor))
         txt = _extract_file_text(real) if real else ""
         if txt:
             await sor.sqlExe(
@@ -624,7 +642,7 @@ async def extract_from_file(project_id, file_id="", model_name="", offset=0,
     o = to_int(offset, 0)
     seg = txt[o:o + to_int(length, 30000)]
     data, err = await llm_json(EXTRACT_PROMPT.replace("__CONTENT__", seg),
-                               model_name=model_name, retries=1)
+                               model_name=model_name, retries=1, project_id=project_id)
     if data is None:
         return False, "LLM 抽取失败: %s" % err
     cnt = {"scoring_items": 0, "qualifications": 0, "doc_requirements": 0}
@@ -662,12 +680,13 @@ async def extract_from_file(project_id, file_id="", model_name="", offset=0,
 
 # ── 四维度拆分抽取（2026-09-02：解析任务拆成四个并行维度，各派一个任务）──
 
-async def _extract_seg(txt, prompt, model_name, offset, length):
+async def _extract_seg(txt, prompt, model_name, offset, length, project_id=""):
     """取分段正文 → LLM 抽取。返回 (data, err)。"""
     o = to_int(offset, 0)
     seg = txt[o:o + to_int(length, 30000)]
     data, err = await llm_json(prompt.replace("__CONTENT__", seg),
-                               model_name=model_name, retries=1)
+                               model_name=model_name, retries=1,
+                               project_id=project_id or '')
     if data is None:
         return None, "LLM 抽取失败: %s" % err
     return data, ""
@@ -685,7 +704,8 @@ async def extract_scoring(project_id, file_id="", model_name="", offset=0,
         txt, fname, err = await _load_tender_text(sor, project_id, file_id=file_id)
         if err:
             return False, err
-    data, err = await _extract_seg(txt, PROMPT_SCORING, model_name, offset, length)
+    data, err = await _extract_seg(txt, PROMPT_SCORING, model_name, offset, length,
+                                   project_id=project_id)
     if data is None:
         return False, err
     cnt = 0
@@ -716,7 +736,8 @@ async def extract_quals(project_id, file_id="", model_name="", offset=0,
         txt, fname, err = await _load_tender_text(sor, project_id, file_id=file_id)
         if err:
             return False, err
-    data, err = await _extract_seg(txt, PROMPT_QUALS, model_name, offset, length)
+    data, err = await _extract_seg(txt, PROMPT_QUALS, model_name, offset, length,
+                                   project_id=project_id)
     if data is None:
         return False, err
     cnt = 0
@@ -745,7 +766,8 @@ async def extract_reqs_outline(project_id, file_id="", model_name="", offset=0,
         txt, fname, err = await _load_tender_text(sor, project_id, file_id=file_id)
         if err:
             return False, err
-    data, err = await _extract_seg(txt, PROMPT_REQS, model_name, offset, length)
+    data, err = await _extract_seg(txt, PROMPT_REQS, model_name, offset, length,
+                                   project_id=project_id)
     if data is None:
         return False, err
     cnt = 0
@@ -819,7 +841,8 @@ async def extract_cost_benefit(project_id, file_id="", model_name="", offset=0,
         txt, fname, err = await _load_tender_text(sor, project_id, file_id=file_id)
         if err:
             return False, err
-    data, err = await _extract_seg(txt, PROMPT_COST_BENEFIT, model_name, offset, length)
+    data, err = await _extract_seg(txt, PROMPT_COST_BENEFIT, model_name, offset, length,
+                                   project_id=project_id)
     if data is None:
         return False, err
     cnt = 0

pipeline_bidding/bid_common.py

@@ -38,13 +38,16 @@ 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 机制）
+HT_TEMPLATE_CONFIRM = "tech_template_confirm"   # 技术方案模版/骨架确认（纯技术方案流程，2026-09-14 批2）
 
-BLOCKING_HT_TYPES = (HT_TENDER_FILE, HT_FLOW_CONFIRM)
+BLOCKING_HT_TYPES = (HT_TENDER_FILE, HT_FLOW_CONFIRM, HT_TEMPLATE_CONFIRM)
 # 2026-09-03 起只保留「等招标文件」为全局阻塞：
 # - HT_HUMAN_DOCS（等人类资质文件）降级为章节级依赖——只阻塞商务章（等资料准备），技术/报价章不受影响；
 # - HT_QC_ESCALATION（QC 轮次用尽）降级为章节级依赖——只阻塞依赖该产出类型的章节，其余章节照常流转。
 # 2026-09-11 增 HT_FLOW_CONFIRM：裁剪计划待用户确认期间全局阻塞（确认后才按计划派发，
 # 防「用户在确认、对账器按旧流程抢跑」）。驳回后的暂停由 bid_flow 查 pipeline_flow_plans 状态兜住。
+# 2026-09-14 增 HT_TEMPLATE_CONFIRM：技术方案模版骨架待用户确认期间全局阻塞（用户明确要
+# 「模版由用户确认后」才开写章节——骨架没确认就编写 = 方向错了全白写）。
 
 # ── QC 契合度审核类型（解析产出逐类审核）──
 QC_TYPE_SCORING = "scoring_items"          # 评分项 + 得分规则
@@ -52,7 +55,11 @@ QC_TYPE_QUALS = "qualifications"           # 所需资质
 QC_TYPE_REQS = "doc_requirements"          # 投标文件要求
 QC_TYPE_OUTLINE = "chapter_outline"        # 章节骨架
 QC_TYPE_COST = "cost_benefit"              # 成本收益分析（2026-09-02 四维度拆分新增）
+QC_TYPE_TECH = "tech_items"                # 技术评估项（2026-09-14 纯技术方案流程新增）
 QC_TYPES = (QC_TYPE_SCORING, QC_TYPE_QUALS, QC_TYPE_REQS, QC_TYPE_OUTLINE, QC_TYPE_COST)
+# 技术方案流程专属 QC 类型（不进 QC_TYPES——标准流程的 _qc_pending_types 只遍历 QC_TYPES，
+# 技术方案流程由 bid_flow 的 tech 分支单独遍历 TECH_QC_TYPES，两条流程互不串审）
+TECH_QC_TYPES = (QC_TYPE_TECH,)
 
 # QC 审核对象 → 产出物落库表（QC 不通过时清空重做的对象；对账器判存在性也用）
 QC_OUTPUT_TABLE = {
@@ -61,6 +68,7 @@ QC_OUTPUT_TABLE = {
     QC_TYPE_REQS: "bid_doc_requirements",
     QC_TYPE_OUTLINE: "bid_chapters",
     QC_TYPE_COST: "bid_cost_benefit",
+    QC_TYPE_TECH: "bid_tech_items",
 }
 
 # ── 流程裁剪（pipeline_flow_plans 机制，2026-09-11）──
@@ -83,6 +91,23 @@ DIM_TO_STAGE = {
     "cost_benefit": STAGE_COST_BENEFIT,
 }
 
+# ── 纯技术方案流程（bid_tech_proposal，2026-09-14 批2）──
+# 阶段 key（bid_ability.TECH_PROPOSAL_STAGES 声明）→ 与标准流程同名字段但含义不同：
+# 该流程以用户提供的技术需求书为唯一基准，QC/评审全部对标需求书原文，不碰评分项。
+FLOW_KEY_TECH_PROPOSAL = "bid_tech_proposal"
+FLOW_KEY_STANDARD = "bid_standard"
+STAGE_TECH_ANALYSIS = "tech_analysis"          # 技术需求解析（六类评估项）
+STAGE_TECH_QC = "tech_qc"                      # 解析产出 QC（对标需求书原文）
+STAGE_TEMPLATE_CONFIRM = "template_confirm"    # 方案模版确定 + 用户确认骨架
+STAGE_TECH_CH_WRITE = "tech_chapter_write"     # 按确认骨架分章编写
+STAGE_TECH_CH_QC = "tech_chapter_qc"           # 章节 QC 评审（对标需求书覆盖度）
+STAGE_TECH_COMPOSE = "tech_compose"            # 合成技术方案书
+STAGE_TECH_DELIVERY = "tech_delivery_confirm"  # 交付确认
+
+# 六类技术评估项（用户 2026-09-11 定义：功能清单/功能点/硬件配置/技术架构/功能架构/部署架构）
+TECH_CATEGORIES = ("function_list", "function_point", "hardware",
+                   "tech_arch", "func_arch", "deploy_arch")
+
 # 任务未终结状态（判断「是否已有在办任务」用）
 # waiting 计入在办（2026-09-11）：PM 复杂度拆解的子任务带 depends_on 时为 waiting，
 # 不算在办会导致对账器重复派发整维度任务/对半成品提前派 QC（09-04 编排堆积同款事故模式）。
@@ -385,13 +410,17 @@ async def find_human_task(sor, project_id, task_type, status=None):
     return rec_to_dict(recs[0]) if recs else {}
 
 
-async def llm_json(prompt, model_name="", retries=1):
-    """调 LLM 并解析 JSON（剥离 markdown code fence）。失败返回 (None, err)。"""
+async def llm_json(prompt, model_name="", retries=1, project_id=""):
+    """调 LLM 并解析 JSON（剥离 markdown code fence）。失败返回 (None, err)。
+
+    project_id：透传到 llm_usage.project_id（按项目统计费用，2026-09-10）。
+    """
     from pipeline_service.llm_bridge import llm_call
     last_err = ""
     for _ in range(max(1, retries + 1)):
         try:
-            raw = await llm_call(prompt, model=model_name or None)
+            raw = await llm_call(prompt, model=model_name or None,
+                                 project_id=project_id or '')
             txt = (raw or "").strip()
             if txt.startswith("```"):
                 parts = txt.split("\n", 1)

pipeline_bidding/bid_flow.py

@@ -25,10 +25,15 @@ from .bid_common import (
     CH_PENDING, CH_WRITING, CH_WRITTEN, CH_REJECTED, CH_APPROVED,
     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,
+    HT_TEMPLATE_CONFIRM,
+    QC_TYPE_SCORING, QC_TYPE_QUALS, QC_TYPE_REQS, QC_TYPE_OUTLINE, QC_TYPE_COST,
+    QC_TYPE_TECH, QC_TYPES, TECH_QC_TYPES,
     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,
+    FLOW_KEY_TECH_PROPOSAL, STAGE_TECH_ANALYSIS, STAGE_TECH_QC,
+    STAGE_TEMPLATE_CONFIRM, STAGE_TECH_CH_WRITE, STAGE_TECH_CH_QC,
+    STAGE_TECH_COMPOSE, STAGE_TECH_DELIVERY,
 )
 
 logger = logging.getLogger("pipeline.bidding")
@@ -456,6 +461,69 @@ async def _latest_flow_plan_status(project_id):
         return "", ""
 
 
+async def _plan_flow_key(project_id):
+    """已确认计划的 base_flow_key（'' = 无计划/单流程标准链路）。"""
+    try:
+        from pipeline_service.flow_plan_capability import get_plan_flow_key
+        return await get_plan_flow_key(project_id)
+    except Exception:
+        return ""
+
+
+async def _auto_detect_flow(sor, project_id, pname):
+    """多流程产线：首个文件导入后 LLM 自动判流 → propose 草案 → 确认待办（阻塞等用户）。
+
+    触发条件：产线声明 ≥2 流程模板 且 项目从未提案（无 pipeline_flow_plans 记录）。
+    判流失败不默认硬跑（语义判断禁静默降级）：发人工介入待办说明原因。
+    单流程产线（模板 <2）本函数直接返回，行为与存量完全一致。
+    """
+    try:
+        from pipeline_service.flow_plan_capability import (
+            detect_flow_template, propose_flow_plan, get_latest_plan)
+    except Exception as e:
+        logger.warning("bid_flow: flow detect unavailable project=%s: %s",
+                       project_id, str(e)[:160])
+        return ""
+    proj = await get_project(sor, project_id)
+    pid = proj.get("pipeline_id") or PIPELINE_ID
+    if await get_latest_plan(project_id):
+        return ""       # 已提案过（任何状态），不重复判流
+    # 读最新文件正文供判流
+    recs = await sor.sqlExe(
+        "SELECT file_name, LEFT(content_text, 6000) AS head FROM bid_tender_files "
+        "WHERE project_id=${p}$ ORDER BY created_at DESC LIMIT 1", {"p": project_id})
+    await sor.sqlExe("COMMIT", {})
+    if not recs:
+        return ""
+    fname = getattr(recs[0], "file_name", "") or ""
+    fhead = getattr(recs[0], "head", "") or ""
+    fk, note = await detect_flow_template(
+        pid, file_name=fname, file_text=fhead,
+        project_id=project_id, org_id=proj.get("org_id") or "")
+    if not fk:
+        # 判流失败：发人工介入待办（不默认选流程硬跑）
+        ht = await find_human_task(sor, project_id, "general", status="pending")
+        if not ht and note:
+            await create_human_task(
+                sor, project_id, "general",
+                "流程自动识别失败，需人工指定流程：%s" % (pname or project_id),
+                ("系统无法自动判断该文件应走哪条流程：%s\n\n"
+                 "请在会话里用 propose_flow_plan(flow_key=...) 指定流程，"
+                 "或联系管理员核对文件内容。" % note),
+                assignee_role="owner.superuser")
+            await sor.sqlExe("COMMIT", {})
+        return ""
+    ok, msg = await propose_flow_plan(
+        project_id, trim_keys=[], user_requirements="（系统自动判流，见识别结论）",
+        propose_note="", pipeline_id=pid, flow_key=fk, detect_note=note,
+        who="system.reconciler", agent_id="bid_flow")
+    if ok:
+        logger.info("bid_flow: auto-detected flow %s for project=%s", fk, project_id)
+        return "created: 自动判流 %s → 流程确认待办（等用户确认）" % fk
+    logger.warning("bid_flow: auto propose failed project=%s: %s", project_id, str(msg)[:160])
+    return ""
+
+
 async def reconcile_project(sor, project_id, project_name=""):
     """对单个投标项目做一次状态对账。返回动作说明列表（幂等，可反复调用）。"""
     acts = []
@@ -522,6 +590,27 @@ async def reconcile_project(sor, project_id, project_name=""):
         return acts + ["blocked: 流程裁剪方案被用户驳回（意见：%s），等会话 agent 修订重提，本轮不派发"
                        % (_fp_comment[:200] or "未填")]
 
+    # ── 0d. 多流程分支（2026-09-14 批2）──
+    # 已确认计划带 base_flow_key=bid_tech_proposal → 走纯技术方案流程（独立对账分支，
+    # 以需求书为唯一基准，不碰评分项/资质/成本收益）。
+    # 多流程产线 + 从未提案 + 已有文件 → LLM 自动判流出草案（确认待办阻塞，等用户拍板）。
+    _fk = await _plan_flow_key(project_id)
+    if _fk == FLOW_KEY_TECH_PROPOSAL:
+        return acts + await _reconcile_tech_proposal(
+            sor, project_id, pname, stage_table or {})
+    if not _fk and not _fp_status:
+        _n_files_now = await _count(sor, "SELECT COUNT(*) AS c FROM bid_tender_files "
+                                         "WHERE project_id=${p}$", {"p": project_id})
+        if _n_files_now > 0:
+            _det = await _auto_detect_flow(sor, project_id, pname)
+            if _det:
+                return acts + [_det]
+            # 判流失败/待人工指定：人工待办已在 _auto_detect_flow 里发（general 非阻塞），
+            # 为避免标准流程抢跑，检测到该待办即本轮不派发；无待办 = 单流程产线（现状不变）。
+            _dj = await find_human_task(sor, project_id, "general", status="pending")
+            if _dj and "流程自动识别失败" in (_dj.get("title") or ""):
+                return acts + ["blocked: 流程自动识别失败，等人工指定流程（会话里 propose_flow_plan）"]
+
     def _stage_on(key):
         """阶段是否执行：无确认计划（None）= 标准全流程都执行。"""
         if stage_table is None:
@@ -1081,6 +1170,266 @@ async def reconcile_all(sor=None):
         return await _reconcile_all_with(s)
 
 
+# ══════════════ 纯技术方案流程分支（bid_tech_proposal，2026-09-14 批2） ══════════════
+# 阶段链：tech_analysis（六类评估项）→ tech_qc（对标需求书原文）→ template_confirm
+# （模版检索+用户确认骨架）→ tech_chapter_write → tech_chapter_qc（对标需求书覆盖度）
+# → tech_compose（合成技术方案书）→ tech_delivery_confirm。
+# 与标准流程的根本差异：**用户提供的技术需求书是唯一基准**——QC/评审全部对标需求书，
+# 不碰评分项/资质/成本收益（那是招标文件的产物，本流程的输入不是招标文件）。
+
+TECH_KIND = "tech"     # 分析维度值（任务 params.analysis_dim，与标准四维度并列）
+
+
+async def _tech_qc_latest(sor, project_id):
+    """tech_items 最新一轮 QC（无记录返回 {}）。"""
+    recs = await sor.sqlExe(
+        "SELECT id, round, fit_score, passed, improvement, task_id FROM bid_qc_reviews "
+        "WHERE project_id=${p}$ AND qc_type=${t}$ ORDER BY round DESC LIMIT 1",
+        {"p": project_id, "t": QC_TYPE_TECH})
+    await sor.sqlExe("COMMIT", {})
+    return rec_to_dict(recs[0]) if recs else {}
+
+
+async def _open_tech_tasks(sor, project_id, task_kind):
+    """某类技术方案任务在办数（按 params.task_kind 匹配）。"""
+    kw = '%"task_kind": "' + task_kind + '"%'
+    recs = await sor.sqlExe(
+        "SELECT COUNT(*) AS c FROM pipeline_tasks WHERE tenant_id=${p}$ "
+        "AND pipeline_id='role_task' AND state IN "
+        "('submitted','running','review','qc_review','waiting') AND params LIKE ${kw}$",
+        {"p": project_id, "kw": kw})
+    await sor.sqlExe("COMMIT", {})
+    return to_int(getattr(recs[0], "c", 0) if recs else 0)
+
+
+async def _reconcile_tech_proposal(sor, project_id, pname, stage_table):
+    """纯技术方案流程对账（幂等）。stage_table=已确认计划的阶段启用表。"""
+    acts = []
+
+    def _on(key):
+        return bool(stage_table.get(key, True))
+
+    th = await get_thresholds(sor)
+
+    n_items = await _count(sor, "SELECT COUNT(*) AS c FROM bid_tech_items "
+                                "WHERE project_id=${p}$ AND kind='item'", {"p": project_id})
+    chs = 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(chs, limit=500)
+    docs = await sor.sqlExe(
+        "SELECT id, version, status FROM bid_documents WHERE project_id=${p}$ "
+        "ORDER BY version DESC LIMIT 1", {"p": project_id})
+    await sor.sqlExe("COMMIT", {})
+    doc = rec_to_dict(docs[0]) if docs else {}
+    latest_qc = await _tech_qc_latest(sor, project_id)
+    qc_passed = (str(latest_qc.get("passed")) == "1") if latest_qc else False
+    # tech_qc 被裁 → T2 段的 _on(STAGE_TECH_QC) 判定即视为通过（用户已在确认待办里放弃该门禁），
+    # 不再单独算 qc_eff 死变量（2026-09-15 清理：原变量算了未被引用）。
+
+    # ── T1. 技术需求解析（六类评估项）──
+    if n_items == 0 and _on(STAGE_TECH_ANALYSIS):
+        if await _open_tech_tasks(sor, project_id, "bid_tech_analysis") > 0:
+            acts.append("waiting: 技术需求解析任务在办")
+        else:
+            qc_imp = (latest_qc.get("improvement") or "") if (latest_qc and str(latest_qc.get("passed")) != "1") else ""
+            attempts = await _count(
+                sor, "SELECT COUNT(*) AS c FROM pipeline_tasks WHERE tenant_id=${p}$ "
+                     "AND pipeline_id='role_task' AND role=${r}$ AND state IN "
+                     "('approved','completed','qc_rejected') AND params LIKE ${kw}$",
+                {"p": project_id, "r": R_ANALYST, "kw": '%bid_tech_analysis%'})
+            if attempts >= MAX_ANALYST_ATTEMPT + th["qc_max_round"]:
+                ht = await find_human_task(sor, project_id, "general", status="pending")
+                if not ht:
+                    await create_human_task(
+                        sor, project_id, "general",
+                        "技术需求解析多次未产出，需人工介入：%s" % pname,
+                        ("技术评估项解析任务已执行 %d 次，产出仍为空。\n"
+                         "请检查需求书是否可读（bid_tender_files.content_text），"
+                         "或用 list_tech_items/add_tech_item 人工补录评估项。" % attempts),
+                        assignee_role="owner.superuser")
+                    await sor.sqlExe("COMMIT", {})
+                acts.append("escalated: 技术需求解析重试超限 → 人工介入")
+            else:
+                tparams = {"stage": "tech_analysis", "task_kind": "bid_tech_analysis",
+                           "analysis_dim": TECH_KIND, "flow_key": FLOW_KEY_TECH_PROPOSAL}
+                if qc_imp:
+                    tparams["qc_redo"] = 1
+                    tparams["qc_improvements"] = ["[tech_items] %s" % qc_imp[:2000]]
+                tid = await create_role_task(
+                    sor, project_id, R_ANALYST,
+                    "%s 技术需求解析（六类评估项%s）" % (pname, "，QC退回重做" if qc_imp else ""),
+                    tparams)
+                await sor.sqlExe("COMMIT", {})
+                acts.append("created: 技术需求解析任务 %s" % tid)
+        return acts or ["idle: 技术需求解析中"]
+
+    # ── T2. 解析产出 QC（对标需求书原文，qc_type=tech_items）──
+    if _on(STAGE_TECH_QC) and n_items > 0 and not qc_passed:
+        rnd = to_int(latest_qc.get("round"), 0) if latest_qc else 0
+        if latest_qc and rnd >= th["qc_max_round"] and str(latest_qc.get("passed")) != "1":
+            # 轮次用尽：产出未更新则不空转，抛人工（同标准流程逃逸阀）
+            ht = await find_human_task(sor, project_id, HT_QC_ESCALATION, status="pending")
+            if not ht:
+                await create_human_task(
+                    sor, project_id, HT_QC_ESCALATION,
+                    "技术评估项 QC 连续不达标需人工介入",
+                    ("「tech_items」已审核 %d 轮仍未达标（最近改进意见：%s）。\n\n"
+                     "请人工核对技术需求书修正评估项（list_tech_items/add_tech_item），"
+                     "或在 QC 审核记录页把该记录 passed 改为 1 强制放行。"
+                     % (rnd, (latest_qc.get("improvement") or "")[:1500])),
+                    assignee_role="owner.superuser")
+                await sor.sqlExe("COMMIT", {})
+            acts.append("escalated: 技术评估项 QC 轮次用尽 → 人工介入")
+            return acts
+        if await _open_tech_tasks(sor, project_id, "bid_qc") > 0:
+            acts.append("waiting: 技术评估项 QC 任务在办")
+            return acts
+        # 竞态守卫：解析/重做任务在办 → 不派审核（防对半成品打分）
+        if await _open_tech_tasks(sor, project_id, "bid_tech_analysis") > 0:
+            acts.append("waiting: 解析任务在办，暂不派 tech QC")
+            return acts
+        tid = await create_role_task(
+            sor, project_id, R_QC,
+            "%s 技术评估项契合度审核（对标需求书原文）" % pname,
+            {"stage": "tech_qc", "task_kind": "bid_qc", "qc_types": [QC_TYPE_TECH],
+             "flow_key": FLOW_KEY_TECH_PROPOSAL})
+        await sor.sqlExe("COMMIT", {})
+        return acts + ["created: 技术评估项 QC 任务 %s" % tid]
+
+    # ── T3. 模版确定 + 用户确认骨架 ──
+    if not chapters:
+        if not _on(STAGE_TEMPLATE_CONFIRM):
+            # 被裁（trim=no 理论上裁不掉，防御）：无人确认不得开写
+            return acts + ["blocked: 模版确认阶段被裁但骨架缺失，无法开写章节"]
+        # 阻塞门禁：pending 的模版确认待办在 0b 已全局阻塞，走到这里只处理「未发/被驳回」
+        ht_pend = await find_human_task(sor, project_id, HT_TEMPLATE_CONFIRM, status="pending")
+        if ht_pend:
+            return acts + ["waiting: 技术方案骨架待用户确认"]
+        if await _open_tech_tasks(sor, project_id, "tech_template") > 0:
+            return acts + ["waiting: 模版/骨架拟定任务在办"]
+        from .bid_tech_capability import get_template_reject_comment
+        rej = await get_template_reject_comment(sor, project_id)
+        tid = await create_role_task(
+            sor, project_id, R_PM,
+            "%s 拟定技术方案模版骨架%s" % (pname, "（按驳回意见修订）" if rej else ""),
+            {"stage": "template_confirm", "task_kind": "tech_template",
+             "flow_key": FLOW_KEY_TECH_PROPOSAL,
+             **({"reject_comment": rej[:2000]} if rej else {})})
+        await sor.sqlExe("COMMIT", {})
+        return acts + ["created: 模版骨架任务 %s（PM 检索模版→拟骨架→发用户确认）" % tid]
+
+    # ── T4. 分章编写（骨架已确认；一章一任务并发派发）──
+    created_w = 0
+    for c in chapters:
+        if created_w >= th["write_concurrency"]:
+            break
+        if c.get("status") not in (CH_PENDING, CH_REJECTED):
+            continue
+        if to_int(c.get("revise_count"), 0) > th["max_revise"]:
+            continue      # 超重做上限，评审阶段已抛人工
+        if await count_open_tasks(sor, project_id, role=R_WRITER, chapter_id=c["id"]) > 0:
+            continue
+        stage = "revise" if c.get("status") == CH_REJECTED else "write"
+        tid = await create_role_task(
+            sor, project_id, R_WRITER,
+            "%s 技术方案 第%s章《%s》%s" % (pname, c.get("chapter_no"), c.get("title"),
+                                          "修改重写" if stage == "revise" else "编写"),
+            {"stage": stage, "task_kind": "bid_write", "chapter_id": c["id"],
+             "chapter_no": c.get("chapter_no"), "section": "technical",
+             "flow_key": FLOW_KEY_TECH_PROPOSAL,
+             "revise_count": c.get("revise_count")})
+        created_w += 1
+        acts.append("created: 技术方案章节编写任务 %s (%s)" % (tid, c.get("chapter_no")))
+    if created_w:
+        await sor.sqlExe("COMMIT", {})
+
+    # ── T5. 章节 QC 评审（对标需求书覆盖度）──
+    if not _on(STAGE_TECH_CH_QC):
+        _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: tech_chapter_qc 已裁剪，%d 个章节 written→approved 直通"
+                        % len(_auto_appr))
+            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 _on(STAGE_TECH_CH_QC):
+            break
+        if c.get("status") != CH_WRITTEN:
+            continue
+        if await count_open_tasks(sor, project_id, role=R_REVIEWER, chapter_id=c["id"]) > 0:
+            continue
+        tid = await create_role_task(
+            sor, project_id, R_REVIEWER,
+            "%s 技术方案 第%s章《%s》评审（对标需求书）" % (pname, c.get("chapter_no"), c.get("title")),
+            {"stage": "review", "task_kind": "bid_review", "chapter_id": c["id"],
+             "chapter_no": c.get("chapter_no"), "flow_key": FLOW_KEY_TECH_PROPOSAL})
+        created_r += 1
+        acts.append("created: 技术方案章节评审任务 %s (%s)" % (tid, c.get("chapter_no")))
+    if created_r:
+        await sor.sqlExe("COMMIT", {})
+    if created_w or created_r:
+        return acts
+
+    # ── T6. 合成技术方案书 ──
+    all_approved = all(c.get("status") == CH_APPROVED for c in chapters)
+    if all_approved:
+        if not doc or doc.get("status") == DOC_REJECTED:
+            if not _on(STAGE_TECH_COMPOSE):
+                return acts + ["blocked: 合成阶段被裁但流程要求产出技术方案书，无法完成"]
+            if await count_open_tasks(sor, project_id, role=R_COMPOSITOR) > 0:
+                return acts + ["waiting: 技术方案书合成任务在办"]
+            tid = await create_role_task(
+                sor, project_id, R_COMPOSITOR, "%s 合成技术方案书" % pname,
+                {"stage": "compose", "task_kind": "bid_compose",
+                 "flow_key": FLOW_KEY_TECH_PROPOSAL,
+                 "prev_version": doc.get("version") if doc else 0})
+            await sor.sqlExe("COMMIT", {})
+            return acts + ["created: 技术方案书合成任务 %s" % tid]
+
+        # ── T7. 交付确认 ──
+        if doc.get("status") in (DOC_DRAFT, DOC_REVIEWING, DOC_PASSED):
+            if not _on(STAGE_TECH_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: tech_delivery_confirm 已裁剪，技术方案书直接置 delivered，项目完成"]
+            ht = await find_human_task(sor, project_id, HT_DELIVERY_CONFIRM)
+            if not ht:
+                hid = await create_human_task(
+                    sor, project_id, HT_DELIVERY_CONFIRM,
+                    "技术方案书已完成，请确认交付",
+                    "技术方案书已合成并通过章节评审，请下载核对后确认交付。")
+                await sor.sqlExe("COMMIT", {})
+                return acts + ["created: 交付确认任务 %s" % hid]
+            if ht.get("status") == "done":
+                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 + ["completed: 技术方案书已交付，项目完成"]
+            return acts + ["waiting: 等待人工交付确认"]
+    return acts or ["idle: 技术方案流程无待办动作"]
+
+
 async def _reconcile_all_with(sor):
     recs = await sor.sqlExe(
         "SELECT id, name FROM sd_projects WHERE pipeline_id=${pl}$ "

pipeline_bidding/bid_qc_capability.py

@@ -15,7 +15,8 @@ import logging
 from .bid_common import (
     get_db, new_id, rec_to_dict, rows_to_dicts, to_float, to_int,
     record, create_human_task, find_human_task, get_thresholds, resolve_project_id,
-    QC_TYPE_SCORING, QC_TYPE_QUALS, QC_TYPE_REQS, QC_TYPE_OUTLINE, QC_TYPE_COST, QC_TYPES,
+    QC_TYPE_SCORING, QC_TYPE_QUALS, QC_TYPE_REQS, QC_TYPE_OUTLINE, QC_TYPE_COST,
+    QC_TYPE_TECH, QC_TYPES, TECH_QC_TYPES,
     QC_OUTPUT_TABLE,
     HT_QC_ESCALATION,
 )
@@ -60,15 +61,23 @@ QC_DIMENSIONS = {
         "3. 推理合理：每项 basis 含原文引用+推理过程，无凭空编造的确定性数字；\n"
         "4. 结论一致：summary 的成本合计/收益合计与单项一致，毛利口径正确；算不出须说明原因，不得硬编。"
     ),
+    QC_TYPE_TECH: (
+        "技术评估项契合度审核（纯技术方案流程，对照【技术需求书】原文——需求书是唯一基准）：\n"
+        "1. 完整性：六类评估项（功能清单/功能点/硬件配置/技术架构/功能架构/部署架构）需求书明确提出的条目全部录入，无漏项；\n"
+        "2. 原文一致：requirement 与需求书原文一致（参数/数量/指标不失真），source_ref 可回溯章节条款；\n"
+        "3. 硬性要求标注：需求书明确必须的条目 mandatory=1 无漏标；\n"
+        "4. 颗粒度：功能点细化到可评估颗粒度（不是把整个模块当一条）；硬件参数逐项拆条；\n"
+        "5. 无编造：需求书未提出的通用最佳实践/臆测需求不得录入；qty_note 与原文数字一致。"
+    ),
 }
 
 # 每类产出对应的落库表：统一从 bid_common.QC_OUTPUT_TABLE 导入（含 cost_benefit）
 
 
 async def _qc_type_valid(sor, qc_type):
-    if qc_type in QC_TYPES:
+    if qc_type in QC_TYPES or qc_type in TECH_QC_TYPES:
         return qc_type, ""
-    return "", "qc_type 非法（应为 %s）" % "/".join(QC_TYPES)
+    return "", "qc_type 非法（应为 %s）" % "/".join(QC_TYPES + TECH_QC_TYPES)
 
 
 async def list_qc(project_id, qc_type="", limit=50):
@@ -209,6 +218,13 @@ async def start_qc(qc_type, project_id="", who=None, agent_id=None, task_id=""):
                 "LEFT(basis, 500) AS basis, confidence, source_ref "
                 "FROM bid_cost_benefit WHERE project_id=${p}$ ORDER BY kind DESC, order_no LIMIT 300",
                 {"p": project_id})
+        elif qc_type == QC_TYPE_TECH:
+            recs = await sor.sqlExe(
+                "SELECT category, item_name, LEFT(requirement, 400) AS requirement, "
+                "qty_note, mandatory, confidence, source_ref "
+                "FROM bid_tech_items WHERE project_id=${p}$ AND kind='item' "
+                "ORDER BY category, order_no LIMIT 400",
+                {"p": project_id})
         else:
             recs = await sor.sqlExe(
                 "SELECT req_type, chapter_no, chapter_title, LEFT(requirement, 400) AS requirement, "
@@ -223,7 +239,9 @@ async def start_qc(qc_type, project_id="", who=None, agent_id=None, task_id=""):
             "pass_score": th["qc_pass_score"], "max_round": th["qc_max_round"],
             "checklist": QC_DIMENSIONS[qc_type],
             "outputs_table": tbl, "outputs": out_rows, "note": note,
-            "instruction": ("请读招标文件原文（list_tender_files → read_tender_file 分段）并逐条对照上述清单核对，"
+            "instruction": (("请读技术需求书原文（list_tender_files → read_tender_file 分段）并逐条对照上述清单核对，"
+                             if qc_type == QC_TYPE_TECH else
+                             "请读招标文件原文（list_tender_files → read_tender_file 分段）并逐条对照上述清单核对，") +
                             "得出 0~10 的契合度分（可带一位小数），然后调 finish_qc 落分。"
                             "不通过（得分未高于通过分）时 improvement 必须写清：漏了什么/错在哪/对照原文哪一处，"
                             "否则工具拒绝提交。"),

pipeline_bidding/bid_tech_capability.py

@@ -0,0 +1,520 @@
+# -*- coding: utf-8 -*-
+"""纯技术方案流程能力（bid_tech_proposal，2026-09-14 批2）。
+
+用户 2026-09-11 定义的产线第二流程：输入=技术需求书（非招标文件），产出=技术方案书。
+分析集中在技术层面六类评估项：功能清单/功能点/硬件配置/技术架构/功能架构/部署架构；
+方案书编写根据预定模版（无则 RAG/网络检索），**模版骨架由用户确认后**按章节分段编写，
+章节 QC 评审**以用户提供的需求书为唯一基准**（对标覆盖度，不碰评分项），最终合成技术方案书。
+
+机制分工（沿用产线既有约定：流转归机制，操作原语归本模块）：
+- bid_flow 对账器按已确认计划的 base_flow_key 走 tech 分支派发任务；
+- 本模块固化操作原语：技术评估项抽取/补录/查询、模版候选生成、模版确认落骨架；
+- QC 复用 bid_qc_reviews 机制（qc_type=tech_items，start_qc/finish_qc 原语不变）；
+- 用户确认走平台待办（tech_template_confirm 类型 + 专用确认端点，agent 无确认工具）。
+"""
+
+import json
+import logging
+
+from .bid_common import (
+    get_db, new_id, rec_to_dict, rows_to_dicts, to_int, to_float, json_loads,
+    record, create_human_task, find_human_task, get_thresholds, get_project,
+    resolve_project_id, llm_json,
+    CH_PENDING, TECH_CATEGORIES, HT_TEMPLATE_CONFIRM, QC_TYPE_TECH,
+)
+
+logger = logging.getLogger("pipeline.bidding.tech")
+
+TABLE = "bid_tech_items"
+
+# 六类评估项中文名（抽取提示词/展示用）
+TECH_CATEGORY_LABELS = {
+    "function_list": "功能清单",
+    "function_point": "功能点",
+    "hardware": "硬件配置",
+    "tech_arch": "技术架构",
+    "func_arch": "功能架构",
+    "deploy_arch": "部署架构",
+}
+
+PROMPT_TECH_ITEMS = """你是技术方案分析师。下面是一份【技术需求书】正文（可能被截断），请抽取六类技术评估项。
+本任务只做技术层面分析，不涉及商务/评分/资质。
+
+【六类评估项】
+- function_list 功能清单：需求书要求的功能模块/子系统清单（一个模块一条）
+- function_point 功能点：功能模块下的具体功能点（细化到可评估颗粒度）
+- hardware 硬件配置：服务器/GPU/存储/网络等硬件要求（型号/数量/参数，需求书没写具体型号的如实标"未指定"）
+- tech_arch 技术架构：需求书对技术栈/框架/协议/性能指标/安全合规的要求
+- func_arch 功能架构：系统分层/模块划分/集成关系要求
+- deploy_arch 部署架构：部署模式（私有化/云/混合）/环境/容灾/扩缩容要求
+
+【技术需求书正文】
+__CONTENT__
+
+严格只输出一个 JSON 对象：
+{
+  "items": [
+    {"category": "function_list|function_point|hardware|tech_arch|func_arch|deploy_arch",
+     "item_name": "条目名称（简短可索引）",
+     "requirement": "需求书原文要求（引用原文，可截断到要点）",
+     "qty_note": "数量/规模说明（功能点数/台数/容量等，没有就空）",
+     "mandatory": "1|0（需求书明确必须的为1）",
+     "source_ref": "出处（章节/条款号）"}
+  ]
+}
+要求：
+1. 只录需求书**明确提出或可直接推出**的条目，禁止编造、禁止把通用最佳实践当需求；
+2. requirement 必须能回溯原文（source_ref 给章节号），抽取不到原文出处的标 source_ref="未明示"；
+3. 同一功能模块的多个功能点分多条录（function_point），不要合并成大条；
+4. 硬件参数逐项拆条（如"GPU服务器""存储阵列"各一条），参数写进 requirement。"""
+
+
+async def _load_req_text(sor, project_id, file_id=""):
+    """读技术需求书正文（复用招标文件读取——需求书也登记在 bid_tender_files）。"""
+    from .bid_analysis_capability import _load_tender_text
+    return await _load_tender_text(sor, project_id, file_id=file_id)
+
+
+# ══════════════ 技术评估项：抽取 / 补录 / 查询 ══════════════
+
+async def extract_tech_items(project_id, file_id="", model_name="", offset=0,
+                             length=30000, who=None, agent_id=None):
+    """分析维度 tech：LLM 抽取六类技术评估项落库 bid_tech_items（大文件分段调）。"""
+    try:
+        project_id = await resolve_project_id(project_id)
+    except ValueError as e:
+        return False, str(e)
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        txt, fname, err = await _load_req_text(sor, project_id, file_id=file_id)
+        if err:
+            return False, err
+    o = to_int(offset, 0)
+    seg = txt[o:o + to_int(length, 30000)]
+    if not seg.strip():
+        return False, "正文分段为空（offset=%s 超出文件长度），无需再抽" % o
+    data, err = await llm_json(PROMPT_TECH_ITEMS.replace("__CONTENT__", seg),
+                               model_name=model_name, retries=1, project_id=project_id)
+    if data is None:
+        return False, "LLM 抽取失败: %s" % err
+    items = data.get("items") or []
+    if not isinstance(items, list):
+        return False, "LLM 返回格式错误（items 非数组）"
+    cnt, skipped = 0, 0
+    async with db.sqlorContext(dbname) as sor:
+        # 已有条目（同段重复调用去重：同名同类不重复插）
+        recs = await sor.sqlExe(
+            "SELECT category, item_name FROM " + TABLE + " WHERE project_id=${p}$",
+            {"p": project_id})
+        await sor.sqlExe("COMMIT", {})
+        exist = {(getattr(r, "category", ""), getattr(r, "item_name", ""))
+                 for r in (recs or [])}
+        order_base = await _max_order(sor, project_id)
+        for it in items:
+            if not isinstance(it, dict):
+                continue
+            cat = str(it.get("category") or "").strip()
+            name = str(it.get("item_name") or "").strip()[:240]
+            if cat not in TECH_CATEGORIES or not name:
+                skipped += 1
+                continue
+            if (cat, name) in exist:
+                skipped += 1
+                continue
+            order_base += 1
+            await sor.C(TABLE, {
+                "id": new_id(), "project_id": project_id, "kind": "item",
+                "category": cat, "item_name": name,
+                "requirement": str(it.get("requirement") or "")[:20000],
+                "qty_note": str(it.get("qty_note") or "")[:200],
+                "mandatory": "1" if str(it.get("mandatory") or "") in ("1", "true", "True") else "0",
+                "confidence": "medium",
+                "source_ref": str(it.get("source_ref") or "")[:480],
+                "order_no": order_base,
+            })
+            exist.add((cat, name))
+            cnt += 1
+        await sor.sqlExe("COMMIT", {})
+        await record(sor, project_id, TABLE, project_id, "extract_tech_items",
+                     to_state="extracted", who=who, agent_id=agent_id,
+                     detail="新增%d条(去重跳过%d) file=%s offset=%s" % (cnt, skipped, fname, o))
+        await sor.sqlExe("COMMIT", {})
+    return True, ("技术评估项抽取完成：新增 %d 条（重复/非法跳过 %d 条）。"
+                  "六类分布用 list_tech_items 核对，漏项用 add_tech_item 补录。" % (cnt, skipped))
+
+
+async def _max_order(sor, project_id):
+    recs = await sor.sqlExe(
+        "SELECT COALESCE(MAX(order_no),0) AS m FROM " + TABLE + " WHERE project_id=${p}$",
+        {"p": project_id})
+    await sor.sqlExe("COMMIT", {})
+    return to_int(getattr(recs[0], "m", 0) if recs else 0)
+
+
+async def add_tech_item(project_id, category, item_name, requirement="",
+                        qty_note="", mandatory="0", source_ref="",
+                        who=None, agent_id=None):
+    """补录一条技术评估项（抽取漏项或人工修正）。"""
+    try:
+        project_id = await resolve_project_id(project_id)
+    except ValueError as e:
+        return False, str(e)
+    cat = str(category or "").strip()
+    name = str(item_name or "").strip()[:240]
+    if cat not in TECH_CATEGORIES:
+        return False, "category 非法（应为 %s）" % "/".join(TECH_CATEGORIES)
+    if not name:
+        return False, "缺少 item_name"
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        order_no = await _max_order(sor, project_id) + 1
+        tid = new_id()
+        await sor.C(TABLE, {
+            "id": tid, "project_id": project_id, "kind": "item",
+            "category": cat, "item_name": name,
+            "requirement": str(requirement or "")[:20000],
+            "qty_note": str(qty_note or "")[:200],
+            "mandatory": "1" if str(mandatory) in ("1", "true", "True") else "0",
+            "confidence": "high",
+            "source_ref": str(source_ref or "")[:480],
+            "order_no": order_no,
+        })
+        await sor.sqlExe("COMMIT", {})
+        await record(sor, project_id, TABLE, tid, "add_tech_item",
+                     who=who, agent_id=agent_id, detail="%s/%s" % (cat, name))
+        await sor.sqlExe("COMMIT", {})
+    return True, "已补录技术评估项：%s（%s）" % (name, TECH_CATEGORY_LABELS.get(cat, cat))
+
+
+async def list_tech_items(project_id, category="", kind="item", limit=300):
+    """列出技术评估项（默认只列 item 行；含六类分布统计头部）。"""
+    try:
+        project_id = await resolve_project_id(project_id)
+    except ValueError as e:
+        return "ERROR: %s" % e
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        sql = ("SELECT id, kind, category, item_name, LEFT(requirement, 300) AS requirement, "
+               "qty_note, mandatory, confidence, source_ref, order_no "
+               "FROM " + TABLE + " WHERE project_id=${p}$")
+        p = {"p": project_id}
+        if category:
+            sql += " AND category=${c}$"
+            p["c"] = category
+        if kind:
+            sql += " AND kind=${k}$"
+            p["k"] = kind
+        sql += " ORDER BY category, order_no LIMIT " + str(to_int(limit, 300))
+        recs = await sor.sqlExe(sql, p)
+        await sor.sqlExe("COMMIT", {})
+        rows = rows_to_dicts(recs, limit=to_int(limit, 300))
+        stat = await sor.sqlExe(
+            "SELECT category, COUNT(*) AS c FROM " + TABLE +
+            " WHERE project_id=${p}$ AND kind='item' GROUP BY category", {"p": project_id})
+        await sor.sqlExe("COMMIT", {})
+    dist = {getattr(r, "category", ""): to_int(getattr(r, "c", 0)) for r in (stat or [])}
+    head = "六类分布：" + "、".join(
+        "%s=%d" % (TECH_CATEGORY_LABELS.get(k, k), v) for k, v in
+        sorted(dist.items(), key=lambda kv: TECH_CATEGORIES.index(kv[0])
+               if kv[0] in TECH_CATEGORIES else 99))
+    missing = [TECH_CATEGORY_LABELS[c] for c in TECH_CATEGORIES if not dist.get(c)]
+    if missing and kind == "item":
+        head += "（缺类：%s——需求书确实没有则如实说明，有漏项用 add_tech_item 补录）" % "、".join(missing)
+    if not rows:
+        return head + "\n（无记录）"
+    return head + "\n" + "\n".join(
+        json.dumps(r, ensure_ascii=False, default=str) for r in rows)
+
+
+# ══════════════ 模版确定 + 用户确认骨架（template_confirm 阶段） ══════════════
+
+async def find_template_candidates(project_id, query="", model_name="",
+                                   who=None, agent_id=None):
+    """检索技术方案模版候选：知识库(doc_type=tech_template/sample_bid) → RAG → 网络。
+
+    返回候选清单文本（PM/analyst 角色工具）。三级降级链每级如实标注来源，
+    全部落空时如实说明——由角色 agent 按行业惯例自拟骨架（也要经用户确认）。
+    """
+    try:
+        project_id = await resolve_project_id(project_id)
+    except ValueError as e:
+        return "ERROR: %s" % e
+    q = (query or "技术方案书 目录 模版").strip()
+    lines = ["## 模版候选检索（%s）" % q, ""]
+    # 1. 公司知识库
+    try:
+        from .bid_kb_capability import search_bid_kb
+        kb = await search_bid_kb(query=q, doc_type="tech_template")
+        if kb and "无命中" not in str(kb):
+            lines += ["### ① 公司知识库（doc_type=tech_template，预定模版优先）", str(kb)[:3000], ""]
+        else:
+            kb2 = await search_bid_kb(query=q, doc_type="sample_bid")
+            if kb2 and "无命中" not in str(kb2):
+                lines += ["### ① 公司知识库（样板标书参照）", str(kb2)[:3000], ""]
+            else:
+                lines += ["### ① 公司知识库：无预定模版", ""]
+    except Exception as e:
+        lines += ["### ① 公司知识库：查询失败（%s）" % str(e)[:100], ""]
+    # 2. RAG 检索
+    try:
+        from pipeline_service.rag_client import rag_search
+        data, err = await rag_search(project_id, q, top_k=5)
+        if err:
+            lines += ["### ② RAG 检索：失败（%s）" % err[:150], ""]
+        else:
+            chunks = (data or {}).get("results") or (data or {}).get("chunks") or []
+            if chunks:
+                brief = "\n".join("- [%s] %s" % (
+                    (c.get("doc_name") or c.get("kb_name") or "?"),
+                    str(c.get("content") or c.get("text") or "")[:200])
+                    for c in chunks[:5] if isinstance(c, dict))
+                lines += ["### ② RAG 知识库检索", brief, ""]
+            else:
+                lines += ["### ② RAG 检索：无命中", ""]
+    except Exception as e:
+        lines += ["### ② RAG 检索：不可用（%s）" % str(e)[:100], ""]
+    # 3. 网络检索
+    try:
+        from pipeline_service.web_tools import tool_web_search
+        web = await tool_web_search(q + " 目录结构", limit=5)
+        lines += ["### ③ 网络检索", str(web)[:2500], ""]
+    except Exception as e:
+        lines += ["### ③ 网络检索：不可用（%s）" % str(e)[:100], ""]
+    lines += ["---", "下一步：结合六类评估项（list_tech_items）与候选模版拟定章节骨架，"
+              "调 propose_tech_template 发用户确认待办（骨架确认前不得开写章节）。"]
+    return "\n".join(lines)
+
+
+PROMPT_TECH_OUTLINE = """你是技术方案架构师。根据【技术需求书评估项】和【模版参考】拟定技术方案书的章节骨架。
+
+【六类技术评估项（需求书抽取结果）】
+__ITEMS__
+
+【模版参考（可能为空——为空则按行业惯例自拟）】
+__TEMPLATE__
+
+严格只输出一个 JSON 对象：
+{"chapters": [
+  {"chapter_no": "1", "title": "章节标题", "outline": "本章要写什么要点（覆盖哪些评估项，写到操作层）",
+   "cover_categories": ["function_list","tech_arch"]}
+]}
+要求：
+1. 章节结构完整覆盖六类评估项：每个 category 至少被一章的 cover_categories 覆盖，漏覆盖 = 骨架不合格；
+2. 有模版参考时优先沿用它章节顺序与命名惯例，按本项目评估项裁剪/补充（模版是蓝本不是成品）；
+3. 每章 outline 写清要响应的需求书条目（引用 item_name），不写空话；
+4. 章节数控制在 6~15 章，粒度以"一章一任务可写完"为准；
+5. chapter_no 用数字字符串（"1","2",...），子节写进 outline 不单独成章。"""
+
+
+async def propose_tech_template(project_id, chapters="", template_source="",
+                                template_ref="", note="", model_name="",
+                                who=None, agent_id=None):
+    """生成技术方案章节骨架并发「模版确认」人类待办（用户确认后才落 bid_chapters 开写）。
+
+    chapters：JSON 数组 [{chapter_no,title,outline,cover_categories?}]；
+      为空时自动按六类评估项 + 模版候选 LLM 生成。
+    template_source：kb/rag/web/manual（如实标注骨架依据来源）。
+    确认动作是用户专属（专用端点 tech_template_confirm.dspy），agent 无确认工具。
+    """
+    try:
+        project_id = await resolve_project_id(project_id)
+    except ValueError as e:
+        return False, str(e)
+    items = json_loads(chapters, []) if isinstance(chapters, str) else (chapters or [])
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        proj = await get_project(sor, project_id)
+        if not proj:
+            return False, "项目不存在: %s" % project_id
+        if not items:
+            # 自动生成：六类评估项 + 模版候选 → LLM 拟骨架
+            irecs = await sor.sqlExe(
+                "SELECT category, item_name, LEFT(requirement,200) AS requirement, mandatory "
+                "FROM " + TABLE + " WHERE project_id=${p}$ AND kind='item' "
+                "ORDER BY category, order_no LIMIT 400", {"p": project_id})
+            await sor.sqlExe("COMMIT", {})
+            irows = rows_to_dicts(irecs, limit=400)
+            if not irows:
+                return False, ("无技术评估项——先完成 tech_analysis 维度抽取"
+                               "（extract_tech_items），骨架必须覆盖评估项，不能凭空拟")
+            items_txt = "\n".join("- [%s]%s %s（%s）" % (
+                TECH_CATEGORY_LABELS.get(r.get("category"), r.get("category")),
+                "★" if str(r.get("mandatory")) == "1" else "",
+                r.get("item_name"), str(r.get("requirement") or "")[:120]) for r in irows)
+            tpl_txt = str(template_ref or "")[:6000] or "（无模版参考，按行业惯例自拟）"
+            data, err = await llm_json(
+                PROMPT_TECH_OUTLINE.replace("__ITEMS__", items_txt).replace("__TEMPLATE__", tpl_txt),
+                model_name=model_name, retries=1, project_id=project_id)
+            if data is None:
+                return False, "LLM 生成骨架失败: %s" % err
+            items = data.get("chapters") or []
+        if not isinstance(items, list) or not items:
+            return False, "章节骨架为空（chapters 解析失败），请传 JSON 数组或修正后重试"
+        # 规范化 + 覆盖度校验（机制硬校验：六类评估项必须全覆盖，缺类拒绝——
+        # 骨架漏类 = 方案书必漏需求 = QC 对标必退，不如在确认门之前就挡住）
+        norm, covered = [], set()
+        for idx, ch in enumerate(items):
+            if not isinstance(ch, dict):
+                continue
+            no = str(ch.get("chapter_no") or (idx + 1)).strip()[:30]
+            title = str(ch.get("title") or "").strip()[:250]
+            if not title:
+                continue
+            cats = ch.get("cover_categories") or []
+            if isinstance(cats, str):
+                cats = [c.strip() for c in cats.replace("，", ",").split(",") if c.strip()]
+            cats = [c for c in cats if c in TECH_CATEGORIES]
+            covered.update(cats)
+            norm.append({"chapter_no": no, "title": title,
+                         "outline": str(ch.get("outline") or "")[:4000],
+                         "cover_categories": cats})
+        if not norm:
+            return False, "章节骨架全部非法（缺 title）"
+        # 项目实际存在的评估类别（需求书没提硬件就不强制覆盖硬件）
+        crecs = await sor.sqlExe(
+            "SELECT DISTINCT category FROM " + TABLE + " WHERE project_id=${p}$ AND kind='item'",
+            {"p": project_id})
+        await sor.sqlExe("COMMIT", {})
+        need = {getattr(r, "category", "") for r in (crecs or [])} & set(TECH_CATEGORIES)
+        miss = sorted(need - covered)
+        if miss:
+            return False, ("骨架未覆盖以下评估类别：%s。每类至少一章 cover_categories 覆盖，"
+                           "请补章节或修正 cover_categories 后重提。"
+                           % "、".join(TECH_CATEGORY_LABELS.get(m, m) for m in miss))
+        # 骨架预览 markdown（进待办正文）
+        md = ["## 技术方案书章节骨架（待确认）", ""]
+        if template_source:
+            md.append("**模版来源：** %s%s" % (
+                {"kb": "公司知识库预定模版", "rag": "RAG 知识库检索",
+                 "web": "网络检索", "manual": "按行业惯例自拟"}.get(template_source, template_source),
+                "（%s）" % template_ref[:200] if template_ref else ""))
+        md += ["", "| # | 章节 | 要点 | 覆盖评估类 |", "|---|---|---|---|"]
+        for ch in norm:
+            md.append("| %s | %s | %s | %s |" % (
+                ch["chapter_no"], ch["title"],
+                (ch["outline"] or "").replace("\n", " ")[:120],
+                "、".join(TECH_CATEGORY_LABELS.get(c, c) for c in ch["cover_categories"]) or "-"))
+        md += ["", "---", "",
+               "确认后骨架落库并开始分章编写（每章对标需求书原文，QC 按需求覆盖度评审）；",
+               "驳回请附意见（换模版/调章节/补覆盖），助手按意见修订后重新提案。"]
+        if note:
+            md += ["", "**助手说明：**", note[:2000]]
+        hid = await create_human_task(
+            sor, project_id, HT_TEMPLATE_CONFIRM,
+            "技术方案书章节骨架确认：%s" % (proj.get("name") or project_id),
+            "\n".join(md),
+            assignee_role="owner.superuser",
+            form_schema={"chapters": norm, "template_source": template_source or "",
+                         "template_ref": (template_ref or "")[:2000]})
+        await sor.sqlExe("COMMIT", {})
+        await record(sor, project_id, "pipeline_human_tasks", hid, "propose_tech_template",
+                     to_state="pending_confirm", who=who, agent_id=agent_id,
+                     detail="%d章 source=%s" % (len(norm), template_source or "-"))
+        await sor.sqlExe("COMMIT", {})
+    return True, ("已生成技术方案章节骨架（%d 章）并发「模版确认」待办（%s）。"
+                  "用户确认前流程阻塞，你不能代替确认；确认后系统自动落骨架开写。"
+                  % (len(norm), hid))
+
+
+async def confirm_tech_template(human_task_id, decision, comment="",
+                                operator_id=None):
+    """用户确认/驳回章节骨架（专用端点调用，agent 无此工具）。
+
+    confirm：骨架落 bid_chapters(section=technical, status=pending) + 关待办 → 对账器开写；
+    reject：关待办（status=rejected，意见存 result_data）→ 对账器重派模版任务带意见。
+    """
+    if not human_task_id:
+        return False, "缺少 human_task_id"
+    if not operator_id:
+        return False, "未登录"
+    decision = str(decision or "").strip().lower()
+    if decision not in ("confirm", "reject"):
+        return False, "decision 必须是 confirm 或 reject"
+    if decision == "reject" and not (comment or "").strip():
+        return False, "驳回必须填写意见（助手按意见修订骨架）"
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.sqlExe(
+            "SELECT * FROM pipeline_human_tasks WHERE id=${h}$", {"h": human_task_id})
+        await sor.sqlExe("COMMIT", {})
+        if not recs:
+            return False, "待办不存在: %s" % human_task_id
+        ht = rec_to_dict(recs[0])
+        if ht.get("task_type") != HT_TEMPLATE_CONFIRM:
+            return False, "该待办不是模版确认类型（%s）" % ht.get("task_type")
+        if ht.get("status") != "pending":
+            return False, "待办已处理（当前 %s）" % ht.get("status")
+        project_id = ht.get("project_id") or ""
+        # 同机构校验
+        prec = await sor.sqlExe("SELECT org_id FROM sd_projects WHERE id=${p}$",
+                                {"p": project_id})
+        await sor.sqlExe("COMMIT", {})
+        proj_org = getattr(prec[0], "org_id", "") if prec else ""
+        from pipeline_service.human_task_capability import _get_user_org
+        user_org = await _get_user_org(sor, operator_id)
+        if proj_org and user_org != proj_org:
+            return False, "仅同机构用户可确认该项目模版"
+
+        if decision == "confirm":
+            schema = json_loads(ht.get("form_schema"), {}) or {}
+            if not isinstance(schema, dict):
+                schema = {}
+            chapters = schema.get("chapters") or []
+            if not chapters:
+                return False, "待办里没有骨架数据（form_schema.chapters 空），无法落库"
+            # 幂等：已有骨架先清（驳回重提后再确认的场景）
+            await sor.sqlExe(
+                "DELETE FROM bid_chapters WHERE project_id=${p}$", {"p": project_id})
+            n = 0
+            for idx, ch in enumerate(chapters):
+                if not isinstance(ch, dict):
+                    continue
+                await sor.C("bid_chapters", {
+                    "id": new_id(), "project_id": project_id,
+                    "chapter_no": str(ch.get("chapter_no") or (idx + 1))[:32],
+                    "title": str(ch.get("title") or "")[:255],
+                    "section": "technical",
+                    "outline": str(ch.get("outline") or "")[:60000],
+                    "status": CH_PENDING, "version": 1, "revise_count": 0,
+                    "order_no": idx + 1,
+                })
+                n += 1
+            await sor.sqlExe(
+                "UPDATE pipeline_human_tasks SET status='done', qc_status='passed', "
+                "qc_comment='用户已确认技术方案骨架', result_data=${rd}$, "
+                "submitted_by=${u}$, submitted_at=NOW() WHERE id=${h}$",
+                {"rd": json.dumps({"decision": "confirm", "chapters": n}, ensure_ascii=False),
+                 "u": operator_id, "h": human_task_id})
+            await sor.sqlExe("COMMIT", {})
+            await record(sor, project_id, "bid_chapters", project_id, "template_confirmed",
+                         to_state="pending", who=operator_id,
+                         detail="骨架落库 %d 章" % n)
+            await sor.sqlExe("COMMIT", {})
+            logger.info("tech_template confirmed: project=%s chapters=%d by=%s",
+                        project_id, n, operator_id)
+            return True, "骨架已确认并落库（%d 章），系统将自动开始分章编写。" % n
+
+        # reject
+        await sor.sqlExe(
+            "UPDATE pipeline_human_tasks SET status='rejected', result_data=${rd}$, "
+            "submitted_by=${u}$, submitted_at=NOW() WHERE id=${h}$",
+            {"rd": json.dumps({"decision": "reject", "comment": (comment or "")[:2000]},
+                              ensure_ascii=False),
+             "u": operator_id, "h": human_task_id})
+        await sor.sqlExe("COMMIT", {})
+        await record(sor, project_id, "pipeline_human_tasks", human_task_id,
+                     "template_rejected", to_state="rejected", who=operator_id,
+                     detail=(comment or "")[:500])
+        await sor.sqlExe("COMMIT", {})
+        return True, "已驳回，助手将按你的意见修订骨架后重新提案。"
+
+
+async def get_template_reject_comment(sor, project_id):
+    """最新一次模版驳回意见（对账器重派模版任务时注入 params，无则 ''）。"""
+    ht = await find_human_task(sor, project_id, HT_TEMPLATE_CONFIRM, status="rejected")
+    if not ht:
+        return ""
+    rd = json_loads(ht.get("result_data"), {}) or {}
+    if not isinstance(rd, dict):
+        return ""
+    return str(rd.get("comment") or "")

pipeline_bidding/bid_tender_capability.py

@@ -9,6 +9,7 @@
 
 import json
 import logging
+import os
 
 from .bid_common import (
     get_db, new_id, rec_to_dict, rows_to_dicts, json_loads, to_float, to_int,
@@ -146,6 +147,22 @@ async def import_tender_file(project_id, content_text="", file_path="", file_nam
             p = await get_project(sor, pid)
             if not p:
                 return False, "项目不存在: %s" % pid
+        # 2026-09-15 修复（bug A 根治半边）：相对路径在落库前归一为磁盘绝对路径。
+        # 上传链路可能给出相对工作空间根的路径（如 "0/xxx.docx"），而读取侧
+        # _resolve_file_path 按项目目录解析——口径不一致导致文件在盘上却永远读不到正文
+        # （首医项目实测）。落库时磁盘绝对路径是已知的，存相对路径本身就是隐患。
+        if fp and not os.path.isabs(fp):
+            real = ""
+            try:
+                from .bid_analysis_capability import _resolve_file_path, _project_dir_of, _workspace_base_of
+                pdir = await _project_dir_of(sor, pid)
+                real = _resolve_file_path(pdir, fp, ws_base=await _workspace_base_of(sor))
+            except Exception as e:
+                logger.warning("tender path normalize failed fp=%s: %s", fp, str(e)[:120])
+            if real:
+                fp = real
+            else:
+                logger.warning("tender path not resolvable on disk, stored as-is: %s", fp)
         await sor.C("bid_tender_files", {
             "id": new_id(), "project_id": pid,
             "file_name": str(file_name or ("招标文件（对话提供）" if txt else fp.split("/")[-1]))[:250],

pipeline_bidding/role_tool_schemas.py

@@ -349,6 +349,48 @@ BID_ROLE_TOOL_SCHEMAS = {
                    "project_id": "项目ID(可选,默认当前)"},
         "required": ["dim"],
     },
+    # ── bid_tech_capability：纯技术方案流程（bid_tech_proposal，2026-09-14 批2）──
+    # 输入=技术需求书（非招标文件），六类评估项：功能清单/功能点/硬件配置/技术架构/功能架构/部署架构
+    "extract_tech_items": {
+        "module": f"{_M}.bid_tech_capability",
+        "description": "技术需求解析（analysis_dim=tech 时用）：LLM 从技术需求书抽取六类技术评估项落库（大文件分段调 offset 递增），同名同类自动去重。需求书是本项目唯一基准，禁止编造需求",
+        "params": {"project_id": "项目ID(可选,默认当前)", "file_id": "需求书文件ID(可选)",
+                   "offset": "分段起始字符(可选)", "length": "分段长度(可选)"},
+        "required": [],
+    },
+    "add_tech_item": {
+        "module": f"{_M}.bid_tech_capability",
+        "description": "补录一条技术评估项（抽取漏项或人工修正；category 必须是六类之一）",
+        "params": {"project_id": "项目ID(可选,默认当前)",
+                   "category": "function_list|function_point|hardware|tech_arch|func_arch|deploy_arch",
+                   "item_name": "条目名称", "requirement": "需求书原文要求(可选)",
+                   "qty_note": "数量/规模说明(可选)", "mandatory": "是否硬性要求 1/0(可选)",
+                   "source_ref": "需求书出处(可选)"},
+        "required": ["category", "item_name"],
+    },
+    "list_tech_items": {
+        "module": f"{_M}.bid_tech_capability",
+        "description": "列出技术评估项（头部含六类分布统计与缺类提示；category 可按类过滤）",
+        "params": {"project_id": "项目ID(可选,默认当前)", "category": "按类过滤(可选)",
+                   "limit": "条数上限(可选,默认300)"},
+        "required": [],
+    },
+    "find_template_candidates": {
+        "module": f"{_M}.bid_tech_capability",
+        "description": "检索技术方案模版候选（三级降级：公司知识库预定模版→RAG→网络），每级如实标注来源。拟骨架前先调它",
+        "params": {"project_id": "项目ID(可选,默认当前)", "query": "检索词(可选,默认按技术方案书目录模版)"},
+        "required": [],
+    },
+    "propose_tech_template": {
+        "module": f"{_M}.bid_tech_capability",
+        "description": "拟定技术方案书章节骨架并发「模版确认」待办（用户确认前流程阻塞，你不能代替确认；确认后系统自动落骨架开写）。机制硬校验：骨架必须覆盖全部已抽取评估类别，漏类拒绝。chapters 为空时自动按评估项+模版候选生成",
+        "params": {"project_id": "项目ID(可选,默认当前)",
+                   "chapters": "章节JSON数组[{chapter_no,title,outline,cover_categories}](可选,空=自动生成)",
+                   "template_source": "骨架依据来源 kb|rag|web|manual(如实标注)",
+                   "template_ref": "模版参考内容/名称(可选)",
+                   "note": "给用户的说明(可选)"},
+        "required": [],
+    },
 }
 
 

scripts/import_init_bidding.py

@@ -45,6 +45,8 @@ def main():
         ("qualifications", "所需资质"),
         ("doc_requirements", "投标文件要求"),
         ("chapter_outline", "章节骨架"),
+        ("cost_benefit", "成本收益分析"),
+        ("tech_items", "技术评估项（纯技术方案流程，对标技术需求书）"),
     ]
     mysql_exec(kw, pwd,
                "INSERT IGNORE INTO appcodes (id, name, hierarchy_flg) "

scripts/load_path.py

@@ -43,12 +43,13 @@ for t in TABLES:
         "/%s/%s/update_%s.dspy" % (MOD, t, t),
         "/%s/%s/delete_%s.dspy" % (MOD, t, t),
     ]
-# 产线业务 API
+# 产线业务 API（⚠️ RBAC 精确 path 匹配：/api/ 前缀不覆盖具体 dspy，逐个显式注册）
 PATHS_LOGINED += [
     "/%s/api/" % MOD,
     "/%s/api/bid_probe.dspy" % MOD,
     "/%s/api/bid_task_tree.dspy" % MOD,
     "/%s/api/bid_task_io.dspy" % MOD,
+    "/%s/api/tech_template_confirm.dspy" % MOD,
 ]
 
 

wwwroot/api/tech_template_confirm.dspy

@@ -0,0 +1,40 @@
+# tech_template_confirm.dspy - 技术方案书章节骨架的用户确认/驳回端点
+# （纯技术方案流程 bid_tech_proposal，2026-09-15 批2收口）
+#
+# 入参（form-encoded）：
+#   human_task_id  模版确认待办 id（pipeline_human_tasks.id，task_type=tech_template_confirm）
+#   decision       confirm | reject
+#   comment        驳回意见（reject 时必填；confirm 可空）
+#
+# 安全：确认动作是用户专属（同机构校验在 bid_tech_capability.confirm_tech_template）；
+# agent 无对应工具，无法代替确认。确认后骨架落 bid_chapters(section=technical) 开写；
+# 驳回意见存 result_data，对账器重派模版任务时注入（get_template_reject_comment）。
+
+import json as _json
+
+user_id = await get_user()
+if not user_id:
+    return _json.dumps({"success": False, "error": "未登录"}, ensure_ascii=False)
+
+human_task_id = str((params_kw or {}).get('human_task_id') or '').strip()
+decision = str((params_kw or {}).get('decision') or '').strip().lower()
+comment = str((params_kw or {}).get('comment') or '').strip()
+
+if not human_task_id:
+    return _json.dumps({"success": False, "error": "缺少 human_task_id"}, ensure_ascii=False)
+if decision not in ('confirm', 'reject'):
+    return _json.dumps({"success": False, "error": "decision 必须是 confirm 或 reject"},
+                       ensure_ascii=False)
+
+try:
+    from pipeline_bidding.bid_tech_capability import confirm_tech_template
+except Exception as e:
+    return _json.dumps({"success": False,
+                        "error": "投标产线技术能力模块不可用：%s" % str(e)[:150]},
+                       ensure_ascii=False)
+
+ok, msg = await confirm_tech_template(human_task_id, decision, comment,
+                                      operator_id=user_id)
+if ok:
+    return _json.dumps({"success": True, "message": msg}, ensure_ascii=False)
+return _json.dumps({"success": False, "error": msg}, ensure_ascii=False)