docs(factory): refresh phase-0 inventory plan

PLAN.md

@@ -1,89 +1,72 @@
-# Bid-MVP Factory Control Planes Implementation Plan (ljwx-deploy)
+# Bid-MVP Factory Control Planes Plan (ljwx-deploy, refreshed on 2026-03-03)
 
 ## Phase 0 Inventory
 
 ### Existing
-- GitOps repo with Argo CD apps under `argocd-apps/` and workload manifests under `apps/`.
+- Workflows exist under `.github/workflows/`:
-- Service images currently mostly controlled in overlay `kustomization.yaml` `images` blocks.
+  - `nightly-evidence.yml`
-- Existing utility script: `scripts/create-app-secret.sh`.
+  - `nightly-auto-repair.yml`
-- No `.github/workflows/` currently.
+  - `deploy-repo-gate.yml`
-
+  - `build-promoter-image.yml`
-### Missing (to add)
+- Evidence control plane exists:
-- Evidence control plane:
-  - `evidence/README.md`
   - `evidence/schema/evidence-record.schema.json`
-  - `evidence/records/*.json`
+  - `evidence/schema/evidence.schema.json`
-  - generated `evidence/index.json`
+  - `evidence/records/` (directory tracked)
-- Evidence scripts:
   - `scripts/evidence/collect.py`
   - `scripts/evidence/validate.py`
-- Pages site source:
+  - `evidence/index.json`, `evidence/summary/latest.md`
-  - `site/index.html`
+- Pages dashboard source exists in both `pages/` and `site/` (needs consolidation decision).
-  - `site/app.js`
+- Release queue + promoter exists:
-  - optional `site/styles.css`
+  - `release/queue.yaml`, `release/services.yaml`
-- GitHub Actions workflows:
+  - `scripts/promoter/promote.py`, `scripts/promoter/deploy_promoter.py`, `scripts/promoter/validate_queue.py`
-  - nightly evidence collect + gh-pages publish
+  - CronJob manifests: `infra/factory/deploy-promoter-cronjob.yaml`, `k8s/promoter/deploy-promoter-cronjob.yaml`
-  - auto-repair (phase 5)
+- Smoke loop exists:
-- Release queue and promoter:
+  - `scripts/smoke/run_smoke.py`, `scripts/smoke/targets.json`
-  - `release/queue.yaml`
+  - `infra/factory/smoke-runner-cronjob.yaml`
-  - `scripts/promoter/deploy_promoter.py`
+- Auto-repair loop exists:
-  - `scripts/promoter/README.md`
+  - `repairs/recipes.yaml`
-  - k8s CronJob manifests for promoter
+  - `scripts/repair/run.sh`, `scripts/repair/run_repair.py`, `scripts/repair/diagnose.py`
-- Dev image pin files:
+  - orchestration docs/scripts: `infra/factory/INTEGRATION_RUN_ALL.md`, `scripts/factory/run_all_dev.sh`, `TEST.md`
-  - `envs/dev/<service>.yaml`
+
-- Smoke checks:
+### Current Gap / To Improve
-  - `scripts/smoke/run_smoke.py`
+- Two-cluster invariant hardening (required):
-  - k8s smoke runner CronJob or Job manifest
+  - Same GitOps code must support both local server `k3s` and China mainland dev server `OrbStack k3s`.
-- Auto repair loop:
+  - Differences must be represented via overlays/values/env only, no forked logic or duplicated pipelines.
-  - `scripts/repair/recipes.yaml`
+- Promoter implementation appears duplicated (`promote.py` and `deploy_promoter.py`):
-  - `scripts/repair/run_repair.py`
+  - choose one canonical entrypoint to avoid drift and behavioral mismatch.
+- Queue state alignment:
+  - ensure docs/code agree on `pending/promoted/failed/superseded` semantics and data placement.
+- Pages source alignment:
+  - choose one source directory (`pages/` or `site/`) and document workflow contract.
+- Deploy gate hardening:
+  - verify gate blocks unauthorized file changes and validates queue/evidence schemas deterministically.
 
 ### Existing promoter/test-runner/evidence/pages status
-- Promoter: not present.
+- Promoter: present.
-- Test-runner/smoke runner: not present.
+- Test-runner/smoke runner: present.
-- Evidence model/feed: not present.
+- Evidence model/feed: present.
-- Pages/gh-pages publishing workflow: not present.
+- Pages/gh-pages publishing: present.
+- Conclusion: Phases 2-5 are largely implemented; next step is consistency hardening and elimination of drift.
 
-## Phase 2 Scope (Evidence + Pages)
+## Planned File Changes (next deploy iterations)
-- Add evidence schema/model and scripts.
+- `scripts/promoter/*` (canonicalize single promoter path and behavior)
-- Add nightly workflow that validates records, builds `evidence/index.json`, and publishes dashboard to `gh-pages` branch root.
+- `release/README.md`, `release/queue.yaml` (shape/field consistency)
-- Keep GitHub as source of truth; no external state required.
+- `.github/workflows/nightly-evidence.yml` + docs (single pages source contract)
+- `docs/pages.md`, `docs/ops-runbook.md` (two-k3s same-code rule and runbook precision)
+- `argocd-apps/` and/or `apps/*/overlays/*` (if overlay naming normalization is needed for dual k3s)
 
-## Phase 3 Scope (Release Queue + Auto Promote)
+## Local Validation Commands
-- Add queue state machine (`pending/promoted/failed`) with retry metadata.
+- `test -f PLAN.md && sed -n '1,260p' PLAN.md`
-- Add promoter logic for per-service serialization and supersede semantics.
+- `python3 scripts/evidence/validate.py`
-- Verify Harbor digest existence via v2 manifest API (HEAD/GET) to avoid replication timing assumptions.
+- `python3 scripts/evidence/collect.py --out evidence/index.json --summary evidence/summary/latest.md`
-- Pin dev image in `envs/dev/<svc>.yaml` as digest reference.
+- `python3 scripts/promoter/validate_queue.py --queue release/queue.yaml`
-- Write evidence records for promotion attempts/results.
+- `python3 scripts/promoter/promote.py --dry-run --skip-registry-check`
+- `python3 scripts/smoke/run_smoke.py --dry-run`
+- `python3 scripts/repair/diagnose.py --help`
+- `python3 scripts/repair/run_repair.py --help`
 
-## Phase 4 Scope (Post-deploy Smoke)
+## Commit Plan
-- Add smoke runner to check Argo app health/sync (or endpoint readiness fallback).
+- `phase0(deploy): refresh inventory plan`
-- Write smoke results into evidence record chain.
+- `phase3(deploy): promoter/queue canonicalization + dual-k3s overlay rule hardening`
-
+- `phase4(deploy): smoke/evidence consistency hardening`
-## Phase 5 Scope (Auto Repair Loop)
+- `phase5(deploy): auto-repair determinism and gate tightening`
-- Add repair recipes + runner for format/lint/codegen style fixes.
-- Add automation notes/scripts for commit→PR→checks→auto-fix→recheck→merge (dev).
-- Add fail-after-N behavior to open issue with minimal repro/logs.
-- Add final `TEST.md` end-to-end dry run instructions.
-
-## Phase-by-Phase Verification Commands (deploy repo)
-- Phase 0:
-  - `test -f PLAN.md && sed -n '1,120p' PLAN.md`
-- Phase 2:
-  - `python3 scripts/evidence/validate.py`
-  - `python3 scripts/evidence/collect.py --out evidence/index.json`
-  - `python3 -m json.tool evidence/index.json >/dev/null`
-- Phase 3:
-  - `python3 scripts/promoter/deploy_promoter.py --help`
-  - `python3 scripts/promoter/deploy_promoter.py --dry-run`
-- Phase 4:
-  - `python3 scripts/smoke/run_smoke.py --help`
-- Phase 5:
-  - `python3 scripts/repair/run_repair.py --help`
-
-## Commit Plan (deploy repo)
-- `phase0(deploy): inventory + PLAN.md`
-- `phase2(deploy): evidence model/scripts/pages workflow/site`
-- `phase3(deploy): release queue + promoter + cronjob + runbook`
-- `phase4(deploy): smoke runner + cronjob + evidence integration docs`
-- `phase5(deploy): auto-repair recipes/runner + workflow + TEST.md`

scripts/evidence/collect.py

@@ -41,7 +41,11 @@ def record_timestamp(record: dict[str, Any]) -> datetime:
     deploy = record.get("deploy", {})
     synced_at = deploy.get("syncedAt") if isinstance(deploy, dict) else None
     promoted_at = record.get("promotedAt")
-    return parse_ts(synced_at) if parse_ts(synced_at) != datetime.min.replace(tzinfo=timezone.utc) else parse_ts(promoted_at)
+    return (
+        parse_ts(synced_at)
+        if parse_ts(synced_at) != datetime.min.replace(tzinfo=timezone.utc)
+        else parse_ts(promoted_at)
+    )
 
 
 def short_digest(image_ref: str) -> str:
@@ -96,14 +100,18 @@ def write_summary(records: list[dict[str, Any]], out_path: Path) -> None:
             if isinstance(smoke_obj, dict):
                 smoke = str(smoke_obj.get("status", "unknown"))
 
-        lines.append(f"| {service} | {env} | `{digest}` | {synced_at} | {smoke} | {links_cell(record)} |")
+        lines.append(
+            f"| {service} | {env} | `{digest}` | {synced_at} | {smoke} | {links_cell(record)} |"
+        )
 
     out_path.parent.mkdir(parents=True, exist_ok=True)
     out_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
 
 
 def main() -> int:
-    parser = argparse.ArgumentParser(description="Collect evidence YAML records into JSON index")
+    parser = argparse.ArgumentParser(
+        description="Collect evidence YAML records into JSON index"
+    )
     parser.add_argument("--records-dir", default="evidence/records", type=Path)
     parser.add_argument("--out", default="evidence/index.json", type=Path)
     parser.add_argument("--summary", default="evidence/summary/latest.md", type=Path)
@@ -123,7 +131,9 @@ def main() -> int:
     records.sort(key=record_timestamp, reverse=True)
 
     args.out.parent.mkdir(parents=True, exist_ok=True)
-    args.out.write_text(json.dumps(records, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+    args.out.write_text(
+        json.dumps(records, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
+    )
     write_summary(records, args.summary)
 
     print(f"wrote {args.out} ({len(records)} records)")

scripts/evidence/validate.py

@@ -37,7 +37,9 @@ def record_files(records_dir: Path) -> list[Path]:
 def main() -> int:
     parser = argparse.ArgumentParser(description="Validate evidence YAML records")
     parser.add_argument("--records-dir", default="evidence/records", type=Path)
-    parser.add_argument("--schema", default="evidence/schema/evidence.schema.json", type=Path)
+    parser.add_argument(
+        "--schema", default="evidence/schema/evidence.schema.json", type=Path
+    )
     args = parser.parse_args()
 
     schema = json.loads(args.schema.read_text(encoding="utf-8"))
@@ -65,7 +67,9 @@ def main() -> int:
                     node: Any = data
                     for key in key_path:
                         if not isinstance(node, dict) or key not in node:
-                            raise ValueError(f"missing required field: {'.'.join(key_path)}")
+                            raise ValueError(
+                                f"missing required field: {'.'.join(key_path)}"
+                            )
                         node = node[key]
                     if node in ("", None):
                         raise ValueError(f"empty required field: {'.'.join(key_path)}")

scripts/promoter/deploy_promoter.py

@@ -35,7 +35,9 @@ def read_json_like(path: Path) -> dict[str, Any]:
 
 def write_json_like(path: Path, payload: dict[str, Any]) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+    path.write_text(
+        json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
+    )
 
 
 def ensure_queue_shape(payload: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
@@ -81,7 +83,9 @@ def fetch_manifest_digest(
     }
 
     if username and password:
-        token = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")
+        token = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode(
+            "ascii"
+        )
         headers["Authorization"] = f"Basic {token}"
 
     methods = ["HEAD", "GET"]
@@ -100,14 +104,20 @@ def fetch_manifest_digest(
         except urllib.error.HTTPError as exc:
             last_error = exc
             if exc.code == 404:
-                raise RuntimeError(f"manifest not found in Harbor: {repository}:{reference}") from exc
+                raise RuntimeError(
+                    f"manifest not found in Harbor: {repository}:{reference}"
+                ) from exc
         except Exception as exc:  # noqa: BLE001
             last_error = exc
 
-    raise RuntimeError(f"failed to query Harbor manifest {repository}:{reference}: {last_error}")
+    raise RuntimeError(
+        f"failed to query Harbor manifest {repository}:{reference}: {last_error}"
+    )
 
 
-def choose_latest_pending(entries: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+def choose_latest_pending(
+    entries: list[dict[str, Any]],
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
     """Keep only latest pending per service and mark older as superseded."""
     grouped: dict[str, list[dict[str, Any]]] = {}
     for entry in entries:
@@ -128,7 +138,9 @@ def choose_latest_pending(entries: list[dict[str, Any]]) -> tuple[list[dict[str,
         for stale in service_entries[:-1]:
             stale_copy = deepcopy(stale)
             stale_copy["status"] = "superseded"
-            stale_copy["last_error"] = "superseded by newer pending entry for same service"
+            stale_copy["last_error"] = (
+                "superseded by newer pending entry for same service"
+            )
             stale_copy["updated_at"] = now_utc()
             superseded.append(stale_copy)
 
@@ -141,13 +153,22 @@ def get_deploy_commit() -> str:
         return env_sha
 
     try:
-        output = subprocess.check_output(["git", "rev-parse", "HEAD"], stderr=subprocess.DEVNULL)
+        output = subprocess.check_output(
+            ["git", "rev-parse", "HEAD"], stderr=subprocess.DEVNULL
+        )
         return output.decode("utf-8").strip()
     except Exception:  # noqa: BLE001
         return "unknown"
 
 
-def upsert_env_manifest(env_dir: Path, service: str, environment: str, image_ref: str, queue_id: str, dry_run: bool) -> str:
+def upsert_env_manifest(
+    env_dir: Path,
+    service: str,
+    environment: str,
+    image_ref: str,
+    queue_id: str,
+    dry_run: bool,
+) -> str:
     path = env_dir / f"{service}.yaml"
     payload = read_json_like(path)
 
@@ -189,7 +210,11 @@ def upsert_evidence_record(
     image = entry.get("image", {})
     repository = str(image.get("repository", ""))
     tag = str(image.get("tag", ""))
-    existing_smoke = existing.get("tests", {}).get("smoke") or existing.get("smoke") or {"status": "unknown"}
+    existing_smoke = (
+        existing.get("tests", {}).get("smoke")
+        or existing.get("smoke")
+        or {"status": "unknown"}
+    )
 
     payload: dict[str, Any] = {
         "record_id": record_id,
@@ -204,7 +229,9 @@ def upsert_evidence_record(
             "queue_id": queue_id,
             "promoter": "deploy-promoter",
             "commit": commit,
-            "promoted_at": now_utc() if outcome_status == "promoted" else existing.get("deploy", {}).get("promoted_at"),
+            "promoted_at": now_utc()
+            if outcome_status == "promoted"
+            else existing.get("deploy", {}).get("promoted_at"),
         },
         "tests": {
             "smoke": existing_smoke,
@@ -301,7 +328,9 @@ def promote(
                 logs.append(f"manifest available for {queue_id}: {digest}")
 
             image_ref = f"{repository}@{digest}"
-            env_manifest_path = upsert_env_manifest(args.env_dir, service, environment, image_ref, queue_id, args.dry_run)
+            env_manifest_path = upsert_env_manifest(
+                args.env_dir, service, environment, image_ref, queue_id, args.dry_run
+            )
 
             entry["attempts"] = attempts + 1
             entry["status"] = "promoted"
@@ -346,23 +375,34 @@ def promote(
                 logs.append(f"{queue_id} failed after {attempts} attempt(s): {error}")
             else:
                 queue["pending"].append(entry)
-                logs.append(f"{queue_id} retry scheduled ({attempts}/{max_attempts}): {error}")
+                logs.append(
+                    f"{queue_id} retry scheduled ({attempts}/{max_attempts}): {error}"
+                )
 
     return queue, logs
 
 
 def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(description="Auto-promote pending queue items for dev")
+    parser = argparse.ArgumentParser(
+        description="Auto-promote pending queue items for dev"
+    )
     parser.add_argument("--queue", default="release/queue.yaml", type=Path)
     parser.add_argument("--env-dir", default="envs/dev", type=Path)
     parser.add_argument("--evidence-dir", default="evidence/records", type=Path)
     parser.add_argument("--max-attempts", default=3, type=int)
-    parser.add_argument("--harbor-url", default=os.getenv("HARBOR_URL", "https://harbor.omniverseai.net"))
+    parser.add_argument(
+        "--harbor-url",
+        default=os.getenv("HARBOR_URL", "https://harbor.omniverseai.net"),
+    )
     parser.add_argument("--harbor-username", default=os.getenv("HARBOR_USERNAME"))
     parser.add_argument("--harbor-password", default=os.getenv("HARBOR_PASSWORD"))
     parser.add_argument("--timeout", default=8.0, type=float)
     parser.add_argument("--dry-run", action="store_true")
-    parser.add_argument("--skip-registry-check", action="store_true", help="use synthetic digest in non-network dry runs")
+    parser.add_argument(
+        "--skip-registry-check",
+        action="store_true",
+        help="use synthetic digest in non-network dry runs",
+    )
     return parser.parse_args()
 
 

scripts/promoter/promote.py

@@ -49,7 +49,9 @@ def yaml_load(path: Path, default: Any) -> Any:
 
 def yaml_dump(path: Path, data: Any) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
+    path.write_text(
+        yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8"
+    )
 
 
 def ensure_queue_shape(queue: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
@@ -130,7 +132,9 @@ def harbor_repo_path(harbor_image: str, harbor_url: str) -> str:
     return image
 
 
-def harbor_manifest_ready(harbor_image: str, digest: str, harbor_url: str, harbor_user: str, harbor_pass: str) -> bool:
+def harbor_manifest_ready(
+    harbor_image: str, digest: str, harbor_url: str, harbor_user: str, harbor_pass: str
+) -> bool:
     repo = harbor_repo_path(harbor_image, harbor_url)
     if not repo or not digest:
         return False
@@ -156,7 +160,11 @@ def ghcr_repo(ghcr_ref: str) -> str:
 
 
 def commit_from_tag(tag: str) -> str:
-    return tag[4:] if isinstance(tag, str) and tag.startswith("sha-") and len(tag) > 4 else "unknown"
+    return (
+        tag[4:]
+        if isinstance(tag, str) and tag.startswith("sha-") and len(tag) > 4
+        else "unknown"
+    )
 
 
 def safe_tag(tag: str) -> str:
@@ -183,7 +191,9 @@ def load_service_map(path: Path) -> dict[str, Any]:
     return data
 
 
-def resolve_target(service_map: dict[str, Any], service: str, env: str) -> dict[str, Any]:
+def resolve_target(
+    service_map: dict[str, Any], service: str, env: str
+) -> dict[str, Any]:
     services = service_map.get("services", {})
     svc = services.get(service, {}) if isinstance(services, dict) else {}
     if not isinstance(svc, dict):
@@ -198,7 +208,9 @@ def resolve_target(service_map: dict[str, Any], service: str, env: str) -> dict[
     return {}
 
 
-def normalize_pending(queue: dict[str, list[dict[str, Any]]], now: str) -> tuple[dict[str, list[dict[str, Any]]], bool]:
+def normalize_pending(
+    queue: dict[str, list[dict[str, Any]]], now: str
+) -> tuple[dict[str, list[dict[str, Any]]], bool]:
     changed = False
     pending = queue["pending"]
     promoted_ids = set(by_id(queue["promoted"]).keys())
@@ -215,7 +227,10 @@ def normalize_pending(queue: dict[str, list[dict[str, Any]]], now: str) -> tuple
 
     grouped: dict[tuple[str, str], list[dict[str, Any]]] = {}
     for entry in filtered:
-        key = (str(entry.get("service", "")).strip(), str(entry.get("env", "dev")).strip())
+        key = (
+            str(entry.get("service", "")).strip(),
+            str(entry.get("env", "dev")).strip(),
+        )
         grouped.setdefault(key, []).append(entry)
 
     new_pending: list[dict[str, Any]] = []
@@ -263,7 +278,11 @@ def update_overlay_kustomization(
             idx = i
             break
 
-    target = copy.deepcopy(images[idx]) if idx is not None and isinstance(images[idx], dict) else {}
+    target = (
+        copy.deepcopy(images[idx])
+        if idx is not None and isinstance(images[idx], dict)
+        else {}
+    )
     target["name"] = image_name
     target["newName"] = harbor_image
     target["digest"] = digest
@@ -289,7 +308,13 @@ def process_pending(
     harbor_user: str,
     harbor_pass: str,
     skip_registry_check: bool,
-) -> tuple[dict[str, list[dict[str, Any]]], dict[Path, dict[str, Any]], dict[Path, dict[str, Any]], list[dict[str, str]], bool]:
+) -> tuple[
+    dict[str, list[dict[str, Any]]],
+    dict[Path, dict[str, Any]],
+    dict[Path, dict[str, Any]],
+    list[dict[str, str]],
+    bool,
+]:
     now = now_rfc3339()
     queue, changed = normalize_pending(queue, now)
 
@@ -300,15 +325,31 @@ def process_pending(
     for entry in list(queue["pending"]):
         service = str(entry.get("service", "")).strip()
         env = str(entry.get("env", "dev")).strip() or "dev"
-        source = entry.get("source", {}) if isinstance(entry.get("source"), dict) else {}
+        source = (
+            entry.get("source", {}) if isinstance(entry.get("source"), dict) else {}
+        )
         tag = str(source.get("tag", "unknown"))
         digest = get_digest(entry)
 
         target = resolve_target(service_map, service, env)
-        overlay_rel = str(target.get("overlayPath", "")).strip() if isinstance(target, dict) else ""
+        overlay_rel = (
-        image_name = str(target.get("kustomizeImageName", "")).strip() if isinstance(target, dict) else ""
+            str(target.get("overlayPath", "")).strip()
-        harbor_image = str(target.get("harborImage", "")).strip() if isinstance(target, dict) else ""
+            if isinstance(target, dict)
-        argocd_app = str(target.get("argocdApp", "")).strip() if isinstance(target, dict) else ""
+            else ""
+        )
+        image_name = (
+            str(target.get("kustomizeImageName", "")).strip()
+            if isinstance(target, dict)
+            else ""
+        )
+        harbor_image = (
+            str(target.get("harborImage", "")).strip()
+            if isinstance(target, dict)
+            else ""
+        )
+        argocd_app = (
+            str(target.get("argocdApp", "")).strip() if isinstance(target, dict) else ""
+        )
 
         if not harbor_image:
             harbor_image = f"harbor.omniverseai.net/app/{service}"
@@ -316,16 +357,22 @@ def process_pending(
         if not service or not env or not digest or not overlay_rel or not image_name:
             attempts = int(entry.get("attempts", 0)) + 1
             entry["attempts"] = attempts
-            entry["lastError"] = "invalid entry or missing service mapping in release/services.yaml"
+            entry["lastError"] = (
+                "invalid entry or missing service mapping in release/services.yaml"
+            )
             if attempts >= retry_max:
                 entry["status"] = "failed"
                 entry["failedAt"] = now_rfc3339()
-                queue["pending"] = [e for e in queue["pending"] if entry_id(e) != entry_id(entry)]
+                queue["pending"] = [
+                    e for e in queue["pending"] if entry_id(e) != entry_id(entry)
+                ]
                 upsert_entry(queue["failed"], entry)
             changed = True
             continue
 
-        if not skip_registry_check and not harbor_manifest_ready(harbor_image, digest, harbor_url, harbor_user, harbor_pass):
+        if not skip_registry_check and not harbor_manifest_ready(
+            harbor_image, digest, harbor_url, harbor_user, harbor_pass
+        ):
             continue
 
         overlay_path = repo_dir / overlay_rel
@@ -336,7 +383,9 @@ def process_pending(
             if attempts >= retry_max:
                 entry["status"] = "failed"
                 entry["failedAt"] = now_rfc3339()
-                queue["pending"] = [e for e in queue["pending"] if entry_id(e) != entry_id(entry)]
+                queue["pending"] = [
+                    e for e in queue["pending"] if entry_id(e) != entry_id(entry)
+                ]
                 upsert_entry(queue["failed"], entry)
             changed = True
             continue
@@ -364,9 +413,13 @@ def process_pending(
             "service": service,
             "env": env,
             "source": {
-                "repo": ghcr_repo(ghcr_ref) if ghcr_ref else f"ghcr.io/unknown/{service}",
+                "repo": ghcr_repo(ghcr_ref)
+                if ghcr_ref
+                else f"ghcr.io/unknown/{service}",
                 "commit": commit_from_tag(tag),
-                "workflowRun": str(source.get("workflowRun", "")) if source.get("workflowRun") else "",
+                "workflowRun": str(source.get("workflowRun", ""))
+                if source.get("workflowRun")
+                else "",
             },
             "image": {
                 "ghcr": ghcr_ref,
@@ -375,7 +428,10 @@ def process_pending(
             "deploy": {
                 "deployRepoCommit": "__PENDING_COMMIT__",
                 "queueId": entry_id(entry),
-                "argocdApp": argocd_app or existing.get("deploy", {}).get("argocdApp", f"{service}-{env}") if isinstance(existing.get("deploy"), dict) else f"{service}-{env}",
+                "argocdApp": argocd_app
+                or existing.get("deploy", {}).get("argocdApp", f"{service}-{env}")
+                if isinstance(existing.get("deploy"), dict)
+                else f"{service}-{env}",
                 "overlayPath": overlay_rel,
                 "syncedAt": promoted_at,
             },
@@ -385,7 +441,9 @@ def process_pending(
                     "checkedAt": "",
                 }
             },
-            "approvals": existing.get("approvals", {}) if isinstance(existing.get("approvals"), dict) else {},
+            "approvals": existing.get("approvals", {})
+            if isinstance(existing.get("approvals"), dict)
+            else {},
         }
 
         evidence_changes[evidence_path] = record
@@ -394,23 +452,31 @@ def process_pending(
         entry["status"] = "promoted"
         entry["promotedAt"] = promoted_at
         entry["lastError"] = ""
-        queue["pending"] = [e for e in queue["pending"] if entry_id(e) != entry_id(entry)]
+        queue["pending"] = [
+            e for e in queue["pending"] if entry_id(e) != entry_id(entry)
+        ]
         upsert_entry(queue["promoted"], entry)
 
-        promoted_meta.append({
+        promoted_meta.append(
-            "service": service,
+            {
-            "tag": tag,
+                "service": service,
-            "evidence_path": str(evidence_path),
+                "tag": tag,
-            "overlay_path": overlay_rel,
+                "evidence_path": str(evidence_path),
-        })
+                "overlay_path": overlay_rel,
+            }
+        )
 
     return queue, overlay_changes, evidence_changes, promoted_meta, changed
 
 
-def run(cmd: list[str], cwd: Path | None = None, check: bool = True) -> subprocess.CompletedProcess[str]:
+def run(
+    cmd: list[str], cwd: Path | None = None, check: bool = True
+) -> subprocess.CompletedProcess[str]:
     proc = subprocess.run(cmd, cwd=cwd, text=True, capture_output=True)
     if check and proc.returncode != 0:
-        raise RuntimeError(f"command failed: {' '.join(cmd)}\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr}")
+        raise RuntimeError(
+            f"command failed: {' '.join(cmd)}\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr}"
+        )
     return proc
 
 
@@ -450,12 +516,24 @@ def commit_and_push(
     run(["python3", "scripts/evidence/validate.py"], cwd=repo_dir)
 
     run(["git", "config", "user.name", "deploy-promoter[bot]"], cwd=repo_dir)
-    run(["git", "config", "user.email", "deploy-promoter[bot]@users.noreply.github.com"], cwd=repo_dir)
+    run(
+        [
+            "git",
+            "config",
+            "user.email",
+            "deploy-promoter[bot]@users.noreply.github.com",
+        ],
+        cwd=repo_dir,
+    )
 
     stage_paths = ["release/queue.yaml", "evidence/records"]
-    stage_paths.extend(item["overlay_path"] for item in promoted_meta if item.get("overlay_path"))
+    stage_paths.extend(
+        item["overlay_path"] for item in promoted_meta if item.get("overlay_path")
+    )
     run(["git", "add", *sorted(set(stage_paths))], cwd=repo_dir)
-    diff = run(["git", "diff", "--cached", "--name-only"], cwd=repo_dir, check=False).stdout.strip()
+    diff = run(
+        ["git", "diff", "--cached", "--name-only"], cwd=repo_dir, check=False
+    ).stdout.strip()
     if not diff:
         print("No changes to commit")
         return
@@ -476,7 +554,9 @@ def commit_and_push(
         record = yaml_load(path, default={})
         if not isinstance(record, dict):
             continue
-        deploy = record.get("deploy", {}) if isinstance(record.get("deploy"), dict) else {}
+        deploy = (
+            record.get("deploy", {}) if isinstance(record.get("deploy"), dict) else {}
+        )
         if deploy.get("deployRepoCommit") != sha:
             deploy["deployRepoCommit"] = sha
             record["deploy"] = deploy
@@ -485,27 +565,58 @@ def commit_and_push(
 
     if touched:
         run(["python3", "scripts/evidence/validate.py"], cwd=repo_dir)
-        run(["git", "add", *[str(p.relative_to(repo_dir)) for p in touched]], cwd=repo_dir)
+        run(
+            ["git", "add", *[str(p.relative_to(repo_dir)) for p in touched]],
+            cwd=repo_dir,
+        )
         run(["git", "commit", "--amend", "--no-edit"], cwd=repo_dir)
 
     run(["git", "push", "origin", "HEAD:main"], cwd=repo_dir)
 
 
 def main() -> int:
-    parser = argparse.ArgumentParser(description="Promote ready release queue entries to Argo-consumed overlays")
+    parser = argparse.ArgumentParser(
+        description="Promote ready release queue entries to Argo-consumed overlays"
+    )
     parser.add_argument("--local-repo-dir", default=os.getenv("LOCAL_REPO_DIR", ""))
-    parser.add_argument("--deploy-repo-url", default=os.getenv("DEPLOY_REPO_URL", "https://github.com/BrunoGaoSZ/ljwx-deploy.git"))
+    parser.add_argument(
-    parser.add_argument("--deploy-repo-token", default=os.getenv("DEPLOY_REPO_TOKEN", ""))
+        "--deploy-repo-url",
-    parser.add_argument("--service-map", default=os.getenv("SERVICE_MAP_PATH", "release/services.yaml"))
+        default=os.getenv(
-    parser.add_argument("--harbor-url", default=os.getenv("HARBOR_URL", "https://harbor.omniverseai.net"))
+            "DEPLOY_REPO_URL", "https://github.com/BrunoGaoSZ/ljwx-deploy.git"
+        ),
+    )
+    parser.add_argument(
+        "--deploy-repo-token", default=os.getenv("DEPLOY_REPO_TOKEN", "")
+    )
+    parser.add_argument(
+        "--service-map", default=os.getenv("SERVICE_MAP_PATH", "release/services.yaml")
+    )
+    parser.add_argument(
+        "--harbor-url",
+        default=os.getenv("HARBOR_URL", "https://harbor.omniverseai.net"),
+    )
     parser.add_argument("--harbor-user", default=os.getenv("HARBOR_USER", ""))
     parser.add_argument("--harbor-pass", default=os.getenv("HARBOR_PASS", ""))
-    parser.add_argument("--retry-max", type=int, default=int(os.getenv("RETRY_MAX", "10")))
+    parser.add_argument(
-    parser.add_argument("--dry-run", action="store_true", help="simulate without writing/committing")
+        "--retry-max", type=int, default=int(os.getenv("RETRY_MAX", "10"))
-    parser.add_argument("--skip-registry-check", action="store_true", help="skip Harbor manifest check (local dry-run only)")
+    )
+    parser.add_argument(
+        "--dry-run", action="store_true", help="simulate without writing/committing"
+    )
+    parser.add_argument(
+        "--skip-registry-check",
+        action="store_true",
+        help="skip Harbor manifest check (local dry-run only)",
+    )
     args = parser.parse_args()
 
-    dry_run = bool(args.dry_run) or os.getenv("DRY_RUN", "0") in {"1", "true", "True", "yes", "YES"}
+    dry_run = bool(args.dry_run) or os.getenv("DRY_RUN", "0") in {
+        "1",
+        "true",
+        "True",
+        "yes",
+        "YES",
+    }
 
     repo_dir, tmp_root = repo_workdir(args)
     try:
@@ -518,15 +629,17 @@ def main() -> int:
 
         service_map = load_service_map(repo_dir / args.service_map)
 
-        queue, overlay_changes, evidence_changes, promoted_meta, changed = process_pending(
+        queue, overlay_changes, evidence_changes, promoted_meta, changed = (
-            queue=queue,
+            process_pending(
-            repo_dir=repo_dir,
+                queue=queue,
-            service_map=service_map,
+                repo_dir=repo_dir,
-            retry_max=args.retry_max,
+                service_map=service_map,
-            harbor_url=args.harbor_url,
+                retry_max=args.retry_max,
-            harbor_user=args.harbor_user,
+                harbor_url=args.harbor_url,
-            harbor_pass=args.harbor_pass,
+                harbor_user=args.harbor_user,
-            skip_registry_check=args.skip_registry_check,
+                harbor_pass=args.harbor_pass,
+                skip_registry_check=args.skip_registry_check,
+            )
         )
 
         if not changed:

scripts/promoter/validate_queue.py

@@ -34,7 +34,19 @@ def main() -> int:
             print(f"missing list key: {key}")
             return 1
 
-    required = ["id", "service", "env", "source", "createdAt", "status", "attempts", "lastError", "promotedAt", "supersededAt", "failedAt"]
+    required = [
+        "id",
+        "service",
+        "env",
+        "source",
+        "createdAt",
+        "status",
+        "attempts",
+        "lastError",
+        "promotedAt",
+        "supersededAt",
+        "failedAt",
+    ]
     for state in ["pending", "promoted", "failed", "superseded"]:
         for idx, item in enumerate(data[state]):
             if not isinstance(item, dict):

scripts/repair/diagnose.py

@@ -38,7 +38,11 @@ def gh_checks_summary(pr_url: str) -> dict[str, Any]:
     if proc_json.returncode == 0:
         try:
             payload = json.loads(proc_json.stdout)
-            failures = [item for item in payload if str(item.get("state", "")).lower() not in {"success", "passed"}]
+            failures = [
+                item
+                for item in payload
+                if str(item.get("state", "")).lower() not in {"success", "passed"}
+            ]
             return {
                 "available": True,
                 "mode": "json",
@@ -71,15 +75,27 @@ def select_recipes(recipes: list[dict[str, Any]], merged_text: str) -> list[str]
     if selected:
         return selected
 
-    return [str(recipe.get("id", "")) for recipe in recipes if bool(recipe.get("default_on_unknown", True))]
+    return [
+        str(recipe.get("id", ""))
+        for recipe in recipes
+        if bool(recipe.get("default_on_unknown", True))
+    ]
 
 
 def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(description="Diagnose failed checks and recommend repair recipes")
+    parser = argparse.ArgumentParser(
-    parser.add_argument("--check-log", type=Path, required=True, help="path to local check log")
+        description="Diagnose failed checks and recommend repair recipes"
+    )
+    parser.add_argument(
+        "--check-log", type=Path, required=True, help="path to local check log"
+    )
     parser.add_argument("--recipes", type=Path, default=Path("repairs/recipes.yaml"))
-    parser.add_argument("--pr-url", default="", help="GitHub PR URL for gh checks summary")
+    parser.add_argument(
-    parser.add_argument("--out", type=Path, default=Path(".factory/repair/diagnose.json"))
+        "--pr-url", default="", help="GitHub PR URL for gh checks summary"
+    )
+    parser.add_argument(
+        "--out", type=Path, default=Path(".factory/repair/diagnose.json")
+    )
     return parser.parse_args()
 
 
@@ -110,8 +126,15 @@ def main() -> int:
     }
 
     args.out.parent.mkdir(parents=True, exist_ok=True)
-    args.out.write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+    args.out.write_text(
-    print(json.dumps({"recommended_recipe_ids": recommended, "diagnose_out": str(args.out)}, ensure_ascii=False))
+        json.dumps(result, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
+    )
+    print(
+        json.dumps(
+            {"recommended_recipe_ids": recommended, "diagnose_out": str(args.out)},
+            ensure_ascii=False,
+        )
+    )
     return 0
 
 

scripts/repair/run_repair.py

@@ -25,7 +25,9 @@ def read_json(path: Path) -> Any:
 
 
 def write_json(path: Path, payload: Any) -> None:
-    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+    path.write_text(
+        json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
+    )
 
 
 def run_check_command(check_cmd: str, log_path: Path) -> tuple[int, str]:
@@ -72,7 +74,9 @@ def action_format_json(repo_root: Path) -> int:
 def action_strip_trailing_whitespace(repo_root: Path) -> int:
     changed = 0
     pattern = re.compile(r"[ \t]+$", re.MULTILINE)
-    for path in iter_files(repo_root, (".py", ".sh", ".md", ".yaml", ".yml", ".js", ".ts")):
+    for path in iter_files(
+        repo_root, (".py", ".sh", ".md", ".yaml", ".yml", ".js", ".ts")
+    ):
         text = path.read_text(encoding="utf-8")
         updated = pattern.sub("", text)
         if updated != text:
@@ -82,7 +86,12 @@ def action_strip_trailing_whitespace(repo_root: Path) -> int:
 
 
 def action_regenerate_evidence_index(repo_root: Path) -> int:
-    cmd = [sys.executable, "scripts/evidence/collect.py", "--out", "evidence/index.json"]
+    cmd = [
+        sys.executable,
+        "scripts/evidence/collect.py",
+        "--out",
+        "evidence/index.json",
+    ]
     proc = subprocess.run(cmd, cwd=repo_root, text=True, capture_output=True)
     if proc.returncode != 0:
         raise RuntimeError(f"collect.py failed: {proc.stderr.strip()}")
@@ -116,7 +125,9 @@ ACTION_MAP = {
 }
 
 
-def select_recipes(recipes: list[dict[str, Any]], check_output: str) -> list[dict[str, Any]]:
+def select_recipes(
+    recipes: list[dict[str, Any]], check_output: str
+) -> list[dict[str, Any]]:
     text = check_output.lower()
     selected: list[dict[str, Any]] = []
 
@@ -128,10 +139,14 @@ def select_recipes(recipes: list[dict[str, Any]], check_output: str) -> list[dic
     if selected:
         return selected
 
-    return [recipe for recipe in recipes if bool(recipe.get("default_on_unknown", True))]
+    return [
+        recipe for recipe in recipes if bool(recipe.get("default_on_unknown", True))
+    ]
 
 
-def run_recipes(recipes: list[dict[str, Any]], repo_root: Path) -> tuple[int, list[str]]:
+def run_recipes(
+    recipes: list[dict[str, Any]], repo_root: Path
+) -> tuple[int, list[str]]:
     total_changed = 0
     logs: list[str] = []
     for recipe in recipes:
@@ -148,7 +163,9 @@ def run_recipes(recipes: list[dict[str, Any]], repo_root: Path) -> tuple[int, li
     return total_changed, logs
 
 
-def open_issue(repo: str, title: str, body: str, labels: list[str], dry_run: bool) -> str:
+def open_issue(
+    repo: str, title: str, body: str, labels: list[str], dry_run: bool
+) -> str:
     if dry_run:
         return "dry-run: issue creation skipped"
 
@@ -156,7 +173,9 @@ def open_issue(repo: str, title: str, body: str, labels: list[str], dry_run: boo
     if not token:
         return "issue skipped: missing GITHUB_TOKEN/GH_TOKEN"
 
-    payload = json.dumps({"title": title, "body": body, "labels": labels}).encode("utf-8")
+    payload = json.dumps({"title": title, "body": body, "labels": labels}).encode(
+        "utf-8"
+    )
     url = f"https://api.github.com/repos/{repo}/issues"
     req = urllib.request.Request(
         url,
@@ -180,7 +199,9 @@ def open_issue(repo: str, title: str, body: str, labels: list[str], dry_run: boo
 
 
 def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(description="Auto-repair common failures and retry checks")
+    parser = argparse.ArgumentParser(
+        description="Auto-repair common failures and retry checks"
+    )
     parser.add_argument("--recipes", type=Path, default=Path("repairs/recipes.yaml"))
     parser.add_argument("--check-cmd", default="bash scripts/ci/run_checks.sh")
     parser.add_argument("--max-attempts", type=int, default=3)
@@ -223,7 +244,9 @@ def main() -> int:
         selected = select_recipes(recipes, output)
         changed, repair_logs = run_recipes(selected, repo_root)
         all_logs.extend(repair_logs)
-        print(f"attempt {attempt} failed; applied {len(selected)} recipe(s), changed={changed}")
+        print(
+            f"attempt {attempt} failed; applied {len(selected)} recipe(s), changed={changed}"
+        )
         for line in repair_logs:
             print(f"  {line}")
 
@@ -251,7 +274,9 @@ def main() -> int:
     print(f"repair failed after {args.max_attempts} attempts; summary: {failure_note}")
 
     if args.open_issue_on_failure and args.issue_repo:
-        labels = [label.strip() for label in args.issue_labels.split(",") if label.strip()]
+        labels = [
+            label.strip() for label in args.issue_labels.split(",") if label.strip()
+        ]
         result = open_issue(
             repo=args.issue_repo,
             title=args.issue_title,

scripts/smoke/run_smoke.py

@@ -38,7 +38,9 @@ def read_yaml(path: Path) -> dict[str, Any]:
 
 
 def write_yaml(path: Path, payload: dict[str, Any]) -> None:
-    path.write_text(yaml.safe_dump(payload, sort_keys=False, allow_unicode=True), encoding="utf-8")
+    path.write_text(
+        yaml.safe_dump(payload, sort_keys=False, allow_unicode=True), encoding="utf-8"
+    )
 
 
 def parse_ts(value: Any) -> datetime:
@@ -68,7 +70,9 @@ def record_files(evidence_dir: Path) -> list[Path]:
     return sorted([p for p in evidence_dir.glob("*.yaml") if p.is_file()])
 
 
-def find_record_path(evidence_dir: Path, service: str, environment: str, queue_id: str | None) -> Path | None:
+def find_record_path(
+    evidence_dir: Path, service: str, environment: str, queue_id: str | None
+) -> Path | None:
     candidates: list[tuple[datetime, Path]] = []
     for path in record_files(evidence_dir):
         try:
@@ -80,7 +84,11 @@ def find_record_path(evidence_dir: Path, service: str, environment: str, queue_i
             continue
 
         deploy = record.get("deploy", {})
-        if queue_id and isinstance(deploy, dict) and str(deploy.get("queueId", "")) != str(queue_id):
+        if (
+            queue_id
+            and isinstance(deploy, dict)
+            and str(deploy.get("queueId", "")) != str(queue_id)
+        ):
             continue
 
         smoke_status = (
@@ -100,7 +108,9 @@ def find_record_path(evidence_dir: Path, service: str, environment: str, queue_i
     return candidates[0][1]
 
 
-def http_get(url: str, headers: dict[str, str] | None = None, timeout: float = 5.0) -> tuple[int, str]:
+def http_get(
+    url: str, headers: dict[str, str] | None = None, timeout: float = 5.0
+) -> tuple[int, str]:
     req = urllib.request.Request(url, headers=headers or {}, method="GET")
     with urllib.request.urlopen(req, timeout=timeout) as resp:
         body = resp.read(4096).decode("utf-8", errors="ignore")
@@ -142,7 +152,9 @@ def wait_for_argocd_health(
     return False, f"argocd wait timeout: {last}"
 
 
-def wait_for_endpoint(endpoint: str, timeout_seconds: int, interval_seconds: int) -> tuple[bool, str]:
+def wait_for_endpoint(
+    endpoint: str, timeout_seconds: int, interval_seconds: int
+) -> tuple[bool, str]:
     end = time.time() + timeout_seconds
     last = ""
 
@@ -191,7 +203,10 @@ def run_target(target: dict[str, Any], args: argparse.Namespace) -> tuple[str, s
 
     record_path = find_record_path(args.evidence_dir, service, environment, queue_id)
     if not record_path:
-        return "skip", f"{service}: no promoted evidence record for {service}/{environment}"
+        return (
+            "skip",
+            f"{service}: no promoted evidence record for {service}/{environment}",
+        )
 
     ok_argocd, argocd_details = wait_for_argocd_health(
         server=args.argocd_server,
@@ -218,8 +233,12 @@ def run_target(target: dict[str, Any], args: argparse.Namespace) -> tuple[str, s
 
 
 def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(description="Run smoke checks and write evidence results")
+    parser = argparse.ArgumentParser(
-    parser.add_argument("--targets", type=Path, default=Path("scripts/smoke/targets.json"))
+        description="Run smoke checks and write evidence results"
+    )
+    parser.add_argument(
+        "--targets", type=Path, default=Path("scripts/smoke/targets.json")
+    )
     parser.add_argument("--evidence-dir", type=Path, default=Path("evidence/records"))
     parser.add_argument("--argocd-server", default=os.getenv("ARGOCD_SERVER", ""))
     parser.add_argument("--argocd-token", default=os.getenv("ARGOCD_TOKEN", ""))
@@ -247,7 +266,16 @@ def main() -> int:
         else:
             skipped += 1
 
-    print(json.dumps({"passed": passed, "failed": failed, "skipped": skipped, "dry_run": args.dry_run}))
+    print(
+        json.dumps(
+            {
+                "passed": passed,
+                "failed": failed,
+                "skipped": skipped,
+                "dry_run": args.dry_run,
+            }
+        )
+    )
     return 0 if failed == 0 else 1