Coverage for scripts / example_job_validation / static_checks.py: 100.00%
266 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Offline validation of every example against its documented contract.
3Runs with no AWS access and no cluster: parses each example, checks the
4spec registry's symmetry with the ``examples/`` directory and the
5``gco_mcp`` catalog, and — for examples documented to travel the API/SQS
6submission paths — proves every document clears the exact transport gates
7(kind/GVK allowlist, image-source trust, target namespace) that the
8deployed services enforce. This is the half that runs in CI on every PR
9(``tests/test_example_job_validation.py``); the live half in ``drivers.py``
10builds on the same parse.
11"""
13from __future__ import annotations
15import copy
16import re
17from dataclasses import dataclass, field
18from pathlib import Path
19from typing import Any
21import yaml
23from .specs import (
24 CAPABILITY_TYPES,
25 COMPANION,
26 DAG_RUN,
27 EXAMPLE_SPECS,
28 SUBMISSION_PATHS,
29 SUBMIT_API,
30 SUBMIT_DIRECT,
31 SUBMIT_SQS,
32 ExampleSpec,
33)
35#: Namespaces the platform provisions for user workloads.
36_WORKLOAD_NAMESPACES = frozenset({"gco-jobs", "gco-inference"})
38#: This repository's clone URLs. An Argo CD example must source one of them so
39#: the offline checks (and the live waiter) can read exactly what it syncs.
40THIS_REPOSITORY_URLS = frozenset(
41 {
42 "https://github.com/aws-solutions-library-samples/global-capacity-orchestrator-on-aws",
43 "https://github.com/aws-solutions-library-samples/global-capacity-orchestrator-on-aws.git",
44 }
45)
47#: The self-managed Argo CD's namespace, fenced project and in-cluster server
48#: (gco/argocd_config.py; post-helm-argocd-access.yaml).
49_ARGOCD_NAMESPACE = "argocd"
50_ARGOCD_PROJECT = "gco-tenants"
51_IN_CLUSTER_SERVER = "https://kubernetes.default.svc"
53#: A whole-line Go template action (``{{ if ... }}`` / ``{{ end }}``). The
54#: static read drops these lines; inline actions sit inside quoted YAML
55#: strings, so what is left parses as the YAML the function renders.
56_TEMPLATE_ACTION_LINE = re.compile(r"^[ \t]*\{\{.*\}\}[ \t]*$", re.MULTILINE)
59@dataclass
60class StaticFinding:
61 """One offline check outcome for one example."""
63 example: str
64 check: str
65 passed: bool
66 detail: str = ""
69@dataclass
70class ParsedExample:
71 """An example file parsed into documents plus its spec."""
73 name: str
74 path: Path
75 spec: ExampleSpec
76 documents: list[dict[str, Any]] = field(default_factory=list)
79def examples_dir(repo_root: Path) -> Path:
80 return repo_root / "examples"
83def example_names(repo_root: Path) -> list[str]:
84 return sorted(path.stem for path in examples_dir(repo_root).glob("*.yaml"))
87def parse_example(repo_root: Path, name: str) -> ParsedExample:
88 """Parse one example's YAML documents (raises on unknown name or bad YAML)."""
89 spec = EXAMPLE_SPECS.get(name)
90 if spec is None:
91 raise KeyError(f"No validation spec for example {name!r} (add one in specs.py)")
92 path = examples_dir(repo_root) / f"{name}.yaml"
93 documents = [
94 doc for doc in yaml.safe_load_all(path.read_text(encoding="utf-8")) if doc is not None
95 ]
96 return ParsedExample(name=name, path=path, spec=spec, documents=documents)
99def _catalog_metadata(repo_root: Path) -> dict[str, dict[str, Any]]:
100 """Read ``gco_mcp``'s EXAMPLE_METADATA literal without executing the module.
102 ``gco_mcp/resources/docs.py`` imports flat sibling modules (the MCP server
103 puts ``gco_mcp/`` itself on ``sys.path``), so importing it from here would
104 require path surgery. The catalog is a pure literal dict, so an AST read
105 is sufficient — and side-effect free.
106 """
107 import ast
109 docs_path = repo_root / "gco_mcp" / "resources" / "docs.py"
110 tree = ast.parse(docs_path.read_text(encoding="utf-8"))
111 for node in ast.walk(tree):
112 value: ast.expr | None = None
113 targets: list[ast.expr] = []
114 if isinstance(node, ast.Assign):
115 targets, value = node.targets, node.value
116 elif isinstance(node, ast.AnnAssign) and node.value is not None:
117 targets, value = [node.target], node.value
118 for target in targets:
119 if isinstance(target, ast.Name) and target.id == "EXAMPLE_METADATA":
120 assert value is not None
121 catalog = ast.literal_eval(value)
122 if not isinstance(catalog, dict):
123 raise RuntimeError("EXAMPLE_METADATA is not a dict literal")
124 return catalog
125 raise RuntimeError(f"EXAMPLE_METADATA literal not found in {docs_path}")
128def check_registry_symmetry(repo_root: Path) -> list[StaticFinding]:
129 """Specs, files, and the MCP catalog must describe the same example set."""
130 findings: list[StaticFinding] = []
131 files = set(example_names(repo_root))
132 specs = set(EXAMPLE_SPECS)
133 catalog = set(_catalog_metadata(repo_root))
134 findings.append(
135 StaticFinding(
136 example="*",
137 check="spec/file symmetry",
138 passed=files == specs,
139 detail=(
140 f"only in examples/: {sorted(files - specs)}; only in specs: {sorted(specs - files)}"
141 if files != specs
142 else ""
143 ),
144 )
145 )
146 findings.append(
147 StaticFinding(
148 example="*",
149 check="spec/catalog symmetry",
150 passed=catalog == specs,
151 detail=(
152 f"only in catalog: {sorted(catalog - specs)}; only in specs: {sorted(specs - catalog)}"
153 if catalog != specs
154 else ""
155 ),
156 )
157 )
158 return findings
161def check_submission_matches_catalog(repo_root: Path, name: str) -> StaticFinding:
162 """The spec's submission path must agree with the catalog's documented command."""
163 meta = _catalog_metadata(repo_root).get(name, {})
164 documented = str(meta.get("submission", ""))
165 spec = EXAMPLE_SPECS[name]
166 expectations = {
167 SUBMIT_DIRECT: "gco jobs submit-direct",
168 SUBMIT_SQS: "gco jobs submit-sqs",
169 SUBMIT_API: "gco jobs submit ",
170 DAG_RUN: "gco dag run",
171 }
172 if spec.submission == SUBMIT_DIRECT:
173 # Inference examples document `gco inference deploy` as the
174 # recommended path and manifest-direct submission as the alternative;
175 # both are valid documented shapes for a submit-direct spec.
176 ok = "gco jobs submit-direct" in documented or "gco inference deploy" in documented
177 detail = "" if ok else f"catalog documents {documented!r}, spec says {spec.submission}"
178 elif spec.submission in expectations:
179 ok = expectations[spec.submission] in documented
180 detail = "" if ok else f"catalog documents {documented!r}, spec says {spec.submission}"
181 elif spec.submission == COMPANION:
182 ok = True
183 detail = ""
184 else: # kubectl-apply
185 ok = "kubectl apply" in documented or documented == ""
186 detail = "" if ok else f"catalog documents {documented!r}, spec says kubectl-apply"
187 return StaticFinding(example=name, check="documented submission path", passed=ok, detail=detail)
190def check_transport_acceptance(parsed: ParsedExample) -> list[StaticFinding]:
191 """API/SQS-documented examples must clear the deployed validation gates."""
192 if parsed.spec.submission not in {SUBMIT_DIRECT, SUBMIT_SQS, SUBMIT_API}:
193 return []
194 # Module-level pure functions: no Kubernetes client construction, so the
195 # checks run on machines with no kubeconfig (CI runners, fresh laptops).
196 from gco.services.manifest_processor import validate_image_sources, validate_resource_kind
198 findings: list[StaticFinding] = []
199 for doc in parsed.documents:
200 label = f"{doc.get('kind')}/{(doc.get('metadata') or {}).get('name')}"
201 if parsed.spec.submission in {SUBMIT_SQS, SUBMIT_API}:
202 # Only the SQS/API services enforce the kind allowlist;
203 # submit-direct is client-side kubectl and takes any kind.
204 kind_ok, kind_reason = validate_resource_kind(doc)
205 findings.append(
206 StaticFinding(
207 example=parsed.name,
208 check=f"transport kind allowlist ({label})",
209 passed=kind_ok,
210 detail=kind_reason or "",
211 )
212 )
213 image_ok, image_reason = validate_image_sources(doc)
214 findings.append(
215 StaticFinding(
216 example=parsed.name,
217 check=f"trusted image sources ({label})",
218 passed=image_ok,
219 detail=image_reason or "",
220 )
221 )
222 return findings
225def _is_argocd_application(doc: dict[str, Any]) -> bool:
226 return doc.get("kind") == "Application" and str(doc.get("apiVersion", "")).startswith(
227 "argoproj.io/"
228 )
231def application_source_documents(repo_root: Path, doc: dict[str, Any]) -> list[dict[str, Any]]:
232 """The manifests an Argo CD Application syncs from this repository, as Argo CD applies them.
234 Documents without a namespace get the Application's destination namespace
235 (what Argo CD does). Returns ``[]`` when the Application sources another
236 repository or a path outside this checkout: there is nothing to read.
237 """
238 spec = doc.get("spec") or {}
239 source = spec.get("source") or {}
240 if str(source.get("repoURL", "")) not in THIS_REPOSITORY_URLS:
241 return []
242 root = repo_root.resolve()
243 directory = (root / str(source.get("path", ""))).resolve()
244 if not directory.is_relative_to(root) or not directory.is_dir():
245 return []
246 destination = str((spec.get("destination") or {}).get("namespace", ""))
247 documents: list[dict[str, Any]] = []
248 for path in sorted([*directory.glob("*.yaml"), *directory.glob("*.yml")]):
249 for loaded in yaml.safe_load_all(path.read_text(encoding="utf-8")):
250 if not isinstance(loaded, dict):
251 continue
252 item = copy.deepcopy(loaded)
253 metadata = item.setdefault("metadata", {})
254 if not metadata.get("namespace"):
255 metadata["namespace"] = destination
256 documents.append(item)
257 return documents
260def _tenant_namespace(namespace: object) -> object:
261 """A CEL / Go-template namespace expression resolves to the instance's tenant namespace.
263 Instances live in gco-jobs or gco-inference; gco-jobs carries the
264 governance, so the offline checks evaluate templated namespaces there.
265 """
266 if namespace is None or (
267 isinstance(namespace, str) and ("${" in namespace or "{{" in namespace)
268 ):
269 return "gco-jobs"
270 return namespace
273def embedded_workload_documents(repo_root: Path, doc: dict[str, Any]) -> list[dict[str, Any]]:
274 """What an Application, a kro RGD or a go-templating Composition will create.
276 * Argo CD Application: the manifests at its Git path in this checkout.
277 * kro ResourceGraphDefinition: each ``resources[].template``.
278 * Crossplane Composition: each document an inline ``GoTemplate`` renders
279 (the composite's own status document excluded).
281 Templated namespaces resolve to gco-jobs (see :func:`_tenant_namespace`).
282 """
283 kind = doc.get("kind")
284 api_version = str(doc.get("apiVersion", ""))
285 spec = doc.get("spec") or {}
286 if _is_argocd_application(doc):
287 return application_source_documents(repo_root, doc)
288 templates: list[dict[str, Any]] = []
289 if kind == "ResourceGraphDefinition" and api_version.startswith("kro.run/"):
290 templates = [
291 copy.deepcopy(resource["template"])
292 for resource in spec.get("resources") or []
293 if isinstance(resource, dict) and isinstance(resource.get("template"), dict)
294 ]
295 elif kind == "Composition" and api_version.startswith("apiextensions.crossplane.io/"):
296 composite = spec.get("compositeTypeRef") or {}
297 for step in spec.get("pipeline") or []:
298 step_input = (step or {}).get("input") or {}
299 if step_input.get("kind") != "GoTemplate" or step_input.get("source") != "Inline":
300 continue
301 text = _TEMPLATE_ACTION_LINE.sub(
302 "", str((step_input.get("inline") or {}).get("template", ""))
303 )
304 templates.extend(
305 item
306 for item in yaml.safe_load_all(text)
307 if isinstance(item, dict)
308 and (item.get("apiVersion"), item.get("kind"))
309 != (composite.get("apiVersion"), composite.get("kind"))
310 )
311 for template in templates:
312 metadata = template.setdefault("metadata", {})
313 metadata["namespace"] = _tenant_namespace(metadata.get("namespace"))
314 return templates
317def check_namespaces(parsed: ParsedExample) -> list[StaticFinding]:
318 """Namespaced example documents must target a provisioned workload namespace.
320 An Argo CD Application is the one exception: it lives in the argocd
321 namespace by design, so it must instead stay inside the fence — the
322 gco-tenants project, the in-cluster server and a tenant destination.
323 """
324 findings: list[StaticFinding] = []
325 for doc in parsed.documents:
326 metadata = doc.get("metadata") or {}
327 namespace = metadata.get("namespace")
328 kind = str(doc.get("kind", ""))
329 if kind in {"ResourceFlavor", "ClusterQueue"}: # cluster-scoped
330 continue
331 if _is_argocd_application(doc):
332 spec = doc.get("spec") or {}
333 destination = spec.get("destination") or {}
334 problems = [
335 problem
336 for problem, failed in (
337 (
338 f"namespace {namespace!r} is not {_ARGOCD_NAMESPACE!r}",
339 namespace != _ARGOCD_NAMESPACE,
340 ),
341 (
342 f"project {spec.get('project')!r} is not {_ARGOCD_PROJECT!r}",
343 spec.get("project") != _ARGOCD_PROJECT,
344 ),
345 (
346 f"destination server {destination.get('server')!r} is not the in-cluster server",
347 destination.get("server") != _IN_CLUSTER_SERVER,
348 ),
349 (
350 f"destination namespace {destination.get('namespace')!r} is not a workload namespace",
351 destination.get("namespace") not in _WORKLOAD_NAMESPACES,
352 ),
353 )
354 if failed
355 ]
356 findings.append(
357 StaticFinding(
358 example=parsed.name,
359 check=f"Argo CD fence ({kind}/{metadata.get('name')})",
360 passed=not problems,
361 detail="; ".join(problems),
362 )
363 )
364 continue
365 if namespace is None:
366 continue
367 findings.append(
368 StaticFinding(
369 example=parsed.name,
370 check=f"workload namespace ({kind}/{metadata.get('name')})",
371 passed=namespace in _WORKLOAD_NAMESPACES,
372 detail="" if namespace in _WORKLOAD_NAMESPACES else f"namespace {namespace!r}",
373 )
374 )
375 return findings
378def check_spec_shape(name: str) -> StaticFinding:
379 """Spec fields must use known enumerations and name real companions."""
380 spec = EXAMPLE_SPECS[name]
381 problems: list[str] = []
382 if spec.submission not in SUBMISSION_PATHS:
383 problems.append(f"unknown submission path {spec.submission!r}")
384 unknown_capabilities = sorted(set(spec.capability_overrides) - set(CAPABILITY_TYPES))
385 if unknown_capabilities:
386 problems.append(f"unknown EKS capability type(s) {unknown_capabilities}")
387 if spec.ack_iam_policy_arns and "ack" not in spec.capability_overrides:
388 problems.append("ack_iam_policy_arns needs capability_overrides to include 'ack'")
389 if spec.companion:
390 companion = EXAMPLE_SPECS.get(spec.companion)
391 if companion is None or companion.submission != COMPANION:
392 problems.append(f"companion {spec.companion!r} is not a companion-artifact spec")
393 return StaticFinding(
394 example=name,
395 check="spec shape",
396 passed=not problems,
397 detail="; ".join(problems),
398 )
401def check_embedded_workloads(repo_root: Path, parsed: ParsedExample) -> list[StaticFinding]:
402 """What an Application syncs, or an RGD / Composition composes, meets the example rules.
404 These workloads never cross the GCO API, so nothing else checks them
405 before they reach a cluster: they must land in a workload namespace, pull
406 from trusted image sources and fit the default gco-jobs governance. An
407 Application must also source a non-empty path of this repository.
408 """
409 from gco.services.manifest_processor import validate_image_sources
411 findings: list[StaticFinding] = []
412 for doc in parsed.documents:
413 embedded = embedded_workload_documents(repo_root, doc)
414 label = f"{doc.get('kind')}/{(doc.get('metadata') or {}).get('name')}"
415 if _is_argocd_application(doc):
416 findings.append(
417 StaticFinding(
418 example=parsed.name,
419 check=f"Git source resolves ({label})",
420 passed=bool(embedded),
421 detail=""
422 if embedded
423 else "the Application must source a non-empty directory of this repository",
424 )
425 )
426 for item in embedded:
427 item_label = f"{label} -> {item.get('kind')}/{(item.get('metadata') or {}).get('name')}"
428 namespace = (item.get("metadata") or {}).get("namespace")
429 findings.append(
430 StaticFinding(
431 example=parsed.name,
432 check=f"embedded workload namespace ({item_label})",
433 passed=namespace in _WORKLOAD_NAMESPACES,
434 detail="" if namespace in _WORKLOAD_NAMESPACES else f"namespace {namespace!r}",
435 )
436 )
437 image_ok, image_reason = validate_image_sources(item)
438 findings.append(
439 StaticFinding(
440 example=parsed.name,
441 check=f"trusted image sources ({item_label})",
442 passed=image_ok,
443 detail=image_reason or "",
444 )
445 )
446 findings.extend(_governance_findings(parsed, embedded))
447 return findings
450#: Container resource dimensions governed by the gco-jobs LimitRange, mapped
451#: to their per-container default ceiling key in DEFAULT_RESOURCE_QUOTA.
452_LIMIT_RANGE_DIMENSIONS = {
453 "cpu": "container_max_cpu",
454 "memory": "container_max_memory",
455 "nvidia.com/gpu": "container_max_gpu",
456}
458#: Aggregate request dimensions governed by the gco-jobs ResourceQuota.
459_QUOTA_DIMENSIONS = {
460 "cpu": "max_cpu",
461 "memory": "max_memory",
462 "nvidia.com/gpu": "max_gpu",
463}
466def check_resource_governance_fit(parsed: ParsedExample) -> list[StaticFinding]:
467 """Every example must be admissible under the default gco-jobs governance.
469 A container exceeding the LimitRange maxima is rejected at pod creation
470 with only namespace events explaining why, and the Job sits podless until
471 the caller gives up — the previous defaults rejected the platform's own
472 EFA training example exactly that way (live run ex241-df723811). Proving
473 the fit offline keeps the shipped examples and the shipped guardrails
474 from contradicting each other again. Only gco-jobs-namespaced pod specs
475 are checked: the LimitRange and ResourceQuota bind that namespace.
476 """
477 return _governance_findings(parsed, parsed.documents)
480def _governance_findings(
481 parsed: ParsedExample, documents: list[dict[str, Any]]
482) -> list[StaticFinding]:
483 """LimitRange / ResourceQuota (and front-door cap) fit for ``documents``."""
484 from gco.stacks.constants import DEFAULT_RESOURCE_QUOTA, parse_k8s_quantity
486 findings: list[StaticFinding] = []
487 for doc in documents:
488 kind = str(doc.get("kind", ""))
489 metadata = doc.get("metadata") or {}
490 if metadata.get("namespace", "gco-jobs") != "gco-jobs":
491 continue
492 pod_spec, replicas = _pod_spec_and_parallelism(doc, kind)
493 if pod_spec is None:
494 continue
495 containers = list(pod_spec.get("containers") or []) + list(
496 pod_spec.get("initContainers") or []
497 )
498 aggregate: dict[str, float] = dict.fromkeys(_QUOTA_DIMENSIONS, 0.0)
499 for container in containers:
500 resources = container.get("resources") or {}
501 requests = resources.get("requests") or {}
502 limits = resources.get("limits") or {}
503 for dimension, ceiling_key in _LIMIT_RANGE_DIMENSIONS.items():
504 ceiling = parse_k8s_quantity(DEFAULT_RESOURCE_QUOTA[ceiling_key])
505 for source_name, source in (("requests", requests), ("limits", limits)):
506 if dimension not in source:
507 continue
508 value = parse_k8s_quantity(source[dimension])
509 findings.append(
510 StaticFinding(
511 example=parsed.name,
512 check=(
513 f"LimitRange fit ({kind}/{metadata.get('name')}: "
514 f"{container.get('name')} {source_name}.{dimension})"
515 ),
516 passed=value <= ceiling,
517 detail=""
518 if value <= ceiling
519 else (
520 f"{source[dimension]} exceeds the default per-container "
521 f"ceiling {DEFAULT_RESOURCE_QUOTA[ceiling_key]} "
522 f"({ceiling_key})"
523 ),
524 )
525 )
526 for dimension in _QUOTA_DIMENSIONS:
527 if dimension in requests:
528 aggregate[dimension] += parse_k8s_quantity(requests[dimension])
529 for dimension, quota_key in _QUOTA_DIMENSIONS.items():
530 total = aggregate[dimension] * replicas
531 quota = parse_k8s_quantity(DEFAULT_RESOURCE_QUOTA[quota_key])
532 findings.append(
533 StaticFinding(
534 example=parsed.name,
535 check=(
536 f"ResourceQuota fit ({kind}/{metadata.get('name')}: "
537 f"{replicas}x pod requests.{dimension})"
538 ),
539 passed=total <= quota,
540 detail=""
541 if total <= quota
542 else (
543 f"aggregate {total:g} exceeds the default namespace quota "
544 f"{DEFAULT_RESOURCE_QUOTA[quota_key]} ({quota_key})"
545 ),
546 )
547 )
548 if parsed.spec.submission in {SUBMIT_API, SUBMIT_SQS}:
549 findings.extend(
550 _manifest_cap_findings(parsed, kind, str(metadata.get("name")), aggregate, replicas)
551 )
552 return findings
555#: Front-door budget dimensions (manifest/queue processor caps) by
556#: DEFAULT_MANIFEST_RESOURCE_CAPS key.
557_MANIFEST_CAP_DIMENSIONS = {
558 "cpu": "max_cpu_per_manifest",
559 "memory": "max_memory_per_manifest",
560 "nvidia.com/gpu": "max_gpu_per_manifest",
561}
564def _manifest_cap_findings(
565 parsed: ParsedExample,
566 kind: str,
567 name: str,
568 aggregate: dict[str, float],
569 replicas: int,
570) -> list[StaticFinding]:
571 """API/SQS-submitted manifests must also fit the front-door budget.
573 The manifest and queue processors cap what one submitted manifest may
574 total; an example the front door rejects while kubectl admits it (or
575 vice versa) means the layers contradict each other.
576 """
577 from gco.stacks.constants import DEFAULT_MANIFEST_RESOURCE_CAPS, parse_k8s_quantity
579 findings: list[StaticFinding] = []
580 for dimension, cap_key in _MANIFEST_CAP_DIMENSIONS.items():
581 total = aggregate[dimension] * replicas
582 cap = parse_k8s_quantity(DEFAULT_MANIFEST_RESOURCE_CAPS[cap_key])
583 findings.append(
584 StaticFinding(
585 example=parsed.name,
586 check=f"manifest-cap fit ({kind}/{name}: {replicas}x pod requests.{dimension})",
587 passed=total <= cap,
588 detail=""
589 if total <= cap
590 else (
591 f"aggregate {total:g} exceeds the default per-manifest cap "
592 f"{DEFAULT_MANIFEST_RESOURCE_CAPS[cap_key]} ({cap_key})"
593 ),
594 )
595 )
596 return findings
599def _pod_spec_and_parallelism(doc: dict[str, Any], kind: str) -> tuple[dict[str, Any] | None, int]:
600 """Extract the pod template spec and concurrent-pod count for a workload."""
601 spec = doc.get("spec") or {}
602 if kind == "Job":
603 template_spec = ((spec.get("template") or {}).get("spec")) or None
604 return template_spec, int(spec.get("parallelism", 1) or 1)
605 if kind in {"Deployment", "StatefulSet"}:
606 template_spec = ((spec.get("template") or {}).get("spec")) or None
607 return template_spec, int(spec.get("replicas", 1) or 1)
608 if kind == "Pod":
609 return spec or None, 1
610 if kind == "TrainJob":
611 # A TrainJob runs its spec.trainer view once per node; the shared
612 # decomposition builds the same synthetic pod spec the deployed
613 # validators check, so the offline governance math matches theirs.
614 from gco.services.manifest_processor import extract_trainjob_pod_specs
616 trainjob_specs = extract_trainjob_pod_specs(doc)
617 return trainjob_specs.trainer, trainjob_specs.num_nodes
618 return None, 1
621def run_static_checks(repo_root: Path, names: list[str] | None = None) -> list[StaticFinding]:
622 """Run every offline check; returns findings (all must pass)."""
623 findings = check_registry_symmetry(repo_root)
624 for name in names or example_names(repo_root):
625 findings.append(check_spec_shape(name))
626 if EXAMPLE_SPECS.get(name) is None:
627 continue
628 parsed = parse_example(repo_root, name)
629 findings.append(check_submission_matches_catalog(repo_root, name))
630 findings.extend(check_transport_acceptance(parsed))
631 findings.extend(check_namespaces(parsed))
632 findings.extend(check_resource_governance_fit(parsed))
633 findings.extend(check_embedded_workloads(repo_root, parsed))
634 return findings