"""Closed development GitOps contracts and exact-revision readiness policy.""" from datetime import datetime import hashlib import ipaddress import json from pathlib import Path import re import yaml import optional_services from . import llm_api, llm_chat, mcp, nodes, run_logs, ssh_access ROOT = Path(__file__).resolve().parents[4] MANIFESTS = ROOT / 'gitops/manifests' ROOT_APPLICATION = ROOT / 'gitops/devbox-root-app.yaml' ROUTES_CHART = MANIFESTS / 'routes/chart' CATALOGUE = MANIFESTS / 'routes/base/services.yaml' ENVIRONMENT = MANIFESTS / 'routes/overlays/devbox/values.yaml' REPOSITORY = 'git://pharos-platform-git.argocd.svc.cluster.local/platform.git' SERVER = 'https://kubernetes.default.svc:443' ROUTE_SERVER = 'https://kubernetes.default.svc.cluster.local' # Services use a third view of the same API, limited to their own namespaces (decision 0041). SERVICES_SERVER = 'https://kubernetes.default.svc.cluster.local:443' SERVICES = ('gitlab', 'waldur', 'themelio-compat', 'pharos', 'pharos-infra', 'pharos-aai', 'llm', 'pharos-site-runtime', *(service['application'] for service in optional_services.SERVICES if service['project'] == 'services')) # The optional GPU child targets the registered self-management alias, the only # in-cluster destination Argo caches cluster-scoped resources for. GPU_SERVER = 'https://kubernetes.default.svc:443' # The GPU Operator keeps its own Application, project and server until the composition adopts it. GPU_OPERATOR = 'gpu-operator' # The DRA driver runs beside it on the GH200 nodes, in its own namespace and the same reviewed project. GPU_DRA = 'gpu-dra' GPU_CHILDREN = {GPU_OPERATOR: 'gpu-operator', GPU_DRA: 'gpu-dra'} # The namespaces the GPU cluster view must watch: the Applications live in argocd, and each child writes in # its own. Argo caches only what its view names, so delivery widens the registration to this list; the # Kubespray Argo role declares a subset for a fresh install, and widening never narrows it. GPU_VIEW_NAMESPACES = ('argocd', GPU_CHILDREN[GPU_OPERATOR], GPU_CHILDREN[GPU_DRA]) # The bootstrap documents of those children: the reviewed project, the namespaces and the writer grants. The # Kubespray Argo role applies them; an adopted foundation has no role, so delivery applies them itself. GPU_BOOTSTRAP = ('gpu-operator.yaml', 'gpu-dra.yaml') # The cloud Site Agent's Applications use a fourth view, its namespaces and its named cluster-scoped # objects (project provisioning 5.1); each maps to its bootstrap-owned namespace. SITE_AGENT_SERVER = 'https://kubernetes.default:443' # The Site Agent project's Applications and the namespaces they own. KServe sits in the same project because it # needs the same named cluster-scoped objects; the llm-inference composition selects it, so it is not resident. SITE_AGENT = {'kubeflow-profiles': 'profiles-system', 'waldur-site-agent': 'waldur-site-agent', 'kserve': 'kserve', 'llm-inference': 'llm-inference'} # Services use a third view of the same API, limited to their own namespaces (decision 0041). An optional profile may # instead select a Site Agent Application — the dev box's KServe serving profile is one — and that Application belongs # to the Site Agent's project, not this one. SERVICES = ('gitlab', 'waldur', 'themelio-compat', 'pharos', 'pharos-infra', 'pharos-aai', 'llm', 'pharos-site-runtime', *(service['application'] for service in optional_services.SERVICES if service['application'] not in SITE_AGENT)) # Applications the base root always installs; the rest of SITE_AGENT is selected by a composition. RESIDENT_SITE_AGENT = ('kubeflow-profiles', 'waldur-site-agent') def site_agent_view_namespaces(): """The namespaces the Site Agent's cluster view watches: the provisioners', KServe's and the model delivery's. The role registers the provisioners' own namespaces, and delivery widens the view with a selected composition's. Argo CD caches every kind its view includes in each watched namespace, so the reads are bound at bootstrap for every namespace the view can hold. """ return tuple(dict.fromkeys([*(SITE_AGENT[name] for name in RESIDENT_SITE_AGENT), *RESIDENT_NAMESPACES, *COMPOSITION_NAMESPACES['llm-inference']])) # The Kubeflow composition (ADR-0067) uses the platform's own in-cluster view, in a project of its own whose # namespaced kinds the writer roles below grant. Argo CD refuses to manage an object in a namespace its view omits, # so delivery adds the composition's namespaces to that view before the root applies the five Applications. KUBEFLOW_SERVER = SERVER KUBEFLOW = {'kubeflow-core': 'kubeflow', 'kubeflow-notebooks': 'kubeflow', 'kubeflow-pipelines': 'kubeflow', 'kubeflow-trainer': 'kubeflow-system', 'kubeflow-katib': 'kubeflow'} # KFAM and the profile controller name the gateway that may reach a Profile by one principal, never a list: both read # `ISTIO_INGRESS_GATEWAY_PRINCIPAL` with `GetEnvDefault` and write it as a single principal. The platform's Istio chart # runs its gateway as `GATEWAY_PRINCIPAL`; an adopted foundation keeps the upstream gateway's # `ADOPTED_GATEWAY_PRINCIPAL`. The composition's kustomization names the platform's, and delivery patches the child # Application with the adopted one, because Kustomize has no view of a foundation mode. GATEWAY_PRINCIPAL = 'cluster.local/ns/istio-system/sa/istio-ingressgateway' ADOPTED_GATEWAY_PRINCIPAL = 'cluster.local/ns/istio-system/sa/istio-ingressgateway-service-account' KUBEFLOW_NAMESPACES = ('kubeflow', 'kubeflow-system', 'oauth2-proxy') # What the composition's Applications watch, and what the controller therefore needs a grant in: the bootstrap-owned # namespaces plus the shared mesh namespace their Gateway, VirtualServices and policies live in. KUBEFLOW_VIEW_NAMESPACES = (*KUBEFLOW_NAMESPACES, 'istio-system') # Optional services are Kustomize components the root selects by profile (decision 0040). # Themelio-compat becomes one again once the new Themelio replaces it. COMPONENTS = optional_services.components() # Whole compositions an environment selects for itself (ADR-0067): a Kustomize component of Applications that no # site renders unless its delivery policy names it. A component path is relative to the root overlay. COMPOSITIONS = {'kubeflow': '../../components/kubeflow', 'llm-inference': '../../components/llm-inference'} # What the Site Agent's Argo cluster view has to watch once an environment selects a composition: Argo refuses to # manage an object in a namespace its cluster view does not name, and a composition's Applications live in namespaces # of their own. KServe is a resident child of the distributed composition, so its namespace is watched either way. COMPOSITION_NAMESPACES = {'kubeflow': (*KUBEFLOW_NAMESPACES, 'istio-system'), 'llm-inference': ('kserve', 'llm-inference')} RESIDENT_NAMESPACES = ('kserve',) # A reviewed composition renders exactly these Applications from the base root and nothing else. # The LLM inference composition brings KServe and the model delivery; KServe is what the LLM stack and the Site # Runtime's Spaces need, and the role-based serving manifests need its LLMInferenceService CRD. COMPOSITION_APPLICATIONS = {'kubeflow': tuple(KUBEFLOW), 'llm-inference': ('llm-inference',)} # What a selected composition adds to the cluster view Argo CD uses for its Applications, and the bootstrap # document that declares its project and namespaces on an adopted foundation. Argo CD refuses to manage an # object in a namespace that view does not watch, so delivery widens the view before the root names them. COMPOSITION_VIEWS = {'kubeflow': (KUBEFLOW_SERVER, KUBEFLOW_VIEW_NAMESPACES)} # Services whose images the cluster builds on every delivery (decision 0045). BUILT = ('pharos', 'themelio-compat', 'pharos-aai', 'pharos-site-runtime', 'waldur-site-agent') # Optional services whose images the cluster builds when their profile selects them. BUILT_COMPONENTS = ('kserve', 'mcp') # The root passes each service the images the catalogue builds for it, by digest (decision 0045). REFERENCE = re.compile(r'registry\.pharos-build\.svc\.cluster\.local:5000/[a-z0-9-]+@sha256:[0-9a-f]{64}') # A record from before the builds were named by the cluster FQDN names the same registry by its short # Service name; it is read so an installed box upgrades, and the next delivery records the FQDN. SHORT_REFERENCE = re.compile(r'registry\.pharos-build\.svc:5000/[a-z0-9-]+@sha256:[0-9a-f]{64}') # A record from before decision 0045 names the workstation-built images by role; the next # make deploy replaces it, and nothing else may reuse those images. LEGACY_IMAGES = {'web', 'gitlabRootActor', 'k8sRootActor'} LEGACY_REFERENCE = re.compile(r'docker\.io/pharos/themelio-compat-[a-z0-9-]+@sha256:[0-9a-f]{64}') APPLICATIONS = ('pharos-platform', 'argocd', 'pharos-platform-routes') PATHS = {'mcp': 'gitops/manifests/mcp/base', 'pharos-aai': 'gitops/manifests/pharos-aai/base', 'pharos-infra': 'gitops/manifests/pharos-infra/base', 'pharos-platform': 'gitops/apps/overlays/devbox', 'argocd': 'gitops/manifests/argocd/chart', 'pharos-platform-routes': 'gitops/manifests/routes/chart', 'gpu-operator': 'gitops/manifests/gpu-operator/chart', 'themelio-compat': 'gitops/manifests/themelio-compat/overlays/devbox', 'pharos': 'gitops/manifests/pharos/overlays/devbox', 'pharos-site-runtime': 'gitops/manifests/pharos-site-runtime/overlays/devbox', 'kubeflow-profiles': 'gitops/manifests/kubeflow-profiles', 'waldur-site-agent': 'gitops/manifests/waldur-site-agent-devbox', # The site-agent compositions' Applications. A site's delivery states their paths itself; the dev box can # select them as profiles, and then `llm-inference` is its own render of the same catalogue. 'kserve': 'gitops/manifests/kserve', 'llm-inference': 'gitops/manifests/llm-inference-devbox'} # The dev box's opt-in serving profile renders its catalogue twice: `catalogue.yaml` is the chart's own CPU server and # the gateway entry that points at it, `catalogue-kserve.yaml` turns that server off and points the gateway at the # LLMInferenceService. The second has to be applied last, and only a root patch is applied after the overlay's own # patches (a root's components are added to the overlay's kustomization, where kustomize runs them first), so the # ordering lives here rather than in the component beside the profile. DEVBOX_LLM_VALUES = ('../base/values.yaml', '../pins.yaml', '../overlays/devbox/values.yaml', '../overlays/devbox/catalogue.yaml') DEVBOX_KSERVE_VALUES = (*DEVBOX_LLM_VALUES, '../overlays/devbox/catalogue-kserve.yaml') OWNER_NAME = 'pharos-platform-ownership' CONTROLLER = 'system:serviceaccount:argocd:argocd-application-controller' NAMESPACES = ('argocd', 'istio-system', 'pharos-aai', 'rook-ceph', 'gitlab', 'waldur', 'themelio-compat', # An optional profile that selects a Site Agent Application owns its namespace through the Site Agent's # project, not this one. *(service['namespace'] for service in optional_services.SERVICES if service['application'] not in SITE_AGENT), 'pharos') # The routes Application writes backend policies around the gateway, but the pharos # Application owns ingress into its own namespace. ROUTE_NAMESPACES = tuple(namespace for namespace in NAMESPACES if namespace != 'pharos') PROFILES = {'pharos', 'platform', 'aai', 'storage', 'argocd', 'gitlab', 'waldur', 'themelio-compat', 'pharos-site', 'llm-chat', 'llm-api', 'kubeflow', *COMPONENTS} def application_path(name): path = PATHS.get(name, f'gitops/manifests/{name}/chart' if name in SERVICES else None) if name in KUBEFLOW and path is None: path = f'gitops/manifests/{name}' if name in {'pharos', 'themelio-compat'}: return path + '-api-auth' return path def revision(value): if not isinstance(value, str) or not re.fullmatch('[0-9a-f]{40}', value): raise ValueError('GitOps requires an exact delivered commit SHA') return value def compositions(value): """The reviewed compositions a delivery selects, in the environment's own order.""" if value is None: return () if (not isinstance(value, list) or any(not isinstance(name, str) for name in value) or len(set(value)) != len(value) or set(value) - set(COMPOSITIONS)): raise ValueError('A selected composition must be a reviewed composition name') return tuple(value) def named(images, names, pattern): return (isinstance(images, dict) and set(images) == names and all(isinstance(ref, str) and pattern.fullmatch(ref) for ref in images.values())) def catalogued(): return yaml.safe_load((MANIFESTS / 'images.yaml').read_text())['images'] def applications(profiles): """The services a delivery builds images for: every built service, and the optional ones its profiles select.""" return [*BUILT, *(name for name in BUILT_COMPONENTS if name in profiles)] def aai_profiles(profiles): """The profiles whose AAI client bindings this delivery writes. `profiles` selects the edge's routes and its optional components; the services every dev box installs have no profile of their own, yet their charts read a Secret the AAI writes (`gitops/manifests//aai-client.yaml`). Delivering only to the selected profiles left `llm-oidc` at its bootstrap placeholders, so Open WebUI never started and its health gate blocked the root sync on the stage that would have repaired it. Deliver to every installed consumer, and keep optional services (a component someone selected) behind their profile. """ optional = {service['application'] for service in optional_services.SERVICES} consumers = {path.parent.name for path in MANIFESTS.glob('*/aai-client.yaml')} return sorted({*profiles, *(name for name in SERVICES if name not in optional and name in consumers)}) def wanted(profiles, application=None): from .images import applications as built_for selected = applications(profiles) if application is None else [application] return {name for name, entry in catalogued().items() if set(built_for(entry)) & set(selected)} def inputs(value, route_only=False, recorded=False): profiles = value.get('profiles') # Images are built in the cluster for every delivery; the edge's route inputs never carry # them. A record from an earlier delivery may name fewer, or the workstation's builds. base = {'profiles', 'sourceChecksum', 'cephCertificateHash'} # Delivery adds the built images and a checksum of AAI's inputs from outside Git. delivered = {'images', 'identity', 'aaiMode', 'providerImages', 'imageProfile', 'edgeNodes'} historical = {'aai'} if recorded and not route_only else set() if set(value) - base - delivered - historical or base - set(value) or ( bool(delivered & set(value)) if route_only else 'images' not in value and not recorded): raise ValueError('Unexpected GitOps route inputs') if (not isinstance(profiles, list) or not profiles or len(set(profiles)) != len(profiles) or set(profiles) - PROFILES): raise ValueError('Invalid resolved route profiles') if 'images' in value: images = value['images'] if recorded: known = set(catalogued()) | ({'aai-admin', 'aai-keycloak'} if 'aai' in value else set()) if not (named(images, set(images) & known, REFERENCE) or named(images, set(images) & known, SHORT_REFERENCE) or named(images, LEGACY_IMAGES, LEGACY_REFERENCE)): raise ValueError('Recorded images are not digests of known builds') elif not named(images, wanted(profiles), REFERENCE): raise ValueError('Images must be digests of the in-cluster builds for the selected services') if value.get('aaiMode', 'local') not in ('local', 'external'): raise ValueError('AAI mode must be local or external') # A record from before the builds were named by the cluster FQDN also names the provider # images by the short Service name, and it holds the parameters the waldur Application runs # with until the next delivery replaces them. It is read for that reason only. if 'providerImages' in value and not ( named(value['providerImages'], {'waldur-mastermind', 'waldur-homeport'}, REFERENCE) or recorded and named(value['providerImages'], {'waldur-mastermind', 'waldur-homeport'}, SHORT_REFERENCE)): raise ValueError('Provider overrides require both pinned Waldur image digests') if 'imageProfile' in value and value['imageProfile'] != 'arm64-singledrive': raise ValueError('Unrecognised optional image profile') if 'edgeNodes' in value: found = value['edgeNodes'] if (not isinstance(found, list) or not 1 <= len(found) <= 8 or len(set(map(str, found))) != len(found) or not all(nodes.node_address(item) for item in found)): raise ValueError('Invalid node addresses: one to eight distinct private IPv4 /32 blocks') if 'aai' in value: # Read the previous AAI record so conversion can retire it. Current # delivery and route inputs still reject this removed input contract. legacy = value['aai'] if (not isinstance(legacy, dict) or set(legacy) != {'revision', 'node', 'service'} or not isinstance(legacy['revision'], str) or not re.fullmatch('[0-9a-f]{64}', legacy['revision'])): raise ValueError('Invalid recorded AAI inputs') for name in ('node', 'service'): if not isinstance(legacy[name], str): raise ValueError('Invalid recorded AAI address') address = ipaddress.IPv4Address(legacy[name]) if not address.is_private or address.is_loopback or address.is_unspecified: raise ValueError('Recorded AAI addresses must be private IPv4') for key in ('sourceChecksum', 'cephCertificateHash', 'identity'): if key in value and (not isinstance(value[key], str) or not re.fullmatch('[0-9a-f]{64}', value[key])): raise ValueError('Invalid authoritative route checksum: ' + key) return value def overrides(images): """Kustomize image overrides that replace each manifest image name with its built digest.""" names = {name: entry['image'] for name, entry in catalogued().items()} return sorted(names[name] + '=' + reference for name, reference in images.items()) def administration_image(images): """The built administration image the Waldur chart's provisioning Jobs run, on the dev box and in a delivery.""" try: return images['database-admin'] except KeyError: raise ValueError('The Waldur provisioning Jobs need the built administration image') from None def patches(commit, route_inputs): """Kustomize patches that bind the reviewed child Applications to one delivery.""" if 'mcp' in route_inputs['profiles'] and route_inputs.get('aaiMode', 'local') != 'local': raise ValueError('The MCP laboratory requires local AAI') ssh = ssh_access.devbox(ROOT) values = {'activeProfiles': route_inputs['profiles'], 'sourceChecksum': route_inputs['sourceChecksum'], 'cephCertificateHash': route_inputs['cephCertificateHash']} from .identity import patches as identity_patches application = {'group': 'argoproj.io', 'version': 'v1alpha1', 'kind': 'Application'} parameters = [{'name': 'pharos.images.databaseAdmin', 'value': administration_image(route_inputs['images']), 'forceString': True}] if ssh['enabled']: parameters.append({'name': 'pharos.metallo.enabled', 'value': 'true'}) if providers := route_inputs.get('providerImages'): for name, prefix in [('waldur-mastermind', 'waldur.waldur'), ('waldur-homeport', 'waldur.waldur.homeport')]: repository, digest = providers[name].split('@') for key, value in [('imageRegistry', repository.split('/', 1)[0]), ('imageName', repository.split('/', 1)[1]), ('imageDigest', digest)]: parameters.append({'name': prefix + '.' + key, 'value': value, 'forceString': True}) return [{'target': application, 'patch': json.dumps( [{'op': 'replace', 'path': '/spec/source/targetRevision', 'value': commit}])}, {'target': {**application, 'name': 'pharos-platform-routes'}, 'patch': json.dumps( [{'op': 'replace', 'path': '/spec/source/helm/valuesObject', 'value': values}])}, {'target': {**application, 'name': 'waldur'}, 'patch': json.dumps( [{'op': 'add', 'path': '/spec/source/helm/parameters', 'value': parameters}])}, *([{'target': {**application, 'name': 'gitlab'}, 'patch': json.dumps( [{'op': 'add', 'path': '/spec/source/helm/valueFiles/-', 'value': '../overlays/arm64-singledrive/values.yaml'}])}] if route_inputs.get('imageProfile') == 'arm64-singledrive' else []), # The dev box's KServe profile is a values-file order, and a root patch is the only thing applied after the # overlay's own patches: without it the overlay appends `catalogue.yaml` again and the chart's CPU server # stays on, serving the same name beside the LLMInferenceService. *([{'target': {**application, 'name': 'llm'}, 'patch': json.dumps( [{'op': 'replace', 'path': '/spec/source/helm/valueFiles', 'value': list(DEVBOX_KSERVE_VALUES)}])}] if 'llm-inference' in route_inputs['profiles'] else []), *({'target': {**application, 'name': name}, 'patch': json.dumps( [{'op': 'replace', 'path': '/spec/source/kustomize/images', 'value': overrides({image: reference for image, reference in route_inputs['images'].items() if image in wanted(route_inputs['profiles'], name)})}])} for name in applications(route_inputs['profiles'])), *({'target': {**application, 'name': name}, 'patch': json.dumps( [{'op': 'replace', 'path': '/spec/source/path', 'value': application_path(name)}])} for name in ('pharos', 'themelio-compat')), # Repository callers reach the node edge. The personal profile also uses Traefik for canonical LFS URLs. {'target': {**application, 'name': 'pharos'}, 'patch': json.dumps( [{'op': 'add', 'path': '/spec/source/kustomize/patches', 'value': [*(nodes.edge_egress(route_inputs['edgeNodes'], canonical=route_inputs.get('imageProfile') == 'arm64-singledrive') if 'edgeNodes' in route_inputs else []), *mcp.patches(route_inputs['profiles']), *ssh_access.patches(ssh), *run_logs.patches(run_logs.devbox(ROOT)), llm_chat.devbox(route_inputs['profiles']), llm_api.devbox(route_inputs['profiles']), # Cross-namespace credential copies must be published when a consumer is selected, # including when setup keeps the same commit and images. credential_profile_patch(route_inputs['profiles'])]}])}, *([{'target': {**application, 'name': 'mcp'}, 'patch': json.dumps( [{'op': 'add', 'path': '/spec/source/kustomize/patches', 'value': mcp.edge(route_inputs.get('edgeNodes', []))}])}] if 'mcp' in route_inputs['profiles'] else []), # The Site Agent opens the same port, at the same addresses, in the Project namespaces it makes. *([{'target': {**application, 'name': 'waldur-site-agent'}, 'patch': json.dumps( [{'op': 'add', 'path': '/spec/source/kustomize/patches', 'value': nodes.edge_agent(route_inputs['edgeNodes'])}])}] if 'edgeNodes' in route_inputs else []), # The Runtime renders native workloads with the fixed batch image, by its built digest. *([{'target': {**application, 'name': 'pharos-site-runtime'}, 'patch': json.dumps( [{'op': 'add', 'path': '/spec/source/kustomize/patches', 'value': [ {'target': {'kind': 'ConfigMap', 'name': 'runtime-config'}, 'patch': json.dumps( [{'op': 'add', 'path': '/data/BATCH_IMAGE', 'value': route_inputs['images']['batch']}])}]}])}] if 'batch' in route_inputs['images'] else []), # A selected optional consumer can appear after infrastructure's last # successful sync. Make the profile set desired state on a normal resource # so Argo reruns the credential and database hooks for the new namespace. {'target': {**application, 'name': 'pharos-infra'}, 'patch': json.dumps( [{'op': 'add', 'path': '/spec/source/kustomize', 'value': {'patches': [ {'target': {'kind': 'ServiceAccount', 'name': 'infrastructure-bootstrap'}, 'patch': json.dumps([{'op': 'add', 'path': '/metadata/annotations/platform.pharos.ai~1active-profiles', 'value': json.dumps(route_inputs['profiles'], separators=(',', ':'))}])}]}}])}, # Preserve consumer selection while applying the private provider fingerprint. *identity_patches(route_inputs.get('identity'), route_inputs.get('aaiMode', 'local'), aai_profiles(route_inputs['profiles']))] def root(commit, route_inputs, automated=True): value = yaml.safe_load(ROOT_APPLICATION.read_text()) source = value['spec']['source'] source['targetRevision'] = revision(commit) source['kustomize'] = {'patches': patches(commit, inputs(route_inputs))} # The dev box's `llm-inference` profile nests the KServe component itself, and a box reports `kserve` beside it # (the Application is installed). Naming both components would add the same Application twice and the root build # would stop with `may not add resource with an already registered id: Application...kserve`; KServe still rides # along inside the profile, and the profile keeps `kserve` selected for the CPU image it builds. profiles = route_inputs['profiles'] components = [path for name, path in COMPONENTS.items() if name in profiles and not (name == 'kserve' and 'llm-inference' in profiles)] if components: source['kustomize']['components'] = components value['spec']['syncPolicy']['automated']['enabled'] = automated return value def batch_patch(patches): """The Runtime takes BATCH_IMAGE and, when distributed, pod placement only.""" found = False for patch in patches: if not isinstance(patch, dict): return False if patch.get('target') == {'kind': '(Deployment|StatefulSet|Job)'}: try: operations = json.loads(patch['patch']) except (KeyError, TypeError, ValueError): return False if (isinstance(operations, list) and len(operations) == 1 and operations[0].get('op') == 'add' and operations[0].get('path') == '/spec/template/spec/nodeSelector' and isinstance(operations[0].get('value'), dict)): continue return False if patch.get('target') != {'kind': 'ConfigMap', 'name': 'runtime-config'} or found: return False try: operations = json.loads(patch['patch']) except (KeyError, TypeError, ValueError): return False if (isinstance(operations, list) and len(operations) == 1 and set(operations[0]) == {'op', 'path', 'value'} and operations[0]['op'] == 'add' and operations[0]['path'] == '/data/BATCH_IMAGE' and isinstance(operations[0]['value'], str) and bool(REFERENCE.fullmatch(operations[0]['value']))): found = True continue return False return found def view_namespaces(compositions=()): """The namespaces the Site Agent's cluster view watches for the compositions an environment selects.""" names = list(RESIDENT_NAMESPACES) for composition in compositions: if composition not in COMPOSITION_NAMESPACES: raise ValueError('Unknown delivery composition: ' + str(composition)) names += [name for name in COMPOSITION_NAMESPACES[composition] if name not in names] return tuple(names) def credential_profile_patch(profiles): return {'target': {'kind': 'ServiceAccount', 'name': 'pharos-credentials'}, 'patch': json.dumps([{'op': 'add', 'path': '/metadata/annotations/platform.pharos.ai~1active-profiles', 'value': json.dumps(profiles, separators=(',', ':'))}])} def without_credential_profiles(patches): """Accept one exact marker on the credential producer; other patches keep their own checks.""" if not isinstance(patches, list) or any(not isinstance(value, dict) for value in patches): return None markers = [value for value in patches if value.get('target') == credential_profile_patch([])['target']] try: marker, = markers profiles = json.loads(json.loads(marker['patch'])[0]['value']) if (not isinstance(profiles, list) or not profiles or any(not isinstance(name, str) or name not in PROFILES for name in profiles) or len(set(profiles)) != len(profiles) or marker != credential_profile_patch(profiles)): return None except (ValueError, KeyError, TypeError, IndexError): return None return [value for value in patches if value != marker] def application_errors(value, commit, *, paths=None): errors = [] spec, metadata = value.get('spec', {}), value.get('metadata', {}) name = metadata.get('name') if metadata.get('namespace') != 'argocd' or name not in ( *APPLICATIONS, *SERVICES, *SITE_AGENT, *GPU_CHILDREN, *KUBEFLOW): errors.append('Application identity is not selected') service, agent, gpu = name in SERVICES, name in SITE_AGENT, name in GPU_CHILDREN kubeflow = name in KUBEFLOW if spec.get('project') != ('gpu' if gpu else 'services' if service else 'site-agent' if agent else 'kubeflow' if kubeflow else 'platform'): errors.append('Application must select its reviewed project') expected_namespace = (GPU_CHILDREN[name] if gpu else name if service else SITE_AGENT[name] if agent else KUBEFLOW[name] if kubeflow else 'istio-system' if name == 'pharos-platform-routes' else 'argocd') expected_server = (GPU_SERVER if gpu else SERVICES_SERVER if service else SITE_AGENT_SERVER if agent else KUBEFLOW_SERVER if kubeflow else ROUTE_SERVER if name == 'pharos-platform-routes' else SERVER) if spec.get('destination') != {'server': expected_server, 'namespace': expected_namespace}: errors.append('Application destination is not its reviewed namespace') source = spec.get('source', {}) if (source.get('repoURL') != REPOSITORY or source.get('targetRevision') != revision(commit) or source.get('path') != (application_path(name) if paths is None else paths.get(name, f'gitops/manifests/{name}/chart' if service else None)) or spec.get('sources')): errors.append('Application source must select the delivered repository, path and revision') if name == 'pharos-platform' and (source.get('helm') or set(source.get('kustomize', {})) - {'patches', 'components'}): errors.append('The root selects services only through Kustomize patches and components') # The Site Agent takes the environment's identity source as a reviewed input patch, beside its image digests. accepted = {'images', 'patches'} if name in ('pharos-aai', 'pharos', 'pharos-site-runtime', 'themelio-compat', 'waldur-site-agent', 'mcp') else {'images'} if name == 'kserve' and paths is not None: accepted = {'images', 'patches'} if name in (*BUILT, *BUILT_COMPONENTS) and (source.get('helm') or set(source.get('kustomize', {})) - accepted): errors.append(name + ' takes only its image digests and reviewed input patches from the root') if name == 'mcp' and not mcp.edge_valid(source.get('kustomize', {}).get('patches', [])): errors.append('MCP takes only exact node HTTPS egress from the root') if name == 'pharos-site-runtime' and not batch_patch(source.get('kustomize', {}).get('patches', [])): errors.append('The Runtime takes only its batch image digest from the root') # Distributed delivery judges its own placement and origins. A dev box accepts # the declared SSH activation and its node-scoped repository egress. if paths is None and name == 'pharos': remaining = without_credential_profiles(source.get('kustomize', {}).get('patches', [])) remaining = ssh_access.remaining(remaining, ssh_access.devbox(ROOT)) if remaining is not None else None remaining = llm_chat.remaining(remaining) remaining = run_logs.remaining(remaining, run_logs.devbox(ROOT)) remaining = llm_api.remaining(remaining) remaining = mcp.remaining(remaining) if remaining is None or (remaining and not nodes.edge_patch(remaining)): errors.append('The platform takes only reviewed credential profiles, SSH and Run log activation, LLM public origins and scoped repository egress') if (paths is None and name == 'waldur-site-agent' and source.get('kustomize', {}).get('patches') and not nodes.agent_patches(source['kustomize']['patches'])): errors.append('The Site Agent takes only the nodes of the repository egress and its identity source from the root') policy = spec.get('syncPolicy', {}) if policy.get('automated', {}).get('prune') or any( option in {'Replace=true', 'Force=true', 'CreateNamespace=true'} for option in policy.get('syncOptions', [])): errors.append('Destructive sync or namespace creation is forbidden') return errors # A model entry's weights job. Argo applies it as a Sync hook, so the sync operation stays `Running` until the last # byte lands; the suffix is what separates that wait from every other hook. DOWNLOAD_HOOK = '-download' # What a running Sync hook looks like in `operationState.syncResult.resources`. The operation record itself carries # only `sync`, `initiatedBy`, `info` and `retry` -- there is no list of hooks there -- and `status.sync.status` is # `Synced`, `OutOfSync` or `Unknown`, never `Running`. A hook that is still running is a resource entry with # `hookType: Sync` and `hookPhase: Running`. SYNC_HOOK, RUNNING_PHASE = 'Sync', 'Running' def downloading(operation): """Whether a running sync operation is waiting only on the catalogue's weight downloads. A checkpoint can be 120 GB and the entry budgets hours for it, while a delivery's readiness wait is measured in minutes. Holding the delivery to the hook would bound the manifests by the download; this answers the narrower question the gate needs: is everything this delivery renders applied and compared, with only the weights still arriving? The running hooks are read from `syncResult.resources`, which is where Argo records the result of each resource a sync touched: the sync itself reports `Running` while a hook it started is still going, and every running hook has to be one of the entries' download Jobs. Anything else that is still running is a delivery that has not finished. """ if operation.get('phase') != 'Running': return False resources = (operation.get('syncResult') or {}).get('resources') or [] running = [resource for resource in resources if resource.get('hookType') == SYNC_HOOK and resource.get('hookPhase') == RUNNING_PHASE] return bool(running) and all(resource.get('kind') == 'Job' and str(resource.get('name', '')).endswith(DOWNLOAD_HOOK) for resource in running) def loading_errors(value, commit, *, paths=None, since=None): """The model delivery loads checkpoints for hours: applied, compared and without a failed operation, while the serving pods are still `Progressing`. A child Application never prunes, and a model's weights live on a PVC that Argo owns; holding the delivery to `Healthy` would keep every delivery of a site with a large catalogue open until the last checkpoint had loaded, hours past the manifests and past any sensible deadline. Every other child keeps the full readiness gate (`readiness_errors`), and a failing Sync hook Job still fails this one: a failed operation at the delivered commit, or an error condition, is reported here too. A wait that has to prove the status is its own passes `since`, exactly as `readiness_errors` accepts it: the loading child is held to the same delivery evidence. The weight download is a Sync hook, and Argo keeps the sync operation `Running` while it runs. This gate accepts that one wait (`downloading`), because the delivery answers for the manifests and the Job carries its own deadline; every other running hook or operation stays pending. """ errors = application_errors(value, commit, paths=paths) status = value.get('status', {}) sync = status.get('sync', {}) operation = status.get('operationState', {}) # The manifests are applied and compared; only the weight download is still arriving. weights = downloading(operation) if sync.get('status') != 'Synced' and not weights: errors.append('Application is not Synced') # The comparison is the normalised one `readiness_errors` uses: Argo drops null members from the source it # records, and a distributed route carries `domain: null`. if sync.get('revision') != commit or (compared_source(sync.get('comparedTo', {}).get('source')) != compared_source(value.get('spec', {}).get('source'))): errors.append('Argo has not compared the delivered source') if value.get('operation') or (operation.get('phase') in {'Running', 'Terminating'} and not weights): errors.append('Application operation is pending') elif operation.get('phase') in {'Failed', 'Error'} and operation.get('syncResult', {}).get('revision') == commit: errors.append('Application operation failed at the delivered commit') if any(c.get('type', '').endswith('Error') for c in status.get('conditions', [])): errors.append('Application has an error condition') if since is not None: errors.extend(stale_errors(value, commit, since)) return errors def compared_source(value): """Normalise a source for the Argo comparison. Argo stores `status.sync.comparedTo.source` from its own Go structures, which drop JSON null members, while `spec.source` keeps them. Every distributed route value carries `domain: null`, so an exact comparison never matches. Remove null members on both sides; any real difference in the source still differs. """ if isinstance(value, dict): return {name: compared_source(item) for name, item in value.items() if item is not None} if isinstance(value, list): return [compared_source(item) for item in value] return value def moment(value): """Argo's RFC 3339 timestamps as seconds since the epoch, with or without fractional seconds.""" if not value: return None try: return datetime.fromisoformat(str(value).replace('Z', '+00:00')).timestamp() except ValueError: return None def condition_errors(status): """Argo's own failure conditions, with their message: `ComparisonError` carries why it could not render.""" failed = [value for value in status.get('conditions', []) if (value.get('type') or '').endswith('Error')] if not failed: return [] reasons = [value.get('message') or value.get('type') for value in failed] return ['Application has an error condition: ' + '; '.join(reasons)] def stale_errors(value, commit, since): """Proof that Argo reported on the delivered source after the delivery began, not from an earlier one. A delivery at a revision the Applications already name changes no child, so Argo starts no operation and the status it holds is the previous delivery's. A `Synced` status alone therefore proves nothing: an adopted site's delivery passed its gate on exactly such a status while the `pharos` Application's next comparison failed with duplicate volume names, leaving the child `Unknown` with the previous delivery's operation. An operation of the delivered revision that started after `since`, or a comparison Argo made after it, is the evidence that the status describes this delivery; a failed comparison never is, because Argo leaves it in a condition and the child `Unknown`. """ status = value.get('status', {}) state = status.get('operationState') or {} operation = state.get('operation') or {} revision = ((state.get('syncResult') or {}).get('revision') or (operation.get('sync') or {}).get('revision')) started = moment(state.get('startedAt')) if revision == commit and started is not None and started >= since: return [] sync = status.get('sync', {}) compared = moment(status.get('reconciledAt')) if (sync.get('status') == 'Synced' and sync.get('revision') == commit and compared is not None and compared >= since): return [] errors = [] if revision not in {None, commit}: errors.append('Application last operated on another revision') errors.append('Application has no operation or comparison of the delivered source since this delivery began') return errors def readiness_errors(value, commit, *, paths=None, since=None): """Decision 0039: Synced and Healthy at the commit, with no queued, running or failed operation. Argo starts every operation. Unchanged desired state at a new commit needs none, so a wait that has to prove the status is its own passes `since` and accepts a comparison made after it just as it accepts an operation. """ errors = application_errors(value, commit, paths=paths) status = value.get('status', {}) sync = status.get('sync', {}) if sync.get('status') != 'Synced': errors.append('Application is not Synced') if status.get('health', {}).get('status') != 'Healthy': errors.append('Application is not Healthy') # The comparison must be of the applied source, not a status from before the root's change. if sync.get('revision') != commit or (compared_source(sync.get('comparedTo', {}).get('source')) != compared_source(value.get('spec', {}).get('source'))): errors.append('Argo has not compared the delivered source') operation = status.get('operationState', {}) if value.get('operation') or operation.get('phase') in {'Running', 'Terminating'}: errors.append('Application operation is pending') elif operation.get('phase') in {'Failed', 'Error'} and operation.get('syncResult', {}).get('revision') == commit: errors.append('Application operation failed at the delivered commit') errors.extend(condition_errors(status)) if since is not None: errors.extend(stale_errors(value, commit, since)) return errors def verify_chart(): from .gpu_chart import verify as verify_gpu_chart verify_gpu_chart(MANIFESTS / 'gpu-operator') pins = yaml.safe_load((MANIFESTS / 'argocd/pins.yaml').read_text()) archive = MANIFESTS / 'argocd/chart/charts' / ('argo-cd-' + pins['chart']['version'] + '.tgz') if hashlib.sha256(archive.read_bytes()).hexdigest() != pins['chart']['sha256']: raise ValueError('Vendored Argo chart checksum mismatch') for daemon in (pins['gitDaemon'], pins['gitDaemonArm64']): if hashlib.sha256((MANIFESTS / 'argocd/artifacts' / daemon['file']).read_bytes()).hexdigest() != daemon['sha256']: raise ValueError('Vendored Git daemon checksum mismatch') return pins def record(data): if data.get('routes') not in {'ansible', 'adopting', 'argo', 'restoring'}: raise ValueError('Unknown routes ownership state') if data.get('argocd') not in {'bootstrap', 'adopting', 'argo'}: raise ValueError('Unknown Argo ownership state') if data.get('revision'): revision(data['revision']) if data.get('inputs'): inputs(json.loads(data['inputs']), recorded=True) if data.get('lease'): from .lease import current current(data) return data def profile_errors(): """Every profile the delivery accepts must also pass the route chart's and the edge's validation.""" from platform_edge import catalogue as edge routes = json.loads((MANIFESTS / 'routes/base/schema.json').read_text())['$defs']['profile']['enum'] chart = json.loads((ROUTES_CHART / 'values.schema.json').read_text())['$defs']['profile']['enum'] if not set(routes) == set(chart) == PROFILES == edge.PROFILES: return ['Route schema, route chart, edge and GitOps model must accept the same profiles'] return []