Merge pull request #6036 from loopx-project/codex/installed-canary-repository-root-r144

fix(canary): use the caller checkout for installed validation

Author: huangruiteng

docs/development/testing-and-quality.md

@@ -526,6 +526,28 @@ Canary 使用启动 LoopX 的 `sys.executable` 执行 Python 检查，显示的
 目标 worktree。已有兼容虚拟环境也可用 `python -m pip install -e ".[test]"` 安装源码，
 随后直接使用该环境的命令。
 
+Canary is a repository development tool even when its CLI comes from a wheel.
+Run `premerge`, `run` and `smoke-suite` inside the intended source checkout
+(a subdirectory is also supported):
+the catalog, discovered examples, subprocess cwd and tracked-write guard share
+that checkout root. `canary premerge --catalog PATH` overrides the catalog only;
+it does not change the execution root or authorize commands outside its examples.
+Standalone read-only catalog commands retain their default catalog resolution;
+use their existing `--catalog PATH` when invoking an installed CLI.
+The wheel does not bundle the repository's docs/examples. Missing development
+inputs fail with a checkout/catalog recovery instruction, rather than an empty
+successful validation. `--no-execute` still previews; required CQR verification
+and merge authority remain separate gates.
+
+即使用 wheel 的 CLI，Canary 仍是仓库开发工具。`premerge`、`run`、`smoke-suite`
+在目标完整源码 checkout
+（也支持子目录）后运行；catalog、example 发现、子进程 cwd 与 tracked 写入保护
+共用该仓库根。`canary premerge --catalog PATH` 仅指定 catalog，不切执行根，也不
+授权执行 examples 之外的命令。独立的只读 catalog 命令保留既有默认解析，安装态
+可用其现有 `--catalog PATH` 指定输入。wheel 不携带仓库 docs/examples；缺少输入时明确
+提示补齐 checkout/catalog，不会以空检查冒充成功。`--no-execute` 仍是预览，
+required CQR 校验与合并权限继续分别判断。
+
 The repository does not currently track `.python-version` or `uv.lock`. `uv`
 creates a local lockfile during resolution; keep that generated file out of
 unrelated PRs. Introducing a shared lock or interpreter pin is a separate

examples/canary/catalog-run-e2e-smoke.py

@@ -114,6 +114,7 @@ def assert_git_required_smoke_skips_without_git_worktree() -> None:
         runner_module.REPO_ROOT = temp_root
         try:
             payload = build_catalog_canary_run(
+                catalog_path=REPO_ROOT / "docs/concepts/interaction-pattern-catalog.md",
                 profiles=["repo-architecture-budget"],
                 max_checks_per_profile=1,
                 check_limit=1,

examples/canary/premerge-validation-gate-smoke.py

@@ -424,6 +424,8 @@ def assert_external_dirty_worktree_uses_caller_repo() -> None:
                 "--from-git-diff",
                 "--git-diff-base",
                 "HEAD",
+                "--catalog",
+                str(REPO_ROOT / "docs/concepts/interaction-pattern-catalog.md"),
                 "--tier",
                 "quick",
                 "--no-progress",
@@ -465,6 +467,8 @@ def assert_external_dirty_worktree_uses_caller_repo() -> None:
                 "--from-git-diff",
                 "--git-diff-base",
                 "HEAD",
+                "--catalog",
+                str(REPO_ROOT / "docs/concepts/interaction-pattern-catalog.md"),
                 "--tier",
                 "quick",
                 "--no-progress",

examples/canary/smoke-suite-runner-smoke.py

@@ -425,7 +425,8 @@ def assert_parallel_jobs_execute_and_preserve_report_order() -> None:
     original_tracked_change_paths = canary_runner._tracked_change_paths
     events: list[dict[str, object]] = []
 
-    def fake_tracked_change_paths() -> tuple[bool, list[str], str]:
+    def fake_tracked_change_paths(repo_root: Path | None = None) -> tuple[bool, list[str], str]:
+        assert repo_root == REPO_ROOT.resolve(), repo_root
         return True, [], ""
 
     try:
@@ -473,7 +474,8 @@ def assert_parallel_jobs_keep_marked_smokes_serial() -> None:
     original_tracked_change_paths = canary_runner._tracked_change_paths
     observed: list[tuple[str, int | None]] = []
 
-    def fake_tracked_change_paths() -> tuple[bool, list[str], str]:
+    def fake_tracked_change_paths(repo_root: Path | None = None) -> tuple[bool, list[str], str]:
+        assert repo_root == REPO_ROOT.resolve(), repo_root
         return True, [], ""
 
     def fake_run_check(
@@ -482,8 +484,9 @@ def fake_run_check(
         timeout_seconds: float,
         check_index: int | None = None,
         check_count: int | None = None,
+        repo_root: Path | None = None,
     ) -> dict[str, object]:
-        normalized = canary_runner.normalize_canary_command(str(check.get("command") or ""))
+        normalized = canary_runner.normalize_canary_command(str(check.get("command") or ""), repo_root=repo_root)
         observed.append((str(normalized.get("script") or ""), check_index))
         return _fake_passed_check(
             check,
@@ -531,7 +534,8 @@ def assert_fail_fast_keeps_parallel_jobs_serial() -> None:
     original_run_check = canary_runner._run_check
     original_tracked_change_paths = canary_runner._tracked_change_paths
 
-    def fake_tracked_change_paths() -> tuple[bool, list[str], str]:
+    def fake_tracked_change_paths(repo_root: Path | None = None) -> tuple[bool, list[str], str]:
+        assert repo_root == REPO_ROOT.resolve(), repo_root
         return True, [], ""
 
     try:
@@ -572,8 +576,9 @@ def _fake_passed_check(
     timeout_seconds: float,
     check_index: int | None = None,
     check_count: int | None = None,
+    repo_root: Path | None = None,
 ) -> dict[str, object]:
-    normalized = canary_runner.normalize_canary_command(str(check.get("command") or ""))
+    normalized = canary_runner.normalize_canary_command(str(check.get("command") or ""), repo_root=repo_root)
     result: dict[str, object] = {
         **check,
         "normalized": normalized,
@@ -596,13 +601,14 @@ def assert_readonly_run_rejects_and_restores_tracked_side_effects() -> None:
     restored: list[list[str]] = []
     calls = {"tracked": 0}
 
-    def fake_tracked_change_paths() -> tuple[bool, list[str], str]:
+    def fake_tracked_change_paths(repo_root: Path | None = None) -> tuple[bool, list[str], str]:
+        assert repo_root == REPO_ROOT.resolve(), repo_root
         calls["tracked"] += 1
         if calls["tracked"] == 1 or restored:
             return True, [], ""
         return True, ["examples/generated-tracked-side-effect.txt"], ""
 
-    def fake_restore_tracked_paths(paths: list[str]) -> dict[str, object]:
+    def fake_restore_tracked_paths(paths: list[str], repo_root: Path | None = None) -> dict[str, object]:
         restored.append(paths)
         return {"ok": True, "restored_paths": paths}
 
@@ -644,7 +650,8 @@ def assert_unavailable_git_worktree_is_reported_explicitly() -> None:
     original_run_check = canary_runner._run_check
     original_tracked_change_paths = canary_runner._tracked_change_paths
 
-    def fake_tracked_change_paths() -> tuple[bool, list[str], str]:
+    def fake_tracked_change_paths(repo_root: Path | None = None) -> tuple[bool, list[str], str]:
+        assert repo_root == REPO_ROOT.resolve(), repo_root
         return False, [], "not_a_git_worktree: fixture"
 
     try:
@@ -680,7 +687,8 @@ def assert_tracked_side_effects_require_explicit_allow() -> None:
     original_tracked_change_paths = canary_runner._tracked_change_paths
     original_restore_tracked_paths = canary_runner._restore_tracked_paths
 
-    def fake_tracked_change_paths() -> tuple[bool, list[str], str]:
+    def fake_tracked_change_paths(repo_root: Path | None = None) -> tuple[bool, list[str], str]:
+        assert repo_root == REPO_ROOT.resolve(), repo_root
         return True, [], ""
 
     def fail_restore_tracked_paths(paths: list[str]) -> dict[str, object]:

loopx/canary/planner.py

@@ -1363,8 +1363,24 @@ def _split_semicolon(value: str) -> list[str]:
     return [item.strip() for item in value.split(";") if item.strip()]
 
 
+def resolve_canary_catalog_path(
+    catalog_path: Path | None = None, *, repo_root: Path | None = None,
+) -> Path:
+    """Resolve development inputs without treating an installed wheel as a checkout."""
+    path = (catalog_path or (
+        repo_root / "docs/concepts/interaction-pattern-catalog.md"
+        if repo_root is not None else DEFAULT_CATALOG_PATH
+    )).expanduser()
+    if not path.is_file():
+        raise ValueError(
+            f"canary catalog is missing: {path}; run from a complete LoopX source "
+            "checkout or supply --catalog PATH. The wheel does not bundle development inputs."
+        )
+    return path
+
+
 def _read_catalog(catalog_path: Path | None = None) -> tuple[Path, str]:
-    path = (catalog_path or DEFAULT_CATALOG_PATH).expanduser()
+    path = resolve_canary_catalog_path(catalog_path)
     return path, path.read_text(encoding="utf-8")
 
 
@@ -1831,7 +1847,8 @@ def render_catalog_canary_profiles_markdown(payload: dict[str, Any]) -> str:
 
 
 def render_catalog_canary_plan_markdown(payload: dict[str, Any]) -> str:
-    inputs = payload.get("selection_inputs") if isinstance(payload.get("selection_inputs"), dict) else {}
+    inputs_value = payload.get('selection_inputs')
+    inputs = inputs_value if isinstance(inputs_value, dict) else {}
     lines = [
         "# Catalog Canary Plan",
         "",

loopx/canary/premerge.py

@@ -7,7 +7,7 @@
 from typing import Any, Callable
 
 from ..contract import scan_public_boundary
-from .planner import REPO_ROOT
+from .planner import REPO_ROOT, resolve_canary_catalog_path
 from .runner import build_canary_smoke_suite_run
 
 
@@ -172,8 +172,8 @@ def _public_boundary_changed_files_run(
         except ValueError:
             display_paths.append(str(path))
     display_argv = ["loopx", "check"]
-    for path in display_paths:
-        display_argv.extend(["--scan-path", path])
+    for display_path in display_paths:
+        display_argv.extend(["--scan-path", display_path])
     check: dict[str, Any] = {
         "source": "premerge_public_boundary",
         "tier": "default",
@@ -621,11 +621,8 @@ def _gate_status(
 
 
 def _check_command(check: dict[str, Any]) -> str:
-    normalized = (
-        check.get("normalized")
-        if isinstance(check.get("normalized"), dict)
-        else {}
-    )
+    normalized_value = check.get('normalized')
+    normalized = normalized_value if isinstance(normalized_value, dict) else {}
     display_argv = normalized.get("display_argv")
     if isinstance(display_argv, list) and display_argv:
         return " ".join(str(part) for part in display_argv)
@@ -717,12 +714,10 @@ def apply_change_quality_verification(
         verification.get("enforcement_applied")
         and verification.get("ok") is False
     )
-    gate = payload.get("gate") if isinstance(payload.get("gate"), dict) else {}
-    summary = (
-        payload.get("validation_summary")
-        if isinstance(payload.get("validation_summary"), dict)
-        else {}
-    )
+    gate_value = payload.get('gate')
+    gate = gate_value if isinstance(gate_value, dict) else {}
+    summary_value = payload.get('validation_summary')
+    summary = summary_value if isinstance(summary_value, dict) else {}
     gate["quality_receipt_failure_count"] = 1 if enforced_failure else 0
     summary["policy_failure_count"] = 1 if enforced_failure else 0
     if enforced_failure:
@@ -830,9 +825,12 @@ def build_premerge_validation_gate(
     include_deep_checks: bool | None = None,
     progress_callback: ProgressCallback | None = None,
     repo_root: Path | None = None,
+    catalog_path: Path | None = None,
 ) -> dict[str, Any]:
     target_repo_root = (repo_root or REPO_ROOT).resolve()
     files = _dedupe(list(changed_files or []))
+    if files:
+        catalog_path = resolve_canary_catalog_path(catalog_path, repo_root=target_repo_root)
     classification = classify_premerge_surfaces(files, repo_root=target_repo_root)
     limits = _tier_limits(tier)
     include_deep = bool(limits["deep"] if include_deep_checks is None else include_deep_checks)
@@ -874,6 +872,7 @@ def build_premerge_validation_gate(
     if module_colocation is not None:
         direct_checks.append(module_colocation)
 
+    catalog_run: dict[str, Any]
     if files:
         catalog_progress = _section_progress_callback(
             progress_callback,
@@ -886,6 +885,8 @@ def build_premerge_validation_gate(
         )
         catalog_run = build_canary_smoke_suite_run(
             suite="catalog-plan",
+            repo_root=target_repo_root,
+            catalog_path=catalog_path,
             changed_files=files,
             include_deep_checks=include_deep,
             max_checks_per_family=3,
@@ -905,10 +906,12 @@ def build_premerge_validation_gate(
             executed_check_count=catalog_run.get("executed_check_count"),
             failure_count=catalog_run.get("failure_count"),
         )
-        catalog_run = downgrade_inherited_baseline_failures(
+        adjusted_catalog_run = downgrade_inherited_baseline_failures(
             catalog_run,
             changed_files=files,
         )
+        assert adjusted_catalog_run is not None  # A dictionary input retains a result.
+        catalog_run = adjusted_catalog_run
     else:
         catalog_run = _empty_smoke_suite_run(
             suite="catalog-plan",
@@ -940,6 +943,8 @@ def build_premerge_validation_gate(
         )
         risk_profile_run = build_canary_smoke_suite_run(
             suite="default-public",
+            repo_root=target_repo_root,
+            catalog_path=catalog_path,
             profiles=risk_profiles,
             include_deep_checks=include_deep,
             limit=profile_limit,
@@ -1094,12 +1099,10 @@ def _emit_section_progress(
 
 
 def render_premerge_validation_gate_markdown(payload: dict[str, Any]) -> str:
-    gate = payload.get("gate") if isinstance(payload.get("gate"), dict) else {}
-    classification = (
-        payload.get("classification")
-        if isinstance(payload.get("classification"), dict)
-        else {}
-    )
+    gate_value = payload.get('gate')
+    gate = gate_value if isinstance(gate_value, dict) else {}
+    classification_value = payload.get('classification')
+    classification = classification_value if isinstance(classification_value, dict) else {}
     lines = [
         "# Pre-Merge Validation Gate",
         "",
@@ -1152,11 +1155,8 @@ def append_run(title: str, run: Any) -> None:
         for check in run.get("selected_checks", [])[:12]:
             if not isinstance(check, dict):
                 continue
-            normalized = (
-                check.get("normalized")
-                if isinstance(check.get("normalized"), dict)
-                else {}
-            )
+            normalized_value = check.get('normalized')
+            normalized = normalized_value if isinstance(normalized_value, dict) else {}
             command = " ".join(str(part) for part in normalized.get("display_argv") or [])
             lines.append(f"- `{check.get('status') or 'ready'}` {command or check.get('command')}")
             if check.get("advisory_reason"):
@@ -1172,11 +1172,8 @@ def append_run(title: str, run: Any) -> None:
         else None
     )
     if quality is not None:
-        evidence = (
-            quality.get("pr_evidence")
-            if isinstance(quality.get("pr_evidence"), dict)
-            else {}
-        )
+        evidence_value = quality.get('pr_evidence')
+        evidence = evidence_value if isinstance(evidence_value, dict) else {}
         lines.extend(
             [
                 "",