Skip to content

symfonic.diagnostics

diagnostics

symfonic doctor -- adopter-config audit package (v7.16, item 5).

Public API:

  • :func:audit_hms -- async core; returns an :class:~symfonic.diagnostics.result.AuditReport for a constructed :class:~symfonic.agent.engine.SymfonicAgent.
  • :func:audit_hms_sync -- sync wrapper around :func:audit_hms; what the CLI subcommand and most adopter scripts will call.

Re-exports the data types so adopters can from symfonic.diagnostics import AuditReport, CheckResult, Severity without traversing the internal package layout.

The audit closes the silent-fail traps Jarvio's innocent_nightingale incident + nurse-navigator's first-hour friction surfaced:

  • Stub embedder (EMBEDDING_PROVIDER empty)
  • OpenAI-base-URL trap (chat proxy routed through OPENAI_API_BASE)
  • Hardcoded tenant_id="default" in the consolidate beat schedule
  • agent.get_chat_model('action') returning None
  • lazy_tooling=True orphan when prerequisites are missing
  • EPISODIC / SEMANTIC layers omitted from enabled_layers
  • Unreachable POSTGRES_DSN

AuditReport dataclass 

AuditReport(results: list[CheckResult] = list())

Aggregate of every :class:CheckResult produced by one run.

The runner builds this object after invoking every check function; the CLI consumes it to render either human-readable text or --json output.

counts 

counts() -> dict[str, int]

Return per-severity counts plus ok (= no findings).

Source code in src/symfonic/diagnostics/result.py 
52
53
54
55
56
57
58
59
60
61
62
def counts(self) -> dict[str, int]:
    """Return per-severity counts plus ``ok`` (= no findings)."""
    info = sum(1 for r in self.results if r.severity is Severity.INFO)
    warn = sum(1 for r in self.results if r.severity is Severity.WARN)
    error = sum(1 for r in self.results if r.severity is Severity.ERROR)
    return {
        "ok": 1 if not self.results else 0,
        "info": info,
        "warn": warn,
        "error": error,
    }

exit_code 

exit_code(*, strict: bool = False) -> int

Compute the CLI exit code.

Semantics (mypy-style):

  • No findings, or only INFO -> 0.
  • WARN present, ERROR absent -> 0 (or 2 if strict=True).
  • ERROR present -> 1.
Source code in src/symfonic/diagnostics/result.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def exit_code(self, *, strict: bool = False) -> int:
    """Compute the CLI exit code.

    Semantics (mypy-style):

    * No findings, or only ``INFO`` -> ``0``.
    * ``WARN`` present, ``ERROR`` absent -> ``0`` (or ``2`` if
      ``strict=True``).
    * ``ERROR`` present -> ``1``.
    """
    worst = self.worst_severity()
    if worst is None or worst is Severity.INFO:
        return 0
    if worst is Severity.WARN:
        return 2 if strict else 0
    # Severity.ERROR
    return 1

to_json_dict 

to_json_dict(*, strict: bool = False) -> dict[str, Any]

Serialise to the documented --json shape.

Source code in src/symfonic/diagnostics/result.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def to_json_dict(self, *, strict: bool = False) -> dict[str, Any]:
    """Serialise to the documented ``--json`` shape."""
    return {
        "checks": [
            {
                "name": r.name,
                "severity": r.severity.value,
                "message": r.message,
                "fix_hint": r.fix_hint,
            }
            for r in self.results
        ],
        "summary": self.counts(),
        "exit_code": self.exit_code(strict=strict),
    }

worst_severity 

worst_severity() -> Severity | None

Return the highest severity present, or None if clean.

Source code in src/symfonic/diagnostics/result.py 
64
65
66
67
68
69
70
def worst_severity(self) -> Severity | None:
    """Return the highest severity present, or ``None`` if clean."""
    if not self.results:
        return None
    return max(
        (r.severity for r in self.results), key=lambda s: s.rank,
    )

AuditRunner 

AuditRunner(*, no_network: bool = False)

Invoke every registered check against an agent and aggregate results.

The runner is constructed with the audit configuration (currently just no_network). run() is the async entry point; :func:audit_hms and :func:audit_hms_sync in the package __init__ are the public wrappers.

Source code in src/symfonic/diagnostics/runner.py 
63
64
def __init__(self, *, no_network: bool = False) -> None:
    self._no_network = no_network

run async 

run(agent: SymfonicAgent) -> AuditReport

Invoke every check; return an :class:AuditReport.

Source code in src/symfonic/diagnostics/runner.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def run(self, agent: SymfonicAgent) -> AuditReport:
    """Invoke every check; return an :class:`AuditReport`."""
    results: list[CheckResult] = []
    for fn in _LOCAL_CHECKS:
        findings = await fn(agent)
        results.extend(findings)

    if not self._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)

CheckResult dataclass 

CheckResult(
    name: str,
    severity: Severity,
    message: str,
    fix_hint: str | None = None,
)

Single audit finding emitted by a check function.

Attributes

name: Stable short identifier (e.g. "embedding_provider.stub"). Used in the JSON output and in human-readable headers. severity: :class:Severity of the finding. message: One-line human-readable summary. Rendered as the headline of the human output line. fix_hint: Optional remediation hint. Rendered on the line after the message, and surfaced in the JSON payload.

Severity

Bases: StrEnum

Audit-check severity.

Inherits from str (via :class:enum.StrEnum) so the JSON output (--json mode) emits the short uppercase label without a custom encoder.

rank property 

rank: int

Numeric rank for comparison.

Higher means more severe. Used by :meth:AuditReport.worst_severity and the runner's exit-code calculator without exposing the internal ordering to call sites.

audit_hms async 

audit_hms(
    agent: SymfonicAgent, *, no_network: bool | None = None
) -> AuditReport

Run the v7.16 audit suite against agent and return an :class:AuditReport.

Parameters

agent: The constructed :class:~symfonic.agent.engine.SymfonicAgent to audit. The runner inspects agent._config, agent._orchestrator, and process-level environment variables (EMBEDDING_PROVIDER, POSTGRES_DSN, ...). no_network: When True, skip the POSTGRES_DSN reachability probe and emit an INFO instead. When None (the default), defers to :func:resolve_no_network which honours CI=true in the environment.

Source code in src/symfonic/diagnostics/__init__.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
async def audit_hms(
    agent: SymfonicAgent,
    *,
    no_network: bool | None = None,
) -> AuditReport:
    """Run the v7.16 audit suite against ``agent`` and return an :class:`AuditReport`.

    Parameters
    ----------
    agent:
        The constructed :class:`~symfonic.agent.engine.SymfonicAgent`
        to audit.  The runner inspects ``agent._config``,
        ``agent._orchestrator``, and process-level environment
        variables (``EMBEDDING_PROVIDER``, ``POSTGRES_DSN``, ...).
    no_network:
        When ``True``, skip the ``POSTGRES_DSN`` reachability probe and
        emit an ``INFO`` instead.  When ``None`` (the default), defers
        to :func:`resolve_no_network` which honours ``CI=true`` in the
        environment.
    """
    runner = AuditRunner(no_network=resolve_no_network(explicit_flag=no_network))
    return await runner.run(agent)

audit_hms_sync 

audit_hms_sync(
    agent: SymfonicAgent, *, no_network: bool | None = None
) -> AuditReport

Sync wrapper around :func:audit_hms.

Wraps the async coroutine in :func:asyncio.run. This is what the symfonic doctor Typer command calls; most adopter scripts will prefer this entry point over the async core because the audit is invoked from a CLI or a one-shot validation helper, not inside a running event loop.

Adopters who already hold an event loop should call :func:audit_hms directly instead.

Source code in src/symfonic/diagnostics/__init__.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def audit_hms_sync(
    agent: SymfonicAgent,
    *,
    no_network: bool | None = None,
) -> AuditReport:
    """Sync wrapper around :func:`audit_hms`.

    Wraps the async coroutine in :func:`asyncio.run`.  This is what the
    ``symfonic doctor`` Typer command calls; most adopter scripts will
    prefer this entry point over the async core because the audit is
    invoked from a CLI or a one-shot validation helper, not inside a
    running event loop.

    Adopters who already hold an event loop should call
    :func:`audit_hms` directly instead.
    """
    return asyncio.run(audit_hms(agent, no_network=no_network))