feat: hard merge gate by default (CI + directory + contract + coordinator + reviews)

- cloud/merge_gate_hard.py (new): pure hard gate; mode resolution is
  env-only (unset/invalid -> hard, fail-closed); five checks plus
  branch protection must be protected; unknown/unprotected block
  even with explicit confirm; no-evidence blocks (fail-closed);
  directory check reuses #209 check_directory_isolation (v0.3 rule);
  blocks codes + merge_gate_blocked audit detail are the track-3
  pushback interface (no pushback state machine in this track).
- cloud/server.py: POST /tasks/{id}/merge uses hard mode by default;
  record/enforce/off only when explicitly set via env (preserving
  #209 semantics); body merge_gate_mode is ignored server-side so
  callers cannot downgrade the gate; every gate block writes a
  merge_gate_blocked audit before returning 409, status unchanged;
  hard-mode body type errors return 400.
- docs/merge-gate-branch-protection.md (new): protection payload
  (PR + strict status checks + no force-push + no deletion +
  enforce_admins), empty required_checks = unconfigured, Pro/public
  prerequisites, 403 causes; states protection is NOT claimed enabled.
- SKILL.md (gate-mechanism wording only): four checks -> five checks;
  body + coordinator + client appendix gate paragraphs updated to
  default hard, merge_gate_blocked audit, payload doc reference.
- tests: 31 new hard-gate tests (pure + endpoint + skill text);
  #209 endpoint tests now set record env explicitly (their intent);
  test_server_merge 200-path calls now carry full hard evidence.

cloud/merge_gate_hard.py

@@ -0,0 +1,271 @@
+"""服务端合并硬门禁：五项全查、缺一 409、fail-closed（纯函数）。
+
+本模块堆叠在 #209（cloud/merge_gate.py）之上，不改 #209 的语义：
+- #209 的 ``evaluate_merge_gate`` / ``gate_from_request`` 原样保留：
+  record 默认、无证据不阻断、unknown 只记风险不阻断。那些是
+  #209 的过渡语义，本模块不碰它们，所以 #209 的测试一字不用改。
+- 本模块只新增一个模式值 ``hard``，以及服务端用它时的模式解析：
+  ``hard_mode_from_env``。服务端（POST /tasks/{id}/merge）以本模块
+  的解析为准，#209 的 ``gate_from_request`` 不再是服务端的模式来源
+  （它仍可作纯函数使用，但服务端不再经由它决定模式——否则调用方
+  可用请求体 ``merge_gate_mode=record`` 把硬门禁降级，那是
+  fail 但照样合并的路径，本模块堵的就是它）。
+
+模式（环境变量 CLOUD_MERGE_GATE_MODE，运维定、调用方改不了）：
+- hard（未设或非法值时都落到 hard，fail-closed，绝不回落 record/off）：
+  下面五项必须全部通过，缺一就阻断。调用方请求体里的
+  ``merge_gate_mode`` 在服务端被忽略——模式只由环境变量决定。
+- record / enforce / off（必须显式设置）：原样沿用 #209 的语义，
+  属过渡/显式降级。record 下无证据调用仍按旧接口放行（只附带门禁
+  信息），enforce 下未保护一律阻断，off 下只记风险不阻断。切到
+  这三者是运维的显式决定，不是调用方的默认权利。
+
+硬门禁五项（全部是调用方在 /merge 请求体里提交的证据；
+本服务没有可信渠道直连 GitHub 核实，所以门禁强制的是“证据齐全
+且全绿”，不是“服务端已向 GitHub 求证”——这条诚实边界必须写清，
+调用方谎报证据不在本模块的验证范围内，由审计（第 3 轨）与
+PR/Checks 的只读核对去覆盖）：
+  ① CI 全绿：``ci_status`` 必须为 ``"success"``，且 ``ci_checks``
+     非空、每项 status 均为 ``"success"``，且 ``required_checks``
+     非空、每项都在 ``ci_checks`` 里且为 success。空
+     ``required_checks``（``[]``）= 未配置，按分支保护 payload
+     的同一口径处理（见 docs/merge-gate-branch-protection.md），
+     不是“没有必须项所以通过”。
+  ② 目录隔离：复用 #209 的 ``check_directory_isolation``（v0.3 口径：
+     merge-base 范围 ``git diff --name-only``，拥有目录前缀检查，
+     含 ``apps/web2`` 非子目录边界与 ``../``、绝对路径逃逸）。
+     ``files_changed`` 缺失 = 无证据，直接阻断（#209 里是跳过，
+     硬模式里是阻断，这是本轨改动的核心之一）；越目录只能靠
+     ``approved_exceptions``（协调员批准的契约变更文件）放行，
+     显式确认绝不覆盖目录违规。
+  ③ 合同版本：``contract_version`` 非空，且 ``contract_frozen``
+     为 true；若同时给了 ``contract_expected_version``（冻结版），
+     两者必须相等，否则阻断。服务端没有独立的合同版本存储，
+     冻结版由调用方（协调员侧）声明， mismatch 时阻断。
+  ④ 协调员批准：``coordinator_approved`` 必须为 true；缺失与
+     显式 false 用不同的 block 码区分（缺证据 vs 已否决）。
+  ⑤ 无未解决评审意见：``unresolved_review_comments`` 必须为 0
+     （整数）或空列表；缺失 = 无证据，阻断；大于 0 / 非空 = 阻断。
+
+分支保护（硬模式下的第六条隐含项，来自 #209 的保护定义）：
+  ``branch_protection`` 必须为 ``"protected"``。``"unprotected"``
+  即使带 ``merge_gate_confirm`` 也阻断——硬模式没有“确认即放行”
+  的逃生口，确认只在 record 模式下覆盖保护风险。``"unknown"``
+  （未声明或非法值）同样阻断：unknown 不等于 protected，这是
+  #209 里“unknown 只记风险”的过渡语义在硬模式下的收紧。
+
+与第 3 轨（独立审计 + 自动推回）的接口约定（本模块不造推回状态机）：
+  - 阻塞依据 = 409 响应里的 ``merge_gate`` 对象：``mode``、
+    ``blocks``（稳定的机器可读码，下面的 _BLOCK_DESCRIPTIONS 有
+    逐条中文说明）、以及 ``ci`` / ``directory`` / ``contract`` /
+    ``coordinator`` / ``reviews`` 五个分项。推回包逐条取 blocks，
+    不要解析审计 detail 文本。
+  - 审计依据 = 每次门禁阻断写一条 ``merge_gate_blocked`` 审计
+    （action 名固定，detail 为 ``"<task_id> mode=<mode>
+    blocks=<逗号分隔的 blocks>"``），任务状态不变（UPDATE 不执行）。
+  - 门禁结论是单次调用的判定，不持久化为任务状态：一次 200 不
+    代表之后另一次合并自动通过；修复后必须重新提交完整证据、
+    重新走全部五项（完整重审，非只审改动点）。
+  - 本模块不建推回任务、不发通知、不改任务状态机：推回执行、
+    轮次上限、升级路径全部归第 3 轨。
+"""
+
+from cloud.merge_gate import (
+    PROTECTION_STATES,
+    RISK_UNKNOWN,
+    RISK_UNPROTECTED,
+    check_directory_isolation,
+)
+
+HARD_MODES = ("hard", "record", "enforce", "off")
+
+_BLOCK_DESCRIPTIONS = {
+    "ci_evidence_missing": "CI 证据缺失（ci_status / ci_checks / required_checks 有一项未给）",
+    "ci_not_green": "CI 状态不是 success",
+    "ci_check_failed": "CI 检查项中有非 success 项",
+    "ci_required_checks_unconfigured": "required_checks 为空 = 未配置",
+    "ci_required_check_missing": "required_checks 中有项不在 CI 检查里或不是 success",
+    "directory_evidence_missing": "目录证据缺失（files_changed 未给）",
+    "directory_owned_dir_unknown": "拥有目录未知，无法核对目录隔离",
+    "directory_isolation_violation": "目录隔离违规（越拥有目录）",
+    "contract_version_missing": "合同版本缺失",
+    "contract_not_frozen": "合同未冻结（contract_frozen 不是 true）",
+    "contract_version_mismatch": "合同版本与冻结版不一致",
+    "coordinator_approval_missing": "协调员批准证据缺失",
+    "coordinator_not_approved": "协调员未批准",
+    "review_evidence_missing": "评审意见证据缺失",
+    "unresolved_review_comments": "存在未解决评审意见",
+    "branch_protection_unknown": "分支保护状态未知（unknown 不等于 protected）",
+    "branch_protection_unprotected": "分支未保护（硬模式下显式确认也不放行）",
+}
+
+
+def hard_mode_from_env(env=None):
+    """服务端有效的门禁模式：环境变量定，调用方改不了。
+
+    未设、空值或非法值一律回落 hard（fail-closed）。注意这与
+    #209 的 gate_mode_from_env 不同：后者非法值回落 record，那是
+    #209 纯函数的过渡语义；服务端模式解析必须 fail-closed，所以
+    这里绝不回落到 record/off。
+    """
+    import os
+
+    raw = (env if env is not None else os.environ).get(
+        "CLOUD_MERGE_GATE_MODE", "")
+    mode = (raw or "").strip().lower()
+    return mode if mode in HARD_MODES else "hard"
+
+
+def _ci_checks_index(ci_checks):
+    """ci_checks 列表 -> {name: status}；条目不合规时返回 None。"""
+    index = {}
+    for entry in ci_checks:
+        if not isinstance(entry, dict):
+            return None
+        name = entry.get("name")
+        status = entry.get("status")
+        if not isinstance(name, str) or not name \
+                or not isinstance(status, str):
+            return None
+        index[name] = status
+    return index
+
+
+def evaluate_hard_merge_gate(files_changed=None, owned_dir=None,
+                             approved_exceptions=(), protection="unknown",
+                             ci_status=None, ci_checks=None,
+                             required_checks=None, contract_version=None,
+                             contract_frozen=False,
+                             contract_expected_version=None,
+                             coordinator_approved=None,
+                             unresolved_review_comments=None):
+    """硬门禁纯函数：五项 + 保护，缺一即 allowed=False。
+
+    files_changed=None、ci_checks=None、required_checks=None、
+    unresolved_review_comments=None 全部是“无证据”，不是“跳过”。
+    protection 非法值归一化为 unknown，随后按 unknown 阻断。
+    返回 dict：allowed / mode / protection / directory / ci /
+    contract / coordinator / reviews / risks / blocks。
+    """
+    protection = protection if protection in PROTECTION_STATES \
+        else "unknown"
+    risks = []
+    blocks = []
+
+    # ① CI 全绿。
+    if ci_status is None or ci_status == "" \
+            or ci_checks is None or required_checks is None:
+        blocks.append("ci_evidence_missing")
+    else:
+        if ci_status != "success":
+            blocks.append("ci_not_green")
+        if not isinstance(ci_checks, list) or not ci_checks:
+            blocks.append("ci_evidence_missing")
+            index = None
+        else:
+            index = _ci_checks_index(ci_checks)
+            if index is None:
+                blocks.append("ci_check_failed")
+            elif any(status != "success" for status in index.values()):
+                blocks.append("ci_check_failed")
+        if isinstance(required_checks, list) and not required_checks:
+            blocks.append("ci_required_checks_unconfigured")
+        elif isinstance(required_checks, list) and index is not None:
+            missing = [name for name in required_checks
+                       if index.get(name) != "success"]
+            if missing:
+                blocks.append("ci_required_check_missing")
+        elif not isinstance(required_checks, list):
+            blocks.append("ci_evidence_missing")
+    ci = {"status": ci_status, "checks": ci_checks,
+          "required_checks": required_checks,
+          "all_green": False}  # 末尾按 CI 分项 blocks 重算，见下。
+
+    # ② 目录隔离（复用 #209，不另造口径）。
+    directory = None
+    if files_changed is None:
+        blocks.append("directory_evidence_missing")
+    else:
+        directory = check_directory_isolation(
+            files_changed, owned_dir, approved_exceptions)
+        if directory["inconclusive"]:
+            blocks.append("directory_owned_dir_unknown")
+        elif not directory["passed"]:
+            blocks.append("directory_isolation_violation")
+
+    # ③ 合同版本。
+    if contract_version is None or (
+            isinstance(contract_version, str)
+            and not contract_version.strip()):
+        blocks.append("contract_version_missing")
+    if contract_frozen is not True:
+        blocks.append("contract_not_frozen")
+    if contract_expected_version is not None \
+            and contract_version is not None \
+            and contract_version != contract_expected_version:
+        blocks.append("contract_version_mismatch")
+    contract = {"version": contract_version, "frozen": contract_frozen,
+                "expected": contract_expected_version}
+
+    # ④ 协调员批准。
+    if coordinator_approved is None:
+        blocks.append("coordinator_approval_missing")
+    elif coordinator_approved is not True:
+        blocks.append("coordinator_not_approved")
+    coordinator = {"approved": coordinator_approved}
+
+    # ⑤ 无未解决评审意见。
+    if unresolved_review_comments is None:
+        blocks.append("review_evidence_missing")
+        reviews = {"unresolved": None}
+    elif isinstance(unresolved_review_comments, bool):
+        blocks.append("review_evidence_missing")
+        reviews = {"unresolved": None}
+    elif isinstance(unresolved_review_comments, list):
+        reviews = {"unresolved": unresolved_review_comments}
+        if unresolved_review_comments:
+            blocks.append("unresolved_review_comments")
+    elif isinstance(unresolved_review_comments, int):
+        reviews = {"unresolved": unresolved_review_comments}
+        if unresolved_review_comments > 0:
+            blocks.append("unresolved_review_comments")
+        elif unresolved_review_comments < 0:
+            blocks.append("review_evidence_missing")
+    else:
+        blocks.append("review_evidence_missing")
+        reviews = {"unresolved": None}
+
+    # 分支保护：硬模式下 unknown 与 unprotected 一律阻断，确认无效。
+    if protection == "unprotected":
+        risks.append(RISK_UNPROTECTED)
+        blocks.append("branch_protection_unprotected")
+    elif protection == "unknown":
+        risks.append(RISK_UNKNOWN)
+        blocks.append("branch_protection_unknown")
+
+    # ci.all_green 只反映 CI 分项本身，不被其它项的 blocks 污染。
+    ci_block_codes = ("ci_evidence_missing", "ci_not_green",
+                      "ci_check_failed", "ci_required_checks_unconfigured",
+                      "ci_required_check_missing")
+    ci["all_green"] = not any(code in blocks for code in ci_block_codes)
+
+    return {"allowed": not blocks, "mode": "hard",
+            "protection": protection, "directory": directory,
+            "ci": ci, "contract": contract,
+            "coordinator": coordinator, "reviews": reviews,
+            "risks": risks, "blocks": blocks}
+
+
+def block_descriptions(blocks):
+    """blocks 码 -> 中文说明（供第 3 轨推回包逐条引用）。"""
+    return [_BLOCK_DESCRIPTIONS.get(code, code) for code in blocks or []]
+
+
+def blocked_audit_detail(task_id, gate):
+    """merge_gate_blocked 审计的 detail：只含任务 id、模式与 blocks 码。
+
+    推回包不要解析这段文本，取 409 响应的 merge_gate 对象为准。
+    """
+    return "%s mode=%s blocks=%s" % (
+        task_id, gate.get("mode", "hard"),
+        ",".join(gate.get("blocks") or []))

cloud/server.py

@@ -74,7 +74,11 @@ from cloud.peer_messages import ensure_followups_table as _peer_ensure_followups
 from cloud.peer_messages import peer_message_prefix as _peer_prefix
 from cloud.peer_messages import row_get as _row_get
 from cloud.marker_tools import parse_spawn_defaults as _parse_spawn_defaults
-from cloud.merge_gate import gate_from_request as _merge_gate_from_request
+from cloud.merge_gate import evaluate_merge_gate as _merge_gate_evaluate
+from cloud.merge_gate import protection_from_env as _merge_gate_protection
+from cloud.merge_gate_hard import blocked_audit_detail as _gate_blocked_detail
+from cloud.merge_gate_hard import evaluate_hard_merge_gate as _merge_gate_hard
+from cloud.merge_gate_hard import hard_mode_from_env as _merge_gate_hard_mode
 from cloud.tree import DEFAULT_MAX_DEPTH as _TREE_DEFAULT_DEPTH
 from cloud.tree import DEFAULT_MAX_ENTRIES as _TREE_DEFAULT_ENTRIES
 from cloud.tree import build_tree as _build_tree
@@ -1960,6 +1964,43 @@ def sweep_stale_tasks(db_path, threshold_seconds=300):
     return swept
 
 
+def _validate_hard_gate_body(body):
+    """硬模式请求体类型校验：畸形字段回 400，不进门禁判定。
+
+    只在 hard 模式调用；record/enforce/off 下这些字段按 #209 语义
+    被忽略，不因此 400。缺失字段不是畸形，是无证据，交门禁判 409。
+    返回错误串，无错返回 None。
+    """
+    for field in ("ci_checks", "required_checks", "approved_exceptions"):
+        value = body.get(field)
+        if value is not None and not isinstance(value, list):
+            return "%s must be a list" % field
+    checks = body.get("ci_checks")
+    if isinstance(checks, list):
+        for entry in checks:
+            if not isinstance(entry, dict) \
+                    or not isinstance(entry.get("name"), str) \
+                    or not isinstance(entry.get("status"), str):
+                return "ci_checks entries must be objects " \
+                    "with name and status"
+    for field in ("contract_frozen", "coordinator_approved"):
+        value = body.get(field)
+        if value is not None and not isinstance(value, bool):
+            return "%s must be a boolean" % field
+    version = body.get("contract_version")
+    if version is not None and not isinstance(version, str):
+        return "contract_version must be a string"
+    expected = body.get("contract_expected_version")
+    if expected is not None and not isinstance(expected, str):
+        return "contract_expected_version must be a string"
+    reviews = body.get("unresolved_review_comments")
+    if reviews is not None and not isinstance(reviews, list) \
+            and not (isinstance(reviews, int)
+                     and not isinstance(reviews, bool)):
+        return "unresolved_review_comments must be a list or an integer"
+    return None
+
+
 class CloudControlPlaneHandler(BaseHTTPRequestHandler):
     """控制面请求处理：除 /health 与 /webhooks/* 外一律要求 X-API-Key。"""
 
@@ -7616,17 +7657,34 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         必须是 completed（其它状态一律 409，含已 merged 的重复
         调用，避免二次合并混淆）。成功后写审计 merge_task。
 
-        合并门禁（cloud/merge_gate，补 v0.3 技能文本检查在服务端
-        缺失的缺口；v0.3 之前本接口只改状态，不查目录、不查保护）：
-        请求体可选字段 files_changed / owned_dir /
-        approved_exceptions / branch_protection / merge_gate_confirm
-        / merge_gate_mode；一个都不给时行为与旧接口一致（只在响应
-        里附带门禁结果，不阻断）。给了证据才按门禁判：越目录
-        （v0.3 口径）或未保护且无确认（record 模式）或未保护
-        （enforce 模式）时 409 且不改状态；未保护 + 确认时放行，
-        风险记入响应与 merge_gate_risk 审计。确认只覆盖保护风险，
-        绝不覆盖目录违规。保护状态 unknown 时记风险、不阻断——
-        本服务没有可信渠道核实 GitHub 保护，unknown 不等于已保护。
+        合并门禁（cloud/merge_gate + cloud/merge_gate_hard，补 v0.3
+        技能文本检查在服务端缺失的缺口；v0.3 之前本接口只改状态，
+        不查目录、不查保护）：
+        模式只由环境变量 CLOUD_MERGE_GATE_MODE 决定，请求体里的
+        merge_gate_mode 在服务端被忽略（调用方不能自行降级）：
+        - hard（默认：环境变量未设或非法值都落到 hard）：五项全查，
+          缺一 409 且不改状态，并写 merge_gate_blocked 审计。五项：
+          ① CI 全绿（ci_status=success + ci_checks 全 success +
+          required_checks 非空且全 success）；② 目录隔离（v0.3 口径，
+          files_changed 缺失即无证据阻断，越目录只能靠
+          approved_exceptions 放行）；③ 合同版本（contract_version
+          非空 + contract_frozen=true，与 contract_expected_version
+          给出时须相等）；④ 协调员批准（coordinator_approved=true）；
+          ⑤ 无未解决评审意见（unresolved_review_comments=0 或空列表）。
+          另：branch_protection 必须为 protected；unprotected 即使带
+          merge_gate_confirm 也阻断，unknown 同样阻断（unknown
+          不等于 protected）。无证据不放行（fail-closed）。
+        - record / enforce / off（必须显式设置，属过渡/显式降级）：
+          原样沿用 #209 语义：record 下无证据调用仍按旧接口放行
+          （只在响应里附带门禁结果），未保护 + 无确认阻断、+ 确认
+          放行并记 merge_gate_risk；enforce 下未保护一律阻断；
+          off 下只记风险不阻断。确认只覆盖保护风险，绝不覆盖目录违规。
+        门禁阻断（任一模式）都写 merge_gate_blocked 审计后才回 409，
+        这是第 3 轨自动推回的阻塞依据：推回包取响应 merge_gate 的
+        blocks 码，不要解析审计文本；本接口不建推回任务、不改状态机。
+        本服务没有可信渠道直连 GitHub 核实 CI/保护：门禁强制的是
+        “调用方提交的证据齐全且全绿”，谎报证据由审计与 PR/Checks
+        只读核对覆盖，不在本接口的验证范围内。
         """
         row = self._fetch_task(task_id)
         if row is None:
@@ -7652,8 +7710,41 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 body.get("files_changed"), list):
             self._send_json(400, {"error": "files_changed must be a list"})
             return
-        gate = _merge_gate_from_request(body)
+        mode = _merge_gate_hard_mode()
+        protection = body.get("branch_protection") \
+            or _merge_gate_protection()
+        if mode == "hard":
+            hard_err = _validate_hard_gate_body(body)
+            if hard_err:
+                self._send_json(400, {"error": hard_err})
+                return
+            gate = _merge_gate_hard(
+                files_changed=body.get("files_changed"),
+                owned_dir=body.get("owned_dir"),
+                approved_exceptions=body.get("approved_exceptions") or (),
+                protection=protection,
+                ci_status=body.get("ci_status"),
+                ci_checks=body.get("ci_checks"),
+                required_checks=body.get("required_checks"),
+                contract_version=body.get("contract_version"),
+                contract_frozen=body.get("contract_frozen", False),
+                contract_expected_version=body.get(
+                    "contract_expected_version"),
+                coordinator_approved=body.get("coordinator_approved"),
+                unresolved_review_comments=body.get(
+                    "unresolved_review_comments"))
+        else:
+            gate = _merge_gate_evaluate(
+                files_changed=body.get("files_changed"),
+                owned_dir=body.get("owned_dir"),
+                approved_exceptions=body.get("approved_exceptions") or (),
+                protection=protection, mode=mode,
+                explicit_confirm=bool(body.get("merge_gate_confirm")))
         if not gate["allowed"]:
+            with _auth_store(self.server.db_path) as store:
+                store.audit(user_id, task["project_id"],
+                            "merge_gate_blocked",
+                            _gate_blocked_detail(task_id, gate))
             self._send_json(409, {
                 "error": "merge gate blocked",
                 "merge_gate": gate,

cloud/skills/software-engineering/SKILL.md

@@ -524,10 +524,10 @@ tests/                 # QA 线程专属，含跨端一致性用例
 
 ### 合并门：缺一不可，与主分支保护
 
-- 合并前四项全过：① GitHub Checks 全绿；② 目录归属检查通过
+- 合并前五项全过：① GitHub Checks 全绿；② 目录归属检查通过
   （只改拥有目录，契约变更经批准除外；检查命令与输出文件清单必须
   记入合并记录，见“目录与分支规则”）；③ 契约版本是冻结版；
-  ④ 协调员评审通过，且无未解决评审意见。有一项不过就不合并。
+  ④ 协调员评审通过；⑤ 无未解决评审意见。有一项不过就不合并。
 - CI 怎么查：用只读方式查，例如本平台的 GET /tasks/{id}/pr/checks。
   它是只读诊断，绝不重跑检查、不批准、不合并；还没建 PR 时它会 409
   （查无可查），这时先补 PR，不要把它当成 CI 通过。
@@ -536,17 +536,24 @@ tests/                 # QA 线程专属，含跨端一致性用例
   协调员要在决策记录里写明风险并去开通，不能假装已保护，
   不能靠“大家自觉”代替保护。
 - 合并门禁（2026-10-30 补，代码在 cloud/merge_gate.py，服务端
-  POST /tasks/{id}/merge 已接线）：保护状态只能由调用方或环境变量
+  POST /tasks/{id}/merge 已接线；2026-10-31 硬门禁轨加
+  cloud/merge_gate_hard.py）：保护状态只能由调用方或环境变量
   CLOUD_BRANCH_PROTECTION 显式声明，代码不去 GitHub 核实、不声称
-  已保护；状态 unknown 时记风险、不阻断。模式由
-  CLOUD_MERGE_GATE_MODE 控制，默认 record：未保护时须显式确认
-  （merge_gate_confirm）才放行，无确认阻断；enforce 模式下未保护
-  一律阻断，确认也不放行——切 enforce 前必须先由用户决定并开通
-  保护（升级 Pro、公开仓库、或保持纪律 + 记风险，三者用户侧定，
-  代码不替选）。显式确认只覆盖保护风险，绝不覆盖目录隔离违规；
-  目录检查复用 v0.3 的 git diff --name-only 口径，越目录只能靠
-  契约变更批准文件放行。技能播种仍只复制 SKILL.md，不做整目录复制
-  （租户已改文件绝不被覆盖，见 cloud/skill_seed.py 的评估结论）。
+  已保护。模式只由环境变量 CLOUD_MERGE_GATE_MODE 控制，默认
+  hard：上面五项全查，缺一 409 且任务状态不变，并写
+  merge_gate_blocked 审计；无证据不放行；unknown 不等于
+  protected，同样阻断；未保护即使带显式确认也不放行。record /
+  enforce / off 必须显式设置才生效（过渡/显式降级）：record 下
+  未保护时须显式确认（merge_gate_confirm）才放行，无确认阻断，
+  无证据调用仍按旧接口放行；enforce 模式下未保护一律阻断，
+  确认也不放行——切 enforce 前必须先由用户决定并开通保护
+  （升级 Pro、公开仓库、或保持纪律 + 记风险，三者用户侧定，
+  代码不替选，操作说明与 payload 见
+  docs/merge-gate-branch-protection.md）。显式确认只覆盖保护风险，
+  绝不覆盖目录隔离违规；目录检查复用 v0.3 的 git diff --name-only
+  口径，越目录只能靠契约变更批准文件放行。技能播种仍只复制
+  SKILL.md，不做整目录复制（租户已改文件绝不被覆盖，见
+  cloud/skill_seed.py 的评估结论）。
 
 ### 合并方式与合并记录
 
@@ -828,18 +835,19 @@ G8 的交付物是 v1.1 交接清单终版与下一轮 G0-G6 输入冻结清单
   测试证据、已知问题、关联待办或 PRD 项；评审意见逐条处理，
   未解决的评审意见不能合并。
 - 合并门：GitHub Checks 全绿、目录归属检查通过、契约版本是冻结版、
-  你评审通过且无未解决评审意见，四项缺一不可。
+  你评审通过、无未解决评审意见，五项缺一不可。
   CI 用只读方式查，例如 GET /tasks/{id}/pr/checks，它绝不重跑检查、
   不批准、不合并；无 PR 时它会 409，先补 PR。
 - 主分支保护：要求 PR、要求 CI 通过、禁止直接推，这是 GitHub 仓库设置，
   不是本平台代码能替你开的。没开保护时记为风险并去开通，不假装已保护。
-- 合并门禁：同正文，代码在 cloud/merge_gate.py，模式由
-  CLOUD_MERGE_GATE_MODE 控制，默认 record：未保护时须显式确认
-  才放行，无确认阻断；enforce 模式下未保护一律阻断，切 enforce
-  前须由用户决定并开通保护（升级 Pro、公开仓库、或保持纪律 +
-  记风险，三者用户侧定）。显式确认只覆盖保护风险，绝不覆盖
-  目录隔离违规；目录检查复用 v0.3 的 git diff --name-only 口径；
-  技能播种仍只复制 SKILL.md，不做整目录复制。
+- 合并门禁：同正文，代码在 cloud/merge_gate.py 与
+  cloud/merge_gate_hard.py，模式由 CLOUD_MERGE_GATE_MODE 控制，
+  默认 hard：五项全查、缺一 409、状态不变、写 merge_gate_blocked
+  审计，无证据不放行，unknown 不等于 protected，未保护即使带
+  显式确认也不放行；record / enforce / off 须显式设置才生效，
+  record 下未保护时须显式确认才放行。显式确认只覆盖保护风险，
+  绝不覆盖目录隔离违规；目录检查复用 v0.3 的 git diff --name-only
+  口径；技能播种仍只复制 SKILL.md，不做整目录复制。
 - 合并记录：默认 merge commit；记录必须含 PR 号、合并提交 hash、
   CI 运行链接、执行人（人或宿主操作，不是你自己合）；
   合并后删远程分支，删前确认合并提交已在主线。
@@ -1068,9 +1076,11 @@ TODO-UPDATE 要求，缺一项就不发任务）。派生提示词里的精简
 - 合并门由协调员判定：Checks 全绿、目录归属通过、契约版本是冻结版、
   评审通过且无未解决意见；你不自己合并、不绕过评审推主线。
 - 合并门禁：同正文，模式由 CLOUD_MERGE_GATE_MODE 控制，默认
-  record：未保护时须显式确认才放行；显式确认只覆盖保护风险，
-  绝不覆盖目录隔离违规；目录检查复用 v0.3 的
-  git diff --name-only 口径；技能播种仍只复制 SKILL.md，
+  hard：五项全查、缺一 409、状态不变、写 merge_gate_blocked 审计，
+  无证据不放行，unknown 不等于 protected，未保护即使带显式确认
+  也不放行；record 须显式设置才生效，未保护时须显式确认才放行；
+  显式确认只覆盖保护风险，绝不覆盖目录隔离违规；目录检查复用
+  v0.3 的 git diff --name-only 口径；技能播种仍只复制 SKILL.md，
   不做整目录复制。
 - 安全：密钥、口令、token 绝不进代码、commit、命令行、PR 正文；
   需要调用 GitHub 时 token 只进子进程环境，不写进示例命令。

cloud/tests/test_merge_gate_hard.py

@@ -0,0 +1,453 @@
+"""服务端合并硬门禁（cloud/merge_gate_hard，默认 hard 模式）。
+
+与 #209 的分工（不要把两套语义混成一套断言）：
+- #209 的纯函数与 record/enforce/off 语义原样保留，见
+  test_merge_gate_protectskill.py；本文件不重测 record 的
+  “无证据放行”，只测：record 仍是显式 env 才能选到的过渡模式。
+- 本文件测硬模式：默认（env 未设或非法值）即 hard，五项全查、
+  缺一 409、状态不变、merge_gate_blocked 审计；unknown 保护
+  不等于 protected；显式确认在硬模式下不覆盖任何一项；
+  调用方请求体里的 merge_gate_mode 不能把硬门禁降级。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.merge_gate_hard import (
+        blocked_audit_detail, evaluate_hard_merge_gate, hard_mode_from_env)
+    from cloud.server import create_server
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.merge_gate_hard import (
+        blocked_audit_detail, evaluate_hard_merge_gate, hard_mode_from_env)
+    from cloud.server import create_server
+
+
+def _hard_kwargs(**overrides):
+    kwargs = {
+        "files_changed": ["apps/web/a.py"],
+        "owned_dir": "apps/web",
+        "protection": "protected",
+        "ci_status": "success",
+        "ci_checks": [{"name": "ci", "status": "success"}],
+        "required_checks": ["ci"],
+        "contract_version": "v2",
+        "contract_frozen": True,
+        "coordinator_approved": True,
+        "unresolved_review_comments": 0,
+    }
+    kwargs.update(overrides)
+    return kwargs
+
+
+class HardModeEnvTestCase(unittest.TestCase):
+    def test_unset_and_invalid_mode_fall_to_hard(self):
+        self.assertEqual(hard_mode_from_env({}), "hard")
+        self.assertEqual(hard_mode_from_env(
+            {"CLOUD_MERGE_GATE_MODE": ""}), "hard")
+        self.assertEqual(hard_mode_from_env(
+            {"CLOUD_MERGE_GATE_MODE": "bogus"}), "hard")
+        self.assertEqual(hard_mode_from_env(
+            {"CLOUD_MERGE_GATE_MODE": "record"}), "record")
+        self.assertEqual(hard_mode_from_env(
+            {"CLOUD_MERGE_GATE_MODE": "HARD"}), "hard")
+
+
+class HardGateUnitTestCase(unittest.TestCase):
+    def test_full_evidence_passes(self):
+        gate = evaluate_hard_merge_gate(**_hard_kwargs())
+        self.assertTrue(gate["allowed"])
+        self.assertEqual(gate["blocks"], [])
+        self.assertEqual(gate["mode"], "hard")
+        self.assertTrue(gate["ci"]["all_green"])
+
+    def test_no_evidence_blocks_every_missing_item(self):
+        gate = evaluate_hard_merge_gate()
+        self.assertFalse(gate["allowed"])
+        for code in ("ci_evidence_missing", "directory_evidence_missing",
+                     "contract_version_missing", "contract_not_frozen",
+                     "coordinator_approval_missing",
+                     "review_evidence_missing",
+                     "branch_protection_unknown"):
+            self.assertIn(code, gate["blocks"])
+
+    def test_ci_variants_block(self):
+        pending = evaluate_hard_merge_gate(**_hard_kwargs(ci_status="pending"))
+        self.assertIn("ci_not_green", pending["blocks"])
+        failed_check = evaluate_hard_merge_gate(**_hard_kwargs(
+            ci_checks=[{"name": "ci", "status": "failure"}]))
+        self.assertIn("ci_check_failed", failed_check["blocks"])
+        self.assertFalse(failed_check["ci"]["all_green"])
+        empty_required = evaluate_hard_merge_gate(**_hard_kwargs(
+            required_checks=[]))
+        self.assertIn("ci_required_checks_unconfigured",
+                      empty_required["blocks"])
+        missing_required = evaluate_hard_merge_gate(**_hard_kwargs(
+            required_checks=["ci", "lint"]))
+        self.assertIn("ci_required_check_missing",
+                      missing_required["blocks"])
+        bare_success = evaluate_hard_merge_gate(**_hard_kwargs(
+            ci_checks=None, required_checks=None))
+        self.assertIn("ci_evidence_missing", bare_success["blocks"])
+
+    def test_directory_reuses_v03_rule(self):
+        edge = evaluate_hard_merge_gate(**_hard_kwargs(
+            files_changed=["apps/web2/x.py"]))
+        self.assertIn("directory_isolation_violation", edge["blocks"])
+        escape = evaluate_hard_merge_gate(**_hard_kwargs(
+            files_changed=["../secret.py", "/abs.py"]))
+        self.assertEqual(escape["directory"]["violations"],
+                         ["../secret.py", "/abs.py"])
+        exc = evaluate_hard_merge_gate(**_hard_kwargs(
+            files_changed=["apps/web/a.py", "contracts/api.md"],
+            approved_exceptions=["contracts/api.md"]))
+        self.assertTrue(exc["allowed"])
+        self.assertEqual(exc["directory"]["exceptions_used"],
+                         ["contracts/api.md"])
+        no_dir = evaluate_hard_merge_gate(**_hard_kwargs(
+            files_changed=None))
+        self.assertIn("directory_evidence_missing", no_dir["blocks"])
+        no_owned = evaluate_hard_merge_gate(**_hard_kwargs(owned_dir=None))
+        self.assertIn("directory_owned_dir_unknown", no_owned["blocks"])
+
+    def test_contract_variants_block(self):
+        missing = evaluate_hard_merge_gate(**_hard_kwargs(
+            contract_version=None))
+        self.assertIn("contract_version_missing", missing["blocks"])
+        unfrozen = evaluate_hard_merge_gate(**_hard_kwargs(
+            contract_frozen=False))
+        self.assertIn("contract_not_frozen", unfrozen["blocks"])
+        mismatch = evaluate_hard_merge_gate(**_hard_kwargs(
+            contract_expected_version="v3"))
+        self.assertIn("contract_version_mismatch", mismatch["blocks"])
+        matched = evaluate_hard_merge_gate(**_hard_kwargs(
+            contract_expected_version="v2"))
+        self.assertTrue(matched["allowed"])
+
+    def test_coordinator_and_reviews_block(self):
+        missing = evaluate_hard_merge_gate(**_hard_kwargs(
+            coordinator_approved=None))
+        self.assertIn("coordinator_approval_missing", missing["blocks"])
+        denied = evaluate_hard_merge_gate(**_hard_kwargs(
+            coordinator_approved=False))
+        self.assertIn("coordinator_not_approved", denied["blocks"])
+        unresolved = evaluate_hard_merge_gate(**_hard_kwargs(
+            unresolved_review_comments=["c1"]))
+        self.assertIn("unresolved_review_comments",
+                      unresolved["blocks"])
+        count = evaluate_hard_merge_gate(**_hard_kwargs(
+            unresolved_review_comments=2))
+        self.assertIn("unresolved_review_comments", count["blocks"])
+        missing_reviews = evaluate_hard_merge_gate(**_hard_kwargs(
+            unresolved_review_comments=None))
+        self.assertIn("review_evidence_missing",
+                      missing_reviews["blocks"])
+
+    def test_protection_unknown_and_unprotected_block(self):
+        unknown = evaluate_hard_merge_gate(**_hard_kwargs(
+            protection="unknown"))
+        self.assertIn("branch_protection_unknown", unknown["blocks"])
+        illegal = evaluate_hard_merge_gate(**_hard_kwargs(
+            protection="bogus"))
+        self.assertEqual(illegal["protection"], "unknown")
+        self.assertIn("branch_protection_unknown", illegal["blocks"])
+
+    def test_unprotected_no_confirm_blocks(self):
+        gate = evaluate_hard_merge_gate(**_hard_kwargs(
+            protection="unprotected"))
+        self.assertFalse(gate["allowed"])
+        self.assertIn("branch_protection_unprotected", gate["blocks"])
+
+    def test_confirm_never_overrides_anything_in_hard_mode(self):
+        # 硬模式没有“确认即放行”：确认不是本函数的入参，调用方
+        # 即使在请求体里带 merge_gate_confirm，服务端硬路径也不读它；
+        # 这里钉住纯函数侧：未保护 + 目录违规同时存在时两项都拦。
+        gate = evaluate_hard_merge_gate(**_hard_kwargs(
+            protection="unprotected",
+            files_changed=["apps/web/a.py", "apps/ios/b.py"]))
+        self.assertIn("branch_protection_unprotected", gate["blocks"])
+        self.assertIn("directory_isolation_violation", gate["blocks"])
+
+    def test_blocked_audit_detail_format(self):
+        gate = evaluate_hard_merge_gate()
+        detail = blocked_audit_detail("task-1", gate)
+        self.assertTrue(detail.startswith("task-1 mode=hard blocks="))
+        self.assertIn("ci_evidence_missing", detail)
+
+
+class HardGateEndpointTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+        # 默认即 hard：每个用例开始时清掉模式/保护 env，个别用例
+        # 再显式设置 record/enforce/off/bogus。
+        self._saved_env = {k: os.environ.get(k) for k in
+                           ("CLOUD_MERGE_GATE_MODE",
+                            "CLOUD_BRANCH_PROTECTION")}
+        os.environ.pop("CLOUD_MERGE_GATE_MODE", None)
+        os.environ.pop("CLOUD_BRANCH_PROTECTION", None)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        for key, value in self._saved_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _completed_task(self):
+        status, task = self._req(
+            "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "x"},
+            self.owner_key)
+        self.assertEqual(status, 201)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE tasks SET status = 'completed' WHERE id = ?",
+                (task["id"],))
+            conn.commit()
+        finally:
+            conn.close()
+        return task
+
+    def _hard_body(self, **overrides):
+        body = {
+            "files_changed": ["apps/web/a.py"],
+            "owned_dir": "apps/web",
+            "branch_protection": "protected",
+            "ci_status": "success",
+            "ci_checks": [{"name": "ci", "status": "success"}],
+            "required_checks": ["ci"],
+            "contract_version": "v2",
+            "contract_frozen": True,
+            "coordinator_approved": True,
+            "unresolved_review_comments": 0,
+        }
+        body.update(overrides)
+        return body
+
+    def _status_of(self, task_id):
+        _s, got = self._req("GET", "/tasks/%s" % task_id,
+                            key=self.owner_key)
+        return got["status"]
+
+    def test_default_no_fields_blocks_status_unchanged_and_audits(self):
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertEqual(body["merge_gate"]["mode"], "hard")
+        self.assertIn("ci_evidence_missing",
+                      body["merge_gate"]["blocks"])
+        self.assertEqual(self._status_of(task["id"]), "completed")
+        audit = self.auth.list_audit(self.project["id"])
+        blocked = [a for a in audit
+                   if a["action"] == "merge_gate_blocked"]
+        self.assertTrue(blocked)
+        self.assertIn(task["id"], blocked[0]["detail"])
+        self.assertIn("ci_evidence_missing", blocked[0]["detail"])
+
+    def test_default_full_evidence_merges(self):
+        task = self._completed_task()
+        status, merged = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            self._hard_body(), self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(merged["status"], "merged")
+        self.assertEqual(merged["merge_gate"]["mode"], "hard")
+        self.assertEqual(merged["merge_gate"]["blocks"], [])
+
+    def test_default_missing_one_item_blocks(self):
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            self._hard_body(unresolved_review_comments=2),
+            self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertIn("unresolved_review_comments",
+                      body["merge_gate"]["blocks"])
+        self.assertEqual(self._status_of(task["id"]), "completed")
+
+    def test_default_unknown_protection_blocks(self):
+        task = self._completed_task()
+        body = self._hard_body()
+        del body["branch_protection"]
+        status, resp = self._req(
+            "POST", "/tasks/%s/merge" % task["id"], body, self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertIn("branch_protection_unknown",
+                      resp["merge_gate"]["blocks"])
+
+    def test_default_unprotected_blocks_even_with_confirm(self):
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            self._hard_body(branch_protection="unprotected",
+                            merge_gate_confirm=True),
+            self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertIn("branch_protection_unprotected",
+                      body["merge_gate"]["blocks"])
+        self.assertEqual(self._status_of(task["id"]), "completed")
+
+    def test_default_confirm_never_overrides_directory_violation(self):
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            self._hard_body(
+                files_changed=["apps/web/a.py", "apps/ios/b.py"],
+                merge_gate_confirm=True),
+            self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertIn("directory_isolation_violation",
+                      body["merge_gate"]["blocks"])
+
+    def test_body_mode_cannot_downgrade_hard(self):
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            {"merge_gate_mode": "record"}, self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertEqual(body["merge_gate"]["mode"], "hard")
+
+    def test_invalid_env_mode_falls_to_hard(self):
+        os.environ["CLOUD_MERGE_GATE_MODE"] = "bogus"
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertEqual(body["merge_gate"]["mode"], "hard")
+
+    def test_record_env_preserves_old_no_fields_behavior(self):
+        os.environ["CLOUD_MERGE_GATE_MODE"] = "record"
+        task = self._completed_task()
+        status, merged = self._req(
+            "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(merged["status"], "merged")
+
+    def test_record_env_unprotected_confirm_still_passes_with_risk(self):
+        os.environ["CLOUD_MERGE_GATE_MODE"] = "record"
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            {"branch_protection": "unprotected"}, self.owner_key)
+        self.assertEqual(status, 409)
+        task2 = self._completed_task()
+        status, merged = self._req(
+            "POST", "/tasks/%s/merge" % task2["id"],
+            {"branch_protection": "unprotected",
+             "merge_gate_confirm": True},
+            self.owner_key)
+        self.assertEqual(status, 200)
+        audit = self.auth.list_audit(self.project["id"])
+        self.assertTrue(any(a["action"] == "merge_gate_risk"
+                            for a in audit))
+
+    def test_enforce_env_blocks_unprotected_even_with_confirm(self):
+        os.environ["CLOUD_MERGE_GATE_MODE"] = "enforce"
+        os.environ["CLOUD_BRANCH_PROTECTION"] = "unprotected"
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            {"merge_gate_confirm": True}, self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertIn("branch_protection_enforce",
+                      body["merge_gate"]["blocks"])
+
+    def test_off_env_is_explicit_opt_down_only(self):
+        os.environ["CLOUD_MERGE_GATE_MODE"] = "off"
+        task = self._completed_task()
+        status, merged = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            {"branch_protection": "unprotected"}, self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(merged["merge_gate"]["mode"], "off")
+
+    def test_malformed_hard_body_returns_400(self):
+        task = self._completed_task()
+        status, _ = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            self._hard_body(ci_checks="ci"), self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            self._hard_body(unresolved_review_comments=True),
+            self.owner_key)
+        self.assertEqual(status, 400)
+        self.assertEqual(self._status_of(task["id"]), "completed")
+
+
+class HardGateSkillTextTestCase(unittest.TestCase):
+    def test_hard_keywords_in_body_and_appendix(self):
+        # 正文换行会把“默认 hard”断成两行，断言前先把空白归一化。
+        import re
+        raw = (Path(__file__).resolve().parents[1] / "skills"
+               / "software-engineering" / "SKILL.md").read_text(
+                   encoding="utf-8")
+        text = re.sub(r"\s+", " ", raw)
+        main, sep, appendix = text.partition("## 附录：角色模板")
+        self.assertTrue(sep, "appendix marker missing")
+        coord = text.split("## 附录角色：00-coordinator", 1)[1].split(
+            "## 附录角色：01-", 1)[0]
+        client = text.split("## 附录角色：04-client-developer", 1)[1].split(
+            "## 附录角色：05-", 1)[0]
+        for kw in ("默认 hard", "merge_gate_blocked"):
+            self.assertIn(kw, main, "正文缺失：" + kw)
+            self.assertIn(kw, coord, "附录协调员章节缺失：" + kw)
+            self.assertIn(kw, client, "附录客户端章节缺失：" + kw)
+        self.assertIn("五项", main, "正文缺失：五项")
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_merge_gate_protectskill.py

@@ -261,6 +261,10 @@ class MergeEndpointGateTestCase(unittest.TestCase):
         return task
 
     def test_no_gate_fields_behaves_like_old_endpoint(self):
+        # 本用例测的是 record 模式的过渡语义（#209 原意）：默认已是
+        # hard，见 test_merge_gate_hard.py 的默认 409 用例；这里显式
+        # 切 record 才沿用“无证据放行、只附带门禁信息”。
+        os.environ["CLOUD_MERGE_GATE_MODE"] = "record"
         task = self._completed_task()
         status, merged = self._req(
             "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key)
@@ -283,6 +287,8 @@ class MergeEndpointGateTestCase(unittest.TestCase):
         self.assertEqual(got["status"], "completed")
 
     def test_unprotected_confirm_allows_and_audits_risk(self):
+        # record 模式语义：硬模式下未保护 + 确认仍阻断，见硬门禁测试。
+        os.environ["CLOUD_MERGE_GATE_MODE"] = "record"
         task = self._completed_task()
         status, body = self._req(
             "POST", "/tasks/%s/merge" % task["id"],

cloud/tests/test_server_merge.py

@@ -4,6 +4,10 @@
 （本地或 Docker 容器内）执行，成功后再调本接口把云端
 任务从 completed 改为 merged。规则：404/401/403/409
 语义、maintainer 以上才能合并、审计里有 merge_task。
+门禁注意：默认模式已是 hard（见 test_merge_gate_hard.py），
+所以本文件里期望 200 的合并调用必须带完整硬门禁证据；
+无证据调用在默认下 409（本文件末尾两项新用例钉住它），
+record 下的旧行为由硬门禁测试文件覆盖。
 """
 
 import http.client
@@ -26,9 +30,28 @@ except ImportError:
 
 
 class MergeTestCase(unittest.TestCase):
+    def _hard_body(self):
+        return {
+            "files_changed": ["apps/web/a.py"],
+            "owned_dir": "apps/web",
+            "branch_protection": "protected",
+            "ci_status": "success",
+            "ci_checks": [{"name": "ci", "status": "success"}],
+            "required_checks": ["ci"],
+            "contract_version": "v2",
+            "contract_frozen": True,
+            "coordinator_approved": True,
+            "unresolved_review_comments": 0,
+        }
+
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
         self.db_path = os.path.join(self.tmp.name, "control.db")
+        self._saved_env = {k: os.environ.get(k) for k in
+                           ("CLOUD_MERGE_GATE_MODE",
+                            "CLOUD_BRANCH_PROTECTION")}
+        os.environ.pop("CLOUD_MERGE_GATE_MODE", None)
+        os.environ.pop("CLOUD_BRANCH_PROTECTION", None)
         self.auth = AuthStore(self.db_path)
         self.owner_id, self.owner_key = self.auth.create_user("owner")
         self.maint_id, self.maint_key = self.auth.create_user("maint")
@@ -50,6 +73,11 @@ class MergeTestCase(unittest.TestCase):
         self.server.shutdown()
         self.server.server_close()
         self.thread.join(timeout=5)
+        for key, value in self._saved_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
         try:
             self.auth.conn.close()
         except Exception:
@@ -97,7 +125,8 @@ class MergeTestCase(unittest.TestCase):
     def test_merge_completed_by_owner(self):
         task = self._completed_task()
         status, merged = self._req(
-            "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key)
+            "POST", "/tasks/%s/merge" % task["id"],
+            self._hard_body(), self.owner_key)
         self.assertEqual(status, 200)
         self.assertEqual(merged["status"], "merged")
         status, got = self._req(
@@ -112,7 +141,8 @@ class MergeTestCase(unittest.TestCase):
             "POST", "/tasks/%s/merge" % task["id"], {}, self.contrib_key)
         self.assertEqual(status, 403)
         status, merged = self._req(
-            "POST", "/tasks/%s/merge" % task["id"], {}, self.maint_key)
+            "POST", "/tasks/%s/merge" % task["id"],
+            self._hard_body(), self.maint_key)
         self.assertEqual(status, 200)
         self.assertEqual(merged["status"], "merged")
 
@@ -128,13 +158,32 @@ class MergeTestCase(unittest.TestCase):
         self.assertEqual(body["status"], "queued")
         done = self._completed_task()
         status, _ = self._req(
-            "POST", "/tasks/%s/merge" % done["id"], {}, self.owner_key)
+            "POST", "/tasks/%s/merge" % done["id"],
+            self._hard_body(), self.owner_key)
         self.assertEqual(status, 200)
         status, body = self._req(
             "POST", "/tasks/%s/merge" % done["id"], {}, self.owner_key)
         self.assertEqual(status, 409)
         self.assertEqual(body["status"], "merged")
 
+    def test_merge_default_hard_blocks_no_evidence_call(self):
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertEqual(body["merge_gate"]["mode"], "hard")
+        status, got = self._req(
+            "GET", "/tasks/%s" % task["id"], key=self.owner_key)
+        self.assertEqual(got["status"], "completed")
+
+    def test_merge_record_env_preserves_old_no_evidence_call(self):
+        os.environ["CLOUD_MERGE_GATE_MODE"] = "record"
+        task = self._completed_task()
+        status, merged = self._req(
+            "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(merged["status"], "merged")
+
     def test_merge_missing_and_no_key(self):
         status, _ = self._req(
             "POST", "/tasks/nope/merge", {}, self.owner_key)