Skip to content

symfonic.diagnostics.inspection.probe

probe

Read the adopter's raw configuration once, keeping what normalisation loses.

The T2.2.3 adapter already answers most of what a migration needs to know -- which legacy key won, which native target it landed on, which values were chosen rather than defaulted. What it does not do is survive a bad configuration: it raises, because a framework that normalises a contradictory configuration into a working one has lied to somebody. A diagnostic has the opposite obligation, so this module runs the adapter behind a boundary that turns its exceptions and warnings back into data.

The pass is deliberately permissive about the environment: extras are granted, the filesystem effect is granted, a parent scope is assumed. Those three facts are properties of a deployment, not of a configuration, and answering "is this configuration coherent?" must not depend on which extras happen to be installed on the machine running the inspection. The environment questions are asked separately, against the normalised result, by the missing-extra and lifecycle inspectors.

ConfigProbe dataclass 

ConfigProbe(supplied: bool, raw_keys: tuple[str, ...] = (), normalized: NormalizedConfig | None = None, diagnostics: tuple[CompatibilityDiagnostic, ...] = (), comparisons: Mapping[str, FieldComparison] = dict(), errors: tuple[tuple[str, str], ...] = (), advisories: tuple[tuple[str, str], ...] = (), rejected: str | None = None)

What one adapter pass established about a raw configuration.

legacy_keys property 

legacy_keys: tuple[tuple[str, str], ...]

The supplied keys that are legacy spellings, with their targets.

family 

family(family_id: str) -> Any | None

Return a normalised capability or service family, or None.

Source code in src/symfonic/diagnostics/inspection/probe.py 
81
82
83
84
85
86
87
88
89
90
def family(self, family_id: str) -> Any | None:
    """Return a normalised capability or service family, or ``None``."""
    if self.normalized is None:
        return None
    source = (
        self.normalized.capabilities
        if family_id.startswith("cap.")
        else self.normalized.services
    )
    return source.get(family_id)

flatten_keys 

flatten_keys(values: Mapping[str, Any], prefix: str = '') -> tuple[str, ...]

Every dotted path present in a raw configuration mapping.

Container paths are kept alongside leaves because the legacy surface has keys at both levels (agent.model is a key; so is agent.model.max_tokens), and dropping either would under-report.

Source code in src/symfonic/diagnostics/inspection/probe.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def flatten_keys(values: Mapping[str, Any], prefix: str = "") -> tuple[str, ...]:
    """Every dotted path present in a raw configuration mapping.

    Container paths are kept alongside leaves because the legacy surface has
    keys at both levels (``agent.model`` is a key; so is
    ``agent.model.max_tokens``), and dropping either would under-report.
    """
    paths: list[str] = []
    for key, value in values.items():
        path = f"{prefix}.{key}" if prefix else str(key)
        paths.append(path)
        if isinstance(value, Mapping):
            paths.extend(flatten_keys(value, path))
    return tuple(sorted(paths))

probe_config 

probe_config(config: Mapping[str, Any]) -> ConfigProbe

Run the legacy adapter over config without letting it raise.

.. warning::

Not thread-safe. Capturing the adapter's advisories needs :func:warnings.catch_warnings, which swaps the process-wide filter for the duration of the call: a concurrent thread can lose a warning it would otherwise have seen, or see one it had filtered out. Harmless for the CLI, which is single-threaded and does this once; a served process calling :func:~symfonic.diagnostics.inspection.inspect_migration from a worker pool should serialise the calls.

Source code in src/symfonic/diagnostics/inspection/probe.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def probe_config(config: Mapping[str, Any]) -> ConfigProbe:
    """Run the legacy adapter over *config* without letting it raise.

    .. warning::

       Not thread-safe. Capturing the adapter's advisories needs
       :func:`warnings.catch_warnings`, which swaps the **process-wide** filter
       for the duration of the call: a concurrent thread can lose a warning it
       would otherwise have seen, or see one it had filtered out. Harmless for
       the CLI, which is single-threaded and does this once; a served process
       calling :func:`~symfonic.diagnostics.inspection.inspect_migration` from a
       worker pool should serialise the calls.
    """
    if not config:
        return ConfigProbe(supplied=False)

    raw_keys = flatten_keys(config)
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always")
        try:
            result = FrameworkConfigAdapter.from_mapping(dict(config))
        except ConfigurationError as exc:
            errors = _parse_rules(str(exc))
            return ConfigProbe(
                supplied=True,
                raw_keys=raw_keys,
                errors=errors or (("XV-00", str(exc)),),
            )
        except Exception as exc:  # noqa: BLE001 -- any rejection is a finding
            return ConfigProbe(
                supplied=True,
                raw_keys=raw_keys,
                rejected=f"{type(exc).__name__}: {exc}",
            )
        advisories = tuple(
            rule
            for entry in caught
            if issubclass(entry.category, UserWarning)
            for rule in _parse_rules(str(entry.message))
        )

    return ConfigProbe(
        supplied=True,
        raw_keys=raw_keys,
        normalized=result.normalized,
        diagnostics=result.diagnostics,
        comparisons=dict(result.comparisons),
        advisories=tuple(sorted(set(advisories))),
    )