"""Application-owned diagnostics, version reports, and repair probes.
:func:`build_cli_version_report` and :func:`render_cli_version_text` back the
root ``aeat --version`` surface without loading the bundled registry.
:func:`build_config_repair_report` composes environment checks,
:class:`~application.workflow.state_models.WorkflowState` loading,
:class:`~application.workflow.profile_health.ActiveProfileHealth` profile storage
verdicts, :class:`~application.wizard.status.WizardStatusReport`
readiness and secure-object decryptability into a
:class:`ConfigRepairReport` of :class:`DiagnosticCheck` rows.
Every warn/fail :class:`DiagnosticCheck` carries an application-owned
``precondition_verdict`` whose outcome is either a canonical action reference
or an explicit no-recovery classification. The validator raises
:class:`~application.errors.DiagnosticModelError` if a row is silent or
ambiguous. Renderers and CLI payloads can therefore treat the repair report as
a typed contract, not a best-effort text scan.
Secure-object repair helpers return :class:`SecureObjectIntegrityReport`
instances. Dry-run preview and quarantine use the same decryptability probe so
the committed mutation has the same namespace counts the operator saw before
confirming it.
See Also:
:mod:`application.workflow.profile_health` supplies the redacted
active-profile health verdict when secure workflow state is readable or
degraded.
:mod:`application.wizard.status` supplies semantic profile/auth
readiness once the workflow state has loaded.
:mod:`entrypoints.cli.config._repair_cli` wires these reports into
``aeat config repair`` commands.
"""
from __future__ import annotations
import sys
from typing import TYPE_CHECKING
from ..core.i18n.render import tr
from ..core.logging import default_log_file_path, get_logger
from ..core.operator_action_enums import NoRecoveryOutcome
from ..core.package_version import PACKAGE_VERSION
from ..core.redaction.rules import CLI_PROFILE_ID_PLACEHOLDER
from ..core.requirement import Requirement, RequirementValue
from .diagnostic_models import (
CliVersionReport as _CliVersionReport,
)
from .diagnostic_models import (
ConfigRepairReport as _ConfigRepairReport,
)
from .diagnostic_models import (
DiagnosticAudience as _DiagnosticAudience,
)
from .diagnostic_models import (
DiagnosticCheck as _DiagnosticCheck,
)
from .diagnostic_models import (
DiagnosticFinding as _DiagnosticFinding,
)
from .diagnostic_models import (
DiagnosticStatus as _DiagnosticStatus,
)
from .diagnostic_models import (
DiagnosticStatusValue as _DiagnosticStatusValue,
)
from .diagnostic_models import (
SecureObjectIntegrityReport as _SecureObjectIntegrityReport,
)
from .diagnostic_models import (
diagnostic_action_verdict,
diagnostic_no_recovery_verdict,
missing_verdict_binding,
resolved_verdict_binding,
)
from .diagnostic_models import (
ensure_models_rebuilt as _ensure_models_rebuilt,
)
from .diagnostics_ports import DiagnosticSecureObjectNamespace, DiagnosticsPorts
from .errors import DiagnosticModelError
from .operator_actions.models import PreconditionVerdict
# The registry authority, the secure-object repository, the workflow store,
# and the wizard-status projection are heavy import subtrees (the registry
# parse alone adds substantial cold-start import time). The ``aeat --version``
# fast path imports this module only for ``build_cli_version_report`` /
# ``render_cli_version_text``, neither of which needs any of them.
# Importing them lazily inside the functions that actually run keeps the
# version surface off the heavy import graph.
if TYPE_CHECKING:
from ..domain.calculations.registry.authority_artifact import ProfileDecodeContext
from .wizard.status import WizardStatusReport
from .workflow.profile_health import ActiveProfileHealth
from .workflow.state_models import WorkflowState
_log = get_logger(__name__)
[docs]
def build_cli_version_report() -> _CliVersionReport:
"""Return package identity for the CLI version surfaces."""
return _CliVersionReport(
package_name="cadrumo",
package_version=PACKAGE_VERSION,
)
def _logging_repair_check(log_parent_exists: bool) -> _DiagnosticCheck:
"""Build the read-only log-path diagnostic row."""
return _DiagnosticCheck(
name="logging.file",
status=_DiagnosticStatus.OK if log_parent_exists else _DiagnosticStatus.WARN,
summary=str(default_log_file_path()),
precondition_verdict=(
None
if log_parent_exists
else diagnostic_no_recovery_verdict(
condition_id="diagnostics.logging.file_parent.available",
evidence_id="diagnostics.logging.file_parent.observation",
values={
"log_file": str(default_log_file_path()),
"parent_exists": False,
},
outcome=NoRecoveryOutcome.OPERATOR_DECISION,
)
),
)
def _initial_config_repair_checks() -> list[_DiagnosticCheck]:
"""Build the environment, package, and logging check rows."""
log_parent_exists = default_log_file_path().parent.exists()
return [
_DiagnosticCheck(
name="environment.python",
status=_DiagnosticStatus.OK,
summary=sys.version.split()[0],
),
_DiagnosticCheck(
name="package.version",
status=_DiagnosticStatus.OK,
summary=PACKAGE_VERSION,
),
_logging_repair_check(log_parent_exists),
]
def _readable_secure_state_repair_checks(
*,
profile_decode_context: ProfileDecodeContext | None = None,
) -> tuple[list[_DiagnosticCheck], WizardStatusReport]:
"""Read secure workflow state and build its healthy diagnostic rows."""
from ..domain.calculations.registry.authority import bundled_indexed_authority
from .wizard.status import build_wizard_status
from .workflow.persistence import workflow_state_repository
from .workflow.profile_health import assess_active_profile_health
# Read the secure state through whatever session the operator already
# holds. This probe deliberately opens none of its own: it used to enter
# the shared-master provider whenever no per-profile session served the
# active bucket, which unlocked a taxpayer's bucket with no password so
# a health report could be printed. The locked case is a diagnostic
# verdict, not an obstacle -- the handler below already renders it as a
# warn with the profile-health verdict that tells the operator to log in.
state = workflow_state_repository().load()
checks = [
_DiagnosticCheck(
name="secure_state.load",
status=_DiagnosticStatus.OK,
summary=tr("cli.diagnostics.summary.state_backend_readable"),
),
]
with bundled_indexed_authority().operation() as operation:
profile_health = assess_active_profile_health(state, operation=operation)
setup_report = _repair_safe_wizard_status(
build_wizard_status(state, operation=operation),
active_profile_label=profile_health.active_profile_label,
)
checks.append(_active_profile_storage_check(profile_health))
checks.append(
build_profile_check(
setup_report,
profile_health=profile_health,
state=state,
profile_decode_context=profile_decode_context,
),
)
checks.append(_auth_check(setup_report))
return checks, setup_report
def _secure_state_failure_check(
exc: Exception,
*,
profile_health: ActiveProfileHealth,
missing_active_bucket_session: bool,
) -> _DiagnosticCheck:
"""Build the secure-state refusal row without exposing internal errors."""
return _DiagnosticCheck(
name="secure_state.load",
status=_DiagnosticStatus.WARN if missing_active_bucket_session else _DiagnosticStatus.FAIL,
summary=tr("cli.diagnostics.summary.state_backend_unreadable"),
# A missing bucket session on a cold start is an
# expected diagnostic verdict, not a fault to report
# verbatim. Surfacing the raw storage-session exception
# text leaks internal plumbing; the
# summary + typed profile verdict already guide the operator.
detail=None if missing_active_bucket_session else _compact_exception(exc),
precondition_verdict=(
_required_profile_health_verdict(profile_health)
if missing_active_bucket_session
else diagnostic_action_verdict(
condition_id="diagnostics.secure_state.load.readable",
evidence_id="diagnostics.secure_state.load.observation",
values={"readable": False},
action_id="operator.diagnostics.workflow.reset_progress",
argument_bindings=(resolved_verdict_binding("yes", True),),
)
),
)
def _unreadable_secure_state_repair_checks(
exc: Exception,
*,
ports: DiagnosticsPorts,
) -> list[_DiagnosticCheck]:
"""Build the redacted secure-state, profile, and auth fallback rows."""
from ..domain.calculations.registry.authority import bundled_indexed_authority
from .workflow.profile_health import assess_active_profile_health
_log.debug("config repair secure state probe failed", exc_info=True)
with bundled_indexed_authority().operation() as operation:
profile_health = assess_active_profile_health(operation=operation)
missing_active_bucket_session = is_missing_active_bucket_session_failure(exc, ports=ports)
return [
_secure_state_failure_check(
exc,
profile_health=profile_health,
missing_active_bucket_session=missing_active_bucket_session,
),
_active_profile_storage_check(profile_health),
_profile_unavailable_check(profile_health),
_auth_unavailable_check(profile_health),
]
def _secure_state_repair_checks(
*,
ports: DiagnosticsPorts,
profile_decode_context: ProfileDecodeContext | None = None,
) -> tuple[list[_DiagnosticCheck], WizardStatusReport | None]:
"""Read secure workflow state, falling back to redacted health rows."""
try:
checks, setup_report = _readable_secure_state_repair_checks(
profile_decode_context=profile_decode_context,
)
except Exception as exc: # pragma: no cover - concrete failure mode depends on local secure backend.
return _unreadable_secure_state_repair_checks(exc, ports=ports), None
return checks, setup_report
[docs]
def build_config_repair_report(
*,
ports: DiagnosticsPorts,
profile_decode_context: ProfileDecodeContext | None = None,
) -> _ConfigRepairReport:
"""Return local diagnostics for the ``aeat config repair`` surface.
Returns a :class:`ConfigRepairReport` enumerating every diagnostic
check and any suggested repairs. It is restricted to actionable local
configuration and secure-state health.
The secure-state branch reads
:class:`~application.workflow.state_models.WorkflowState`, derives
:class:`~application.workflow.profile_health.ActiveProfileHealth`, and builds a
:class:`~application.wizard.status.WizardStatusReport`. If that load
fails, the report still emits profile and auth rows from the redacted health
verdict so repair remains usable on a cold or degraded storage root.
Each emitted warning/failure row is validated by :class:`DiagnosticCheck` so
the caller never receives a silent repair finding.
"""
_ensure_models_rebuilt()
checks = _initial_config_repair_checks()
secure_state_checks, setup_report = _secure_state_repair_checks(
ports=ports,
profile_decode_context=profile_decode_context,
)
checks.extend(secure_state_checks)
secure_objects = _probe_secure_objects_integrity(ports)
checks.append(_secure_objects_integrity_check(secure_objects))
return _ConfigRepairReport(
overall=_overall_status(tuple(checks)),
package_name="cadrumo",
package_version=PACKAGE_VERSION,
python_version=sys.version.split()[0],
log_file=str(default_log_file_path()),
setup=setup_report,
secure_objects=secure_objects,
checks=tuple(checks),
)
[docs]
def render_config_repair_text(report: _ConfigRepairReport) -> str:
"""Render a compact human-readable repair report.
Preserves :attr:`DiagnosticCheck.audience` and never reconstructs command
prose from an application-owned :attr:`DiagnosticCheck.precondition_verdict`.
Typed actions are resolved by the entrypoint projection; this application
renderer remains a diagnosis-only view.
"""
lines = [
f"{tr('cli.diagnostics.repair.overall_label')}\t{report.overall}",
(f"{tr('cli.diagnostics.repair.version_label')}\t{report.package_name} {report.package_version}"),
f"{tr('cli.diagnostics.repair.python_label')}\t{report.python_version}",
f"{tr('cli.diagnostics.repair.logs_label')}\t{report.log_file}",
]
if report.setup is not None:
lines.append(
f"{tr('cli.diagnostics.repair.profile_label')}\t{report.setup.active_profile or '-'} "
f"({report.setup.profile_present_keys}/{report.setup.profile_total_keys})",
)
lines.append(f"{tr('cli.diagnostics.repair.auth_label')}\t{report.setup.auth_provider or '-'}")
lines.append(
tr(
"cli.diagnostics.repair.checks_heading",
)
)
for check in report.checks:
scope = (
""
if check.status == _DiagnosticStatus.OK or check.audience == _DiagnosticAudience.OPERATOR
else f" [{tr('cli.diagnostics.repair.audience_internal')}]"
)
lines.append(f"{check.status}\t{check.name}\t{check.summary}{scope}")
if check.detail:
lines.append(f"{tr('cli.diagnostics.repair.detail_label')}\t{check.detail}")
for finding in check.findings:
tag = _finding_tag(finding)
lines.append(f"{tr('cli.diagnostics.repair.finding_label')}\t{tag}{finding.summary}")
if finding.detail:
lines.append(f" {tr('cli.diagnostics.repair.detail_label')}\t{finding.detail}")
return "\n".join(lines) + "\n"
def _repair_safe_wizard_status(
report: WizardStatusReport,
*,
active_profile_label: str | None,
) -> WizardStatusReport:
"""Return a repair status with a label, never an internal bucket UUID."""
if report.active_profile is None:
return report
return report.model_copy(
update={"active_profile": active_profile_label or CLI_PROFILE_ID_PLACEHOLDER},
)
def _finding_tag(finding: _DiagnosticFinding) -> str:
"""Return the requirement prefix rendered ahead of a finding summary."""
if finding.requirement == "required":
return f"{tr('cli.diagnostics.repair.finding_required')}: "
if finding.requirement == "optional":
return f"{tr('cli.diagnostics.repair.finding_optional')}: "
return ""
def _probe_secure_objects_integrity(ports: DiagnosticsPorts) -> _SecureObjectIntegrityReport:
"""Iterate every populated secure-objects namespace and aggregate counts.
Returns an empty report when the table is empty or the engine cannot
be reached. Non-empty results expose per-namespace counts so the
operator can locate which application surface holds rows from a
rotated master-key generation.
The per-namespace rows come from the injected secure-object diagnostic
repository capability.
"""
_ensure_models_rebuilt()
try:
repo = ports.secure_object_repository
namespaces = repo.list_namespaces()
except Exception as exc: # pragma: no cover - engine resolution depends on local backend.
_log.debug(
"secure objects engine unreachable for repair probe: %s: %s",
type(exc).__name__,
exc,
exc_info=True,
)
return _SecureObjectIntegrityReport()
integrity_items: list[DiagnosticSecureObjectNamespace] = []
for ns in namespaces:
try:
integrity_items.append(repo.probe_namespace_integrity(ns))
# BROAD-EXCEPT-RATIONALE-DIAGNOSTICS-INTEGRITY-PROBE:
# storage probes raise heterogeneous backend errors; continue per namespace.
except Exception:
_log.debug("secure objects integrity probe failed for namespace=%s", ns, exc_info=True)
integrity_items.append(
DiagnosticSecureObjectNamespace(
namespace=ns,
readable=0,
unreadable=1,
),
)
integrity = tuple(integrity_items)
readable_total = sum(item.readable for item in integrity)
unreadable_total = sum(item.unreadable for item in integrity)
return _SecureObjectIntegrityReport(
namespaces=integrity,
readable_total=readable_total,
unreadable_total=unreadable_total,
)
def _secure_objects_integrity_check(report: _SecureObjectIntegrityReport) -> _DiagnosticCheck:
"""Render the ``secure_objects.integrity`` repair row.
A non-zero unreadable total becomes a warn :class:`DiagnosticCheck` whose
typed verdict resolves to the guarded quarantine path.
"""
if report.unreadable_total == 0:
if report.readable_total == 0:
return _DiagnosticCheck(
name="secure_objects.integrity",
status=_DiagnosticStatus.OK,
summary=tr("cli.diagnostics.summary.secure_objects_empty"),
)
return _DiagnosticCheck(
name="secure_objects.integrity",
status=_DiagnosticStatus.OK,
summary=tr(
"cli.diagnostics.summary.secure_objects_readable",
readable=report.readable_total,
namespaces=len(report.namespaces),
),
)
affected = ", ".join(
f"{item.namespace} ({item.unreadable}/{item.readable + item.unreadable})"
for item in report.namespaces
if item.unreadable > 0
)
return _DiagnosticCheck(
name="secure_objects.integrity",
status=_DiagnosticStatus.WARN,
summary=tr(
"cli.diagnostics.summary.secure_objects_unreadable",
unreadable=report.unreadable_total,
readable=report.readable_total,
),
detail=affected,
precondition_verdict=diagnostic_action_verdict(
condition_id="diagnostics.secure_objects.integrity.readable",
evidence_id="diagnostics.secure_objects.integrity.observation",
values={
"readable_total": report.readable_total,
"unreadable_total": report.unreadable_total,
},
action_id="operator.diagnostics.secure_objects.quarantine",
argument_bindings=(resolved_verdict_binding("yes", True),),
),
)
def _active_profile_storage_check(health: ActiveProfileHealth) -> _DiagnosticCheck:
"""Render :class:`ActiveProfileHealth` storage status before semantic readiness."""
active_profile = health.active_profile_label or (
CLI_PROFILE_ID_PLACEHOLDER if health.active_profile is not None else "-"
)
summary = tr(
"cli.diagnostics.summary.profile_storage",
active_profile=active_profile,
source=health.source,
status=health.status,
)
if health.status in {"none", "incomplete", "ready"}:
return _DiagnosticCheck(
name="profile.storage",
status=_DiagnosticStatus.OK,
summary=summary,
)
detail = health.profile_record_error or None
return _DiagnosticCheck(
name="profile.storage",
status=_DiagnosticStatus.WARN,
summary=summary,
detail=detail,
precondition_verdict=_required_profile_health_verdict(health),
)
def _profile_unavailable_check(health: ActiveProfileHealth) -> _DiagnosticCheck:
"""Render a profile-readiness row when only :class:`ActiveProfileHealth` is available."""
if health.status == "profile_locked":
# A locked profile is neither absent nor broken: the capsule is
# committed and the record is intact, merely not knowable until
# someone logs in. Falling through to either sibling branch below
# would tell the operator something untrue -- "unreadable" implies
# damage, "no profile configured" implies nothing was ever set up.
return _DiagnosticCheck(
name="profile.readiness",
status=_DiagnosticStatus.WARN,
summary=tr("cli.diagnostics.summary.profile_locked"),
precondition_verdict=_required_profile_health_verdict(health),
)
if health.status in {"dangling_pointer", "missing_profile_record", "profile_record_unreadable"}:
return _DiagnosticCheck(
name="profile.readiness",
status=_DiagnosticStatus.WARN,
summary=tr("cli.diagnostics.summary.profile_unreadable", status=health.status),
detail=health.profile_record_error or None,
precondition_verdict=_required_profile_health_verdict(health),
)
return _DiagnosticCheck(
name="profile.readiness",
status=_DiagnosticStatus.WARN,
summary=tr(
"cli.diagnostics.summary.profile_none",
),
precondition_verdict=_required_profile_health_verdict(health),
)
def _unset_profile_key_findings(state: WorkflowState | None) -> tuple[_DiagnosticFinding, ...]:
"""Return one finding per profile key the active profile leaves unset.
Each finding names the canonical key path, its operator-facing label,
whether the key is required or optional. The parent readiness row owns the
single typed profile-editor action, avoiding per-finding transport prose.
"""
from ..domain.calculations.registry.authority import bundled_indexed_authority
from .user_profile.profile_keys import profile_keys
if state is None:
return ()
try:
record = state.active_profile_record()
except Exception: # pragma: no cover
# BROAD-EXCEPT-RATIONALE-DIAGNOSTICS-RECORD-READ:
# record unreadability is handled by upstream storage checks.
_log.debug("config repair profile-key finding probe could not read record", exc_info=True)
return ()
if record is None:
return ()
from .user_profile.projections import record_to_path_values
values = record_to_path_values(record)
findings: list[_DiagnosticFinding] = []
with bundled_indexed_authority().operation() as operation:
entries = profile_keys(operation)
for entry in entries:
raw = values.get(entry.key)
if raw is not None and raw.strip() != "":
continue
requirement: RequirementValue = (
Requirement.REQUIRED if entry.requirement is Requirement.REQUIRED else Requirement.OPTIONAL
)
label = tr(str(entry.description))
findings.append(
_DiagnosticFinding(
summary=f"{entry.key} — {label}",
requirement=requirement,
),
)
return tuple(findings)
[docs]
def build_profile_check(
report: WizardStatusReport,
*,
profile_health: ActiveProfileHealth | None = None,
state: WorkflowState | None = None,
profile_decode_context: ProfileDecodeContext | None = None,
) -> _DiagnosticCheck:
"""Render semantic profile readiness from wizard status plus workflow state.
``report`` supplies the
:class:`~application.wizard.status.WizardStatusReport` counters and
next action. ``profile_health`` can override the row when
:class:`~application.workflow.profile_health.ActiveProfileHealth` says the active
profile bucket is unavailable. ``state`` lets the check expand missing
profile keys from :class:`~application.workflow.state_models.WorkflowState` into
per-key :class:`DiagnosticFinding` rows.
"""
from .workflow.profile_health import UNREADABLE_PROFILE_STATUSES
if profile_health is not None and profile_health.status == "profile_locked":
# Same distinction as `_profile_unavailable_check`: a locked profile
# is benign, so it gets its own sentence rather than falling through
# to `report.active_profile is None` below, which reports the
# "no profile configured" summary that is untrue of a locked one.
return _DiagnosticCheck(
name="profile.readiness",
status=_DiagnosticStatus.WARN,
summary=tr("cli.diagnostics.summary.profile_locked"),
precondition_verdict=_required_profile_health_verdict(profile_health),
)
if profile_health is not None and profile_health.status in UNREADABLE_PROFILE_STATUSES:
return _DiagnosticCheck(
name="profile.readiness",
status=_DiagnosticStatus.WARN,
summary=tr("cli.diagnostics.summary.profile_unreadable", status=profile_health.status),
detail=profile_health.profile_record_error or None,
precondition_verdict=_required_profile_health_verdict(profile_health),
)
if report.active_profile is None:
return _DiagnosticCheck(
name="profile.readiness",
status=_DiagnosticStatus.WARN,
summary=tr("cli.diagnostics.summary.profile_none"),
precondition_verdict=(
_required_profile_health_verdict(profile_health)
if profile_health is not None
else diagnostic_action_verdict(
condition_id="diagnostics.profile.active.available",
evidence_id="diagnostics.profile.active.observation",
values={"active_profile_present": False},
action_id="operator.profile.create",
argument_bindings=(missing_verdict_binding("profile_name"),),
missing_argument_names=("profile_name",),
)
),
)
unset_findings = _unset_profile_key_findings(state)
if not report.profile_ready:
return _profile_not_ready_check(
report,
unset_findings=unset_findings,
profile_decode_context=profile_decode_context,
)
return _DiagnosticCheck(
name="profile.readiness",
status=_DiagnosticStatus.OK,
summary=tr(
"cli.diagnostics.summary.profile_keys_set",
present=report.profile_present_keys,
total=report.profile_total_keys,
),
findings=unset_findings,
)
def _profile_not_ready_check(
report: WizardStatusReport,
*,
unset_findings: tuple[_DiagnosticFinding, ...],
profile_decode_context: ProfileDecodeContext | None,
) -> _DiagnosticCheck:
"""Render the readiness row for a profile still missing required keys.
Prefers the record probe's own findings; when it is unavailable the row
falls back to the wizard report's missing-required tuple so it still
names what is wrong rather than reporting an empty warning. Enrolment
keys already named by a required finding are not repeated.
"""
findings = _profile_not_ready_findings(
report,
unset_findings,
profile_decode_context=profile_decode_context,
)
return _DiagnosticCheck(
name="profile.readiness",
status=_DiagnosticStatus.WARN,
summary=tr(
"cli.diagnostics.summary.profile_missing_fields",
count=len(findings),
fields=", ".join(finding.summary for finding in findings),
),
precondition_verdict=_profile_not_ready_verdict(report, finding_count=len(findings)),
findings=findings,
)
def _profile_not_ready_findings(
report: WizardStatusReport,
unset_findings: tuple[_DiagnosticFinding, ...],
*,
profile_decode_context: ProfileDecodeContext | None,
) -> tuple[_DiagnosticFinding, ...]:
missing_required = tuple(f for f in unset_findings if f.requirement == "required")
if not missing_required:
missing_required = tuple(
_DiagnosticFinding(
summary=_grounded_profile_key_summary(key, profile_decode_context=profile_decode_context),
requirement=Requirement.REQUIRED,
)
for key in report.missing_required
)
already_named = {finding.summary.split(" — ", 1)[0] for finding in missing_required}
enrolment_findings = tuple(
_DiagnosticFinding(
summary=_grounded_profile_key_summary(key, profile_decode_context=profile_decode_context),
requirement=Requirement.REQUIRED,
)
for key in report.missing_enrolment
if key not in already_named
)
return missing_required + enrolment_findings
def _profile_not_ready_verdict(
report: WizardStatusReport,
*,
finding_count: int,
) -> PreconditionVerdict:
return diagnostic_action_verdict(
condition_id="diagnostics.profile.required_fields.complete",
evidence_id="diagnostics.profile.required_fields.observation",
values={
"missing_required_count": finding_count,
"profile_name": report.active_profile or CLI_PROFILE_ID_PLACEHOLDER,
},
action_id="operator.profile.edit",
argument_bindings=(
resolved_verdict_binding(
"profile_name",
report.active_profile or CLI_PROFILE_ID_PLACEHOLDER,
),
),
)
def _grounded_profile_key_summary(
key: str,
*,
profile_decode_context: ProfileDecodeContext | None = None,
) -> str:
"""Render a bare profile key path as "path - label", or unchanged.
The fallback branch above reads keys straight off the wizard report, which
carries canonical paths and no labels, while the primary branch already
emits the labelled form. Without this the operator's diagnostics would
name the same field two different ways depending on which branch produced
it, and the fallback way would be a raw dotted path.
The path is retained ahead of the label because the enrolment de-duplication
above splits on the separator to compare paths, and because the path is what
an operator types at the profile editor.
A key the schema does not resolve is returned unchanged rather than
guessed at.
"""
from .user_profile.preflight import build_profile_preflight_requirement
if profile_decode_context is None:
return key
schema = profile_decode_context.schema
requirement = build_profile_preflight_requirement(key, schema=schema)
if requirement.label == key:
return key
return f"{key} — {requirement.label}"
def _auth_unavailable_check(health: ActiveProfileHealth) -> _DiagnosticCheck:
"""Render auth readiness when workflow state cannot expose wizard status."""
return _DiagnosticCheck(
name="auth.readiness",
status=_DiagnosticStatus.WARN,
summary=tr("cli.diagnostics.summary.auth_state_unreadable"),
precondition_verdict=_required_profile_health_verdict(health),
)
def _required_profile_health_verdict(health: ActiveProfileHealth) -> PreconditionVerdict:
"""Return the health assessment's required typed refusal outcome.
``ActiveProfileHealth`` owns the applicability decision. A diagnostics
consumer may preserve that verdict, but cannot infer a second recovery from
the health status or from an error message.
"""
verdict = health.precondition_verdict
if verdict is None:
raise DiagnosticModelError(
f"non-ready active-profile health {health.status!r} has no precondition verdict",
)
return verdict
def _auth_check(report: WizardStatusReport) -> _DiagnosticCheck:
"""Render auth readiness from :class:`WizardStatusReport`."""
if not report.auth_provider:
return _DiagnosticCheck(
name="auth.readiness",
status=_DiagnosticStatus.WARN,
summary=tr("cli.diagnostics.summary.auth_none"),
precondition_verdict=diagnostic_action_verdict(
condition_id="diagnostics.auth.provider.configured",
evidence_id="diagnostics.auth.provider.observation",
values={"configured": False},
action_id="operator.auth.configure",
argument_bindings=(
missing_verdict_binding("file"),
resolved_verdict_binding("provider", "certificate"),
),
missing_argument_names=("file",),
),
)
if not report.login_ready:
return _DiagnosticCheck(
name="auth.readiness",
status=_DiagnosticStatus.WARN,
summary=tr(
"cli.diagnostics.summary.auth_no_session",
provider=report.auth_provider,
),
precondition_verdict=diagnostic_action_verdict(
condition_id="diagnostics.auth.session.ready",
evidence_id="diagnostics.auth.session.observation",
values={"login_ready": False, "provider": report.auth_provider},
action_id="operator.auth.login",
argument_bindings=(resolved_verdict_binding("provider", report.auth_provider),),
),
)
return _DiagnosticCheck(
name="auth.readiness",
status=_DiagnosticStatus.OK,
summary=tr(
"cli.diagnostics.summary.auth_session_ready",
provider=report.auth_provider,
),
)
[docs]
def is_missing_active_bucket_session_failure(
exc: BaseException,
*,
ports: DiagnosticsPorts,
) -> bool:
"""Classify a secure-state probe failure as a missing active bucket session.
This verdict decides whether ``secure_state.load`` warns (an expected cold
start) or fails (a real fault), and whether the raw exception text reaches
the operator, so it reads the typed exception chain rather than rendered
text.
Scanning the rendered text for the class name could not survive this
module's own contract: a producer that stops passing an authored sentence
positionally leaves ``str(exc)`` degraded to a locale key which no longer
contains the class name, so the scan would quietly stop matching and every
cold start would report as a hard fault. It also matched in the wrong
direction, classifying any unrelated failure whose message merely mentioned
the class as an expected cold start, which downgrades a genuine fault to a
warning and suppresses its detail.
Both the ``__cause__`` and ``__context__`` edges are followed. An explicit
``raise ... from`` sets only the former and an incidental re-raise inside an
``except`` block sets only the latter, so a chain can fork across the two;
the previous single-edge walk silently abandoned the ``__context__`` branch
whenever a ``__cause__`` was present. Identity marking keeps a self- or
mutually-referential chain from looping.
"""
seen: set[int] = set()
pending: list[BaseException] = [exc]
while pending:
current = pending.pop()
if id(current) in seen:
continue
seen.add(id(current))
if ports.session_failure_classifier(current):
return True
pending.extend(link for link in (current.__cause__, current.__context__) if link is not None)
return False
def _compact_exception(exc: BaseException) -> str:
root = getattr(exc, "orig", None)
if isinstance(root, BaseException):
exc = root
message = str(exc).splitlines()[0] if str(exc) else type(exc).__name__
return f"{type(exc).__name__}: {message}"
def _overall_status(checks: tuple[_DiagnosticCheck, ...]) -> _DiagnosticStatusValue:
if any(check.status == _DiagnosticStatus.FAIL for check in checks):
return _DiagnosticStatus.FAIL
if any(check.status == _DiagnosticStatus.WARN for check in checks):
return _DiagnosticStatus.WARN
return _DiagnosticStatus.OK
[docs]
def render_cli_version_text(report: _CliVersionReport) -> str:
"""Render a compact text line for human-facing version output."""
return f"{report.package_name} {report.package_version}"
[docs]
def secure_object_unreadable_total(*, ports: DiagnosticsPorts) -> int:
"""Return the count of rows the current master key cannot decrypt.
Lightweight wrapper over :func:`_probe_secure_objects_integrity` for
consumers (notably ``aeat app overview status``) that want to surface
a concise "N rows unreadable" pointer towards
``aeat config repair`` without rendering the per-namespace breakdown
themselves. The full breakdown remains the authority of
:class:`ConfigRepairReport`.
"""
return _probe_secure_objects_integrity(ports).unreadable_total
[docs]
def preview_quarantine_unreadable_secure_objects(*, ports: DiagnosticsPorts) -> _SecureObjectIntegrityReport:
"""Report the rows ``repair quarantine`` would move, mutating nothing.
Backs the ``aeat config repair quarantine --dry-run`` preview. Runs
the same per-namespace decryptability probe that
:func:`quarantine_unreadable_secure_objects` uses to decide which
rows to archive, but performs no copy and no delete: the
``secure_objects`` table is left exactly as found. The returned
:class:`SecureObjectIntegrityReport` carries, per namespace, the
``unreadable`` count (= rows the non-dry-run verb would quarantine)
and the ``readable`` count (= rows it would retain), so the
operator can confirm the blast radius before committing - the same
preview shape ``reset-progress --dry-run`` already offers.
"""
return _probe_secure_objects_integrity(ports)
[docs]
def quarantine_unreadable_secure_objects(*, ports: DiagnosticsPorts) -> _SecureObjectIntegrityReport:
"""Move every undecryptable secure-object row into the quarantine table.
Delegates to the injected secure-object diagnostic repository, whose
storage implementation creates the quarantine archive table on first use,
copies each undecryptable row's metadata and (still encrypted) payload into
the archive, then deletes the row from the active ``secure_objects`` table.
Decryptable rows are not touched.
The user's ciphertext is preserved in the archive; nothing is
auto-deleted. If a missing master key is later recovered (e.g.
restored from a recovery-key backup), the operator can manually
re-import rows from the quarantine table.
Returns:
A :class:`SecureObjectIntegrityReport` whose ``namespaces``
report carries per-namespace ``unreadable`` counts (= rows
moved to quarantine) and ``readable`` counts (= rows retained
in ``secure_objects``).
"""
_ensure_models_rebuilt()
namespaces = ports.secure_object_repository.quarantine_unreadable_rows()
quarantined_total = sum(item.unreadable for item in namespaces)
retained_total = sum(item.readable for item in namespaces)
return _SecureObjectIntegrityReport(
namespaces=namespaces,
readable_total=retained_total,
unreadable_total=quarantined_total,
)
__all__ = [
"build_cli_version_report",
"build_config_repair_report",
"build_profile_check",
"is_missing_active_bucket_session_failure",
"preview_quarantine_unreadable_secure_objects",
"quarantine_unreadable_secure_objects",
"render_cli_version_text",
"render_config_repair_text",
"secure_object_unreadable_total",
]