aeat.core.observability._models module

Strict pydantic v2 record types for the run-trace observability layer.

Every type is strict=True, frozen=True, extra="forbid". Closed sets are enum.StrEnum. The RunEventPayload is a tagged union with an exactly-one-variant invariant enforced by a model_validator(mode="after") — bare dict[str, Any] is deliberately absent from the wire so every persisted JSONL line round-trips through the model.

Audit data policy

Run traces are audit artefacts; payloads will contain data that is sensitive in a tax / PII sense:

  • FormFillPayload.value is the literal form-field value — i.e. the tax figure the operator put into an AEAT draft. Treat the file as containing tax-return data.

  • NavigationPayload.url / NavigationPayload.description capture the user’s navigation path through AEAT sede. URLs may embed session identifiers; callers must not record authentication tokens here.

  • ErrorPayload.message is free-form and may contain traceback fragments with file paths or captured user input.

  • ArgumentRecord values are redacted for secret-named parameters by aeat.entrypoints.cli._observability.build_arguments() (password / secret / token / etc. → "***"). Other argument values are recorded verbatim.

  • RunTrace.cert_fingerprint is a SHA-256 of the configured PKCS#12 on disk — a stable identity marker of the operator’s cert, not a secret, but identifying.

Callers that sync var/runs/ to cloud storage must understand that every one of these fields is in scope. The framework does not attempt DLP-style scanning — it trusts callers not to feed secrets into the payload fields they control.

class ArgumentSource(*values)[source]

Bases: StrEnum

Provenance label for a CLI argument captured on a RunTrace.

ENV, CONFIG and DEFAULT values are recorded for audit completeness but are not re-emitted on argv during replay.

Variables:

  • FLAG – Option-style flag (e.g. --since 2026-01-01).

  • POSITIONAL – Positional argument that must be re-emitted in the original order with no -- prefix during replay (e.g. notificacion_id on aeat inbox show).

  • ENV – Value sourced from a process environment variable.

  • CONFIG – Value sourced from a configuration file.

  • DEFAULT – Value sourced from the option’s declared default.

FLAG
POSITIONAL
ENV
CONFIG
DEFAULT
class RunEventKind(*values)[source]

Bases: StrEnum

Closed catalogue of run-event kinds emitted by the observability layer.

Variables:

  • STEP_START – Boundary marker entering a logical step.

  • STEP_END – Boundary marker leaving a logical step.

  • NAVIGATION – A page navigation inside the AEAT sede browser.

  • FORM_FILL – A form-field value written into an AEAT draft form.

  • ASSERTION – A workflow-level expectation evaluation.

  • CACHE_HIT – Indicates a cached lookup served the request.

  • ERROR – A captured failure surfaced during the run.

  • WORKFLOW_STARTED – Links the run to a workflow-engine run id.

  • WORKFLOW_COMPLETED – Marks workflow-engine completion.

STEP_START
STEP_END
NAVIGATION
FORM_FILL
ASSERTION
CACHE_HIT
ERROR
WORKFLOW_STARTED
WORKFLOW_COMPLETED
class RunOutcome(*values)[source]

Bases: StrEnum

Terminal outcome recorded on a RunTrace.

Variables:

  • OK – The yielded body returned cleanly.

  • FAILED – The yielded body raised, or never executed because STEP_START itself failed.

  • ABORTED – The run was cancelled before completion.

OK
FAILED
ABORTED
class ArgumentRecord(**data)[source]

Bases: BaseModel

A single CLI argument captured for replay.

Variables:

  • name – Python parameter name as bound by the wrapped command (e.g. "as_json").

  • value – Stringified argument value.

  • source – Where the value originated; see ArgumentSource.

  • cli_flag – Optional override carrying the actual Typer option spelling (e.g. "--json") when the Python parameter name differs from the user-facing flag. Without the override, aeat.core.observability._replay._argv_from_arguments() derives the flag by replacing underscores with dashes — which is wrong for renamed options like typer.Option(False, "--json") bound to parameter as_json.

Parameters:
name: str
value: str
source: ArgumentSource
cli_flag: str | None
class NavigationPayload(**data)[source]

Bases: BaseModel

Payload for RunEventKind.NAVIGATION.

Variables:

  • url – Destination URL of the navigation event.

  • description – Optional human-readable label for the navigation.

Parameters:
url: str
description: str
class FormFillPayload(**data)[source]

Bases: BaseModel

Payload for RunEventKind.FORM_FILL.

Variables:

  • form_id – Identifier of the AEAT form being filled (e.g. "aeat-130").

  • display_number – Browser-visible box number within the form.

  • value – Literal value written to the box.

Parameters:

  • form_id (str)

  • display_number (str)

  • value (str)

form_id: str
display_number: str
value: str
class AssertionPayload(**data)[source]

Bases: BaseModel

Payload for RunEventKind.ASSERTION.

Variables:

  • expectation – Stable string identifying the assertion.

  • passed – Whether the assertion held.

  • detail – Optional free-form diagnostic text.

Parameters:
expectation: str
passed: bool
detail: str
class CacheHitPayload(**data)[source]

Bases: BaseModel

Payload for RunEventKind.CACHE_HIT.

Variables:

  • cache_name – Stable identifier of the cache that served the value.

  • key – Cache key whose lookup succeeded.

Parameters:
cache_name: str
key: str
class ErrorPayload(**data)[source]

Bases: BaseModel

Payload for RunEventKind.ERROR.

Variables:

  • error_type – Class name of the captured exception.

  • message – Free-form diagnostic text; may include traceback fragments. See the module docstring for the redaction contract this field is subject to.

Parameters:

  • error_type (str)

  • message (str)

error_type: str
message: str
class StepBoundaryPayload(**data)[source]

Bases: BaseModel

Payload for RunEventKind.STEP_START and RunEventKind.STEP_END.

Variables:

  • step_id – Identifier of the step the boundary refers to.

  • label – Human-readable label (typically the entrypoint string).

Parameters:
step_id: str
label: str
class WorkflowLinkPayload(**data)[source]

Bases: BaseModel

Payload for RunEventKind.WORKFLOW_STARTED / WORKFLOW_COMPLETED.

Links the observability run_id to a workflow-engine run_id via a workflow_run_id field; the two identifiers are deliberately distinct so the observability layer can wrap a workflow invocation without conflating its identity.

Variables:

workflow_run_id – Workflow-engine run id linked to this trace.

Parameters:

workflow_run_id (str)

workflow_run_id: str
class GenericPayload(**data)[source]

Bases: BaseModel

Structured-but-typed key/value payload for ad-hoc events.

Fields are a tuple of (name, str_value) pairs so the wire shape stays free of bare dict[str, Any] while still allowing extensibility for downstream call sites.

Variables:

fields – Ordered tuple of (name, str_value) pairs.

Parameters:

fields (tuple[tuple[str, str], ...])

fields: tuple[tuple[str, str], ...]
class RunEventPayload(**data)[source]

Bases: BaseModel

Tagged-union wrapper for the per-event payload variants.

Exactly one variant field must be set; the invariant is enforced post-construction by _exactly_one().

Variables:
Parameters:
navigation: NavigationPayload | None
form_fill: FormFillPayload | None
assertion: AssertionPayload | None
cache_hit: CacheHitPayload | None
error: ErrorPayload | None
step: StepBoundaryPayload | None
generic: GenericPayload | None
class RunEvent(**data)[source]

Bases: BaseModel

A single observability event captured during a run.

Variables:

  • run_id – Owning run identifier (16-char lowercase hex).

  • step_id – Step identifier active when the event was emitted.

  • kind – One of RunEventKind.

  • payload – Tagged-union payload; see RunEventPayload.

  • timestamp – UTC capture time; must be timezone-aware.

  • module – __name__ of the caller that emitted the event.

Parameters:
run_id: str
step_id: str
kind: RunEventKind
payload: RunEventPayload
timestamp: datetime
module: str
class RunTrace(**data)[source]

Bases: BaseModel

Metadata header persisted as trace.json for a CLI invocation.

Variables:

  • run_id – 16-char lowercase hex identifier for the run.

  • started_at – UTC enter time of the outermost run context.

  • finished_at – UTC exit time, or None if persistence happens before exit.

  • entrypoint – Stable CLI entrypoint string.

  • arguments – Tuple of ArgumentRecord captured for replay.

  • corpus_sha256 – Fingerprint of .vault/ plus aeat.core.config.Settings plus env/.env at enter time; gates replay_run().

  • db_sha256 – Fingerprint of the local var/ state tree at enter time.

  • cert_fingerprint – SHA-256 of the configured PKCS#12 cert, or "" when no cert is configured.

  • outcome – Terminal run outcome; see RunOutcome.

  • replay_of – Run id of the immediate original trace when this trace was produced by a replay re-entry, otherwise None. Replaying a replay produces a new trace whose replay_of points at the second-level trace, NOT at the chain root — walk the chain by following each replay_of link until you reach None. Each link is a supervised replay in its own right.

Parameters:
run_id: str
started_at: datetime
finished_at: datetime | None
entrypoint: str
arguments: tuple[ArgumentRecord, ...]
corpus_sha256: str
db_sha256: str
cert_fingerprint: str
outcome: RunOutcome
replay_of: str | None