feat: add QSL workspace report plan

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

README.md

@@ -33,7 +33,7 @@ This repository is the central manifest source for QSL compatibility checks.
 
 - `qsl.toml`: current repo QSL metadata (`tier`, `upgrade_ring`, `bundle`).
 - `scripts/check_qsl_compat.py`: validate a repo's QSL compliance in `pyproject.toml`/`uv.lock` against central bundle.
-- `python/scripts/qslctl.py`: unified QSL version-control CLI for repo checks, workspace checks, and matrix generation.
+- `python/scripts/qslctl.py`: unified QSL version-control CLI for repo checks, workspace checks, ring reports/plans, and matrix generation.
 - `scripts/render_qsl_dependency_graph.py`: render dependency graph for review.
 - `compat/bundles/2026.07.0.toml`: first compatibility bundle (CalVer).
 - `compat/repo-tiers.toml`: repo tier and upgrade ring policy notes.

docs/qsl_version_control.md

@@ -17,6 +17,8 @@ Use `qslctl` for repository and workspace checks:
 ```bash
 python3 python/scripts/qslctl.py check --repo-root ../UsEquityStrategies
 python3 python/scripts/qslctl.py check-all --projects-root /Users/lisiyi/Projects
+python3 python/scripts/qslctl.py report --projects-root /Users/lisiyi/Projects
+python3 python/scripts/qslctl.py plan --projects-root /Users/lisiyi/Projects
 python3 python/scripts/qslctl.py generate-matrix --projects-root /Users/lisiyi/Projects --check
 python3 python/scripts/qslctl.py generate-matrix --projects-root /Users/lisiyi/Projects --sync
 ```
@@ -27,3 +29,9 @@ python3 python/scripts/qslctl.py generate-matrix --projects-root /Users/lisiyi/P
 2. Apply that bundle by ring: core → strategy libraries → pipelines/research → runtime platforms → ops/tooling.
 3. Regenerate `internal_dependency_matrix.json` after consumer pins are updated.
 4. Keep strict checks enabled only after a ring is converged.
+
+## Workspace report / plan
+
+- `qslctl report` is read-only. It groups the current workspace by ring, status, and bundle hotspot.
+- `qslctl plan` is read-only. It renders the ring-by-ring convergence order and highlights which repos should be fixed before the next ring starts.
+- Use `report` to answer “what is broken right now?” and `plan` to answer “what should we fix first?”

internal_dependency_matrix.json

@@ -90,7 +90,7 @@
       "path": "requirements.txt",
       "package": "us-equity-strategies",
       "source_repo": "UsEquityStrategies",
-      "ref": "67e1ed69d7b9c318f94c8809121d2b9481d6b11a"
+      "ref": "1643f37d723c4d0fdc475ce1030b3d911c07aacb"
     },
     {
       "consumer_repo": "InteractiveBrokersPlatform",

python/scripts/qslctl.py

@@ -13,6 +13,8 @@
 import json
 import subprocess
 import sys
+import tomllib
+from collections import Counter
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -51,6 +53,10 @@ class RepoCheckResult:
     warnings: list[str]
     notes: list[str]
     repo_root: str
+    bundle: str
+    tier: str
+    upgrade_ring: str
+    enforce_bundle: bool
 
 
 def _is_quant_repo(repo_dir: Path) -> bool:
@@ -82,18 +88,24 @@ def iter_qsl_repos(projects_root: Path) -> list[Path]:
 def check_repo(repo_root: Path, compat_root: Path) -> RepoCheckResult:
     try:
         ok, issues, warnings, notes = check_qsl_compat._check(repo_root=repo_root, compat_root=compat_root)
+        qsl_cfg = check_qsl_compat._load_qsl_config(repo_root)
     except (FileNotFoundError, ValueError, TypeError) as exc:
         ok = False
         issues = [str(exc)]
         warnings = []
         notes = []
+        qsl_cfg = {"bundle": "", "tier": "", "upgrade_ring": "", "enforce_bundle": False}
     return RepoCheckResult(
         repo=repo_root.name,
         ok=bool(ok),
         issues=list(issues),
         warnings=list(warnings),
         notes=list(notes),
         repo_root=str(repo_root),
+        bundle=str(qsl_cfg["bundle"]),
+        tier=str(qsl_cfg["tier"]),
+        upgrade_ring=str(qsl_cfg["upgrade_ring"]),
+        enforce_bundle=bool(qsl_cfg["enforce_bundle"]),
     )
 
 
@@ -109,6 +121,10 @@ def _result_payload(result: RepoCheckResult) -> dict[str, Any]:
         "warnings": result.warnings,
         "notes": result.notes,
         "repo_root": result.repo_root,
+        "bundle": result.bundle,
+        "tier": result.tier,
+        "upgrade_ring": result.upgrade_ring,
+        "enforce_bundle": result.enforce_bundle,
     }
 
 
@@ -124,6 +140,229 @@ def _summary_payload(results: list[RepoCheckResult]) -> dict[str, Any]:
     }
 
 
+def _load_repo_tiers(compat_root: Path) -> tuple[list[str], dict[str, str], dict[str, str]]:
+    with (compat_root / "compat" / "repo-tiers.toml").open("rb") as handle:
+        payload = tomllib.load(handle)
+    upgrade_rings = payload.get("upgrade_rings")
+    if not isinstance(upgrade_rings, dict) or not upgrade_rings:
+        raise ValueError(f"repo tier manifest missing [upgrade_rings] in {compat_root / 'compat' / 'repo-tiers.toml'}")
+
+    ring_order: list[str] = []
+    ring_to_tier: dict[str, str] = {}
+    for ring, tier in upgrade_rings.items():
+        if not isinstance(ring, str) or not ring.strip():
+            continue
+        if not isinstance(tier, str) or not tier.strip():
+            raise ValueError(f"repo tier manifest invalid tier for {ring}")
+        ring_name = ring.strip()
+        ring_order.append(ring_name)
+        ring_to_tier[ring_name] = tier.strip()
+    tier_to_ring = {tier: ring for ring, tier in ring_to_tier.items()}
+    return ring_order, ring_to_tier, tier_to_ring
+
+
+def _normalize_ring(value: str) -> str:
+    ring = value.strip().lower()
+    aliases = {
+        "0": "ring_a",
+        "ring_a": "ring_a",
+        "1": "ring_b",
+        "ring_b": "ring_b",
+        "2": "ring_c",
+        "ring_c": "ring_c",
+        "3": "ring_d",
+        "ring_d": "ring_d",
+        "4": "ring_e",
+        "ring_e": "ring_e",
+    }
+    return aliases.get(ring, value.strip())
+
+
+def _status_bucket(result: RepoCheckResult) -> str:
+    if result.issues:
+        return "strict"
+    if result.warnings:
+        return "warning"
+    return "clean"
+
+
+def _issue_kind(message: str) -> str:
+    if message.startswith("bundle pin mismatch for "):
+        return "bundle pin mismatch"
+    if message.startswith("forbidden short/invalid ref "):
+        return "forbidden short/invalid ref"
+    if message.startswith("forbidden ref 'main'"):
+        return "forbidden ref 'main'"
+    if message.startswith("legacy dependency files detected"):
+        return "legacy dependency files detected"
+    if message.startswith("unmanaged qsl dependency "):
+        return "unmanaged qsl dependency"
+    return "other"
+
+
+def _parse_bundle_mismatch(message: str) -> tuple[str | None, str | None]:
+    prefix = "bundle pin mismatch for "
+    if not message.startswith(prefix):
+        return None, None
+    try:
+        after_prefix = message[len(prefix) :]
+        package, remainder = after_prefix.split(" in ", 1)
+        source = remainder.split(":", 1)[0]
+    except ValueError:
+        return None, None
+    return package.strip(), source.strip()
+
+
+def _workspace_report(results: list[RepoCheckResult], compat_root: Path) -> dict[str, Any]:
+    ring_order, ring_to_tier, tier_to_ring = _load_repo_tiers(compat_root)
+    ring_index = {ring: idx for idx, ring in enumerate(ring_order)}
+    grouped: dict[str, dict[str, Any]] = {
+        ring: {
+            "ring": ring,
+            "tier": ring_to_tier.get(ring, "unknown"),
+            "total": 0,
+            "strict": 0,
+            "warning": 0,
+            "clean": 0,
+            "repositories": [],
+        }
+        for ring in ring_order
+    }
+    issue_counts: Counter[str] = Counter()
+    package_hotspots: Counter[tuple[str, str]] = Counter()
+
+    for result in sorted(
+        results,
+        key=lambda item: (ring_index.get(_normalize_ring(tier_to_ring.get(item.tier, item.upgrade_ring)), len(ring_order)), item.repo),
+    ):
+        canonical_ring = _normalize_ring(tier_to_ring.get(result.tier, result.upgrade_ring))
+        bucket = grouped.setdefault(
+            canonical_ring,
+            {
+                "ring": canonical_ring,
+                "tier": ring_to_tier.get(canonical_ring, result.tier or "unknown"),
+                "total": 0,
+                "strict": 0,
+                "warning": 0,
+                "clean": 0,
+                "repositories": [],
+            },
+        )
+        status = _status_bucket(result)
+        bucket["total"] += 1
+        bucket[status] += 1
+        bucket["repositories"].append(
+            {
+                "repo": result.repo,
+                "status": status,
+                "issues": len(result.issues),
+                "warnings": len(result.warnings),
+            }
+        )
+
+        for message in result.issues + result.warnings:
+            issue_counts[_issue_kind(message)] += 1
+            package, source = _parse_bundle_mismatch(message)
+            if package and source:
+                package_hotspots[(package, source)] += 1
+
+    rings = [grouped[ring] for ring in ring_order if ring in grouped and grouped[ring]["total"]]
+    extra_rings = [grouped[ring] for ring in grouped if ring not in ring_order and grouped[ring]["total"]]
+
+    return {
+        "schema_version": 1,
+        "total_repositories": len(results),
+        "strict_repositories": sum(1 for result in results if result.issues),
+        "warning_repositories": sum(1 for result in results if result.warnings and not result.issues),
+        "clean_repositories": sum(1 for result in results if not result.issues and not result.warnings),
+        "issue_counts": dict(sorted(issue_counts.items())),
+        "bundle_hotspots": [
+            {"package": package, "source": source, "count": count}
+            for (package, source), count in package_hotspots.most_common(10)
+        ],
+        "rings": rings + extra_rings,
+        "repositories": [_result_payload(result) for result in results],
+    }
+
+
+def _workspace_plan(report: dict[str, Any]) -> dict[str, Any]:
+    phases: list[dict[str, Any]] = []
+    for ring in report["rings"]:
+        strict_repos = [repo for repo in ring["repositories"] if repo["status"] == "strict"]
+        warning_repos = [repo for repo in ring["repositories"] if repo["status"] == "warning"]
+        clean_repos = [repo for repo in ring["repositories"] if repo["status"] == "clean"]
+
+        next_actions: list[str] = []
+        if strict_repos:
+            next_actions.append("先清理 strict mismatch，再推进下一 ring。")
+            next_actions.extend(f"修复 {repo['repo']}（{repo['issues']} 个 issue）" for repo in strict_repos)
+        elif warning_repos:
+            next_actions.append("当前只有 warning；可先保持只读观察，但在 release 前需要清零。")
+            next_actions.extend(f"确认 {repo['repo']} 的 warning 是否允许短期保留（{repo['warnings']} 个 warning）" for repo in warning_repos)
+        else:
+            next_actions.append("当前 ring 已收敛，可以作为下一 ring 的 gate。")
+
+        phases.append(
+            {
+                "ring": ring["ring"],
+                "tier": ring["tier"],
+                "strict_repositories": strict_repos,
+                "warning_repositories": warning_repos,
+                "clean_repositories": clean_repos,
+                "next_actions": next_actions,
+            }
+        )
+
+    return {
+        "schema_version": 1,
+        "phases": phases,
+        "bundle_hotspots": report["bundle_hotspots"],
+        "issue_counts": report["issue_counts"],
+    }
+
+
+def _print_report(report: dict[str, Any]) -> None:
+    print(
+        "QSL workspace report: "
+        f"repos={report['total_repositories']} "
+        f"strict={report['strict_repositories']} "
+        f"warnings={report['warning_repositories']} "
+        f"clean={report['clean_repositories']}"
+    )
+    print("Issue breakdown:")
+    for kind, count in report["issue_counts"].items():
+        print(f"  {kind}: {count}")
+    print("Ring summary:")
+    for ring in report["rings"]:
+        print(
+            f"  {ring['ring']} / {ring['tier']}: "
+            f"total={ring['total']} strict={ring['strict']} warning={ring['warning']} clean={ring['clean']}"
+        )
+        repos = ", ".join(repo["repo"] for repo in ring["repositories"])
+        if repos:
+            print(f"    repos: {repos}")
+    if report["bundle_hotspots"]:
+        print("Hotspots:")
+        for hotspot in report["bundle_hotspots"]:
+            print(f"  {hotspot['package']} @ {hotspot['source']}: {hotspot['count']}")
+
+
+def _print_plan(plan: dict[str, Any]) -> None:
+    print("QSL ring-by-ring convergence plan:")
+    for idx, phase in enumerate(plan["phases"], start=1):
+        print(f"{idx}. {phase['ring']} / {phase['tier']}")
+        for repo in phase["strict_repositories"]:
+            print(f"   - strict: {repo['repo']} ({repo['issues']} issues)")
+        for repo in phase["warning_repositories"]:
+            print(f"   - warning: {repo['repo']} ({repo['warnings']} warnings)")
+        for action in phase["next_actions"]:
+            print(f"   - {action}")
+    if plan["bundle_hotspots"]:
+        print("Top bundle hotspots:")
+        for hotspot in plan["bundle_hotspots"]:
+            print(f"  - {hotspot['package']} @ {hotspot['source']}: {hotspot['count']}")
+
+
 def _print_check_result(result: RepoCheckResult) -> None:
     status = "PASS" if result.ok else "FAIL"
     warning_suffix = f", warnings={len(result.warnings)}" if result.warnings else ""
@@ -164,6 +403,27 @@ def _cmd_check_all(args: argparse.Namespace) -> int:
     return 1 if args.strict and payload["failed_repositories"] else 0
 
 
+def _cmd_report(args: argparse.Namespace) -> int:
+    results = check_all(projects_root=args.projects_root.resolve(), compat_root=args.compat_root.resolve())
+    report = _workspace_report(results, compat_root=args.compat_root.resolve())
+    if args.json:
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+    else:
+        _print_report(report)
+    return 0
+
+
+def _cmd_plan(args: argparse.Namespace) -> int:
+    results = check_all(projects_root=args.projects_root.resolve(), compat_root=args.compat_root.resolve())
+    report = _workspace_report(results, compat_root=args.compat_root.resolve())
+    plan = _workspace_plan(report)
+    if args.json:
+        print(json.dumps(plan, ensure_ascii=False, indent=2))
+    else:
+        _print_plan(plan)
+    return 0
+
+
 def _matrix_payload(projects_root: Path) -> dict[str, Any]:
     return check_internal_dependency_matrix.matrix_payload(
         check_internal_dependency_matrix.collect_dependency_pins_from_projects(projects_root)
@@ -224,6 +484,18 @@ def build_parser() -> argparse.ArgumentParser:
     check_all_parser.add_argument("--verbose", action="store_true", help="Print passing repositories too.")
     check_all_parser.set_defaults(func=_cmd_check_all)
 
+    report = subparsers.add_parser("report", help="Summarize current QSL workspace status by ring and issue type.")
+    report.add_argument("--projects-root", type=Path, default=DEFAULT_PROJECTS_ROOT)
+    report.add_argument("--compat-root", type=Path, default=DEFAULT_COMPAT_ROOT)
+    report.add_argument("--json", action="store_true")
+    report.set_defaults(func=_cmd_report)
+
+    plan = subparsers.add_parser("plan", help="Render a ring-by-ring QSL convergence plan from current workspace state.")
+    plan.add_argument("--projects-root", type=Path, default=DEFAULT_PROJECTS_ROOT)
+    plan.add_argument("--compat-root", type=Path, default=DEFAULT_COMPAT_ROOT)
+    plan.add_argument("--json", action="store_true")
+    plan.set_defaults(func=_cmd_plan)
+
     matrix = subparsers.add_parser("generate-matrix", help="Generate or check the internal dependency matrix.")
     matrix.add_argument("--projects-root", type=Path, default=DEFAULT_PROJECTS_ROOT)
     matrix.add_argument("--matrix", type=Path, default=DEFAULT_MATRIX_PATH)