Skip to content

symfonic.diagnostics.runner

runner

Legacy HMS audit orchestrator (v7.16, item 5); deprecated by CORE-389.

Imports every registered check coroutine from :mod:symfonic.diagnostics.checks, invokes them in a stable order, and aggregates the results into a single :class:AuditReport.

Network-sensitive checks (currently only check_postgres_dsn) run last so the faster pure-introspection checks are not blocked by a slow probe. When no_network=True (the default in CI), the check_postgres_dsn short-circuits and emits an INFO that the probe was skipped.

Three checks read a SymfonicAgent: _config, _orchestrator and get_chat_model, so the audit is typed against :class:LegacyAuditTarget rather than the engine class. The embedding-provider and consolidate-schedule checks ignore the agent and read only the environment and the process's Celery app. symfonic doctor does not use this runner and reproduces none of these five checks: without --offline it lists each as UNKNOWN, and with --offline it omits them.

CheckFn module-attribute 

CheckFn = Callable[[LegacyAuditTarget], Awaitable[list[CheckResult]]]

Type alias for the uniform check coroutine signature.

AuditRunner 

AuditRunner(*, no_network: bool = False)

Invoke every registered check against an agent and aggregate results.

Deprecated (CORE-389): constructing one emits a DeprecationWarning pointing at symfonic doctor; run() is unchanged.

Source code in src/symfonic/diagnostics/runner.py 
96
97
98
def __init__(self, *, no_network: bool = False) -> None:
    warn_legacy_audit("AuditRunner", stacklevel=2)
    self._no_network = no_network

run async 

run(agent: LegacyAuditTarget) -> AuditReport

Invoke every check; return an :class:AuditReport.

Source code in src/symfonic/diagnostics/runner.py 
100
101
102
async def run(self, agent: LegacyAuditTarget) -> AuditReport:
    """Invoke every check; return an :class:`AuditReport`."""
    return await run_legacy_checks(agent, no_network=self._no_network)

LegacyAuditTarget

Bases: Protocol

What the legacy checks read from a SymfonicAgent; never checked at runtime.

resolve_no_network 

resolve_no_network(*, explicit_flag: bool | None = None) -> bool

Return whether network checks should be skipped.

Precedence:

  • If explicit_flag is not None, honour it verbatim.
  • Otherwise, check CI=true in the environment.

Most CI runners (GitHub Actions, GitLab, CircleCI, Buildkite) export CI=true; this gives the runner a sane default that doesn't require adopters to wire --no-network into every job config.

Source code in src/symfonic/diagnostics/runner.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def resolve_no_network(*, explicit_flag: bool | None = None) -> bool:
    """Return whether network checks should be skipped.

    Precedence:

    * If ``explicit_flag`` is not ``None``, honour it verbatim.
    * Otherwise, check ``CI=true`` in the environment.

    Most CI runners (GitHub Actions, GitLab, CircleCI, Buildkite) export
    ``CI=true``; this gives the runner a sane default that doesn't
    require adopters to wire ``--no-network`` into every job config.
    """
    if explicit_flag is not None:
        return explicit_flag
    return os.environ.get("CI", "").strip().lower() == "true"

run_legacy_checks async 

run_legacy_checks(agent: LegacyAuditTarget, *, no_network: bool) -> AuditReport

The audit body shared by :class:AuditRunner and audit_hms; never warns.

Source code in src/symfonic/diagnostics/runner.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def run_legacy_checks(agent: LegacyAuditTarget, *, no_network: bool) -> AuditReport:
    """The audit body shared by :class:`AuditRunner` and ``audit_hms``; never warns."""
    results: list[CheckResult] = []
    for fn in _LOCAL_CHECKS:
        findings = await fn(agent)
        results.extend(findings)

    if not no_network:
        for fn in _NETWORK_CHECKS:
            findings = await fn(agent)
            results.extend(findings)
    else:
        results.append(
            CheckResult(
                name="postgres_dsn.skipped",
                severity=Severity.INFO,
                message=(
                    "POSTGRES_DSN probe skipped (CI=true or --no-network)."
                ),
                fix_hint=None,
            ),
        )
    return AuditReport(results=results)

warn_legacy_audit 

warn_legacy_audit(entry: str, *, stacklevel: int) -> None

Emit the CORE-389 deprecation pointing entry at the native doctor.

Source code in src/symfonic/diagnostics/runner.py 
61
62
63
64
65
66
67
def warn_legacy_audit(entry: str, *, stacklevel: int) -> None:
    """Emit the CORE-389 deprecation pointing ``entry`` at the native doctor."""
    warnings.warn(
        _NATIVE_DOCTOR.format(entry=f"symfonic.diagnostics.{entry}"),
        DeprecationWarning,
        stacklevel=stacklevel + 1,
    )