fix(gitea): surface all product dev prod repo truth
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m12s
CD Pipeline / build-and-deploy (push) Successful in 4m52s
CD Pipeline / post-deploy-checks (push) Successful in 1m47s

This commit is contained in:
Your Name
2026-07-03 08:38:04 +08:00
parent 0dc9e62b6d
commit fbd3ad2ca4
9 changed files with 1286 additions and 144 deletions

View File

@@ -54,6 +54,12 @@ def parse_args() -> argparse.Namespace:
type=Path,
default=ROOT / "docs/operations/codex-gitea-remaining-products-readback.snapshot.json",
)
parser.add_argument(
"--dev-prod-repo-readback",
type=Path,
default=ROOT
/ "docs/operations/gitea-all-product-dev-prod-repo-readback.snapshot.json",
)
parser.add_argument(
"--controlled-closeout-receipt",
type=Path,
@@ -107,6 +113,37 @@ def product_row(row: dict[str, Any], state: str) -> dict[str, Any]:
}
def dev_prod_product_row(row: dict[str, Any]) -> dict[str, Any]:
return {
"product_id": str(row.get("product_id", "")),
"state": "repo_environment_ready"
if row.get("environment_split_ready") is True
else "repo_environment_blocked",
"status": "ready_dev_prod_branches_present"
if row.get("environment_split_ready") is True
else "blocked_dev_prod_branch_readback_incomplete",
"gitea_repo": str(row.get("gitea_repo", "")),
"prod_environment_branch": str(row.get("prod_environment_branch") or "main"),
"prod_branch_sha": str(row.get("prod_branch_sha") or ""),
"prod_branch_present": bool(row.get("prod_branch_present", False)),
"dev_environment_branch": str(row.get("dev_environment_branch") or "dev"),
"dev_branch_sha": str(row.get("dev_branch_sha") or ""),
"dev_branch_present": bool(row.get("dev_branch_present", False)),
"remote_main_present": bool(row.get("prod_branch_present", False)),
"remote_dev_present": bool(row.get("dev_branch_present", False)),
"ssh_refs_readable": bool(row.get("ssh_refs_readable", False)),
"tokenless_http_status": as_int(row.get("tokenless_http_status")),
"public_ui_visible": bool(row.get("public_ui_visible", False)),
"visibility_readback": str(row.get("visibility_readback") or "unknown"),
"environment_split_ready": bool(row.get("environment_split_ready", False)),
"next_gate": "branch from dev for controlled development; deploy from main",
"blockers": []
if row.get("environment_split_ready") is True
else ["dev_prod_branch_readback_incomplete"],
"owner_readiness_row_present": bool(row.get("product_id")),
}
def build_product_rows(remaining: dict[str, Any]) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
for row in as_list(remaining.get("ready_products")):
@@ -118,6 +155,48 @@ def build_product_rows(remaining: dict[str, Any]) -> list[dict[str, Any]]:
return rows
def build_dev_prod_product_rows(readback: dict[str, Any]) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
for row in as_list(readback.get("products")):
if isinstance(row, dict):
rows.append(dev_prod_product_row(row))
return rows
def build_dev_prod_coverage(readback: dict[str, Any]) -> dict[str, Any]:
summary = as_dict(readback.get("summary"))
expected = as_int(summary.get("expected_product_count"))
ssh_present = as_int(summary.get("ssh_repo_present_count"))
main_present = as_int(summary.get("main_branch_present_count"))
dev_present = as_int(summary.get("dev_branch_present_count"))
split_ready = as_int(summary.get("environment_split_ready_count"))
public_visible = as_int(summary.get("public_visible_count"))
private_or_auth = as_int(summary.get("private_or_auth_only_count"))
return {
"schema_version": readback.get("schema_version", ""),
"status": str(readback.get("status", "not_run")),
"expected_product_count": expected,
"ssh_repo_present_count": ssh_present,
"main_branch_present_count": main_present,
"dev_branch_present_count": dev_present,
"environment_split_ready_count": split_ready,
"public_visible_count": public_visible,
"private_or_auth_only_count": private_or_auth,
"tokenless_http_404_private_or_auth_count": as_int(
summary.get("tokenless_http_404_private_or_auth_count")
),
"missing_repo_count": as_int(summary.get("missing_repo_count")),
"missing_main_branch_count": as_int(summary.get("missing_main_branch_count")),
"missing_dev_branch_count": as_int(summary.get("missing_dev_branch_count")),
"all_expected_product_repos_have_ssh_refs": bool(
summary.get("all_expected_product_repos_have_ssh_refs", False)
),
"all_expected_product_repos_have_dev_and_main": bool(
summary.get("all_expected_product_repos_have_dev_and_main", False)
),
}
def find_validation_lane(rollup: dict[str, Any], lane_id: str) -> dict[str, Any]:
for lane in as_list(rollup.get("validation_lanes")):
if isinstance(lane, dict) and lane.get("lane_id") == lane_id:
@@ -199,17 +278,23 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
payload_validation = load_optional_json(args.payload_validation)
closeout_receipt = load_optional_json(args.controlled_closeout_receipt)
remaining_products = load_json(args.remaining_products)
dev_prod_readback = load_optional_json(args.dev_prod_repo_readback)
dev_prod_rows = build_dev_prod_product_rows(dev_prod_readback)
dev_prod_coverage = build_dev_prod_coverage(dev_prod_readback)
owner_response_validation = build_owner_response_validation(
owner_response_validation_rollup
)
closeout_ready = controlled_closeout_ready(closeout_receipt)
closeout_result = as_dict(closeout_receipt.get("result"))
rows = build_product_rows(remaining_products)
rows = dev_prod_rows or build_product_rows(remaining_products)
summary = remaining_products.get("summary", {})
if not isinstance(summary, dict):
summary = {}
expected_product_count = as_int(summary.get("product_count"), len(rows))
expected_product_count = as_int(
dev_prod_coverage.get("expected_product_count"),
as_int(summary.get("product_count"), len(rows)),
)
missing_row_count = max(expected_product_count - len(rows), 0)
gitea_status = str(gitea_inventory.get("status", "unknown"))
@@ -249,6 +334,12 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
for repo in as_list(gitea_inventory.get("repos"))
if isinstance(repo, dict) and repo.get("gitea_repo")
]
if dev_prod_rows:
public_repos = [
str(row.get("gitea_repo", ""))
for row in dev_prod_rows
if row.get("public_ui_visible") is True and row.get("gitea_repo")
]
blockers: list[str] = []
if gitea_status != "ok":
@@ -261,8 +352,18 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
blockers.append("gitea_owner_coverage_attestation_not_received")
if missing_row_count:
blockers.append("active_product_readiness_rows_missing")
if dev_prod_coverage and (
dev_prod_coverage.get("all_expected_product_repos_have_dev_and_main")
is not True
):
blockers.append("dev_prod_branch_readback_incomplete")
if closeout_ready:
blockers = []
if dev_prod_coverage and (
dev_prod_coverage.get("all_expected_product_repos_have_dev_and_main")
is not True
):
blockers.append("dev_prod_branch_readback_incomplete")
return {
"schema_version": SCHEMA_VERSION,
@@ -289,7 +390,16 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
"schema_version": gitea_inventory.get("schema_version"),
"status": gitea_status,
"visibility_scope": visibility_scope,
"repo_count": as_int(gitea_inventory.get("repo_count")),
"repo_count": as_int(
dev_prod_coverage.get("public_visible_count"),
as_int(gitea_inventory.get("repo_count")),
),
"ssh_verified_repo_count": as_int(
dev_prod_coverage.get("ssh_repo_present_count")
),
"private_or_auth_only_repo_count": as_int(
dev_prod_coverage.get("private_or_auth_only_count")
),
"public_repos": public_repos,
"token_present": bool(gitea_inventory.get("token_present", False)),
"blocking_reason": (
@@ -298,6 +408,8 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
else str(gitea_inventory.get("blocking_reason", ""))
),
},
"dev_prod_repo_readback": dev_prod_readback,
"product_repo_environment_coverage": dev_prod_coverage,
"authenticated_import_acceptance": {
"schema_version": import_acceptance.get("schema_version"),
"status": (
@@ -356,13 +468,29 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
"expected_product_count": expected_product_count,
"present_product_row_count": len(rows),
"missing_product_row_count": missing_row_count,
"ready_product_count": len([row for row in rows if row["state"] == "ready"]),
"blocked_product_count": len([row for row in rows if row["state"] == "blocked"]),
"ready_product_count": len(
[
row
for row in rows
if row["state"] in {"ready", "repo_environment_ready"}
]
),
"blocked_product_count": len(
[
row
for row in rows
if row["state"] in {"blocked", "repo_environment_blocked"}
]
),
"internal_or_authenticated_inventory_required_count": as_int(
summary.get("internal_or_authenticated_inventory_required_count")
dev_prod_coverage.get("private_or_auth_only_count"),
as_int(summary.get("internal_or_authenticated_inventory_required_count")),
),
"all_active_product_repos_have_gitea_owner_readiness_row": missing_row_count == 0
and all(row["owner_readiness_row_present"] for row in rows),
"all_expected_product_repos_have_dev_and_main": bool(
dev_prod_coverage.get("all_expected_product_repos_have_dev_and_main")
),
},
"product_rows": rows,
"active_blockers": blockers,
@@ -372,6 +500,7 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
"gitea_repo_inventory.status=ok",
"gitea_repo_inventory.visibility_scope in authenticated/admin_export",
"all_active_product_repos_have_gitea_owner_readiness_row=true",
"all_expected_product_repos_have_dev_and_main=true",
],
"safe_next_step": (
"continue_to_p0_006_source_to_runtime_drift_cleanup"

View File

@@ -41,13 +41,30 @@ def test_scorecard_commits_controlled_closeout_receipt() -> None:
assert scorecard["status"] == "closed_gitea_private_inventory_controlled_closeout"
assert scorecard["gitea_inventory"]["status"] == "ok"
assert scorecard["gitea_inventory"]["visibility_scope"] == "admin_export"
assert scorecard["gitea_inventory"]["repo_count"] == 4
assert scorecard["gitea_inventory"]["repo_count"] == 6
assert scorecard["gitea_inventory"]["ssh_verified_repo_count"] == 12
assert scorecard["gitea_inventory"]["private_or_auth_only_repo_count"] == 6
assert {
"wooo/awoooi",
"wooo/ewoooc",
"wooo/agent-bounty-protocol",
"wooo/2026FIFAWorldCup",
"wooo/vtuber",
"wooo/bitan-pharmacy",
} <= set(scorecard["gitea_inventory"]["public_repos"])
env_coverage = scorecard["product_repo_environment_coverage"]
assert env_coverage["expected_product_count"] == 12
assert env_coverage["ssh_repo_present_count"] == 12
assert env_coverage["main_branch_present_count"] == 12
assert env_coverage["dev_branch_present_count"] == 12
assert env_coverage["environment_split_ready_count"] == 12
assert env_coverage["public_visible_count"] == 6
assert env_coverage["private_or_auth_only_count"] == 6
assert env_coverage["missing_repo_count"] == 0
assert env_coverage["missing_main_branch_count"] == 0
assert env_coverage["missing_dev_branch_count"] == 0
assert env_coverage["all_expected_product_repos_have_ssh_refs"] is True
assert env_coverage["all_expected_product_repos_have_dev_and_main"] is True
assert scorecard["authenticated_import_acceptance"]["accepted_payload_count"] == 1
assert (
scorecard["authenticated_payload_validation"]["status"]
@@ -109,12 +126,22 @@ def test_scorecard_keeps_all_active_product_rows_visible() -> None:
scorecard = load_scorecard()
coverage = scorecard["product_row_coverage"]
assert coverage["expected_product_count"] == 11
assert coverage["present_product_row_count"] == 11
assert coverage["expected_product_count"] == 12
assert coverage["present_product_row_count"] == 12
assert coverage["missing_product_row_count"] == 0
assert coverage["ready_product_count"] == 3
assert coverage["blocked_product_count"] == 8
assert coverage["ready_product_count"] == 12
assert coverage["blocked_product_count"] == 0
assert coverage["all_active_product_repos_have_gitea_owner_readiness_row"] is True
assert coverage["all_expected_product_repos_have_dev_and_main"] is True
product_ids = {row["product_id"] for row in scorecard["product_rows"]}
assert {"awoooi", "momo-pro", "awooogo", "bitan-pharmacy", "tsenyang-website"} <= product_ids
assert {
"awoooi",
"ewoooc",
"awooogo",
"momo-pro-system",
"bitan-pharmacy",
"tsenyang-website",
"clawbot-openclaw",
} <= product_ids
assert all(row["environment_split_ready"] is True for row in scorecard["product_rows"])