diff --git a/cloud/arch_conformance.py b/cloud/arch_conformance.py
new file mode 100644
index 0000000..5b91dca
--- /dev/null
+++ b/cloud/arch_conformance.py
@@ -0,0 +1,299 @@
+"""Architecture-conformance checks for the control plane and for
+skill-driven projects (software-engineering skill, G2 frozen artefacts).
+
+Zero third-party dependencies; standard library only. This module
+imports no ``cloud.*`` module, so it can never violate the layer bans
+it enforces, and it can be reused verbatim against a skill project's
+own directory tree (see ``check_skill_project``).
+
+The four checks, each with a stated rule and a stated reason:
+
+1. Contract version (``check_contract_version``). Rule mirrors the
+   skill's contract-freeze change protocol: after freeze, a version
+   change is allowed only with a coordinator-approved change record
+   that names backward compatibility, the affected clients, and (for
+   a breaking change) a regression task per affected client; the
+   version must advance by exactly one. Reason: a silent contract
+   drift is how parallel client threads diverge; the protocol is the
+   only sanctioned path, so the check enforces the protocol instead
+   of merely documenting it.
+
+2. Forbidden directories (``is_within`` / ``check_task_writes``).
+   Rule reuses the two directory definitions that already exist and
+   invents no third one: on the control plane a task's own directory
+   is its tenant work directory (``worker._tenant_dirs`` ``work``);
+   in a skill project a client thread's own directory is its
+   ``apps/<client>/`` tree. Any write outside the own directory --
+   including ``..`` escapes and symlink escapes, both resolved before
+   comparison -- is a violation. Reason: cross-directory writes are
+   how one tenant/client corrupts another's worktree.
+
+3. Forbidden imports (``check_layer_imports``). Bans were chosen by
+   auditing the current tree, not by wishful layering; every ban
+   below passes on the baseline, and the repo-level test pins that:
+
+   - Foundation modules (``db``, ``models``, ``queue``,
+     ``logging_config``) must not import orchestration/service
+     modules (``server``, ``worker``, ``pr_exec``, ``webhooks``,
+     ``autoscaler``, ``cost``). Reason: the foundation is imported
+     by everything; an upward import is a cycle waiting to happen
+     and drags service state into pure helpers.
+   - The execution module (``sandbox``) must not import persistence
+     or orchestration (``db``, ``server``, ``worker``, ``cost``,
+     ``auth``, ``pr``, ``pr_exec``, ``webhooks``). Reason: executors
+     must stay swappable and testable with fake runners; decisions
+     about mounts and networks arrive as arguments, not by reading
+     the database.
+   - No cloud module except the driver-adjacent ones (``db``,
+     ``backup``, ``migrations``) may open a database connection
+     directly (``sqlite3.connect`` or any ``psycopg*`` import).
+     Service modules must obtain connections through
+     ``cloud.db.connect`` or a Store. Reason: the backend switch
+     (sqlite/postgres) lives in exactly one place; a direct driver
+     call bypasses it. Deliberately NOT banned: ``server.py``'s use
+     of ``sqlite3.Row`` / ``sqlite3.IntegrityError`` as type adapters
+     on connections obtained from ``cloud.db`` -- that is adaptation,
+     not a bypass, and banning it would fail the baseline for no
+     safety gain. The test asserts the narrow form (no ``connect``
+     call, no driver import) rather than the broad form.
+
+   Only static ``ast`` imports are seen; dynamic imports
+   (``importlib``) are out of scope and are noted, not claimed.
+
+4. Frozen stack (``check_frozen_stack``). Rule mirrors the skill's
+   G2 rule "frozen stack must not be swapped mid-flight": a client
+   whose stack entry changes, or disappears, after freeze fails
+   unless an approved change record covers it; adding a brand-new
+   client (no frozen entry) is an addition, not a swap, and passes.
+   Reason: the shared core is built against the frozen table; a
+   mid-flight swap silently invalidates it.
+
+All checkers return plain data (dicts/lists), never raise for a
+violation, so CI can report every violation in one run instead of
+stopping at the first.
+"""
+
+import ast
+import os
+from pathlib import Path
+
+FOUNDATION_MODULES = frozenset({"db", "models", "queue", "logging_config"})
+ORCHESTRATION_MODULES = frozenset(
+    {"server", "worker", "pr_exec", "webhooks", "autoscaler", "cost"})
+EXECUTION_MODULES = frozenset({"sandbox"})
+EXECUTION_BANNED = frozenset(
+    {"db", "server", "worker", "cost", "auth", "pr", "pr_exec", "webhooks"})
+DRIVER_ADJACENT_MODULES = frozenset({"db", "backup", "migrations"})
+
+
+# ---------------------------------------------------------------- contract
+
+def parse_version(text):
+    """Parse ``v1`` / ``1`` to int; anything else is None (unparseable
+    is a failure, never silently treated as the frozen version)."""
+    if text is None:
+        return None
+    text = str(text).strip()
+    if text.startswith("v"):
+        text = text[1:]
+    if text.isdigit():
+        return int(text)
+    return None
+
+
+def _change_record_problems(change, breaking):
+    """Missing pieces of a contract-change record, per the protocol."""
+    problems = []
+    if not isinstance(change, dict):
+        return ["change record missing (frozen contract changed without "
+                "a coordinator-approved change record)"]
+    if not change.get("approved"):
+        problems.append("change not coordinator-approved")
+    if change.get("backward_compatible") is None:
+        problems.append("backward compatibility not declared")
+    affected = change.get("affected_clients")
+    if not affected or not isinstance(affected, list):
+        problems.append("affected clients not listed")
+    if breaking and not change.get("regression_task"):
+        problems.append("breaking change without a regression task")
+    return problems
+
+
+def check_contract_version(frozen, current, change=None):
+    """Return ``{"ok": bool, "reasons": [...]}``.
+
+    Equal versions pass regardless of ``change``. A different version
+    passes only when it is exactly frozen + 1 and ``change`` satisfies
+    the protocol; a breaking change is one where
+    ``backward_compatible`` is False.
+    """
+    frozen_n = parse_version(frozen)
+    current_n = parse_version(current)
+    if frozen_n is None or current_n is None:
+        return {"ok": False,
+                "reasons": ["unparseable contract version: frozen=%r "
+                            "current=%r" % (frozen, current)]}
+    if current_n == frozen_n:
+        return {"ok": True, "reasons": []}
+    reasons = []
+    if current_n != frozen_n + 1:
+        reasons.append("contract version must advance by exactly one "
+                       "(frozen v%d -> v%d, not v%d)"
+                       % (frozen_n, frozen_n + 1, current_n))
+    breaking = isinstance(change, dict) and \
+        change.get("backward_compatible") is False
+    reasons.extend(_change_record_problems(change, breaking))
+    return {"ok": not reasons, "reasons": reasons}
+
+
+# ------------------------------------------------------------- frozen stack
+
+def check_frozen_stack(frozen, current, change=None):
+    """``frozen``/``current`` map client -> stack string.
+
+    Changed or removed entries fail without an approved change record
+    (same shape as the contract-change record, minus the version
+    rules); added clients pass as additions.
+    """
+    frozen = frozen or {}
+    current = current or {}
+    reasons = []
+    swapped = sorted(c for c in frozen
+                     if c not in current or current[c] != frozen[c])
+    if not swapped:
+        return {"ok": True, "reasons": [], "swapped": []}
+    problems = _change_record_problems(change, breaking=False)
+    if problems:
+        reasons.append("frozen stack changed for %s without an approved "
+                       "change record: %s" % (swapped, "; ".join(problems)))
+    return {"ok": not reasons, "reasons": reasons, "swapped": swapped}
+
+
+# ------------------------------------------------------------ directories
+
+def is_within(root, candidate):
+    """True iff ``candidate`` resolves inside ``root`` (symlinks and
+    ``..`` resolved first, so both escape routes are caught)."""
+    try:
+        root_p = Path(root).resolve()
+        cand_p = Path(candidate).resolve()
+    except (OSError, RuntimeError):
+        return False
+    return cand_p == root_p or root_p in cand_p.parents
+
+
+def check_task_writes(own_dir, paths):
+    """Return the subset of ``paths`` written outside ``own_dir``."""
+    return [p for p in paths if not is_within(own_dir, p)]
+
+
+# ----------------------------------------------------------------- imports
+
+def scan_imports(source):
+    """Static import scan of one module's source.
+
+    Returns ``{"cloud": set, "driver_connect": bool, "driver_import": bool}``:
+    ``cloud`` holds the ``cloud.<name>`` top-level names imported;
+    ``driver_connect`` is True on any ``sqlite3.connect`` call;
+    ``driver_import`` is True on any ``psycopg*`` import. A bare
+    ``import sqlite3`` (Row/exception adapters) sets neither flag.
+    """
+    cloud_imports = set()
+    driver_connect = False
+    driver_import = False
+    try:
+        tree = ast.parse(source)
+    except SyntaxError:
+        return {"cloud": cloud_imports,
+                "driver_connect": driver_connect,
+                "driver_import": driver_import,
+                "parse_error": True}
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                if alias.name == "sqlite3":
+                    continue
+                if alias.name.startswith("psycopg"):
+                    driver_import = True
+                if alias.name == "cloud" or alias.name.startswith("cloud."):
+                    parts = alias.name.split(".")
+                    if len(parts) > 1:
+                        cloud_imports.add(parts[1])
+        elif isinstance(node, ast.ImportFrom):
+            if node.module and node.module.startswith("psycopg"):
+                driver_import = True
+            if node.module == "cloud" or (node.module or "").startswith("cloud."):
+                if node.module == "cloud":
+                    for alias in node.names:
+                        cloud_imports.add(alias.name)
+                else:
+                    cloud_imports.add(node.module.split(".")[1])
+        elif isinstance(node, ast.Call):
+            func = node.func
+            if isinstance(func, ast.Attribute) and func.attr == "connect":
+                value = func.value
+                if isinstance(value, ast.Name) and value.id == "sqlite3":
+                    driver_connect = True
+    return {"cloud": cloud_imports,
+            "driver_connect": driver_connect,
+            "driver_import": driver_import}
+
+
+def check_layer_imports(sources):
+    """``sources`` maps module name -> source text. Returns a list of
+    violation dicts; empty means the layer rules hold."""
+    violations = []
+    for module, source in sorted(sources.items()):
+        scanned = scan_imports(source)
+        if scanned.get("parse_error"):
+            violations.append({"module": module, "rule": "parse",
+                               "detail": "source does not parse; "
+                               "imports cannot be verified"})
+            continue
+        if module in FOUNDATION_MODULES:
+            bad = scanned["cloud"] & ORCHESTRATION_MODULES
+            if bad:
+                violations.append({"module": module,
+                                   "rule": "foundation_must_not_import_orchestration",
+                                   "detail": sorted(bad)})
+        if module in EXECUTION_MODULES:
+            bad = scanned["cloud"] & EXECUTION_BANNED
+            if bad:
+                violations.append({"module": module,
+                                   "rule": "execution_must_not_import_persistence_or_orchestration",
+                                   "detail": sorted(bad)})
+        if module not in DRIVER_ADJACENT_MODULES:
+            if scanned["driver_connect"] or scanned["driver_import"]:
+                violations.append({"module": module,
+                                   "rule": "service_must_not_open_driver_connection_directly",
+                                   "detail": "use cloud.db.connect or a Store"})
+    return violations
+
+
+def scan_cloud_package(repo_root):
+    """Run ``check_layer_imports`` over every ``cloud/*.py`` in a repo
+    checkout (tests and caches excluded). This is the repo-level test
+    that makes a layer violation fail CI."""
+    cloud_dir = Path(repo_root) / "cloud"
+    sources = {}
+    if cloud_dir.is_dir():
+        for path in sorted(cloud_dir.glob("*.py")):
+            sources[path.stem] = path.read_text(encoding="utf-8")
+    return check_layer_imports(sources)
+
+
+def check_skill_project(project_root, frozen_version, current_version,
+                        frozen_stack, current_stack, own_dir, writes,
+                        change=None):
+    """One-call conformance for a skill project: contract version,
+    frozen stack, and directory writes together. Returns a dict with
+    one entry per check plus an overall ``ok``."""
+    contract = check_contract_version(frozen_version, current_version,
+                                      change=change)
+    stack = check_frozen_stack(frozen_stack, current_stack, change=change)
+    outside = check_task_writes(own_dir, writes)
+    directory = {"ok": not outside,
+                 "reasons": (["writes outside own directory: %s" % outside]
+                             if outside else [])}
+    return {"ok": contract["ok"] and stack["ok"] and directory["ok"],
+            "contract": contract, "stack": stack, "directory": directory}
diff --git a/cloud/metrics.py b/cloud/metrics.py
new file mode 100644
index 0000000..dfbf09e
--- /dev/null
+++ b/cloud/metrics.py
@@ -0,0 +1,253 @@
+"""Metrics collection and feedback loop for agent-task quality.
+
+Zero third-party dependencies; standard library only. This module
+imports no ``cloud.*`` module: it computes over plain record dicts so
+the collection logic is testable without a database, and so the
+report shape is identical whether the records came from the control
+plane (``server._metrics_report``), a test fixture, or a future
+attempt ledger.
+
+Definitions (pinned here, not left to each caller):
+
+- A task counts toward a rate only when it reached a terminal status
+  (``completed`` / ``merged`` / ``failed`` / ``merge-failed``).
+  ``interrupted`` is deliberately NOT terminal: platform-interrupted
+  tasks auto-resume as fresh runs, so counting them as failures
+  would punish a platform event, not the agent.
+- First-pass green: terminal task whose ``attempts`` == 1 and whose
+  status is ``completed`` / ``merged``.
+- Defect: a task with ``defects`` > 0, where ``defects`` is the count
+  of failed CI/audit rounds or follow-up rounds for that task. The
+  current tasks schema has NO attempts/defects column, so a caller
+  that cannot supply them must pass ``None`` -- and the rates then
+  come back as ``None`` with ``*_known`` False. An unknown rate is
+  never reported as 0: 0 would claim "no defects", which is a
+  decision-grade lie.
+- Rework: terminal task with ``attempts`` > 1. Same None discipline.
+- Per-task cost reuses the #212 dual-column discipline from
+  ``cloud.cost``: priced USD micros and unpriced actual tokens are
+  reported in separate columns. ``total_usd_micros`` is ``None``
+  whenever any task's cost is incomplete (no cost rows, or unpriced
+  actual tokens > 0); the priced partial sum is still shown, labelled
+  ``priced_usd_micros_partial``, so nobody mistakes it for a total.
+  Unpriced cost is never folded into 0.
+
+Feedback (``evaluate_feedback``) is deliberately a pure function
+returning action records, not an auto-applier: an action says WHAT
+should change and WHY, with its evidence attached; applying it to
+routing or to the audit standards is a separate, approved step
+(see ``docs/arch-conformance-metrics.md``). The rules are testable:
+
+- defect rate for a category above threshold -> ``tighten_check``
+  for that category's check;
+- first-pass rate below its floor -> ``route_with_extra_audit``;
+- a check whose violations are overturned on review above
+  threshold -> ``recalibrate_check`` (review the threshold, do NOT
+  auto-apply a new one -- a check that fires wrongly must not
+  silently widen itself).
+
+Thresholds passed in are labelled by source in the report. The
+defaults in ``DEFAULT_THRESHOLDS`` are starting values for tests,
+NOT human-approved production thresholds: the report marks them
+``thresholds_source: "default_unapproved"`` until a human approves a
+set, and a report built on unapproved thresholds is still usable
+for collection -- it is just not decision-grade on its own.
+
+Golden tasks: a report is ``decision_ready`` only when a golden set
+has been human-approved (``approved_by`` + ``approved_at`` present
+and at least one golden task listed). Until then the golden section
+reads ``status: "pending_human_approval"`` and the report says so
+plainly; the sample-only set that exists today must never be
+presented as decision-ready. This module never invents golden tasks.
+"""
+
+import json
+import time
+
+TERMINAL_STATUSES = frozenset(
+    {"completed", "merged", "failed", "merge-failed"})
+SUCCESS_STATUSES = frozenset({"completed", "merged"})
+
+DEFAULT_THRESHOLDS = {
+    "defect_rate": 0.20,
+    "first_pass_rate_min": 0.50,
+    "check_overturn_rate": 0.30,
+}
+
+
+def _rate(numerator, denominator):
+    if denominator is None or denominator == 0:
+        return None
+    return numerator / denominator
+
+
+def compute_metrics(records, costs=None):
+    """``records``: list of dicts with keys ``id``, ``status`` and
+    optional ``attempts`` (int), ``defects`` (int),
+    ``defect_category`` (str). ``costs``: dict task_id -> dict with
+    ``priced_usd_micros`` (int|None), ``unpriced_actual_tokens`` (int),
+    ``has_rows`` (bool). Returns the metrics dict."""
+    costs = costs or {}
+    terminal = [r for r in records if r.get("status") in TERMINAL_STATUSES]
+    known_attempts = [r for r in terminal if r.get("attempts") is not None]
+    first_pass = [r for r in known_attempts
+                  if r["attempts"] == 1
+                  and r.get("status") in SUCCESS_STATUSES]
+    rework = [r for r in known_attempts if r["attempts"] > 1]
+    known_defects = [r for r in terminal if r.get("defects") is not None]
+    defective = [r for r in known_defects if r["defects"] > 0]
+    defect_by_category = {}
+    for r in defective:
+        cat = r.get("defect_category") or "uncategorized"
+        defect_by_category[cat] = defect_by_category.get(cat, 0) + 1
+    attempts_total = sum(r["attempts"] for r in known_attempts)
+
+    per_task_cost = []
+    priced_partial = 0
+    cost_complete = True
+    for r in terminal:
+        c = costs.get(r["id"]) or {}
+        priced = c.get("priced_usd_micros")
+        unpriced = int(c.get("unpriced_actual_tokens") or 0)
+        has_rows = bool(c.get("has_rows"))
+        complete = has_rows and unpriced == 0 and priced is not None
+        if not complete:
+            cost_complete = False
+        if priced is not None:
+            priced_partial += int(priced)
+        per_task_cost.append({
+            "task_id": r["id"],
+            "priced_usd_micros": priced,
+            "unpriced_actual_tokens": unpriced,
+            "cost_complete": complete,
+        })
+    if not terminal:
+        cost_complete = False
+
+    return {
+        "tasks_total": len(records),
+        "terminal_tasks": len(terminal),
+        "first_pass_tasks": len(first_pass),
+        "first_pass_rate": _rate(len(first_pass), len(known_attempts)),
+        "first_pass_known": bool(known_attempts),
+        "rework_tasks": len(rework),
+        "rework_rate": _rate(len(rework), len(known_attempts)),
+        "rework_known": bool(known_attempts),
+        "avg_attempts": (attempts_total / len(known_attempts)
+                         if known_attempts else None),
+        "defect_tasks": len(defective),
+        "defect_rate": _rate(len(defective), len(known_defects)),
+        "defect_known": bool(known_defects),
+        "defect_by_category": defect_by_category,
+        "per_task_cost": per_task_cost,
+        # Dual-column total: None unless every terminal task's cost
+        # is complete; the partial sum is shown separately, labelled.
+        "priced_usd_micros_partial": priced_partial,
+        "total_usd_micros": priced_partial if cost_complete else None,
+        "cost_complete": cost_complete,
+    }
+
+
+def evaluate_feedback(metrics, thresholds=None, check_stats=None):
+    """Testable feedback rules. Returns a list of action dicts; each
+    carries its evidence so a reviewer can audit why it fired."""
+    thresholds = thresholds or DEFAULT_THRESHOLDS
+    actions = []
+    defect_rate = metrics.get("defect_rate")
+    if metrics.get("defect_known") and defect_rate is not None:
+        limit = thresholds.get("defect_rate")
+        if limit is not None and defect_rate > limit:
+            for cat, count in sorted(
+                    metrics.get("defect_by_category", {}).items()):
+                actions.append({
+                    "kind": "tighten_check",
+                    "target": cat,
+                    "reason": "defect_rate %.3f > threshold %.3f "
+                              "(%d defective tasks in category)"
+                              % (defect_rate, limit, count),
+                    "evidence": {"defect_rate": defect_rate,
+                                 "threshold": limit,
+                                 "category_tasks": count},
+                    "auto_apply": False,
+                })
+    first_pass = metrics.get("first_pass_rate")
+    floor = thresholds.get("first_pass_rate_min")
+    if first_pass is not None and floor is not None and first_pass < floor:
+        actions.append({
+            "kind": "route_with_extra_audit",
+            "target": "all",
+            "reason": "first_pass_rate %.3f < floor %.3f"
+                      % (first_pass, floor),
+            "evidence": {"first_pass_rate": first_pass, "floor": floor},
+            "auto_apply": False,
+        })
+    overturn_limit = thresholds.get("check_overturn_rate")
+    for stat in check_stats or []:
+        violations = stat.get("violations") or 0
+        overturned = stat.get("overturned") or 0
+        if violations and overturn_limit is not None:
+            rate = overturned / violations
+            if rate > overturn_limit:
+                actions.append({
+                    "kind": "recalibrate_check",
+                    "target": stat.get("check_id"),
+                    "reason": "overturn_rate %.3f > threshold %.3f "
+                              "(%d/%d violations overturned on review)"
+                              % (rate, overturn_limit, overturned,
+                                 violations),
+                    "evidence": {"overturn_rate": rate,
+                                 "threshold": overturn_limit,
+                                 "violations": violations,
+                                 "overturned": overturned},
+                    "auto_apply": False,
+                })
+    return actions
+
+
+def build_report(metrics, golden=None, thresholds=None,
+                 thresholds_source="default_unapproved",
+                 check_stats=None):
+    """Assemble the JSON report document.
+
+    ``golden``: None, or a dict with ``tasks`` (list of task ids),
+    ``approved_by`` and ``approved_at``. Only a fully approved golden
+    set makes the report decision-ready; anything else is pending.
+    """
+    golden = golden or {}
+    tasks = golden.get("tasks") or []
+    approved = bool(tasks) and bool(golden.get("approved_by")) \
+        and bool(golden.get("approved_at"))
+    return {
+        "generated_at": time.time(),
+        "golden_set": {
+            "status": "approved" if approved else "pending_human_approval",
+            "tasks": list(tasks),
+            "approved_by": golden.get("approved_by"),
+            "approved_at": golden.get("approved_at"),
+        },
+        "decision_ready": approved,
+        "decision_note": ("golden set human-approved; metrics may inform "
+                          "decisions" if approved else
+                          "golden set NOT human-approved: this report "
+                          "collects metrics only and must not be used "
+                          "as a decision basis"),
+        "metrics": metrics,
+        "thresholds": dict(thresholds or DEFAULT_THRESHOLDS),
+        "thresholds_source": thresholds_source,
+        "feedback_actions": evaluate_feedback(
+            metrics, thresholds=thresholds, check_stats=check_stats),
+        # How the loop closes (see docs/arch-conformance-metrics.md):
+        # feedback actions are consumed by (a) routing -- tasks in a
+        # flagged category are routed with an extra audit gate, and
+        # (b) audit standards -- a tighten/recalibrate action opens a
+        # standards-review item; the standards document itself changes
+        # only through the contract-change protocol
+        # (arch_conformance.check_contract_version) with coordinator
+        # approval, never by auto-applying an action record.
+        "feedback_consumers": ["routing", "audit_standards"],
+    }
+
+
+def render_report(report):
+    """Canonical JSON rendering of a report document."""
+    return json.dumps(report, indent=2, sort_keys=True, default=str)
diff --git a/cloud/server.py b/cloud/server.py
index 8880e36..41a1614 100644
--- a/cloud/server.py
+++ b/cloud/server.py
@@ -158,6 +158,8 @@ from cloud.auto_memory import process_due_tasks as _auto_process_due
 from cloud.identity import IdentityStore
 from cloud.kb import KBStore
 from cloud.memory import MemoryStore
+from cloud.metrics import build_report as _build_metrics_report
+from cloud.metrics import compute_metrics as _compute_metrics
 from cloud.pr import PrSkipped, PrStepError
 from cloud.uploads import (
     BUNDLE_TTL_SECONDS,
@@ -3008,6 +3010,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             self._workers_heartbeat(user_id)
         elif path == "/metrics" and method == "GET":
             self._metrics(user_id)
+        elif path == "/metrics-report" and method == "GET":
+            self._metrics_report(user_id)
         elif path == "/models" and method == "GET":
             self._models_overview(user_id)
         elif path == "/relay/models" and method == "GET":
@@ -9629,6 +9633,58 @@ pid 须为整数、address 须为字符串，否则 400（注册表里的
             },
         })
 
+    def _metrics_report(self, user_id):
+        """指标闭环报告：首过绿灯率 / 缺陷率 / 返工率 / 每任务成本。
+
+        权限与 /metrics 同款：只统计调用者是成员的项目。
+        诚实边界：当前 tasks 表没有 attempts/defects 列，所以
+        attempts/defects 一律传 None，首过/缺陷/返工率回 None
+        （未知，不报 0）；金任务集未经人工审定，报告恒为
+        pending_human_approval、decision_ready False。等任务尝试
+        台账落库、金任务人工审定后，这两处再接真实输入——
+        在此之前绝不把样本集或未知率包装成可决策数字。
+        成本走 #212 双列口径：有价合计与无价 actual token 分开，
+        总额在成本不完整时为 None，不把无价折成 0。
+        """
+        conn = self._db()
+        try:
+            member_ids = {
+                row["project_id"]
+                for row in conn.execute(
+                    "SELECT project_id FROM project_members WHERE user_id = ?",
+                    (user_id,),
+                ).fetchall()
+            }
+            rows = conn.execute(
+                "SELECT id, project_id, status FROM tasks").fetchall()
+            records = [{"id": row["id"], "status": row["status"],
+                        "attempts": None, "defects": None}
+                       for row in rows
+                       if row["project_id"] in member_ids]
+            try:
+                cost_rows = conn.execute(
+                    "SELECT task_id, project_id, source, total_tokens,"
+                    " usd_micros FROM cost_records").fetchall()
+            except (sqlite3.OperationalError, _PgOperationalError):
+                cost_rows = []
+            costs = {}
+            for row in cost_rows:
+                if row["project_id"] not in member_ids:
+                    continue
+                entry = costs.setdefault(row["task_id"], {
+                    "priced_usd_micros": 0, "unpriced_actual_tokens": 0,
+                    "has_rows": True})
+                if row["source"] == "actual" \
+                        and row["usd_micros"] is None:
+                    entry["unpriced_actual_tokens"] += \
+                        row["total_tokens"] or 0
+                elif row["usd_micros"] is not None:
+                    entry["priced_usd_micros"] += row["usd_micros"]
+        finally:
+            conn.close()
+        metrics = _compute_metrics(records, costs)
+        self._send_json(200, _build_metrics_report(metrics))
+
     # -- OAuth（pre-auth，靠 state + secret，不走 X-API-Key）-----------
     def _oauth_provider(self, path):
         parts = [p for p in path.split("/") if p]
diff --git a/cloud/tests/test_arch_conformance.py b/cloud/tests/test_arch_conformance.py
new file mode 100644
index 0000000..f68240b
--- /dev/null
+++ b/cloud/tests/test_arch_conformance.py
@@ -0,0 +1,221 @@
+"""Tests for cloud.arch_conformance: architecture consistency checks.
+
+Every check has a positive and a negative case; the repo-level test
+pins that the current tree passes the layer bans, so a future
+violation fails CI (the existing host-verify workflow runs
+``discover -s cloud/tests``; a failure here is a CI failure).
+
+Only the standard library; no real model, no network, no secrets.
+"""
+
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(
+    os.path.dirname(os.path.abspath(__file__)))))
+
+from cloud.arch_conformance import (  # noqa: E402
+    check_contract_version, check_frozen_stack, check_layer_imports,
+    check_skill_project, check_task_writes, is_within,
+    scan_cloud_package, scan_imports)
+
+
+class TestContractVersion(unittest.TestCase):
+    def test_equal_versions_pass(self):
+        self.assertTrue(check_contract_version("v1", "v1")["ok"])
+
+    def test_approved_step_bump_passes(self):
+        change = {"approved": True, "backward_compatible": True,
+                  "affected_clients": ["web"]}
+        self.assertTrue(
+            check_contract_version("v1", "v2", change=change)["ok"])
+
+    def test_breaking_bump_needs_regression_task(self):
+        change = {"approved": True, "backward_compatible": False,
+                  "affected_clients": ["web"]}
+        verdict = check_contract_version("v1", "v2", change=change)
+        self.assertFalse(verdict["ok"])
+        self.assertTrue(any("regression" in r for r in verdict["reasons"]))
+
+    def test_breaking_bump_with_regression_task_passes(self):
+        change = {"approved": True, "backward_compatible": False,
+                  "affected_clients": ["web"], "regression_task": "t-1"}
+        self.assertTrue(
+            check_contract_version("v1", "v2", change=change)["ok"])
+
+    def test_unapproved_bump_fails(self):
+        change = {"approved": False, "backward_compatible": True,
+                  "affected_clients": ["web"]}
+        self.assertFalse(
+            check_contract_version("v1", "v2", change=change)["ok"])
+
+    def test_bump_without_change_record_fails(self):
+        self.assertFalse(check_contract_version("v1", "v2")["ok"])
+
+    def test_skip_bump_fails_even_when_approved(self):
+        change = {"approved": True, "backward_compatible": True,
+                  "affected_clients": ["web"]}
+        verdict = check_contract_version("v1", "v3", change=change)
+        self.assertFalse(verdict["ok"])
+
+    def test_unparseable_version_fails(self):
+        self.assertFalse(check_contract_version("v1", "draft")["ok"])
+
+
+class TestFrozenStack(unittest.TestCase):
+    def test_unchanged_stack_passes(self):
+        frozen = {"web": "React", "ios": "SwiftUI"}
+        self.assertTrue(check_frozen_stack(frozen, dict(frozen))["ok"])
+
+    def test_new_client_is_addition_not_swap(self):
+        frozen = {"web": "React"}
+        current = {"web": "React", "ios": "SwiftUI"}
+        self.assertTrue(check_frozen_stack(frozen, current)["ok"])
+
+    def test_swap_without_approval_fails(self):
+        verdict = check_frozen_stack({"web": "React"}, {"web": "Vue"})
+        self.assertFalse(verdict["ok"])
+        self.assertEqual(verdict["swapped"], ["web"])
+
+    def test_removed_client_without_approval_fails(self):
+        self.assertFalse(
+            check_frozen_stack({"web": "React", "ios": "SwiftUI"},
+                               {"web": "React"})["ok"])
+
+    def test_swap_with_approved_change_passes(self):
+        change = {"approved": True, "backward_compatible": True,
+                  "affected_clients": ["web"]}
+        self.assertTrue(
+            check_frozen_stack({"web": "React"}, {"web": "Vue"},
+                               change=change)["ok"])
+
+
+class TestDirectoryGuard(unittest.TestCase):
+    def test_inside_passes_and_outside_fails(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            own = os.path.join(tmp, "work")
+            os.makedirs(own)
+            self.assertTrue(is_within(own, os.path.join(own, "a.txt")))
+            self.assertTrue(is_within(own, own))
+            self.assertFalse(is_within(own, os.path.join(tmp, "other")))
+
+    def test_dotdot_escape_fails(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            own = os.path.join(tmp, "work")
+            os.makedirs(own)
+            escape = os.path.join(own, "..", "secret.txt")
+            self.assertFalse(is_within(own, escape))
+            self.assertEqual(check_task_writes(own, [escape]), [escape])
+
+    def test_symlink_escape_fails(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            own = os.path.join(tmp, "work")
+            outside = os.path.join(tmp, "outside")
+            os.makedirs(own)
+            os.makedirs(outside)
+            link = os.path.join(own, "link")
+            try:
+                os.symlink(outside, link)
+            except OSError:
+                self.skipTest("symlinks unavailable")
+            self.assertFalse(is_within(own, link))
+
+    def test_mixed_writes_report_only_violations(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            own = os.path.join(tmp, "work")
+            os.makedirs(own)
+            good = os.path.join(own, "ok.txt")
+            bad = os.path.join(tmp, "bad.txt")
+            self.assertEqual(check_task_writes(own, [good, bad]), [bad])
+
+
+class TestLayerImports(unittest.TestCase):
+    def test_foundation_importing_server_fails(self):
+        verdict = check_layer_imports(
+            {"db": "from cloud.server import foo\n"})
+        self.assertEqual(len(verdict), 1)
+        self.assertEqual(verdict[0]["rule"],
+                         "foundation_must_not_import_orchestration")
+
+    def test_foundation_importing_store_passes(self):
+        # db importing nothing, models importing nothing: the baseline
+        # shape. A foundation module importing another foundation
+        # module is not an upward import.
+        self.assertEqual(check_layer_imports(
+            {"models": "import os\n", "queue": "from cloud.models import x\n"}),
+            [])
+
+    def test_sandbox_importing_db_fails(self):
+        verdict = check_layer_imports(
+            {"sandbox": "from cloud.db import connect\n"})
+        self.assertEqual(len(verdict), 1)
+
+    def test_sandbox_stdlib_only_passes(self):
+        self.assertEqual(check_layer_imports(
+            {"sandbox": "import os\nimport subprocess\n"}), [])
+
+    def test_service_opening_sqlite_directly_fails(self):
+        verdict = check_layer_imports(
+            {"server": "import sqlite3\nconn = sqlite3.connect('x.db')\n"})
+        self.assertTrue(any(v["rule"] == "service_must_not_open_driver_"
+                            "connection_directly" for v in verdict))
+
+    def test_service_sqlite_row_adapter_passes(self):
+        # Deliberate narrow ban: Row/exception adapters on a
+        # cloud.db connection are adaptation, not a bypass.
+        self.assertEqual(check_layer_imports(
+            {"server": "import sqlite3\nconn.row_factory = sqlite3.Row\n"
+                       "try:\n    pass\nexcept sqlite3.IntegrityError:\n"
+                       "    pass\n"}), [])
+
+    def test_service_importing_psycopg_fails(self):
+        verdict = check_layer_imports(
+            {"worker": "import psycopg\n"})
+        self.assertTrue(any(v["rule"] == "service_must_not_open_driver_"
+                            "connection_directly" for v in verdict))
+
+    def test_driver_adjacent_modules_exempt(self):
+        self.assertEqual(check_layer_imports(
+            {"backup": "import sqlite3\nsqlite3.connect('x.db')\n",
+             "migrations": "import sqlite3\nsqlite3.connect('y.db')\n"}), [])
+
+    def test_scan_imports_sees_from_cloud_form(self):
+        scanned = scan_imports("from cloud.worker import Worker\n")
+        self.assertIn("worker", scanned["cloud"])
+
+
+class TestRepoLevelScan(unittest.TestCase):
+    def test_current_tree_has_zero_layer_violations(self):
+        repo_root = Path(__file__).resolve().parents[2]
+        violations = scan_cloud_package(repo_root)
+        self.assertEqual(violations, [])
+
+
+class TestSkillProjectFixture(unittest.TestCase):
+    def test_frozen_project_passes_all_checks(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            own = os.path.join(tmp, "apps", "web")
+            os.makedirs(own)
+            verdict = check_skill_project(
+                tmp, "v1", "v1", {"web": "React"}, {"web": "React"},
+                own, [os.path.join(own, "App.tsx")])
+            self.assertTrue(verdict["ok"])
+
+    def test_drifting_project_fails_each_changed_check(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            own = os.path.join(tmp, "apps", "web")
+            os.makedirs(own)
+            verdict = check_skill_project(
+                tmp, "v1", "v2", {"web": "React"}, {"web": "Vue"},
+                own, [os.path.join(tmp, "apps", "ios", "leak.txt")])
+            self.assertFalse(verdict["ok"])
+            self.assertFalse(verdict["contract"]["ok"])
+            self.assertFalse(verdict["stack"]["ok"])
+            self.assertFalse(verdict["directory"]["ok"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/cloud/tests/test_metrics_loop.py b/cloud/tests/test_metrics_loop.py
new file mode 100644
index 0000000..f1cf712
--- /dev/null
+++ b/cloud/tests/test_metrics_loop.py
@@ -0,0 +1,186 @@
+"""Tests for cloud.metrics: collection, dual-column cost, feedback.
+
+Every metric has a positive case and an unknown/edge case; unknown
+is never reported as zero, and an unapproved golden set never makes
+a report decision-ready.
+
+Only the standard library; no real model, no network, no secrets.
+"""
+
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(
+    os.path.dirname(os.path.abspath(__file__)))))
+
+from cloud.arch_conformance import check_contract_version  # noqa: E402
+from cloud.metrics import (  # noqa: E402
+    build_report, compute_metrics, evaluate_feedback)
+
+
+def _record(task_id, status, attempts=None, defects=None, category=None):
+    return {"id": task_id, "status": status, "attempts": attempts,
+            "defects": defects, "defect_category": category}
+
+
+class TestComputeMetrics(unittest.TestCase):
+    def test_first_pass_rework_defect_rates(self):
+        records = [
+            _record("a", "completed", attempts=1, defects=0),
+            _record("b", "merged", attempts=3, defects=2,
+                    category="contract"),
+            _record("c", "failed", attempts=2, defects=1,
+                    category="contract"),
+            _record("d", "queued", attempts=1),  # not terminal: excluded
+            _record("e", "interrupted", attempts=1),  # not terminal
+        ]
+        m = compute_metrics(records)
+        self.assertEqual(m["terminal_tasks"], 3)
+        self.assertAlmostEqual(m["first_pass_rate"], 1 / 3)
+        self.assertAlmostEqual(m["rework_rate"], 2 / 3)
+        self.assertAlmostEqual(m["defect_rate"], 2 / 3)
+        self.assertEqual(m["defect_by_category"], {"contract": 2})
+
+    def test_unknown_attempts_and_defects_are_none_not_zero(self):
+        records = [_record("a", "completed"),
+                   _record("b", "failed")]
+        m = compute_metrics(records)
+        self.assertIsNone(m["first_pass_rate"])
+        self.assertIsNone(m["rework_rate"])
+        self.assertIsNone(m["defect_rate"])
+        self.assertFalse(m["first_pass_known"])
+        self.assertFalse(m["rework_known"])
+        self.assertFalse(m["defect_known"])
+
+    def test_empty_set_rates_are_none(self):
+        m = compute_metrics([])
+        self.assertEqual(m["terminal_tasks"], 0)
+        self.assertIsNone(m["first_pass_rate"])
+        self.assertIsNone(m["defect_rate"])
+        self.assertIsNone(m["total_usd_micros"])
+        self.assertFalse(m["cost_complete"])
+
+
+class TestDualColumnCost(unittest.TestCase):
+    def test_complete_cost_totals(self):
+        records = [_record("a", "completed", attempts=1)]
+        costs = {"a": {"priced_usd_micros": 5000,
+                        "unpriced_actual_tokens": 0, "has_rows": True}}
+        m = compute_metrics(records, costs)
+        self.assertEqual(m["total_usd_micros"], 5000)
+        self.assertTrue(m["cost_complete"])
+
+    def test_unpriced_cost_never_folds_into_zero_or_total(self):
+        records = [_record("a", "completed", attempts=1)]
+        costs = {"a": {"priced_usd_micros": 5000,
+                        "unpriced_actual_tokens": 700, "has_rows": True}}
+        m = compute_metrics(records, costs)
+        self.assertIsNone(m["total_usd_micros"])
+        self.assertEqual(m["priced_usd_micros_partial"], 5000)
+        self.assertFalse(m["cost_complete"])
+        self.assertEqual(m["per_task_cost"][0]["unpriced_actual_tokens"],
+                         700)
+
+    def test_missing_cost_rows_make_total_unknown(self):
+        records = [_record("a", "completed", attempts=1)]
+        m = compute_metrics(records, costs={})
+        self.assertIsNone(m["total_usd_micros"])
+        self.assertFalse(m["cost_complete"])
+
+
+class TestFeedbackRules(unittest.TestCase):
+    def test_defect_breach_tightens_check(self):
+        m = compute_metrics([
+            _record("a", "completed", attempts=1),
+            _record("b", "failed", attempts=1, defects=1,
+                    category="contract"),
+        ])
+        actions = evaluate_feedback(
+            m, thresholds={"defect_rate": 0.20,
+                           "first_pass_rate_min": 0.50,
+                           "check_overturn_rate": 0.30})
+        tighten = [a for a in actions if a["kind"] == "tighten_check"]
+        self.assertEqual(len(tighten), 1)
+        self.assertEqual(tighten[0]["target"], "contract")
+        self.assertFalse(tighten[0]["auto_apply"])
+
+    def test_no_breach_no_action(self):
+        m = compute_metrics([_record("a", "completed", attempts=1)])
+        actions = evaluate_feedback(
+            m, thresholds={"defect_rate": 0.20,
+                           "first_pass_rate_min": 0.50,
+                           "check_overturn_rate": 0.30})
+        self.assertEqual(actions, [])
+
+    def test_low_first_pass_routes_with_extra_audit(self):
+        m = compute_metrics([
+            _record("a", "completed", attempts=3),
+            _record("b", "failed", attempts=2),
+        ])
+        actions = evaluate_feedback(
+            m, thresholds={"defect_rate": 0.20,
+                           "first_pass_rate_min": 0.50,
+                           "check_overturn_rate": 0.30})
+        self.assertTrue(any(a["kind"] == "route_with_extra_audit"
+                            for a in actions))
+
+    def test_high_overturn_recalibrates_without_auto_apply(self):
+        m = compute_metrics([_record("a", "completed", attempts=1)])
+        actions = evaluate_feedback(
+            m, thresholds={"defect_rate": 0.20,
+                           "first_pass_rate_min": 0.50,
+                           "check_overturn_rate": 0.30},
+            check_stats=[{"check_id": "forbidden-dir", "violations": 10,
+                          "overturned": 5}])
+        recal = [a for a in actions if a["kind"] == "recalibrate_check"]
+        self.assertEqual(len(recal), 1)
+        self.assertFalse(recal[0]["auto_apply"])
+
+    def test_low_overturn_no_recalibration(self):
+        m = compute_metrics([_record("a", "completed", attempts=1)])
+        actions = evaluate_feedback(
+            m, check_stats=[{"check_id": "forbidden-dir",
+                             "violations": 10, "overturned": 1}])
+        self.assertEqual(
+            [a for a in actions if a["kind"] == "recalibrate_check"], [])
+
+
+class TestGoldenSetAndStandardsLoop(unittest.TestCase):
+    def test_unapproved_golden_set_is_not_decision_ready(self):
+        m = compute_metrics([_record("a", "completed", attempts=1)])
+        report = build_report(m, golden={"tasks": ["a", "b"]})
+        self.assertFalse(report["decision_ready"])
+        self.assertEqual(report["golden_set"]["status"],
+                         "pending_human_approval")
+
+    def test_approved_golden_set_is_decision_ready(self):
+        m = compute_metrics([_record("a", "completed", attempts=1)])
+        report = build_report(
+            m, golden={"tasks": ["a"], "approved_by": "human",
+                       "approved_at": "2026-10-31"},
+            thresholds_source="human_approved")
+        self.assertTrue(report["decision_ready"])
+        self.assertEqual(report["thresholds_source"], "human_approved")
+
+    def test_sample_only_set_must_not_be_presented_as_golden(self):
+        # The set that exists today is sample-only: no approval fields
+        # exist for it, so any report built from it stays pending.
+        m = compute_metrics([_record("a", "completed", attempts=1)])
+        report = build_report(m, golden={"tasks": ["sample-1"]})
+        self.assertFalse(report["decision_ready"])
+        self.assertIn("NOT human-approved", report["decision_note"])
+
+    def test_standards_change_goes_through_contract_protocol(self):
+        # The loop closes through the same protocol: an audit-standard
+        # version bump without approval fails; with approval it passes.
+        self.assertFalse(
+            check_contract_version("v1", "v2")["ok"])
+        self.assertTrue(check_contract_version(
+            "v1", "v2",
+            change={"approved": True, "backward_compatible": True,
+                    "affected_clients": ["all"]})["ok"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/cloud/tests/test_metrics_report_endpoint.py b/cloud/tests/test_metrics_report_endpoint.py
new file mode 100644
index 0000000..32e7991
--- /dev/null
+++ b/cloud/tests/test_metrics_report_endpoint.py
@@ -0,0 +1,120 @@
+"""Endpoint test for GET /metrics-report.
+
+Only the standard library http.client; a temporary SQLite DB and a
+temporary port; no real model, no network, no secrets.
+
+Covers: unauthenticated -> 401, empty project -> report shape with
+unknown rates (None, not 0) and pending golden set, and a completed
+task with a priced cost row -> terminal count and dual-column cost.
+"""
+
+import http.client
+import json
+import os
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.cost import CostStore
+    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.cost import CostStore
+    from cloud.server import create_server
+
+
+class MetricsReportEndpointTestCase(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)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        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 test_unauthenticated_is_401(self):
+        status, _ = self._req("GET", "/metrics-report")
+        self.assertEqual(status, 401)
+
+    def test_empty_project_report_is_pending_and_unknown(self):
+        status, report = self._req(
+            "GET", "/metrics-report", key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertFalse(report["decision_ready"])
+        self.assertEqual(report["golden_set"]["status"],
+                         "pending_human_approval")
+        self.assertIsNone(report["metrics"]["first_pass_rate"])
+        self.assertIsNone(report["metrics"]["defect_rate"])
+        self.assertIsNone(report["metrics"]["total_usd_micros"])
+        self.assertEqual(report["thresholds_source"],
+                         "default_unapproved")
+
+    def test_completed_task_with_priced_cost_row(self):
+        from cloud.db import connect as db_connect
+        conn = db_connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, status) VALUES(?, ?, ?)",
+                ("t-1", self.project["id"], "completed"))
+            conn.commit()
+        finally:
+            conn.close()
+        store = CostStore(self.db_path)
+        store.record("t-1", self.project["id"], "gpt-6-sol",
+                     input_tokens=1000, output_tokens=100,
+                     usd_micros=5000, source="actual")
+        status, report = self._req(
+            "GET", "/metrics-report", key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(report["metrics"]["terminal_tasks"], 1)
+        # Attempts are not in the schema: first-pass stays unknown,
+        # it must not be fabricated as 0 or 1 from status alone.
+        self.assertIsNone(report["metrics"]["first_pass_rate"])
+        self.assertEqual(report["metrics"]["total_usd_micros"], 5000)
+        self.assertTrue(report["metrics"]["cost_complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/docs/arch-conformance-metrics.md b/docs/arch-conformance-metrics.md
new file mode 100644
index 0000000..d37951c
--- /dev/null
+++ b/docs/arch-conformance-metrics.md
@@ -0,0 +1,66 @@
+# 架构一致性进 CI 与指标闭环（审计加固第 5 步）
+
+实现：`cloud/arch_conformance.py`（纯检查，标准库）、`cloud/metrics.py`
+（纯计算，标准库）、端点 `GET /metrics-report`（`cloud/server.py`，
+与 `/metrics` 同款认证与成员范围）。测试失败即 CI 失败：现有
+`host-verify` 工作流已跑 `discover -s cloud/tests`，本轨未改任何
+workflow——第 2 轨拥有 lint/类型/安扫，本轨不碰。
+
+## 架构一致性检查（规则、理由、测试）
+
+| 检查 | 规则 | 理由 | 测试 |
+|---|---|---|---|
+| 合同版本 | 冻结后版本只能 +1，且须有协调员批准的变更记录（声明向后兼容、列出受影响客户端；破坏性变更须有回归任务） | 契约静默漂移是并行线程分叉的根源；协议是唯一合法路径，检查执行协议而非只记录协议 | `test_arch_conformance.py::TestContractVersion`（正向：等值/批准+1/破坏性+回归任务；负向：未批准、无记录、跳号、不可解析） |
+| 禁目录 | 任务只能写自有目录；`..` 与符号链接逃逸在解析后判定 | 跨目录写会 corrupt 他人 worktree | `TestDirectoryGuard`（正向：目录内；负向：`..`、symlink、混合写只报违规项）。口径复用已有定义，不另造：控制面 = 租户 work 目录（`worker._tenant_dirs` 的 `work`），技能项目 = `apps/<client>/` |
+| 禁 import | 基础层（db/models/queue/logging_config）不 import 编排层；执行层（sandbox）不 import 持久化/编排层；除 db/backup/migrations 外任何 cloud 模块不得直接 `sqlite3.connect` 或 import `psycopg*`（服务层走 `cloud.db.connect` 或 Store） | 基础层上引是循环依赖前兆；执行器须可换、可假 runner 测试；驱动直连绕过 sqlite/postgres 唯一分发点 | `TestLayerImports`（每项正向+负向）+ 仓库级 `TestRepoLevelScan`（当前主线零违规，未来违规即 CI 失败）。有意不禁：`server.py` 的 `sqlite3.Row`/异常类型适配（那是在 `cloud.db` 连接上的适配，不是绕过；禁它会在基线误报） |
+| 技术栈冻结 | G2 冻结表变更/删除须有批准的变更记录；新增客户端是新增不是换栈，通过 | 共享核心按冻结表构建，中途换栈使其静默失效 | `TestFrozenStack`（正向：不变/新增；负向：换栈/删除无批准） |
+
+技能项目一键版：`check_skill_project` 把合同版本、冻结栈、目录写
+三项合在一起验，`TestSkillProjectFixture` 用合成项目目录正向/负向
+各验一次。静态 AST 扫描看不到 `importlib` 动态导入——这是已知
+边界，不声称覆盖。
+
+## 指标采集与报告
+
+端点：`GET /metrics-report`（需 X-API-Key，成员范围）。报告格式
+见 `cloud/metrics.py::build_report`：`golden_set`、`decision_ready`、
+`metrics`、`thresholds` + `thresholds_source`、`feedback_actions`、
+`feedback_consumers`。
+
+指标口径（`compute_metrics`）：
+
+- 首过绿灯率 = 首过成功任务 / 已知 attempts 的终态任务；缺陷率 =
+  有缺陷任务 / 已知 defects 的终态任务；返工率 = attempts>1 任务 /
+  已知 attempts 的终态任务。终态 = completed/merged/failed/
+  merge-failed；`interrupted` 不算终态（平台中断自动重跑，算失败
+  是罚平台事件）。
+- 每任务成本复用 #212 双列口径：有价 USD micros 与无价 actual
+  token 分列；成本不完整（无成本行或有无价 token）时
+  `total_usd_micros` 为 `None`，有价部分只以
+  `priced_usd_micros_partial` 出现——无价永不记 0、永不折进总额。
+- 未知率回 `None` + `*_known: false`，绝不报 0。
+
+当前诚实边界（端点实测）：tasks 表无 attempts/defects 列，端点
+传 `None`，首过/缺陷/返工率恒为 `None`；金任务集未经人工审定，
+报告恒为 `pending_human_approval`、`decision_ready: false`。接真实
+输入需两项前置（见待审定清单），本轨不编造。
+
+## 回流与审计标准迭代（可测试规则，非口号）
+
+`evaluate_feedback` 是纯函数，只产出 `auto_apply: false` 的行动
+记录，不自动改路由、不自动改标准；行动由两处消费：
+
+- 路由：`tighten_check`（某类缺陷率 > 阈值）→ 该类任务路由时加
+  一道审计门；`route_with_extra_audit`（首过率 < 下限）→ 全量加
+  审计门。消费动作本身须经协调员/人工确认后才生效。
+- 审计标准：`tighten_check` / `recalibrate_check`（某检查的违规被
+  复核推翻率 > 阈值）各开一条标准复核项；标准文档本身的版本
+  变更走同一合同变更协议（`check_contract_version`，批准 +1），
+  测试 `TestGoldenSetAndStandardsLoop` 钉住“无批准改标准失败、
+  有批准通过”。误报复核规则：推翻率高只触发“复核阈值”行动，
+  绝不让检查自动放宽自己。
+
+默认阈值（`DEFAULT_THRESHOLDS`：缺陷率 0.20、首过下限 0.50、
+推翻率 0.30）只是测试用初值，报告标注
+`thresholds_source: "default_unapproved"`；生产阈值须人工拍板后
+以 `thresholds_source: "human_approved"` 传入，见待审定清单。