workspace.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """``ok workspace`` commands (§MR.7).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from argparse import Namespace |
| 6 | from pathlib import Path |
| 7 | |
| 8 | from adapters.config import load_config |
| 9 | from adapters.errors import ConfigError |
| 10 | from cli.context import CliContext |
| 11 | from cli.paths import is_within_repo, resolve_config_path, resolve_repo_root |
| 12 | from cli.sanitize import format_config_error |
| 13 | from tools.workspace import ( |
| 14 | EXIT_WORKSPACE_RELAY, |
| 15 | WorkspaceLoadError, |
| 16 | build_status_report, |
| 17 | check_next, |
| 18 | load_manifest_for_repo, |
| 19 | run_doctor, |
| 20 | ) |
| 21 | from tools.workspace.board_names import check_next_unconfigured_advisory |
| 22 | from tools.workspace.types import EXIT_CONFIG, EXIT_OK, EXIT_USAGE |
| 23 | |
| 24 | |
| 25 | def run_workspace(args: Namespace, ctx: CliContext) -> int: |
| 26 | """Dispatch ``ok workspace status|check-next|doctor``.""" |
| 27 | action = getattr(args, "workspace_action", None) |
| 28 | if action is None: |
| 29 | ctx.output.error("usage: ok workspace status|check-next|doctor") |
| 30 | return EXIT_USAGE |
| 31 | |
| 32 | repo_root = resolve_repo_root(cwd=ctx.cwd, repo_arg=args.repo, command="workspace") |
| 33 | config_path = resolve_config_path(repo_root, args.config) |
| 34 | if not is_within_repo(repo_root, config_path): |
| 35 | ctx.output.error("refused: config path outside repo root") |
| 36 | return 4 |
| 37 | |
| 38 | try: |
| 39 | config = load_config(config_path) |
| 40 | except ConfigError as exc: |
| 41 | ctx.output.error(format_config_error(exc, repo_root)) |
| 42 | return EXIT_CONFIG |
| 43 | |
| 44 | if action == "status": |
| 45 | return _run_status(args, ctx, config, repo_root) |
| 46 | if action == "check-next": |
| 47 | return _run_check_next(args, ctx, config, repo_root) |
| 48 | if action == "doctor": |
| 49 | return _run_doctor(args, ctx, config, repo_root) |
| 50 | |
| 51 | ctx.output.error(f"unknown workspace action: {action}") |
| 52 | return EXIT_USAGE |
| 53 | |
| 54 | |
| 55 | def _run_status(args: Namespace, ctx: CliContext, config, repo_root: Path) -> int: |
| 56 | report = build_status_report( |
| 57 | config, |
| 58 | repo_root, |
| 59 | strict_all=bool(getattr(args, "strict_all", False)), |
| 60 | ) |
| 61 | if ctx.output.json_mode: |
| 62 | payload = report.to_json() |
| 63 | payload["workspace"] = payload # convenience alias for status --workspace compose |
| 64 | ctx.output.emit_json(report.to_json()) |
| 65 | else: |
| 66 | if not report.configured: |
| 67 | ctx.output.emit("workspace: not_configured") |
| 68 | return EXIT_OK |
| 69 | ctx.output.emit(f"workspace.ok: {str(report.ok).lower()}") |
| 70 | ctx.output.emit(f"workspace.state: {report.state}") |
| 71 | ctx.output.emit(f"constellation_id: {report.constellation_id}") |
| 72 | ctx.output.emit(f"product_order_member: {report.product_order_member}") |
| 73 | ctx.output.emit(f"manifest_source: {report.manifest_source}") |
| 74 | if report.authoritative_handover: |
| 75 | ctx.output.emit(f"authoritative_handover: {report.authoritative_handover}") |
| 76 | for member in report.members: |
| 77 | base = member.get("handover_basename") or "?" |
| 78 | ctx.output.emit( |
| 79 | f"member {member['id']}: role={member['role']} " |
| 80 | f"status={member['member_status']} handover={base}" |
| 81 | ) |
| 82 | for warning in report.warnings: |
| 83 | ctx.output.emit(f"warning: {warning}") |
| 84 | if report.check_next and report.check_next.get("messages"): |
| 85 | for msg in report.check_next["messages"]: |
| 86 | ctx.output.emit(msg) |
| 87 | # status itself exits 0 even when workspace.ok is false (§MR.12.1 / S9 separation); |
| 88 | # --exit-code is only on `ok status --workspace --exit-code`. |
| 89 | return EXIT_OK |
| 90 | |
| 91 | |
| 92 | def _run_check_next(args: Namespace, ctx: CliContext, config, repo_root: Path) -> int: |
| 93 | if config.workspace is None: |
| 94 | # §NXP.5 — visibility advisory; exit stays 0 (not a new gate). |
| 95 | advisory = check_next_unconfigured_advisory(config) |
| 96 | payload = { |
| 97 | "ok": True, |
| 98 | "state": "not_configured", |
| 99 | "exit_code": EXIT_OK, |
| 100 | "advisory": advisory, |
| 101 | } |
| 102 | if ctx.output.json_mode: |
| 103 | ctx.output.emit_json(payload) |
| 104 | else: |
| 105 | ctx.output.emit(advisory) |
| 106 | return EXIT_OK |
| 107 | try: |
| 108 | manifest = load_manifest_for_repo(config, repo_root) |
| 109 | except WorkspaceLoadError as exc: |
| 110 | if ctx.output.json_mode: |
| 111 | ctx.output.emit_json({"ok": False, "state": "error", "error": str(exc), "exit_code": EXIT_CONFIG}) |
| 112 | else: |
| 113 | ctx.output.error(str(exc)) |
| 114 | return EXIT_CONFIG |
| 115 | |
| 116 | result = check_next(manifest, lane=getattr(args, "lane", None)) |
| 117 | if ctx.output.json_mode: |
| 118 | ctx.output.emit_json( |
| 119 | { |
| 120 | "ok": result.ok, |
| 121 | "state": result.state, |
| 122 | "exit_code": result.exit_code, |
| 123 | "lane": result.lane, |
| 124 | "primary": result.primary, |
| 125 | "relays": list(result.relays), |
| 126 | "messages": list(result.messages), |
| 127 | } |
| 128 | ) |
| 129 | else: |
| 130 | for msg in result.messages: |
| 131 | ctx.output.emit(msg) |
| 132 | if result.ok: |
| 133 | ctx.output.emit("workspace check-next: ok") |
| 134 | else: |
| 135 | ctx.output.error(f"workspace check-next: {result.state}") |
| 136 | return result.exit_code |
| 137 | |
| 138 | |
| 139 | def _run_doctor(args: Namespace, ctx: CliContext, config, repo_root: Path) -> int: |
| 140 | report = run_doctor(config, repo_root) |
| 141 | if ctx.output.json_mode: |
| 142 | ctx.output.emit_json(report.to_json()) |
| 143 | else: |
| 144 | if not report.configured: |
| 145 | ctx.output.emit("workspace: not_configured") |
| 146 | return EXIT_OK |
| 147 | if not report.findings: |
| 148 | ctx.output.emit("workspace doctor: no findings") |
| 149 | for finding in report.findings: |
| 150 | loc = f" [{finding.member_id}]" if finding.member_id else "" |
| 151 | ctx.output.emit(f"{finding.code}{loc}: {finding.message}") |
| 152 | return EXIT_OK if report.ok or not report.configured else EXIT_OK |