Environment inspectors: what is not installed, and what will not survive.
Both questions are about the deployment rather than the configuration, which
is why they are asked here against the normalised result instead of inside the
adapter pass. An adopter's laptop and their production image disagree about
what is installed, and a configuration is not wrong for being run in both.
The lifecycle half exists because of a gap worth naming. XV-18 requires a
checkpoint service before human pause/resume is allowed -- and when a legacy
configuration enables ask_user, the compatibility layer supplies one, at
its default in-memory backend, along with an in-memory pause-token store. The
rule is satisfied and the deployment still loses every paused run on restart
and cannot serve a resume from a second replica. Nothing raises, nothing warns,
and the failure appears in production as a pause token that does not exist.
lifecycle_risks
lifecycle_risks(probe: ConfigProbe) -> list[InspectionFinding]
State that a restart, a replica, or a swept directory will lose.
Source code in src/symfonic/diagnostics/inspection/service_checks.py
| def lifecycle_risks(probe: ConfigProbe) -> list[InspectionFinding]:
"""State that a restart, a replica, or a swept directory will lose."""
findings = _rule_findings(probe)
if probe.normalized is None:
return findings
human = probe.family("cap.human")
paused = human is not None and (
human.elicitation is not None or human.interrupts is not None
)
checkpoint = probe.family("svc.checkpoint")
background = probe.family("svc.background")
if paused and checkpoint is not None and checkpoint.backend == "memory":
findings.append(
InspectionFinding(
code="LIFE-1",
category="lifecycle-risk",
severity=Severity.WARN,
subject="svc.checkpoint.backend",
detail=(
"human pause/resume is enabled against an in-process "
"checkpoint, so a paused run is lost on restart and invisible "
"to every other replica"
),
migration_action=(
"Select a durable checkpoint backend (sqlite for one process, "
"postgres or mongo for more than one)."
),
)
)
if paused and background is not None and background.token_store == "memory":
findings.append(
InspectionFinding(
code="LIFE-2",
category="lifecycle-risk",
severity=Severity.WARN,
subject="svc.background.token_store",
detail=(
"pause tokens are stored in process memory, so a resume "
"delivered to another replica cannot find its token"
),
migration_action=(
"Use the persistent pause-token store once more than one "
"process serves resumes."
),
)
)
path = getattr(checkpoint, "sqlite_path", None)
if path and str(path).startswith(_EPHEMERAL_PREFIXES):
findings.append(
InspectionFinding(
code="LIFE-3",
category="lifecycle-risk",
severity=Severity.WARN,
subject="svc.checkpoint.sqlite_path",
detail=(
f"the checkpoint database lives at {path}, a directory the "
"operating system may empty at any time"
),
migration_action=(
"Move the checkpoint database onto a persistent volume."
),
)
)
return findings
|
missing_extras(context: InspectionContext, probe: ConfigProbe) -> list[InspectionFinding]
Optional distributions this configuration needs and this install lacks.
Source code in src/symfonic/diagnostics/inspection/service_checks.py
| def missing_extras(
context: InspectionContext, probe: ConfigProbe
) -> list[InspectionFinding]:
"""Optional distributions this configuration needs and this install lacks."""
available = context.resolved_roots()
findings: list[InspectionFinding] = []
for root in required_roots(probe):
if root in available:
continue
requirement = requirement_for_root(root)
if requirement is None: # pragma: no cover - registry is exhaustive
continue
findings.append(
InspectionFinding(
code="MISSING-EXTRA",
category="missing-extra",
severity=Severity.ERROR,
subject=requirement.packaging_extra,
detail=(
f"{requirement.trigger}, which needs the import root "
f"{requirement.import_root!r}; it is not installed"
),
migration_action=install_hint(requirement.packaging_extra),
)
)
findings.extend(
InspectionFinding(
code=rule,
category="missing-extra",
severity=Severity.ERROR,
subject=rule,
detail=message,
migration_action=(
"Install the extra the rule names; the packaging name may differ "
"from the name in the message."
),
)
for rule, message in probe.errors
if rule == "XV-15"
)
return findings
|
required_roots
required_roots(probe: ConfigProbe) -> tuple[str, ...]
Every optional import root this configuration commits the adopter to.
Source code in src/symfonic/diagnostics/inspection/service_checks.py
| def required_roots(probe: ConfigProbe) -> tuple[str, ...]:
"""Every optional import root this configuration commits the adopter to."""
if probe.normalized is None:
return ()
roots: list[str] = []
if probe.normalized.integrations.telemetry is not None:
roots.append("opentelemetry")
embedding = probe.normalized.integrations.embedding
if embedding is not None:
root = _EMBEDDING_ROOTS.get(embedding.provider)
if root is not None:
roots.append(root)
checkpoint = probe.family("svc.checkpoint")
if checkpoint is not None:
root = _CHECKPOINT_ROOTS.get(checkpoint.backend)
if root is not None:
roots.append(root)
consolidation = probe.family("svc.consolidation")
linking = getattr(consolidation, "entity_linking", None)
if linking is not None and linking.extractor == "spacy":
roots.append("spacy")
human = probe.family("cap.human")
if human is not None and (human.elicitation is not None or human.interrupts is not None):
roots.append("jose")
return tuple(dict.fromkeys(roots))
|