#!/usr/bin/env python3 """Process names derived from supported roles; installed configs select serving roles.""" from __future__ import annotations import argparse from pathlib import Path from validate_serviceification_bom import D3_WORKER_ROLE_SHAPES WORK_FAMILY_STEMS = tuple( family.removeprefix('runtime.').removesuffix('.v1').replace('.', '-') for family, _, _ in D3_WORKER_ROLE_SHAPES ) L0_WORKER_STEMS = WORK_FAMILY_STEMS # Legacy command/Catalog publication only; synchronous Layout v2 runs in Runtime API. RETIRED_L0_WORKER_STEMS = ('layout-authoring',) PRE_PERMISSION_WORKER_STEMS = tuple( stem for stem in L0_WORKER_STEMS if stem != 'permission-authoring-command-lifecycle' ) PRE_LAYOUT_WORKER_STEMS = WORK_FAMILY_STEMS PRE_PERMISSION_AND_LAYOUT_WORKER_STEMS = tuple( stem for stem in WORK_FAMILY_STEMS if stem != 'permission-authoring-command-lifecycle' ) def configured_workers(state_root: Path) -> tuple[str, ...]: """Reuse the installer-owned role files, without a second role manifest.""" roles = state_root / 'config/workers' if not roles.exists(): return L0_WORKER_STEMS # First bootstrap prepares current image carriers. selected = frozenset(path.stem for path in roles.glob('*.json')) if not selected: return L0_WORKER_STEMS # Interrupted preparation is bootstrap-only. supported = { frozenset(PRE_PERMISSION_AND_LAYOUT_WORKER_STEMS), frozenset(PRE_LAYOUT_WORKER_STEMS), frozenset(PRE_PERMISSION_WORKER_STEMS), frozenset(L0_WORKER_STEMS), } if selected not in supported: raise RuntimeError('incomplete standalone Worker configuration; use stopped upgrade') return tuple(stem for stem in L0_WORKER_STEMS if stem in selected) if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--state-root', type=Path) args = parser.parse_args() print(' '.join(configured_workers(args.state_root) if args.state_root else L0_WORKER_STEMS))