|
| 1 | +"""Check that an adapter manifest edit changes only named lifecycle keys. |
| 2 | +
|
| 3 | +``kb init`` touches one file outside ``knowledge/``: the adapter manifest, and |
| 4 | +only to switch the repository's knowledge service on (``enabled`` + ``mode``) |
| 5 | +or to name its calibration set. The knowledge L1 never sees that file (its |
| 6 | +whitelist is ``knowledge/``), so this check validates it on its own: both |
| 7 | +documents are parsed, the allowed key paths are removed from both, and what is |
| 8 | +left must be equal. Formatting and comments are free to stay as they are; the |
| 9 | +comparison is on parsed values. |
| 10 | +
|
| 11 | +The caller additionally runs ``config.parse_lifecycle`` on the head adapter |
| 12 | +(it needs an adapter object, not text) so the new values are also valid. |
| 13 | +""" |
| 14 | + |
| 15 | +from __future__ import annotations |
| 16 | + |
| 17 | +from typing import Any, Iterable |
| 18 | + |
| 19 | +import yaml |
| 20 | + |
| 21 | +LIFECYCLE = "knowledge_lifecycle" |
| 22 | +FLIP_KEYS: frozenset[tuple[str, ...]] = frozenset({(LIFECYCLE, "enabled"), (LIFECYCLE, "mode")}) |
| 23 | +CALIBRATION_KEYS: frozenset[tuple[str, ...]] = frozenset({(LIFECYCLE, "calibration_set")}) |
| 24 | + |
| 25 | + |
| 26 | +def _load(text: str, which: str) -> tuple[Any, list[str]]: |
| 27 | + try: |
| 28 | + data = yaml.safe_load(text) |
| 29 | + except yaml.YAMLError as exc: |
| 30 | + return None, [f"{which} manifest is not valid YAML: {exc}"] |
| 31 | + if not isinstance(data, dict): |
| 32 | + return None, [f"{which} manifest is not a mapping"] |
| 33 | + return data, [] |
| 34 | + |
| 35 | + |
| 36 | +def _copy(value: Any, active: frozenset[int] = frozenset()) -> Any: |
| 37 | + """A tree copy with YAML aliases UNshared: ``deepcopy`` keeps ``*alias`` |
| 38 | + nodes shared, so removing an allowed key through one path would also hide |
| 39 | + a change made through another. A recursive alias is refused.""" |
| 40 | + if isinstance(value, (dict, list)): |
| 41 | + if id(value) in active: |
| 42 | + raise ValueError("manifest contains a recursive alias") |
| 43 | + active = active | {id(value)} |
| 44 | + if isinstance(value, dict): |
| 45 | + return {key: _copy(item, active) for key, item in value.items()} |
| 46 | + return [_copy(item, active) for item in value] |
| 47 | + return value |
| 48 | + |
| 49 | + |
| 50 | +def _without(data: dict, paths: Iterable[tuple[str, ...]]) -> dict: |
| 51 | + out = _copy(data) |
| 52 | + for path in paths: |
| 53 | + node: Any = out |
| 54 | + for key in path[:-1]: |
| 55 | + node = node.get(key) if isinstance(node, dict) else None |
| 56 | + if node is None: |
| 57 | + break |
| 58 | + if isinstance(node, dict): |
| 59 | + node.pop(path[-1], None) |
| 60 | + return out |
| 61 | + |
| 62 | + |
| 63 | +def _typed(mapping: dict) -> dict[tuple[str, Any], Any]: |
| 64 | + # 1, 1.0 and True are equal (and collide) as Python dict keys; YAML keeps them apart |
| 65 | + return {(type(key).__name__, key): value for key, value in mapping.items()} |
| 66 | + |
| 67 | + |
| 68 | +def _diff_keys(base: Any, head: Any, prefix: str = "") -> list[str]: |
| 69 | + """Paths where ``base`` and ``head`` differ, type-sensitively: YAML ``1``, |
| 70 | + ``1.0`` and ``true`` are different values even though Python compares |
| 71 | + them equal (and ``parse_lifecycle`` would stringify them differently).""" |
| 72 | + where = prefix or "<root>" |
| 73 | + if isinstance(base, dict) and isinstance(head, dict): |
| 74 | + typed_base, typed_head = _typed(base), _typed(head) |
| 75 | + out = [] |
| 76 | + for key in sorted(set(typed_base) | set(typed_head), key=str): |
| 77 | + child = f"{prefix}.{key[1]}" if prefix else str(key[1]) |
| 78 | + if key not in typed_base or key not in typed_head: |
| 79 | + out.append(child) |
| 80 | + else: |
| 81 | + out.extend(_diff_keys(typed_base[key], typed_head[key], child)) |
| 82 | + return out |
| 83 | + if isinstance(base, list) and isinstance(head, list): |
| 84 | + if len(base) != len(head): |
| 85 | + return [where] |
| 86 | + out = [] |
| 87 | + for index, (left, right) in enumerate(zip(base, head)): |
| 88 | + out.extend(_diff_keys(left, right, f"{where}[{index}]")) |
| 89 | + return out |
| 90 | + same = type(base) is type(head) and (base == head or (base != base and head != head)) # .nan |
| 91 | + return [] if same else [where] |
| 92 | + |
| 93 | + |
| 94 | +def check_lifecycle_flip(base_yaml: str, head_yaml: str, *, |
| 95 | + allowed: Iterable[tuple[str, ...]]) -> list[str]: |
| 96 | + """Problems when the manifest changed anything besides the ``allowed`` |
| 97 | + key paths (empty list: the edit is confined to them).""" |
| 98 | + allowed = list(allowed) |
| 99 | + base, problems = _load(base_yaml, "base") |
| 100 | + head, more = _load(head_yaml, "head") |
| 101 | + problems += more |
| 102 | + if problems: |
| 103 | + return problems |
| 104 | + try: |
| 105 | + changed = _diff_keys(_without(base, allowed), _without(head, allowed)) |
| 106 | + except ValueError as exc: |
| 107 | + return [str(exc)] |
| 108 | + return [f"manifest key {key} changed; only " |
| 109 | + f"{', '.join('.'.join(p) for p in sorted(allowed))} may change" for key in changed] |
| 110 | + |
| 111 | + |
| 112 | +def check_flip_to_shadow(base_yaml: str, head_yaml: str) -> list[str]: |
| 113 | + """The init lifecycle flip: only ``enabled``/``mode`` change, and the head |
| 114 | + turns the service on in shadow mode.""" |
| 115 | + problems = check_lifecycle_flip(base_yaml, head_yaml, allowed=FLIP_KEYS) |
| 116 | + if problems: |
| 117 | + return problems |
| 118 | + section = (yaml.safe_load(head_yaml) or {}).get(LIFECYCLE) |
| 119 | + if not isinstance(section, dict): |
| 120 | + return [f"head manifest has no {LIFECYCLE} mapping"] |
| 121 | + if section.get("enabled") is not True: |
| 122 | + problems.append(f"{LIFECYCLE}.enabled must be true") |
| 123 | + if section.get("mode", "shadow") != "shadow": # parse_lifecycle's default |
| 124 | + problems.append(f"{LIFECYCLE}.mode must be shadow") |
| 125 | + return problems |
0 commit comments