Skip to content

symfonic.agent.configuration.normalizer

normalizer

One-shot native configuration normalization.

normalize_config 

normalize_config(*, runtime: RuntimeConfig, capabilities: Sequence[FamilyConfig] = (), services: Sequence[FamilyConfig] = (), integrations: IntegrationSelections | None = None, compat: LegacyBehavior | None = None, provenance: ConfigProvenance | None = None, bindings: ConfigBindings | None = None, context: ValidationContext | None = None) -> NormalizedConfig

Normalize once, validate once, and return the compiler's sole input.

Source code in src/symfonic/agent/configuration/normalizer.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def normalize_config(
    *,
    runtime: RuntimeConfig,
    capabilities: Sequence[FamilyConfig] = (),
    services: Sequence[FamilyConfig] = (),
    integrations: IntegrationSelections | None = None,
    compat: LegacyBehavior | None = None,
    provenance: ConfigProvenance | None = None,
    bindings: ConfigBindings | None = None,
    context: ValidationContext | None = None,
) -> NormalizedConfig:
    """Normalize once, validate once, and return the compiler's sole input."""
    capability_map = _index_families(capabilities, CAPABILITY_MODELS, "capability")
    service_map = _index_families(services, SERVICE_MODELS, "service")
    selected_integrations = integrations or IntegrationSelections()
    live_bindings = bindings or ConfigBindings()
    validation_context = context or ValidationContext()

    generated = _native_entries(runtime, runtime.family_id)
    for family_id, value in (*capability_map.items(), *service_map.items()):
        generated.update(_native_entries(value, family_id))
    generated.update(
        _native_entries(selected_integrations.provider, selected_integrations.provider.family_id)
    )
    if selected_integrations.embedding:
        generated.update(
            _native_entries(
                selected_integrations.embedding,
                selected_integrations.embedding.family_id,
            )
        )
    if selected_integrations.telemetry:
        generated.update(
            _native_entries(
                selected_integrations.telemetry,
                selected_integrations.telemetry.family_id,
            )
        )
    if compat:
        generated.update(_native_entries(compat, compat.family_id))
    if provenance is None:
        provenance = ConfigProvenance(entries=generated)
    else:
        missing = sorted(set(generated) - set(provenance.entries))
        if missing:
            raise ConfigurationError("XV-01 missing provenance for target " + ", ".join(missing))

    normalized = NormalizedConfig(
        runtime=runtime,
        capabilities=capability_map,
        services=service_map,
        integrations=selected_integrations,
        compat=compat,
        provenance=provenance,
        binding_descriptors=live_bindings.descriptors(),
    )
    from .validation import validate_dependencies

    validate_dependencies(normalized, bindings=live_bindings, context=validation_context)
    return normalized