Add operator UX diagnostics

scripts/ci_test.py

@@ -34,6 +34,7 @@ DEFAULT_TARGETS = [
     "architecture-maintainability-check",
     "yao-cli-check",
     "yao-cli-world-class-check",
+    "operator-ux-check",
     "skill-overview-check",
     "skill-interpretation-check",
     "skill-report-metrics-check",

scripts/yao.py

@@ -38,6 +38,11 @@ from yao_cli_output_commands import (
     command_output_review_import,
     command_output_review_kit,
 )
+from yao_cli_operator_commands import (
+    command_install_status,
+    command_localized_doc_sync_check,
+    command_pr_review_report,
+)
 from yao_cli_parser import build_parser as build_cli_parser
 from yao_cli_report_commands import (
     command_artifact_design_profile,

scripts/yao_cli_operator_commands.py

@@ -0,0 +1,424 @@
+"""Operator UX commands for installation, docs sync, and PR review evidence."""
+
+import argparse
+import json
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Any
+
+from yao_cli_runtime import ROOT
+
+
+SCRIPT_INTERFACE = "internal-module"
+SCRIPT_INTERFACE_REASON = "Imported by yao.py for operator-facing install, localized docs, and PR review diagnostics."
+
+
+DEFAULT_DOC_PAIRS = [
+    {
+        "key": "skill-os-2-upgrade",
+        "source": "## Skill OS 2.0 Upgrade",
+        "localized": "## Skill OS 2.0 升级",
+    },
+    {
+        "key": "from-1-to-2",
+        "source": "## From 1.0 to 2.0",
+        "localized": "## 从 1.0 到 2.0",
+    },
+    {
+        "key": "use-cases",
+        "source": "## 2.0 Use Cases",
+        "localized": "## 2.0 使用场景",
+    },
+    {
+        "key": "operator-ux",
+        "source": "## Operator UX Commands",
+        "localized": "## Operator UX 命令",
+    },
+    {
+        "key": "architecture",
+        "source": "## Architecture",
+        "localized": "## 架构图",
+    },
+    {
+        "key": "benchmark",
+        "source": "## Weighted Quality Benchmark",
+        "localized": "## 加权质量评测",
+    },
+    {
+        "key": "claim-boundary",
+        "source": 'Current posture: the repository is ready for beta and external testing',
+        "localized": "当前发布口径：仓库已经适合进入测试版和外部试用",
+    },
+]
+
+
+def _expand(path: str) -> Path:
+    return Path(path).expanduser().resolve()
+
+
+def _display(path: Path) -> str:
+    try:
+        return str(path.relative_to(Path.home()))
+    except ValueError:
+        return str(path)
+
+
+def _skill_path_status(root: Path, skill_name: str, expected_source: Path) -> dict[str, Any]:
+    path = root / skill_name
+    exists = path.exists()
+    resolved = path.resolve() if exists or path.is_symlink() else path
+    has_skill_md = (path / "SKILL.md").exists()
+    return {
+        "root": str(root),
+        "path": str(path),
+        "exists": exists,
+        "is_symlink": path.is_symlink(),
+        "resolved": str(resolved),
+        "has_skill_md": has_skill_md,
+        "points_to_expected_source": resolved == expected_source,
+    }
+
+
+def _recommend_install_status(codex: dict[str, Any], agents: dict[str, Any], disabled: dict[str, Any]) -> list[str]:
+    recommendations = []
+    if codex["has_skill_md"]:
+        recommendations.append("Codex/Cortex can discover this skill through .codex/skills; restart the app if the list looks stale.")
+    if agents["has_skill_md"]:
+        recommendations.append("The skill is also active under .agents/skills; expect a duplicate entry when this repo is open.")
+    if not codex["has_skill_md"] and not agents["has_skill_md"] and disabled["has_skill_md"]:
+        recommendations.append("Only the disabled mirror exists; run make sync-active-install or add a .codex/skills symlink to activate it.")
+    if not codex["has_skill_md"] and not agents["has_skill_md"] and not disabled["has_skill_md"]:
+        recommendations.append("No active or disabled install was found for this skill name.")
+    if codex["has_skill_md"] and not codex["points_to_expected_source"]:
+        recommendations.append("The .codex/skills entry points somewhere else; inspect the symlink before editing or syncing.")
+    if not recommendations:
+        recommendations.append("No action required.")
+    return recommendations
+
+
+def command_install_status(args: argparse.Namespace) -> int:
+    skill_name = args.skill_name
+    expected_source = _expand(args.expected_source)
+    codex_root = _expand(args.codex_root)
+    agents_root = _expand(args.agents_root)
+    disabled_root = _expand(args.disabled_root)
+
+    codex = _skill_path_status(codex_root, skill_name, expected_source)
+    agents = _skill_path_status(agents_root, skill_name, expected_source)
+    disabled = _skill_path_status(disabled_root, skill_name, expected_source)
+    active_locations = [name for name, item in (("codex", codex), ("agents", agents)) if item["has_skill_md"]]
+    report = {
+        "ok": True,
+        "skill_name": skill_name,
+        "expected_source": str(expected_source),
+        "summary": {
+            "codex_active": codex["has_skill_md"],
+            "agents_active": agents["has_skill_md"],
+            "disabled_mirror": disabled["has_skill_md"],
+            "active_location_count": len(active_locations),
+            "duplicate_active": len(active_locations) > 1,
+            "active_locations": active_locations,
+        },
+        "locations": {
+            "codex": codex,
+            "agents": agents,
+            "disabled": disabled,
+        },
+        "recommendations": _recommend_install_status(codex, agents, disabled),
+    }
+    print(json.dumps(report, ensure_ascii=False, indent=2))
+    return 0
+
+
+def _load_pairs(raw_pairs: list[str]) -> list[dict[str, str]]:
+    if not raw_pairs:
+        return DEFAULT_DOC_PAIRS
+    pairs = []
+    for raw in raw_pairs:
+        parts = raw.split("::")
+        if len(parts) != 3:
+            raise ValueError("--pair must use key::source-marker::localized-marker")
+        pairs.append({"key": parts[0], "source": parts[1], "localized": parts[2]})
+    return pairs
+
+
+def _write_json(path: str | None, payload: dict[str, Any]) -> None:
+    if not path:
+        return
+    target = _expand(path)
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def _write_docs_sync_md(path: str | None, payload: dict[str, Any]) -> None:
+    if not path:
+        return
+    target = _expand(path)
+    target.parent.mkdir(parents=True, exist_ok=True)
+    lines = [
+        "# Localized Docs Sync Check",
+        "",
+        f"- ok: `{str(payload['ok']).lower()}`",
+        f"- source: `{payload['source']}`",
+        f"- localized: `{payload['localized']}`",
+        f"- checked: `{payload['summary']['checked_count']}`",
+        f"- missing: `{payload['summary']['missing_count']}`",
+        f"- skipped: `{payload['summary']['skipped_count']}`",
+        "",
+        "## Pairs",
+        "",
+    ]
+    for item in payload["pairs"]:
+        lines.append(f"- `{item['key']}`: `{item['status']}`")
+    target.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def command_localized_doc_sync_check(args: argparse.Namespace) -> int:
+    source_path = _expand(args.source)
+    localized_path = _expand(args.localized)
+    try:
+        pairs = _load_pairs(args.pair)
+    except ValueError as exc:
+        report = {
+            "ok": False,
+            "source": str(source_path),
+            "localized": str(localized_path),
+            "summary": {"checked_count": 0, "missing_count": 0, "skipped_count": 0},
+            "pairs": [],
+            "missing": [],
+            "failures": [str(exc)],
+        }
+        _write_json(args.output_json, report)
+        _write_docs_sync_md(args.output_md, report)
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+        return 2
+    missing_paths = [str(path) for path in (source_path, localized_path) if not path.exists()]
+    if missing_paths:
+        report = {
+            "ok": False,
+            "source": str(source_path),
+            "localized": str(localized_path),
+            "summary": {"checked_count": 0, "missing_count": 0, "skipped_count": 0},
+            "pairs": [],
+            "missing": [],
+            "failures": [f"Missing docs file: {path}" for path in missing_paths],
+        }
+        _write_json(args.output_json, report)
+        _write_docs_sync_md(args.output_md, report)
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+        return 2
+    source_text = source_path.read_text(encoding="utf-8")
+    localized_text = localized_path.read_text(encoding="utf-8")
+    results = []
+    for pair in pairs:
+        source_present = pair["source"] in source_text
+        localized_present = pair["localized"] in localized_text
+        if not source_present:
+            status = "skipped-source-missing"
+        elif localized_present:
+            status = "pass"
+        else:
+            status = "missing-localized-marker"
+        results.append(
+            {
+                **pair,
+                "source_present": source_present,
+                "localized_present": localized_present,
+                "status": status,
+            }
+        )
+    missing = [item for item in results if item["status"] == "missing-localized-marker"]
+    checked = [item for item in results if item["source_present"]]
+    report = {
+        "ok": not missing,
+        "source": str(source_path),
+        "localized": str(localized_path),
+        "summary": {
+            "checked_count": len(checked),
+            "missing_count": len(missing),
+            "skipped_count": len(results) - len(checked),
+        },
+        "pairs": results,
+        "missing": missing,
+    }
+    _write_json(args.output_json, report)
+    _write_docs_sync_md(args.output_md, report)
+    print(json.dumps(report, ensure_ascii=False, indent=2))
+    return 0 if report["ok"] else 2
+
+
+def _run_gh(args: list[str]) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(["gh", *args], capture_output=True, text=True)
+
+
+def _gh_available() -> bool:
+    return shutil.which("gh") is not None
+
+
+def _review_depth(changed_files: int, changed_lines: int) -> str:
+    if changed_files <= 5 and changed_lines < 100:
+        return "quick"
+    if changed_files <= 10 and changed_lines < 500:
+        return "standard"
+    return "deep"
+
+
+def _check_summary(checks: list[dict[str, Any]]) -> dict[str, Any]:
+    pending = []
+    failed = []
+    passed = []
+    for item in checks:
+        name = str(item.get("name") or item.get("workflowName") or "check")
+        status = str(item.get("status") or "").upper()
+        conclusion = str(item.get("conclusion") or "").upper()
+        if status and status != "COMPLETED":
+            pending.append(name)
+        elif conclusion in {"SUCCESS", "SKIPPED", "NEUTRAL"}:
+            passed.append(name)
+        else:
+            failed.append(name)
+    return {
+        "present": bool(checks),
+        "passed_count": len(passed),
+        "pending_count": len(pending),
+        "failed_count": len(failed),
+        "passed": passed,
+        "pending": pending,
+        "failed": failed,
+    }
+
+
+def _pr_decision(view: dict[str, Any], checks: dict[str, Any], require_checks: bool) -> str:
+    if view.get("state") != "OPEN":
+        return "not-open"
+    if view.get("isDraft"):
+        return "draft"
+    if view.get("mergeable") not in {"MERGEABLE", "UNKNOWN"}:
+        return "not-mergeable"
+    if checks["failed_count"]:
+        return "fix-failing-checks"
+    if checks["pending_count"]:
+        return "wait-for-checks"
+    if require_checks and not checks["present"]:
+        return "checks-required"
+    if not checks["present"]:
+        return "local-verification-required"
+    return "mergeable-after-review"
+
+
+def _write_pr_review_md(path: str | None, payload: dict[str, Any]) -> None:
+    if not path:
+        return
+    target = _expand(path)
+    target.parent.mkdir(parents=True, exist_ok=True)
+    lines = [
+        "# PR Review Report",
+        "",
+        f"- ok: `{str(payload['ok']).lower()}`",
+        f"- PR: `{payload['pr']}`",
+        f"- repo: `{payload.get('repo') or 'current'}`",
+        f"- decision: `{payload['decision']}`",
+        f"- review depth: `{payload['review_depth']}`",
+        f"- changed files: `{payload['summary']['changed_files']}`",
+        f"- additions: `{payload['summary']['additions']}`",
+        f"- deletions: `{payload['summary']['deletions']}`",
+        f"- checks present: `{str(payload['checks']['present']).lower()}`",
+        "",
+        "## Files",
+        "",
+    ]
+    for path_item in payload["files"]:
+        lines.append(f"- `{path_item}`")
+    lines.extend(["", "## Suggested Commands", ""])
+    for command in payload["suggested_commands"]:
+        lines.append(f"- `{command}`")
+    target.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def command_pr_review_report(args: argparse.Namespace) -> int:
+    if not _gh_available():
+        report = {
+            "ok": False,
+            "pr": args.pr,
+            "repo": args.repo,
+            "decision": "missing-gh-cli",
+            "failures": ["GitHub CLI `gh` is not available on PATH."],
+        }
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+        return 2
+
+    fields = ",".join(
+        [
+            "number",
+            "title",
+            "state",
+            "isDraft",
+            "mergeable",
+            "author",
+            "baseRefName",
+            "headRefName",
+            "url",
+            "additions",
+            "deletions",
+            "changedFiles",
+            "commits",
+            "statusCheckRollup",
+            "reviewDecision",
+            "maintainerCanModify",
+        ]
+    )
+    view_args = ["pr", "view", args.pr, "--json", fields]
+    diff_args = ["pr", "diff", args.pr, "--name-only"]
+    if args.repo:
+        view_args.extend(["--repo", args.repo])
+        diff_args.extend(["--repo", args.repo])
+    view_proc = _run_gh(view_args)
+    if view_proc.returncode != 0:
+        report = {
+            "ok": False,
+            "pr": args.pr,
+            "repo": args.repo,
+            "decision": "gh-pr-view-failed",
+            "returncode": view_proc.returncode,
+            "stderr": view_proc.stderr.strip(),
+        }
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+        return 2
+    view = json.loads(view_proc.stdout)
+    diff_proc = _run_gh(diff_args)
+    files = [line.strip() for line in diff_proc.stdout.splitlines() if line.strip()] if diff_proc.returncode == 0 else []
+    checks = _check_summary(view.get("statusCheckRollup") or [])
+    changed_lines = int(view.get("additions") or 0) + int(view.get("deletions") or 0)
+    depth = _review_depth(int(view.get("changedFiles") or len(files)), changed_lines)
+    decision = _pr_decision(view, checks, args.require_checks)
+    report = {
+        "ok": decision not in {"missing-gh-cli", "gh-pr-view-failed", "checks-required", "fix-failing-checks", "not-mergeable", "draft", "not-open"},
+        "pr": str(view.get("number", args.pr)),
+        "repo": args.repo,
+        "url": view.get("url", ""),
+        "title": view.get("title", ""),
+        "state": view.get("state", ""),
+        "mergeable": view.get("mergeable", ""),
+        "decision": decision,
+        "review_depth": depth,
+        "summary": {
+            "changed_files": int(view.get("changedFiles") or len(files)),
+            "additions": int(view.get("additions") or 0),
+            "deletions": int(view.get("deletions") or 0),
+            "commit_count": len(view.get("commits") or []),
+            "maintainer_can_modify": bool(view.get("maintainerCanModify")),
+        },
+        "checks": checks,
+        "files": files,
+        "suggested_commands": [
+            f"gh pr view {args.pr}{' --repo ' + args.repo if args.repo else ''} --json number,title,state,mergeable,statusCheckRollup",
+            f"gh pr diff {args.pr}{' --repo ' + args.repo if args.repo else ''} --name-only",
+            f"git fetch origin pull/{args.pr}/head:refs/tmp/pr-{args.pr}",
+            f"git diff --check origin/main...refs/tmp/pr-{args.pr}",
+        ],
+    }
+    _write_json(args.output_json, report)
+    _write_pr_review_md(args.output_md, report)
+    print(json.dumps(report, ensure_ascii=False, indent=2))
+    return 0 if report["ok"] else 2

scripts/yao_cli_parser.py

@@ -5,6 +5,7 @@ import argparse
 from collections.abc import Callable
 
 from yao_cli_parser_evidence import add_evidence_commands
+from yao_cli_parser_operator import add_operator_commands
 from yao_cli_parser_operations import add_operating_loop_commands
 
 
@@ -299,6 +300,7 @@ def build_parser(command_handlers: dict[str, Callable[[argparse.Namespace], int]
     feedback_cmd.add_argument("--recommended-action", default="review")
     feedback_cmd.set_defaults(func=_handler(command_handlers, "command_feedback"))
 
+    add_operator_commands(subparsers, command_handlers)
     add_operating_loop_commands(subparsers, command_handlers)
 
     baseline_compare_cmd = subparsers.add_parser(

scripts/yao_cli_parser_operator.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+"""Operator UX command declarations for the Yao CLI parser."""
+
+import argparse
+from collections.abc import Callable
+
+
+SCRIPT_INTERFACE = "internal-module"
+SCRIPT_INTERFACE_REASON = "Imported by yao_cli_parser.py to keep operator UX command declarations out of the main parser module."
+
+
+def _handler(command_handlers: dict[str, Callable[[argparse.Namespace], int]], name: str) -> Callable[[argparse.Namespace], int]:
+    if name not in command_handlers:
+        raise KeyError(f"Missing CLI command handler: {name}")
+    return command_handlers[name]
+
+
+def add_operator_commands(
+    subparsers: argparse._SubParsersAction,
+    command_handlers: dict[str, Callable[[argparse.Namespace], int]],
+) -> None:
+    install_status_cmd = subparsers.add_parser(
+        "install-status",
+        help="Diagnose whether yao-meta-skill is active in Codex/Cortex, .agents, or a disabled mirror.",
+    )
+    install_status_cmd.add_argument("--skill-name", default="yao-meta-skill")
+    install_status_cmd.add_argument("--expected-source", default=".")
+    install_status_cmd.add_argument("--codex-root", default="~/.codex/skills")
+    install_status_cmd.add_argument("--agents-root", default="~/.agents/skills")
+    install_status_cmd.add_argument("--disabled-root", default="~/.agents/skills.disabled")
+    install_status_cmd.set_defaults(func=_handler(command_handlers, "command_install_status"))
+
+    localized_docs_cmd = subparsers.add_parser(
+        "localized-doc-sync-check",
+        help="Check that localized README docs contain the public homepage sections introduced in README.md.",
+    )
+    localized_docs_cmd.add_argument("--source", default="README.md")
+    localized_docs_cmd.add_argument("--localized", default="docs/README.zh-CN.md")
+    localized_docs_cmd.add_argument(
+        "--pair",
+        action="append",
+        default=[],
+        help="Custom sync marker in key::source-marker::localized-marker form. Replaces defaults when present.",
+    )
+    localized_docs_cmd.add_argument("--output-json")
+    localized_docs_cmd.add_argument("--output-md")
+    localized_docs_cmd.set_defaults(func=_handler(command_handlers, "command_localized_doc_sync_check"))
+
+    pr_review_cmd = subparsers.add_parser(
+        "pr-review-report",
+        help="Build a read-only GitHub PR review report with mergeability, checks, files, and suggested commands.",
+    )
+    pr_review_cmd.add_argument("pr")
+    pr_review_cmd.add_argument("--repo")
+    pr_review_cmd.add_argument("--require-checks", action="store_true")
+    pr_review_cmd.add_argument("--output-json")
+    pr_review_cmd.add_argument("--output-md")
+    pr_review_cmd.set_defaults(func=_handler(command_handlers, "command_pr_review_report"))