cadrumo.core.json_contract module

Shared primitives for the CLI’s strict --json output contract.

Defines the strict pydantic v2 bases (OutputSchema, OutputRootSchema), the canonical success envelope (SchemaEnvelope), the typed diagnostic channel (Notice), and the emit helper (emit_json_success()) used by every authored machine-output path. CLI payload modules import these primitives directly from this module and route JSON mode through entrypoints.cli.common.emit_envelope().

emit_json_success() derives EnvelopeStatus from supplied Notice values via derive_status() and applies core.redaction.rules.redact_structured_for_cli_output() to the entire envelope before writing stdout. Text output remains owned by core.output_rendering.render_command_output(), so redaction and the reveal_cli_identifiers_opt_in switch stay consistent across text and JSON surfaces.

Living in core keeps domain and adapter packages free of any dependency on entrypoints.cli: a wrapped command emits its strict-validated payload through emit_json_success() without having to know how the CLI itself wires Click options.

This module owns the stdout success contract and schema registry. The stderr failure document is rendered by core.errors using the same ENVELOPE_SCHEMA_VERSION, and text-mode layout remains owned by core.output_rendering.

ENVELOPE_SCHEMA_VERSION

Envelope contract version shared by the success SchemaEnvelope and the stderr error envelope. Both documents carry the same outer spine (schema_version, command, status, notices), so the version is pinned once here and bumped only on a backwards-incompatible change to that spine.

class EnvelopeStatus(*values)[source]

Bases: StrEnum

Outcome discriminator carried on every CLI return document.

success and warning ride on the stdout SchemaEnvelope (warning when the command attached at least one warning-severity Notice); error rides on the stderr error envelope. A machine consumer reads this single field to learn the outcome instead of branching on stdout-vs-stderr, and derive_status() is the success-envelope authority for computing it.

emit_json_success() never emits ERROR; blocking failures route through the shared CadrumoError boundary instead of being smuggled into stdout notices.

SUCCESS
WARNING
ERROR
class NoticeSeverity(*values)[source]

Bases: StrEnum

Severity of a single operator-facing Notice.

info is a non-fatal next-step hint or informational advisory; warning is a non-blocking advisory the operator should act on. A command that attaches any warning notice resolves to EnvelopeStatus.WARNING through derive_status().

INFO
WARNING
class ActionConditionEvidence(**data)[source]

Bases: PreconditionEvidence

One evaluated failed-condition fact projected onto the wire.

The model deliberately contains observed facts only. It does not select a condition, decide applicability, or resolve an action catalogue.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict]

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ResolvedActionReference(**data)[source]

Bases: PreconditionActionIdentity

One catalogue action after an outer resolver supplied its canonical target.

The resolver is outside cadrumo.core; this record neither looks up the action ID nor discovers the live command surface.

Parameters:

data (Any)

target_command_key: str
cli_path: tuple[str, ...] | None
arguments: Mapping[str, str] | None
model_config: ClassVar[ConfigDict]

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ResolvedActionArgument(**data)[source]

Bases: ActionArgumentResolution

One resolved or missing target argument in the action wire projection.

Parameters:
model_config: ClassVar[ConfigDict]

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ResolvedPreconditionAction(**data)[source]

Bases: PreconditionOutcomeInvariant[ActionConditionEvidence, ResolvedActionReference, ResolvedActionArgument]

A resolved precondition verdict projected for a machine consumer.

This is intentionally a wire DTO. Application/domain guards own the rejected-condition policy and evidence; the operator surface owns the catalogue and live-schema resolution. The DTO only proves that the action target and each argument were resolved before presentation.

Parameters:

  • failed_condition_id (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=3, max_length=160, pattern=^[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)+$, ascii_only=None)])

  • evidence (tuple[EvidenceT, ...])

  • action (ActionT | None)

  • argument_bindings (tuple[ArgumentT, ...])

  • missing_argument_names (tuple[str, ...])

  • conditionality (ActionConditionality)

  • no_recovery_outcome (NoRecoveryOutcome | None)

model_config: ClassVar[ConfigDict]

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ResolvedNoticeAction(**data)[source]

Bases: BaseModel

A fully materialised next action carried by a successful notice.

This transport record is deliberately distinct from ResolvedPreconditionAction: success notices do not report a failed condition, conditional recovery state, or a terminal outcome. The operator-surface resolver must establish the catalogue target and every required input before constructing this record; this DTO in turn refuses any unresolved argument so a consumer never receives a partial action.

Parameters:
model_config: ClassVar[ConfigDict]

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

action: ResolvedActionReference
argument_bindings: tuple[ResolvedActionArgument, ...]
class Notice(**data)[source]

Bases: BaseModel

One typed, non-blocking diagnostic on the envelope notices channel.

The single uniform surface for operator-facing warnings, advisories, and next-step hints across every command. Domain diagnostics (e.g. ModeloValidationFinding, source-resolution advisories) are projected into this shape rather than re-modelled as bespoke per-command payload fields. CLI helpers such as entrypoints.cli.common.emit_envelope() pass these values to emit_json_success(), while text renderers fold equivalent prose into their line output.

Variables:

  • severity – info or warning; drives the envelope status.

  • code – Stable machine-readable notice identifier (e.g. "modelo.calculate.unconsumed_iva").

  • message – Localized operator-facing presentation text. It cannot carry an executable command identity.

  • action –

    Optional schema-resolved action projection, and still the ONLY notice field that may identify an executable action. It carries a ResolvedPreconditionAction for a failed precondition or a fully materialised ResolvedNoticeAction after success.

    The success shape keeps forward guidance out of locale prose and retains every concrete target argument and its provenance.

  • context – Optional deterministic non-action diagnostic metadata (e.g. source-resolution reason / source_kind). Reserved action keys and executable command prose are rejected here.

Parameters:

Blocking failures are not notices; they raise an CadrumoError and emit on stderr. Command payload schemas should also avoid reintroducing bespoke advisory, hint, or warning fields inside result when a Notice can carry the same non-blocking diagnostic.

model_config: ClassVar[ConfigDict]

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

severity: NoticeSeverity
code: str
message: str
action: ResolvedPreconditionAction | ResolvedNoticeAction | None
context: Mapping[str, str] | None
derive_status(notices)[source]

Return EnvelopeStatus.WARNING if any notice is warning-severity.

Success documents never carry EnvelopeStatus.ERROR; that status is reserved for the stderr error envelope. The returned EnvelopeStatus is the stdout SchemaEnvelope status used by emit_json_success().

Return type:

EnvelopeStatus

Parameters:

notices (Sequence[Notice])

exception OutputSchemaError(message=None, *, context=None, translated_message=None)[source]

Bases: CadrumoError

Raised when a strict CLI output contract is violated.

It deliberately inherits core.errors.hierarchy.CadrumoError so registry defects route through the shared CLI error boundary instead of bypassing structured output.

Parameters:

  • message (str | None)

  • context (Mapping[str, object] | None)

  • translated_message (str | None)

Return type:

None

code
class OutputSchema(**data)[source]

Bases: BaseModel

Strict, frozen base class for every command-specific --json payload.

Subclasses inherit extra="forbid", frozen=True, strict=True, and validate_assignment=True so accidental field drift between contract and implementation surfaces as a validation error rather than a silently-extended payload. Each concrete result model should be referenced by the authored CLI command specification.

The class describes the inner result payload only; the outer SchemaEnvelope is applied later by emit_json_success().

model_config: ClassVar[ConfigDict]

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class OutputRootSchema(root=PydanticUndefined, **data)[source]

Bases: RootModel[TypeVar], Generic

Strict root/list base class for --json payloads with a non-mapping root.

Use this for commands whose top-level JSON value is a list or scalar rather than an object. Carries the same strict / frozen / validate-on- assignment configuration as OutputSchema, and participates in the same strict command-spec result contract.

Parameters:

root (RootModelRootType)

model_config: ClassVar[ConfigDict]

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

strict_round_trip(cls, obj)[source]

Project obj into cls through a genuine JSON-text round trip.

cls.model_validate(obj.model_dump(mode="json")) reads as a round trip but is not one under strict pydantic v2 validation: model_dump(mode="json") projects a StrEnum to its bare string, a datetime to its isoformat string, and a tuple to a JSON array, and model_validate on the resulting plain dict does not coerce any of those back to their native type – only model_validate_json’s genuine JSON-text parse gets that leniency. A model that gains one of those field types after its call sites were written breaks silently at the two-step dict form with no local signal at the call site itself.

Use this helper (or call cls.model_validate_json(obj.model_dump_json()) directly) at every typed cross-model projection boundary. It is the migration target the model_dump(mode="json")-into-model_validate inventory gate enforces, not a substitute for the gate: the codebase already had model_dump_json/model_validate_json one method-name swap away and around forty call sites independently reached for the unsafe two-step form anyway, so a convenience helper alone would only add a second available idiom rather than retiring the first.

Return type:

TypeVar(ModelT, bound= BaseModel)

Parameters:

  • cls (type[ModelT])

  • obj (BaseModel)

class SchemaEnvelope(**data)[source]

Bases: BaseModel, Generic

Stable outer envelope wrapping a successful command’s payload.

Every successful --json response is rendered through this envelope so consumers can rely on the same outer keys regardless of the inner payload shape. The outer spine (schema_version, command, status, notices) is shared with the stderr error envelope so one shape describes success, warning, and error outcomes. emit_json_success() constructs the runtime mapping and the JSON-contract conformance gate specialises this generic envelope over every schema authored by the command-spec graph.

The envelope is a wire contract, not the command dispatcher. It does not discover Click/Typer leaves, choose text output, or own command authorization; those layers supply a strict result and stable command path before entering this contract.

Variables:

  • schema_version – Envelope version; bumped only on backwards-incompatible changes to the shared spine.

  • command – Stable command path string (e.g. "workflow list").

  • active_profile – Human-readable label of the active taxpayer profile (the operator-chosen display name), or None before any profile exists and for non-profile-bound commands. The identity anchor a caller reconciles against; it is the label, never the redacted profile/bucket UUID. Resolved and injected at the CLI transport boundary (the core layer never scans profile manifests), so it stays None for any emitter that does not supply it.

  • status – Outcome discriminator (success or warning here).

  • result – The strict-validated command result.

  • notices – Typed non-blocking diagnostics (warnings, advisories, next-step hints) surfaced to the caller. Replaces the former free-form warnings string list.

Parameters:
model_config: ClassVar[ConfigDict]

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

schema_version: str
command: str
active_profile: str | None
status: EnvelopeStatus
result: ResultT
notices: list[Notice]
emit_json_success(command, result, *, notices=None, active_profile=None, indent=2, sort_keys=False, stream=None)[source]

Wrap result in the success spine and emit it via _write_json_document().

The envelope’s schema_version is pinned to ENVELOPE_SCHEMA_VERSION; bumping it is a contract-breaking change handled by the JSON-contract test suite, not a casual edit. The status is derived from the supplied notices (derive_status()) so the JSON outcome and the shell exit code never disagree. The assembled envelope is redacted through core.redaction.rules.redact_structured_for_cli_output() before _write_json_document() writes it.

This helper is stdout-only. Any raised CadrumoError is handled by the CLI error boundary, which renders the sibling stderr envelope instead of returning a success document with an error-shaped result.

Parameters:

  • command (str) – Stable command path string (e.g. "workflow list").

  • result (object) – The strict-validated command payload to surface as envelope.result.

  • notices (Sequence[Notice] | None) – Optional typed Notice diagnostics (warnings, advisories, next-step hints); defaults to an empty list.

  • active_profile (str | None) – Optional human label of the active taxpayer profile placed on the shared spine (the identity anchor). The core layer never scans profile manifests, so the CLI transport resolves the label and passes it here; None for non-profile-bound emitters. It rides through the same redaction pass as the rest of the envelope, but it is the non-secret display name, not the redacted profile/bucket UUID.

  • indent (int | None) – Indent width forwarded to _write_json_document().

  • sort_keys (bool) – Sort-keys flag forwarded to _write_json_document().

  • stream (Optional[IO[str]]) – Target text stream; defaults to sys.stdout.

Return type:

None

validate_registered_result(command, result)[source]

Strictly revalidate an already typed operator result.

The low-level JSON emitter intentionally remains a transport primitive for metadata and diagnostic surfaces. Operator command output must enter through this check so an envelope cannot claim a registered command while carrying an arbitrary result shape.

Return type:

OutputSchema | OutputRootSchema[Any]

Parameters:
validate_registered_envelope_document(document, schema)[source]

Strictly validate one emitted CLI success or error JSON document.

Return type:

dict[str, object]

Parameters:

  • document (object)

  • schema (RegisteredSchema | None)