#!/usr/bin/env python3 """Validate the product-group AI-native collaboration workspace.""" from __future__ import annotations import argparse import csv from pathlib import Path import sys REQUIRED_FILES = [ "README.md", "ROUTING_NOTE.md", "OPERATING_MODEL.md", "WORKFLOW_MAP.md", "WORK_ITEM_CONTRACT.md", "FIRST_RUN.md", "ONBOARDING.md", "ACCESS_AND_ROLES.md", "KNOWLEDGE_SHARING.md", "source-index.csv", "summary.md", "lanes/README.md", "control/projects.csv", "control/work-items.csv", "control/decisions.csv", "control/risks-changes.csv", "control/reviews.csv", "control/metrics.csv", "control/contributions.csv", "control/members.csv", "control/role-pools.csv", "control/access-requests.csv", ] CSV_HEADERS = { "control/projects.csv": [ "project_id", "name", "product_line", "owner", "stage", "health", "objective", "evidence_path", "updated_at", ], "control/work-items.csv": [ "work_item_id", "project_id", "title", "primary_lane", "lifecycle_stage", "decision_owner", "contributors", "due_date", "expected_artifact", "acceptance_criteria", "evidence_path", "risk_level", "human_gate", "state", "external_id", "updated_at", ], "control/decisions.csv": [ "decision_id", "work_item_id", "decision_date", "question", "decision", "decision_owner", "reviewers", "evidence_path", "assumptions", "next_review_date", "status", ], "control/risks-changes.csv": [ "record_id", "work_item_id", "type", "severity", "description", "impact", "owner", "due_date", "decision_id", "close_evidence", "state", ], "control/reviews.csv": [ "review_id", "work_item_id", "review_type", "reviewer", "review_date", "result", "requested_changes", "evidence_path", "next_gate", ], "control/metrics.csv": [ "metric_id", "project_id", "work_item_id", "metric", "definition", "baseline", "target", "current_value", "owner", "data_source", "refresh_frequency", "updated_at", "status", ], "control/contributions.csv": [ "contribution_id", "work_item_id", "contributor", "role", "artifact", "complexity", "collaboration", "evidence_path", "recorded_at", ], "control/members.csv": [ "member_id", "name", "codeup_account", "primary_role", "team", "codeup_access", "approval_scope", "work_item_id", "join_status", "last_verified_at", "notes", ], "control/role-pools.csv": [ "role_id", "role_name", "default_codeup_access", "approval_scope", "requires_work_item", "activation_gate", "active", ], "control/access-requests.csv": [ "access_request_id", "member_name", "codeup_account", "member_type", "requested_repo_access", "requested_role", "requested_by", "approved_by", "status", "expires_at", "evidence_link", "notes", ], } VALID_LANES = { "strategy", "market-research", "user-discovery", "requirements", "prioritization-planning", "product-design", "delivery-collaboration", "analytics-experiments", "gtm-growth", "feedback-learning", } VALID_STATES = { "intake", "triage", "ready", "in-progress", "review", "approved", "execution", "verified", "shipped", "measuring", "learned", "blocked", "cancelled", "closed", } VALID_MEMBER_STATES = {"pending", "active", "inactive"} VALID_CODEUP_ACCESS = {"viewer", "developer", "admin"} YES_NO = {"yes", "no"} UNRESOLVED = {"", "待确认", "待指定", "tbd", "n/a"} def read_csv(path: Path) -> tuple[list[str], list[dict[str, str]]]: with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) return reader.fieldnames or [], list(reader) def validate(root: Path, strict: bool) -> list[str]: errors: list[str] = [] for relative in REQUIRED_FILES: if not (root / relative).is_file(): errors.append(f"missing file: {relative}") tables: dict[str, list[dict[str, str]]] = {} for relative, expected in CSV_HEADERS.items(): path = root / relative if not path.is_file(): continue actual, rows = read_csv(path) tables[relative] = rows if actual != expected: errors.append( f"invalid CSV header: {relative}\n" f" expected: {','.join(expected)}\n" f" actual: {','.join(actual)}" ) work_items = tables.get("control/work-items.csv", []) seen_ids: set[str] = set() for index, row in enumerate(work_items, start=2): item_id = row.get("work_item_id", "").strip() if item_id in seen_ids: errors.append(f"duplicate work_item_id at row {index}: {item_id}") seen_ids.add(item_id) lane = row.get("primary_lane", "").strip() state = row.get("state", "").strip() if lane and lane not in VALID_LANES: errors.append(f"invalid primary_lane at row {index}: {lane}") if state and state not in VALID_STATES: errors.append(f"invalid state at row {index}: {state}") members = tables.get("control/members.csv", []) role_pools = tables.get("control/role-pools.csv", []) active_role_names: set[str] = set() seen_role_ids: set[str] = set() for index, row in enumerate(role_pools, start=2): role_id = row.get("role_id", "").strip() role_name = row.get("role_name", "").strip() default_access = row.get("default_codeup_access", "").strip() requires_work_item = row.get("requires_work_item", "").strip() active = row.get("active", "").strip() if not role_id: errors.append(f"missing role_id at control/role-pools.csv row {index}") elif role_id in seen_role_ids: errors.append( f"duplicate role_id at control/role-pools.csv row {index}: {role_id}" ) seen_role_ids.add(role_id) if not role_name: errors.append(f"missing role_name at control/role-pools.csv row {index}") if default_access not in VALID_CODEUP_ACCESS: errors.append( f"invalid default_codeup_access at control/role-pools.csv row {index}: " f"{default_access}" ) if requires_work_item not in YES_NO: errors.append( f"invalid requires_work_item at control/role-pools.csv row {index}: " f"{requires_work_item}" ) if active not in YES_NO: errors.append( f"invalid active at control/role-pools.csv row {index}: {active}" ) if active == "yes" and role_name: active_role_names.add(role_name) seen_member_ids: set[str] = set() for index, row in enumerate(members, start=2): member_id = row.get("member_id", "").strip() if not member_id: errors.append(f"missing member_id at control/members.csv row {index}") elif member_id in seen_member_ids: errors.append(f"duplicate member_id at control/members.csv row {index}: {member_id}") seen_member_ids.add(member_id) join_status = row.get("join_status", "").strip() codeup_access = row.get("codeup_access", "").strip() if join_status not in VALID_MEMBER_STATES: errors.append( f"invalid join_status at control/members.csv row {index}: {join_status}" ) if codeup_access not in VALID_CODEUP_ACCESS: errors.append( f"invalid codeup_access at control/members.csv row {index}: {codeup_access}" ) primary_role = row.get("primary_role", "").strip() if primary_role and primary_role not in active_role_names: errors.append( f"unknown or inactive primary_role at control/members.csv row {index}: " f"{primary_role}" ) if strict: eligible = [ row for row in work_items if row.get("state") not in {"intake", "cancelled", "closed", ""} ] if not eligible: errors.append("strict gate: at least one active real work item is required") required_fields = [ "decision_owner", "due_date", "expected_artifact", "acceptance_criteria", "evidence_path", "risk_level", "human_gate", ] for index, row in enumerate(work_items, start=2): if row.get("state") in {"intake", "cancelled", "closed", ""}: continue for field in required_fields: value = row.get(field, "").strip().lower() if value in UNRESOLVED: errors.append( f"strict gate: control/work-items.csv row {index} " f"has unresolved {field}" ) reviews = tables.get("control/reviews.csv", []) if eligible and not any(row.get("result") in {"approved", "changes-requested"} for row in reviews): errors.append("strict gate: at least one human review result is required") active_members = [ row for row in members if row.get("join_status", "").strip() == "active" ] if not active_members: errors.append("strict gate: at least one verified active teammate is required") member_required_fields = [ "name", "codeup_account", "primary_role", "approval_scope", "work_item_id", "last_verified_at", ] known_work_items = { row.get("work_item_id", "").strip() for row in work_items } for index, row in enumerate(members, start=2): if row.get("join_status", "").strip() != "active": continue for field in member_required_fields: value = row.get(field, "").strip().lower() if value in UNRESOLVED: errors.append( f"strict gate: control/members.csv row {index} " f"has unresolved {field}" ) member_work_item = row.get("work_item_id", "").strip() if member_work_item not in known_work_items: errors.append( f"strict gate: control/members.csv row {index} references " f"unknown work_item_id {member_work_item}" ) return errors def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("workspace", type=Path) parser.add_argument("--strict", action="store_true") args = parser.parse_args() root = args.workspace.resolve() if not root.is_dir(): print(f"ERROR: workspace does not exist: {root}", file=sys.stderr) return 2 errors = validate(root, args.strict) if errors: print("VALIDATION FAILED") for error in errors: print(f"- {error}") return 1 mode = "strict" if args.strict else "bootstrap" print(f"VALIDATION PASSED ({mode}): {root}") return 0 if __name__ == "__main__": raise SystemExit(main())