"""LLM-assisted ledger classification: suggest / apply / provider availability.
Wires the existing :class:`~domain.transactions.LLMClassifier` engine into
the operator suggest -> review -> confirm / override / reject loop without
rebuilding the classifier. The contract is deliberately thin:
* :func:`suggest_llm_classification` loads one transaction, runs the
(injected, default-resolved) classifier with the category-enabled prompt
spec, and returns a typed
:class:`~application.ledger._llm_suggestions.LLMClassificationSuggestion`
**without persisting anything**. Rejecting a suggestion is simply not
applying it.
* :func:`apply_llm_classification` persists an accepted suggestion through the
established classification write (:func:`~domain.transactions.set_classification`),
stamping ``classified_by`` with the classifier's ``decided_by`` (``llm:<model>``
provenance, distinct from manual / ``rule:``) and recording the model's
``confidence`` and ``reason``. The accepted decision is appended to the
profile audit trail through a
:class:`~adapters.persistence.profile.buckets.BucketEventHistoryRepository` as a
``ledger.transaction.classified`` event.
* :func:`available_llm_providers` reports which subprocess providers have a
usable CLI on ``PATH`` so the CLI can refuse instructively rather than crash.
Hallucination containment stays inside the engine: the classifier's
``classify`` runs the allow-list-guarded
:func:`~domain.transactions.parse_response`, so an out-of-allow-list
value is rejected before it ever reaches this module.
**Stage-1 constraint.** :func:`suggest_llm_classification` /
:func:`apply_llm_classification` persist only the non-regulated
``business_classification`` and optional expense ``category``; they never set a
regulated tax value.
**Stage-2 saturation.** :func:`saturate_llm_classification` /
:func:`apply_saturated_llm_classification` additionally persist the
model-selected ``iva_category`` and the system-DERIVED ``taxable_base`` /
``iva_rate`` / ``iva_amount``. The model still never emits a number — the rate
is looked up from the registry and the base and amount are derived with
``round_to_cents`` (see ``2026-06-04-llm-ledger-classification-adr``).
``irpf_category`` remains operator-only.
"""
from __future__ import annotations
import base64
import shutil
from collections.abc import Callable
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from uuid import uuid4
from ...adapters.outbound.llm import rasterise_pdf_pages_to_base64_png
from ...adapters.persistence.profile.buckets import BucketEventHistoryRepository
from ...adapters.persistence.storage import AttachmentStore, secure_object_repository_for_bucket
from ...core.config import Settings, load_settings
from ...core.logging import get_logger
from ...core.time import coerce_utc_aware, now
from ...domain.buckets import BucketEventHistoryRepositoryProtocol, BucketEventObjectType, BucketEventType
from ...domain.categories import SpendingCategory
from ...domain.iva import IvaCategory, resolve_category_rate, split_gross_at_rate
from ...domain.transactions import (
BusinessClassification,
LLMClassificationResponse,
LLMClassifier,
LLMSplitProposer,
LLMSplitResponse,
PromptSpec,
Transaction,
TransactionCatalogueRepositoryProtocol,
TransactionLifecycleState,
TransactionNotFoundError,
TransactionValidationError,
prompt_spec_with_every_spending_category,
prompt_spec_with_saturation_fields,
resolve_classifier,
resolve_split_proposer,
set_classification,
)
from ._actions_common import (
_build_bucket_event,
_result,
_save_transaction_catalogue_and_events,
_transaction_repository,
)
from ._actions_manual import update_manual_transaction_fields
from ._actions_split_merge import split_transaction
from ._evidence import MediaKind, PurchaseInvoiceEvidenceInputError, PurchaseInvoiceEvidenceService
from ._evidence_advisory import printed_iva_advisory
from ._evidence_input import (
cloud_evidence_read_permitted,
resolve_attachment_evidence_input,
resolve_purchase_invoice_evidence_input,
)
from ._evidence_split import derive_child_amounts
from ._evidence_textlayer import extract_evidence_text
from ._llm_suggestions import (
LLMClassificationSuggestion,
LLMProvider,
LLMProviderAvailability,
LLMSaturatedSuggestion,
LLMSplitApplyResult,
LLMSplitChildSuggestion,
LLMSplitSuggestion,
LLMSuggestionRejectionResult,
OperatorIvaDerivationResult,
)
from ._models import ManualLedgerTransactionPatch, ManualLedgerTransactionResult, SplitChildCommand
from ._vision_classifier import LocalVisionLLMClassifier
_logger = get_logger(__name__)
_BUCKET_EVENT_PAYLOAD_VERSION = 1
# The CLI binary each subprocess provider shells out to. Used by
# :func:`available_llm_providers` to probe PATH without spawning the process.
# ``antigravity`` resolves to the ``agy`` binary.
_PROVIDER_CLI_BINARY: dict[LLMProvider, str] = {
LLMProvider.CLAUDE: "claude",
LLMProvider.ANTIGRAVITY: "agy",
LLMProvider.CODEX: "codex",
}
[docs]
def available_llm_providers() -> tuple[LLMProviderAvailability, ...]:
"""Report which subprocess LLM providers are usable on this host.
Probes ``PATH`` for each provider's CLI binary with ``shutil.which``
(no process is spawned). The CLI surfaces this so an operator can discover
which providers are installed before classifying.
Returns:
One :class:`~application.ledger._llm_suggestions.LLMProviderAvailability`
per :class:`~application.ledger._llm_suggestions.LLMProvider`, ordered
by enum declaration.
"""
listings: list[LLMProviderAvailability] = []
for provider in LLMProvider:
binary = _PROVIDER_CLI_BINARY[provider]
resolved = shutil.which(binary)
listings.append(
LLMProviderAvailability(
provider=provider,
cli_binary=binary,
available=resolved is not None,
resolved_path=resolved,
),
)
return tuple(listings)
[docs]
def is_llm_provider_available(provider: LLMProvider) -> bool:
"""Return whether ``provider``'s CLI binary is resolvable on ``PATH``."""
return shutil.which(_PROVIDER_CLI_BINARY[provider]) is not None
def _resolve_default_classifier(provider: LLMProvider) -> LLMClassifier:
"""Resolve the production classifier for ``provider`` with the category prompt.
Builds the classifier with
:func:`~domain.transactions.prompt_spec_with_every_spending_category`
so the model also suggests an expense
:class:`~domain.categories.SpendingCategory`, and keeps
the allow-list-guarded ``parse_response`` path intact.
"""
return resolve_classifier(provider.value, spec=prompt_spec_with_every_spending_category())
@dataclass(frozen=True)
class _ResolvedEvidence:
"""A transaction's linked evidence resolved for an on-host read.
Exactly one read mode is populated: ``text`` for a text-layer PDF (inlined into
the prompt and fed to the cloud subprocess classifier, consent-gated) or base64
``images`` for a scan-only PDF / image (read in memory by the LOCAL vision
model, on-host, gestor-allowed). The ``images`` are transient
FINANCIAL-derived bytes and MUST never be persisted or logged
(``sensitive-financial-data-secure-storage-only``).
"""
reference: str
text: str | None
images: tuple[str, ...]
@property
def is_images(self) -> bool:
"""Whether this evidence routes to the on-host vision reader."""
return bool(self.images)
def _resolve_evidence(
transaction: Transaction,
*,
bucket_id: str,
settings: Settings,
evidence_acknowledged: bool,
) -> _ResolvedEvidence | None:
"""Resolve a transaction's linked evidence to an on-host read, or ``None``.
Returns ``None`` when the transaction has no linked evidence. A text-layer PDF
yields ``text`` and routes to the cloud subprocess classifier, so it is gated by
the cloud-upload consent posture (default-off, gestor-barred, per-invocation). A
scan-only PDF or an image yields base64 ``images`` and is read in memory by the
LOCAL Ollama vision model -- fully on-host, needing no cloud consent and
permitted for gestor deployments. Bytes are read from secure storage into memory
only; nothing is written to disk
(``sensitive-financial-data-secure-storage-only``).
Raises:
PurchaseInvoiceEvidenceInputError: When a text-layer read would transmit to a
cloud model but the per-invocation consent gate is not satisfied.
"""
evidence_id = transaction.purchase_invoice_evidence_id
attachment_ids = transaction.attachment_ids
if evidence_id is None and not attachment_ids:
return None
store = AttachmentStore(objects=secure_object_repository_for_bucket(bucket_id, settings))
if evidence_id is not None:
record = PurchaseInvoiceEvidenceService(settings=settings).view(bucket_id=bucket_id, evidence_id=evidence_id)
evidence_input = resolve_purchase_invoice_evidence_input(record, store=store)
reference = evidence_id
else:
reference = attachment_ids[0]
evidence_input = resolve_attachment_evidence_input(reference, store=store)
if evidence_input.media_kind is MediaKind.PDF:
try:
text = extract_evidence_text(evidence_input)
except PurchaseInvoiceEvidenceInputError:
text = "" # scan-only / no usable text layer -> on-host vision path
if text:
if not cloud_evidence_read_permitted(settings, acknowledged=evidence_acknowledged):
raise PurchaseInvoiceEvidenceInputError(
"reading text-layer evidence sends it to a cloud model, which requires the "
"explicit per-invocation consent acknowledgement; it is off by default and barred "
"for gestor deployments (scanned/image evidence is read on-host and needs no consent)",
suggestion="enable the cloud-upload consent posture and acknowledge the upload",
)
return _ResolvedEvidence(reference=reference, text=text, images=())
images = rasterise_pdf_pages_to_base64_png(evidence_input.data)
else:
images = (base64.b64encode(evidence_input.data).decode("ascii"),)
# The on-host vision read is the only path that reaches here. Gate it on the
# profile's llm_vision capability — opting out disables scanned/image reading
# entirely (a typed refusal, never a silent skip).
from ...core import ServiceCapability
from ..user_profile import resolve_active_capability
if not resolve_active_capability(ServiceCapability.LLM_VISION, settings=settings).enabled:
raise PurchaseInvoiceEvidenceInputError(
"on-host LLM vision reading is disabled for this profile; enable it to read scanned or image evidence",
suggestion="aeat config profile capabilities set llm_vision on",
)
return _ResolvedEvidence(reference=reference, text=None, images=images)
# Raised when a transaction must be read by a cloud subprocess provider (text-layer
# evidence, or no readable image evidence) but no ``--llm`` provider was supplied.
# The on-host vision path needs no provider, so this names that distinction.
_TEXT_PATH_NEEDS_PROVIDER = (
"classifying this transaction needs a cloud provider: pass --llm with claude, antigravity, or codex. "
"(--read-evidence reads a scanned or image invoice on-host with no provider, but this transaction has "
"no readable image evidence to route there.)"
)
def _run_vision_or_refuse[T](run: Callable[[], T], *, settings: Settings) -> T:
"""Run an on-host vision call, converting a missing/unreachable Ollama to a typed refusal.
The local adapter only guards HTTP *status* errors; a connection-refused or a
model-missing failure escaped every CLI ``except`` clause as a raw
``httpx.ConnectError`` / ``LLMProviderError`` traceback. This converts both into
an ``LLMClassifierError`` (which the classify CLI already renders) carrying the
exact remediation from :func:`probe_ollama_vision`
(``dependency-provisioning`` ADR: probe -> typed refusal with the fix).
"""
import httpx
from ...adapters.outbound.llm import LLMProviderError
from ...domain.transactions import LLMClassifierError
try:
return run()
except (httpx.HTTPError, LLMProviderError) as exc:
from ..provisioning import probe_ollama_vision
status = probe_ollama_vision(settings)
fix = status.remediation or "ensure the local Ollama vision model is reachable"
detail = status.detail if not status.available else str(exc)
raise LLMClassifierError(f"on-host vision reading failed: {detail}. Fix: {fix}") from exc
def _record_subprocess_run[T](run: Callable[[], T], *, provider: str) -> T:
"""Run a subprocess CLI classify/split call, recording local run-timing telemetry.
Wraps :class:`~domain.transactions.SubprocessLLMClassifier` calls (which
stay pure and time-unaware, per hexagonal layering -- the domain layer must
not import the storage-touching recorder). Records duration and outcome via
:class:`~adapters.outbound.llm.LLMRunTelemetryRecorder`, mirroring the
recording :class:`~adapters.outbound.llm.LLMClient.complete` performs
for the on-host vision transport. A run-telemetry write failure never masks
the real classification result or a real classifier error.
"""
import time
from ...adapters.outbound.llm import LLMCacheError, LLMRunRecord, LLMRunTelemetryRecorder
started_at = now()
clock_start = time.monotonic()
recorder = LLMRunTelemetryRecorder()
def _write(*, succeeded: bool, error_kind: str) -> None:
try:
recorder.record(
LLMRunRecord(
run_id=uuid4().hex,
caller="aeat.application.ledger.llm_classification",
provider=provider,
duration_ms=max(0, round((time.monotonic() - clock_start) * 1000)),
succeeded=succeeded,
error_kind=error_kind,
started_at=started_at,
),
)
except LLMCacheError:
_logger.debug("llm run-telemetry write failed; continuing without it", exc_info=True)
try:
result = run()
except Exception as exc:
_write(succeeded=False, error_kind=type(exc).__name__)
raise
_write(succeeded=True, error_kind="")
return result
def _classify_with_evidence(
transaction: Transaction,
evidence: _ResolvedEvidence | None,
*,
text_classifier: LLMClassifier | None,
spec: PromptSpec,
vision_classifier: LocalVisionLLMClassifier | None,
vision_model: str | None,
settings: Settings,
) -> tuple[LLMClassificationResponse, str]:
"""Classify, routing scan/image evidence to the on-host vision classifier.
Returns ``(response, provenance)``. Image evidence is read by the local vision
model (``llm:local-vision:<model>`` provenance) and needs no ``text_classifier``;
text or no evidence runs the cloud subprocess ``text_classifier``
(``llm:<provider>:<model>`` provenance), which must be present. ``vision_model``
overrides the settings default vision model for this read.
Raises:
TransactionValidationError: When the text path is taken but no
``text_classifier`` was resolved (no ``--llm`` provider supplied).
LLMClassifierError: When the on-host Ollama vision model is unreachable or
the configured model is not pulled.
"""
if evidence is not None and evidence.is_images:
# The vision path shells out through LLMClient.complete, which records
# its own run-timing telemetry -- do not double-record here.
vision = vision_classifier or LocalVisionLLMClassifier(spec=spec, settings=settings, model=vision_model)
images = evidence.images
response = _run_vision_or_refuse(
lambda: vision.classify(transaction, evidence_images=images),
settings=settings,
)
return response, vision.decided_by
if text_classifier is None:
raise TransactionValidationError(
_TEXT_PATH_NEEDS_PROVIDER,
context={"transaction_id": transaction.transaction_id},
)
text = evidence.text if evidence is not None else None
return _record_subprocess_run(
lambda: text_classifier.classify(transaction, evidence_text=text),
provider=text_classifier.decided_by,
), text_classifier.decided_by
def _split_with_evidence(
transaction: Transaction,
evidence: _ResolvedEvidence | None,
*,
proposer: LLMSplitProposer | None,
spec: PromptSpec,
vision_classifier: LocalVisionLLMClassifier | None,
vision_model: str | None,
settings: Settings,
) -> tuple[LLMSplitResponse, str]:
"""Propose a split, routing scan/image evidence to the on-host vision classifier.
``vision_model`` overrides the settings default vision model for this read.
Raises:
TransactionValidationError: When the text path is taken but no ``proposer``
was resolved (no ``--llm`` provider supplied).
"""
if evidence is not None and evidence.is_images:
# The vision path shells out through LLMClient.complete, which records
# its own run-timing telemetry -- do not double-record here.
vision = vision_classifier or LocalVisionLLMClassifier(spec=spec, settings=settings, model=vision_model)
images = evidence.images
response = _run_vision_or_refuse(
lambda: vision.propose_split(transaction, evidence_images=images),
settings=settings,
)
return response, vision.decided_by
if proposer is None:
raise TransactionValidationError(
_TEXT_PATH_NEEDS_PROVIDER,
context={"transaction_id": transaction.transaction_id},
)
text = evidence.text if evidence is not None else None
return _record_subprocess_run(
lambda: proposer.propose_split(transaction, evidence_text=text),
provider=proposer.decided_by,
), proposer.decided_by
[docs]
def suggest_llm_classification(
*,
bucket_id: str,
transaction_id: str,
provider: LLMProvider | None,
classifier: LLMClassifier | None = None,
vision_classifier: LocalVisionLLMClassifier | None = None,
vision_model: str | None = None,
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None,
read_evidence: bool = False,
evidence_acknowledged: bool = False,
settings: Settings | None = None,
) -> LLMClassificationSuggestion:
"""Run the LLM classifier for one transaction and return a suggestion.
Loads the transaction, runs the injected classifier (default-resolved from
``provider`` with the category-enabled prompt spec), and returns the typed
suggestion. **Persists nothing** — this is the suggest step of the
suggest / review / confirm / reject loop.
Args:
bucket_id: Active profile bucket id.
transaction_id: Stable id of the transaction to classify.
provider: Subprocess provider to resolve when ``classifier`` is None.
classifier: Injected classifier (dependency injection for tests). When
None, resolved via :func:`resolve_classifier` for ``provider``.
vision_classifier: Injected on-host vision classifier used when the
evidence is a scan-only PDF or image; default-resolved otherwise.
vision_model: Overrides the settings default local vision model (e.g.
``qwen2.5vl:7b``) for an image/scan read; ``None`` uses the default.
transaction_repository: Injected catalogue repository.
read_evidence: When True, resolve the transaction's linked evidence and read
it on-host — a text-layer PDF is inlined and sent to the cloud
classifier (consent-gated), a scan-only PDF or image is read by the
local vision model (no consent needed). Off by default.
evidence_acknowledged: Per-invocation acknowledgement that sending text-layer
evidence to a cloud model is accepted; required by the cloud-upload consent
gate for the text path (the on-host vision path needs no acknowledgement).
settings: Injected settings; defaults to ``load_settings()``.
Returns:
A :class:`~application.ledger._llm_suggestions.LLMClassificationSuggestion`.
Raises:
TransactionNotFoundError: When the transaction id is unknown.
LLMClassifierError: When the classifier fails (e.g. provider CLI
unavailable, hallucinated out-of-allow-list value).
"""
repository = _transaction_repository(bucket_id=bucket_id, repository=transaction_repository)
transaction = repository.load().get(transaction_id)
if transaction is None:
raise TransactionNotFoundError(
translated_message="application.ledger.errors.transaction_not_found",
context={"transaction_id": transaction_id},
)
resolved_settings = settings if settings is not None else load_settings()
resolved_classifier = (
classifier
if classifier is not None
else (_resolve_default_classifier(provider) if provider is not None else None)
)
evidence = (
_resolve_evidence(
transaction,
bucket_id=bucket_id,
settings=resolved_settings,
evidence_acknowledged=evidence_acknowledged,
)
if read_evidence
else None
)
response, provenance = _classify_with_evidence(
transaction,
evidence,
text_classifier=resolved_classifier,
spec=prompt_spec_with_every_spending_category(),
vision_classifier=vision_classifier,
vision_model=vision_model,
settings=resolved_settings,
)
_logger.info(
"llm suggest: transaction=%s provider=%s classification=%s confidence=%s",
transaction_id,
provider.value if provider is not None else "local-vision",
response.classification.value,
response.confidence,
)
return LLMClassificationSuggestion(
transaction_id=transaction_id,
provider=provider,
provenance=provenance,
classification=response.classification,
category=response.category,
confidence=response.confidence,
reason=response.reason,
evidence_id=evidence.reference if evidence is not None else None,
multiple_components=response.multiple_components,
)
[docs]
def apply_llm_classification(
suggestion: LLMClassificationSuggestion,
*,
bucket_id: str,
business_pct: Decimal | None = None,
actor: str = "operator",
source_command: str = "aeat app ledger classify --llm",
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None,
bucket_event_repository: BucketEventHistoryRepositoryProtocol | None = None,
occurred_at: datetime | None = None,
) -> ManualLedgerTransactionResult:
"""Persist an accepted LLM suggestion with ``llm:`` provenance.
Writes the decision through :func:`~domain.transactions.set_classification`,
stamping ``classified_by`` with the suggestion's ``provenance`` (the
classifier's ``decided_by``, e.g. ``llm:<model>``) and recording the
model's ``confidence`` and ``reason``. Persists the catalogue and emits a
:attr:`~domain.buckets.BucketEventType.LEDGER_TRANSACTION_CLASSIFIED`
event atomically.
The MVP persists only the non-regulated ``business_classification`` and
optional expense ``category``. It never sets a regulated tax value.
A ``MIXED`` suggestion requires an explicit ``business_pct`` (the LLM does
not produce one); apply refuses instructively when it is absent. The
expense ``category`` is recorded only for ``BUSINESS`` / ``MIXED``
classifications.
Args:
suggestion: The accepted
:class:`~application.ledger._llm_suggestions.LLMClassificationSuggestion`.
bucket_id: Active profile bucket id.
business_pct: Required when ``suggestion.classification`` is ``MIXED``.
actor: Operator identity for the audit event.
source_command: Source-command label for the audit event.
transaction_repository: Injected catalogue repository.
bucket_event_repository: Injected audit-event repository.
occurred_at: Override clock for deterministic tests.
Returns:
A :class:`~application.ledger._models.ManualLedgerTransactionResult`
reflecting the persisted decision.
Raises:
TransactionNotFoundError: When the transaction id is unknown.
TransactionValidationError: When the transaction is not ACTIVE or a
``MIXED`` suggestion is applied without a ``business_pct``.
"""
classification = suggestion.classification
if classification is BusinessClassification.MIXED and business_pct is None:
raise TransactionValidationError(
"applying a MIXED LLM suggestion requires --business-pct; "
"the LLM proposes the split direction but not the business-use percentage",
context={"transaction_id": suggestion.transaction_id},
)
if classification is not BusinessClassification.MIXED and business_pct is not None:
raise TransactionValidationError(
"--business-pct only applies to a MIXED classification",
context={"transaction_id": suggestion.transaction_id},
)
occurred = coerce_utc_aware(occurred_at or now())
repository = _transaction_repository(bucket_id=bucket_id, repository=transaction_repository)
_event_repo_arg = bucket_event_repository or BucketEventHistoryRepository()
assert isinstance(_event_repo_arg, BucketEventHistoryRepository), (
"apply_llm_classification requires a concrete BucketEventHistoryRepository "
"(to_secure_object_write is not on the protocol)"
)
event_repository = _event_repo_arg
catalogue = repository.load()
current = catalogue.get(suggestion.transaction_id)
if current is None:
raise TransactionNotFoundError(
translated_message="application.ledger.errors.transaction_not_found",
context={"transaction_id": suggestion.transaction_id},
)
if current.lifecycle_state is not TransactionLifecycleState.ACTIVE:
raise TransactionValidationError(
"only active ledger transactions can be classified; archived, stashed, and split-parent rows are immutable",
context={
"transaction_id": suggestion.transaction_id,
"lifecycle_state": current.lifecycle_state.value,
},
)
category_id: str | None = None
if classification in {BusinessClassification.BUSINESS, BusinessClassification.MIXED}:
category_id = suggestion.category.value if suggestion.category is not None else None
updated_catalogue = set_classification(
catalogue,
suggestion.transaction_id,
classification=classification,
business_pct=business_pct,
category_id=category_id,
classified_by=suggestion.provenance,
reason=suggestion.reason,
confidence=suggestion.confidence,
)
updated_transaction = updated_catalogue.get(suggestion.transaction_id)
assert updated_transaction is not None # set_classification preserves the id
event = _build_bucket_event(
bucket_id=bucket_id,
event_type=BucketEventType.LEDGER_TRANSACTION_CLASSIFIED,
occurred_at=occurred,
actor=actor,
object_type=BucketEventObjectType.LEDGER_TRANSACTION,
object_id=suggestion.transaction_id,
payload={
"source_command": source_command,
"classification": classification.value,
"category_id": category_id or "",
"classified_by": suggestion.provenance,
"provider": suggestion.provider.value if suggestion.provider is not None else "local-vision",
"confidence": format(suggestion.confidence, "f"),
"mutation_kind": "llm_classification",
},
)
_save_transaction_catalogue_and_events(
transaction_repository=repository,
event_repository=event_repository,
catalogue=updated_catalogue,
events=(event,),
)
_logger.info(
"llm apply: transaction=%s classified_by=%s classification=%s",
suggestion.transaction_id,
suggestion.provenance,
classification.value,
)
return _result(bucket_id, updated_transaction, (event.event_id,))
# ── stage-2 saturation: grounded rich tax metadata ────────────────
def _resolve_saturation_classifier(provider: LLMProvider) -> LLMClassifier:
"""Resolve the production classifier for ``provider`` with the saturation prompt.
Builds the classifier with
:func:`~domain.transactions.prompt_spec_with_saturation_fields` so the
model also selects an expense :class:`~domain.categories.SpendingCategory`
and an :class:`~domain.iva.IvaCategory` from the registry-grounded allow-list,
keeping the allow-list-guarded ``parse_response`` path intact.
"""
return resolve_classifier(provider.value, spec=prompt_spec_with_saturation_fields())
def _derive_iva_substrate(
iva_category: IvaCategory,
*,
gross: Decimal,
on_date: date,
) -> tuple[Decimal | None, Decimal | None, Decimal | None, bool, str]:
"""Derive ``(iva_rate, taxable_base, iva_amount, derivable, note)`` for a category.
Resolves the registry rate for ``iva_category`` via
:func:`~domain.iva.resolve_category_rate` and, when derivable, splits
the absolute ``gross`` at that rate with
:func:`~domain.iva.split_gross_at_rate`. The model never supplies these
numbers; they trace to the registry rate and a deterministic inverse split.
Returns the derived rate/base/amount (or ``None`` for each when the
category has no simple derivable Spanish domestic rate), the
``derivable`` flag, and an operator-facing ``note`` explaining a
non-derivable category.
"""
resolution = resolve_category_rate(iva_category, on_date=on_date)
if not resolution.derivable or resolution.rate is None:
return None, None, None, False, resolution.reason
taxable_base, iva_amount = split_gross_at_rate(abs(gross), resolution.rate)
return resolution.rate, taxable_base, iva_amount, True, ""
[docs]
def saturate_llm_classification(
*,
bucket_id: str,
transaction_id: str,
provider: LLMProvider | None,
classifier: LLMClassifier | None = None,
vision_classifier: LocalVisionLLMClassifier | None = None,
vision_model: str | None = None,
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None,
on_date: date | None = None,
read_evidence: bool = False,
evidence_acknowledged: bool = False,
settings: Settings | None = None,
) -> LLMSaturatedSuggestion:
"""Run the saturating LLM classifier for one transaction and return a suggestion.
Loads the transaction, runs the injected classifier (default-resolved from
``provider`` with the saturation prompt spec), then DERIVES the regulated
tax substrate from the model's selected :class:`~domain.iva.IvaCategory`
using the registry rate and a deterministic inverse split. **Persists
nothing** — this is the suggest step; rejecting a suggestion is simply not
applying it.
Args:
bucket_id: Active profile bucket id.
transaction_id: Stable id of the transaction to classify.
provider: Subprocess provider to resolve when ``classifier`` is None.
classifier: Injected classifier (dependency injection for tests). When
None, resolved via :func:`resolve_classifier` for ``provider`` with
the saturation prompt spec.
vision_classifier: Injected on-host vision classifier used when the
evidence is a scan-only PDF or image; default-resolved otherwise.
vision_model: Overrides the settings default local vision model (e.g.
``qwen2.5vl:7b``) for an image/scan read; ``None`` uses the default.
transaction_repository: Injected catalogue repository.
on_date: Effective date used to resolve the registry rate; defaults to
the transaction's value date (or booked date).
read_evidence: When True, resolve the transaction's linked evidence, extract
its text on-host, and inject it into the prompt. Off by default.
evidence_acknowledged: Per-invocation acknowledgement that sending the evidence
to a cloud model is accepted; required by the cloud-upload consent gate when
``read_evidence`` is set.
settings: Injected settings; defaults to ``load_settings()``.
Returns:
A :class:`~application.ledger._llm_suggestions.LLMSaturatedSuggestion`
carrying the model's selections and the system-derived euro substrate.
Raises:
TransactionNotFoundError: When the transaction id is unknown.
LLMClassifierError: When the classifier fails (provider CLI
unavailable, hallucinated out-of-allow-list value).
"""
repository = _transaction_repository(bucket_id=bucket_id, repository=transaction_repository)
transaction = repository.load().get(transaction_id)
if transaction is None:
raise TransactionNotFoundError(
translated_message="application.ledger.errors.transaction_not_found",
context={"transaction_id": transaction_id},
)
resolved_settings = settings if settings is not None else load_settings()
resolved_classifier = (
classifier
if classifier is not None
else (_resolve_saturation_classifier(provider) if provider is not None else None)
)
evidence = (
_resolve_evidence(
transaction,
bucket_id=bucket_id,
settings=resolved_settings,
evidence_acknowledged=evidence_acknowledged,
)
if read_evidence
else None
)
response, provenance = _classify_with_evidence(
transaction,
evidence,
text_classifier=resolved_classifier,
spec=prompt_spec_with_saturation_fields(),
vision_classifier=vision_classifier,
vision_model=vision_model,
settings=resolved_settings,
)
evidence_text = evidence.text if evidence is not None else None
evidence_reference = evidence.reference if evidence is not None else None
effective_date = on_date or transaction.raw.value_date or transaction.raw.booked_date
iva_rate: Decimal | None = None
taxable_base: Decimal | None = None
iva_amount: Decimal | None = None
rate_derivable = False
derivation_note = ""
if response.iva_category is not None:
iva_rate, taxable_base, iva_amount, rate_derivable, derivation_note = _derive_iva_substrate(
response.iva_category,
gross=transaction.raw.amount,
on_date=effective_date,
)
_logger.info(
"llm saturate: transaction=%s provider=%s classification=%s iva_category=%s derivable=%s",
transaction_id,
provider.value if provider is not None else "local-vision",
response.classification.value,
response.iva_category.value if response.iva_category is not None else "",
rate_derivable,
)
return LLMSaturatedSuggestion(
transaction_id=transaction_id,
provider=provider,
provenance=provenance,
classification=response.classification,
category=response.category,
confidence=response.confidence,
reason=response.reason,
iva_category=response.iva_category,
business_pct=response.business_pct,
iva_rate=iva_rate,
taxable_base=taxable_base,
iva_amount=iva_amount,
rate_derivable=rate_derivable,
derivation_note=derivation_note,
evidence_id=evidence_reference,
evidence_advisory=printed_iva_advisory(evidence_text, iva_amount) or "",
multiple_components=response.multiple_components,
)
[docs]
def apply_saturated_llm_classification(
suggestion: LLMSaturatedSuggestion,
*,
bucket_id: str,
business_pct: Decimal | None = None,
actor: str = "operator",
source_command: str = "aeat app ledger classify --llm --saturate --apply",
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None,
bucket_event_repository: BucketEventHistoryRepositoryProtocol | None = None,
occurred_at: datetime | None = None,
) -> ManualLedgerTransactionResult:
"""Persist an accepted saturated suggestion through the manual write path.
Composes the established single-writer manual-command write
(:func:`update_manual_transaction_fields`) rather than re-implementing it,
so the regulated fields land with their existing validators plus the
``gross == taxable_base + iva_amount`` invariant, and stamps
``classified_by`` with the suggestion's ``llm:<model>`` provenance via
``classified_by_override``.
The non-regulated business decision (classification, expense category) and
the model-selected ``iva_category`` are persisted; the regulated euro
figures are persisted only when the category was derivable (a
non-derivable category leaves the operator to complete the numbers). A
``MIXED`` suggestion requires a business percentage — the model's proposed
``business_pct`` is used unless the caller overrides it; apply refuses
instructively when neither is present.
Args:
suggestion: The accepted
:class:`~application.ledger._llm_suggestions.LLMSaturatedSuggestion`.
bucket_id: Active profile bucket id.
business_pct: Operator override for the MIXED business percentage;
falls back to the model's proposed ``business_pct``.
actor: Operator identity for the audit event.
source_command: Source-command label recording the operator's verb.
transaction_repository: Injected catalogue repository.
bucket_event_repository: Injected audit-event repository.
occurred_at: Override clock for deterministic tests.
Returns:
A :class:`~application.ledger._models.ManualLedgerTransactionResult`
reflecting the persisted state.
Raises:
TransactionValidationError: When a ``MIXED`` suggestion is applied with
no business percentage available.
"""
classification = suggestion.classification
effective_business_pct = business_pct if business_pct is not None else suggestion.business_pct
if classification is BusinessClassification.MIXED and effective_business_pct is None:
raise TransactionValidationError(
"applying a MIXED saturated suggestion requires a business percentage; "
"pass --business-pct (the model proposes the split direction but the percentage is operator-owned)",
context={"transaction_id": suggestion.transaction_id},
)
patch_fields: dict[str, object] = {"business_classification": classification}
if classification is BusinessClassification.MIXED:
patch_fields["business_pct"] = effective_business_pct
category_carrying = classification in {BusinessClassification.BUSINESS, BusinessClassification.MIXED}
if category_carrying and suggestion.category is not None:
patch_fields["category_id"] = suggestion.category.value
if suggestion.iva_category is not None:
patch_fields["iva_category"] = suggestion.iva_category
if suggestion.rate_derivable:
patch_fields["taxable_base"] = suggestion.taxable_base
patch_fields["iva_rate"] = suggestion.iva_rate
patch_fields["iva_amount"] = suggestion.iva_amount
patch = ManualLedgerTransactionPatch.model_validate(patch_fields)
# Compose the single-writer manual write rather than re-implementing the
# regulated-field persistence (composition-service-no-parallel-write-path).
# The operator's verb is recorded via ``source_command`` on the manual
# write's own classification event, and model provenance via
# ``classified_by_override``; we deliberately do not emit a second,
# parallel LLM-specific event here.
result = update_manual_transaction_fields(
bucket_id=bucket_id,
transaction_id=suggestion.transaction_id,
patch=patch,
actor=actor,
source_command=source_command,
classified_by_override=suggestion.provenance,
transaction_repository=transaction_repository,
bucket_event_repository=bucket_event_repository,
occurred_at=occurred_at,
)
_logger.info(
"llm saturate apply: transaction=%s classified_by=%s iva_category=%s derived=%s",
suggestion.transaction_id,
suggestion.provenance,
suggestion.iva_category.value if suggestion.iva_category is not None else "",
suggestion.rate_derivable,
)
return result
# ── operator-initiated derivation (no LLM) ────────────────────────
[docs]
def derive_operator_iva_substrate(
*,
bucket_id: str,
transaction_id: str,
iva_category: IvaCategory,
on_date: date | None = None,
actor: str = "operator",
source_command: str = "aeat app ledger classify --iva-category --saturate",
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None,
bucket_event_repository: BucketEventHistoryRepositoryProtocol | None = None,
occurred_at: datetime | None = None,
) -> OperatorIvaDerivationResult:
"""Derive and persist the IVA substrate for an OPERATOR-chosen category.
The same grounded derivation the saturating LLM path uses
(:func:`~domain.iva.resolve_category_rate` +
:func:`~domain.iva.split_gross_at_rate`), but initiated by the operator
rather than the model — the fallback for when the model declines (returns
``unknown``) or the operator simply knows the category. Given a transaction
already classified BUSINESS or MIXED and the selected
:class:`~domain.iva.IvaCategory`, it resolves the registry rate, splits
the gross into taxable base and IVA amount, and persists them through the
manual write with ``derived:`` provenance. Only the IVA substrate is
touched; the business classification stays as-is. A non-derivable category
persists nothing and returns an explanatory note.
Returns:
The
:class:`~application.ledger._llm_suggestions.OperatorIvaDerivationResult`
recording the persisted IVA substrate, or an explanatory note when the
category is non-derivable.
Raises:
TransactionNotFoundError: When the transaction id is unknown.
TransactionValidationError: When the transaction is not classified
BUSINESS or MIXED (IVA applies only to business activity).
"""
repository = _transaction_repository(bucket_id=bucket_id, repository=transaction_repository)
transaction = repository.load().get(transaction_id)
if transaction is None:
raise TransactionNotFoundError(
translated_message="application.ledger.errors.transaction_not_found",
context={"transaction_id": transaction_id},
)
if transaction.business_classification not in {
BusinessClassification.BUSINESS,
BusinessClassification.MIXED,
}:
raise TransactionValidationError(
"IVA derivation applies only to a business transaction; classify it as "
"BUSINESS or MIXED first, then derive the IVA substrate",
context={"transaction_id": transaction_id},
)
effective_date = on_date or transaction.raw.value_date or transaction.raw.booked_date
iva_rate, taxable_base, iva_amount, derivable, note = _derive_iva_substrate(
iva_category,
gross=transaction.raw.amount,
on_date=effective_date,
)
if not derivable:
return OperatorIvaDerivationResult(
transaction_id=transaction_id,
iva_category=iva_category,
derivable=False,
note=note,
)
patch = ManualLedgerTransactionPatch.model_validate(
{
"iva_category": iva_category,
"iva_rate": iva_rate,
"taxable_base": taxable_base,
"iva_amount": iva_amount,
}
)
result = update_manual_transaction_fields(
bucket_id=bucket_id,
transaction_id=transaction_id,
patch=patch,
actor=actor,
source_command=source_command,
classified_by_override="derived:iva-category",
transaction_repository=transaction_repository,
bucket_event_repository=bucket_event_repository,
occurred_at=occurred_at,
)
_logger.info(
"operator iva derive: transaction=%s iva_category=%s rate=%s base=%s amount=%s",
transaction_id,
iva_category.value,
iva_rate,
taxable_base,
iva_amount,
)
return OperatorIvaDerivationResult(
transaction_id=transaction_id,
iva_category=iva_category,
derivable=True,
iva_rate=iva_rate,
taxable_base=taxable_base,
iva_amount=iva_amount,
result=result,
)
# ── stage-3b: evidence-driven N-way split ─────────────────────────
def _resolve_default_split_proposer(provider: LLMProvider) -> LLMSplitProposer:
"""Resolve the production split proposer for ``provider`` with the saturation prompt.
Uses :func:`~domain.transactions.prompt_spec_with_saturation_fields` so each
proposed child carries the same allow-list-guarded expense-category and
IVA-category selections the saturate path uses.
"""
return resolve_split_proposer(provider.value, spec=prompt_spec_with_saturation_fields())
def _split_child_description(child_index: int, *, citation: str, category: SpendingCategory | None) -> str:
"""Build a distinct, operator-facing description for one split child.
The 1-based ordinal prefix guarantees distinct child descriptions (hence
distinct split-child ids) even when two children share an amount and a
category; the label prefers the model's evidence citation, then the expense
category, then a neutral fallback.
"""
label = citation.strip() or (category.value if category is not None else "línea")
return f"{child_index + 1}. {label}"
[docs]
def suggest_evidence_split(
*,
bucket_id: str,
transaction_id: str,
provider: LLMProvider | None,
proposer: LLMSplitProposer | None = None,
vision_classifier: LocalVisionLLMClassifier | None = None,
vision_model: str | None = None,
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None,
on_date: date | None = None,
read_evidence: bool = True,
evidence_acknowledged: bool = False,
settings: Settings | None = None,
) -> LLMSplitSuggestion:
"""Propose an evidence-driven N-way split for one transaction.
Loads the transaction, runs the injected proposer (default-resolved from
``provider`` with the saturation prompt spec) over the optional on-host
evidence text, DERIVES each child's euro amount from the parent gross and the
model's proportion (summing exactly to the parent), and DERIVES each child's
regulated tax substrate from the registry rate for the model-selected IVA
category. **Persists nothing** — this is the suggest step.
Args:
bucket_id: Active profile bucket id.
transaction_id: Stable id of the transaction to split.
provider: Subprocess provider to resolve when ``proposer`` is None.
proposer: Injected split proposer (dependency injection for tests). When
None, resolved via :func:`resolve_split_proposer` for ``provider``.
vision_classifier: Injected on-host vision classifier used when the
evidence is a scan-only PDF or image; default-resolved otherwise.
vision_model: Overrides the settings default local vision model (e.g.
``qwen2.5vl:7b``) for an image/scan read; ``None`` uses the default.
transaction_repository: Injected catalogue repository.
on_date: Effective date used to resolve each child's registry rate;
defaults to the transaction's value date (or booked date).
read_evidence: When True (default for splitting), resolve the
transaction's linked evidence, extract its text on-host, and inject it
into the prompt.
evidence_acknowledged: Per-invocation acknowledgement that sending the
evidence to a cloud model is accepted; required by the cloud-upload
consent gate when ``read_evidence`` is set and evidence is linked.
settings: Injected settings; defaults to ``load_settings()``.
Returns:
A :class:`~application.ledger._llm_suggestions.LLMSplitSuggestion`
whose child amounts sum exactly to the parent.
Raises:
TransactionNotFoundError: When the transaction id is unknown.
LLMClassifierError: When the proposer fails (provider CLI unavailable,
hallucinated out-of-allow-list value, or a malformed split response).
"""
repository = _transaction_repository(bucket_id=bucket_id, repository=transaction_repository)
transaction = repository.load().get(transaction_id)
if transaction is None:
raise TransactionNotFoundError(
translated_message="application.ledger.errors.transaction_not_found",
context={"transaction_id": transaction_id},
)
resolved_settings = settings if settings is not None else load_settings()
resolved_proposer = (
proposer
if proposer is not None
else (_resolve_default_split_proposer(provider) if provider is not None else None)
)
evidence = (
_resolve_evidence(
transaction,
bucket_id=bucket_id,
settings=resolved_settings,
evidence_acknowledged=evidence_acknowledged,
)
if read_evidence
else None
)
response, provenance = _split_with_evidence(
transaction,
evidence,
proposer=resolved_proposer,
spec=prompt_spec_with_saturation_fields(),
vision_classifier=vision_classifier,
vision_model=vision_model,
settings=resolved_settings,
)
evidence_reference = evidence.reference if evidence is not None else None
proportions = tuple(child.proportion for child in response.children)
amounts = derive_child_amounts(transaction.raw.amount, proportions)
effective_date = on_date or transaction.raw.value_date or transaction.raw.booked_date
children: list[LLMSplitChildSuggestion] = []
for index, (child, amount) in enumerate(zip(response.children, amounts, strict=True)):
iva_rate: Decimal | None = None
taxable_base: Decimal | None = None
iva_amount: Decimal | None = None
rate_derivable = False
derivation_note = ""
if child.iva_category is not None:
iva_rate, taxable_base, iva_amount, rate_derivable, derivation_note = _derive_iva_substrate(
child.iva_category,
gross=amount,
on_date=effective_date,
)
children.append(
LLMSplitChildSuggestion(
proportion=child.proportion,
amount=amount,
description=_split_child_description(index, citation=child.evidence_citation, category=child.category),
category=child.category,
iva_category=child.iva_category,
iva_rate=iva_rate,
taxable_base=taxable_base,
iva_amount=iva_amount,
rate_derivable=rate_derivable,
derivation_note=derivation_note,
evidence_citation=child.evidence_citation,
),
)
_logger.info(
"llm split suggest: transaction=%s provider=%s children=%d",
transaction_id,
provider.value if provider is not None else "local-vision",
len(children),
)
return LLMSplitSuggestion(
transaction_id=transaction_id,
provider=provider,
provenance=provenance,
reason=response.reason,
parent_amount=transaction.raw.amount,
children=tuple(children),
evidence_id=evidence_reference,
)
[docs]
def apply_evidence_split(
suggestion: LLMSplitSuggestion,
*,
bucket_id: str,
actor: str = "operator",
source_command: str = "aeat app ledger split --llm --apply",
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None,
bucket_event_repository: BucketEventHistoryRepositoryProtocol | None = None,
occurred_at: datetime | None = None,
) -> LLMSplitApplyResult:
"""Apply a reviewed evidence-driven split through the single-writer split path.
Composes the established single writers rather than re-implementing them
(``composition-service-no-parallel-write-path``): first
:func:`split_transaction` redistributes the parent into children whose
magnitudes sum exactly to the parent, then for each child
:func:`update_manual_transaction_fields` stamps the model-selected expense
category and IVA category, the registry-DERIVED regulated numbers, the parent
invoice's evidence link, and the ``llm:<model>`` provenance.
The split path enforces children-sum-to-parent and the non-negative-magnitude
invariant; the per-child write enforces the
``gross == taxable_base + iva_amount`` invariant. The LLM never supplies a
persisted euro amount or regulated number.
Args:
suggestion: The accepted
:class:`~application.ledger._llm_suggestions.LLMSplitSuggestion`.
bucket_id: Active profile bucket id.
actor: Operator identity for the audit events.
source_command: Source-command label recording the operator's verb.
transaction_repository: Injected catalogue repository.
bucket_event_repository: Injected audit-event repository.
occurred_at: Override clock for deterministic tests.
Returns:
An :class:`~application.ledger._llm_suggestions.LLMSplitApplyResult`
naming the split group and its children.
Raises:
TransactionNotFoundError: When the parent transaction id is unknown.
TransactionValidationError: When the split invariants are violated.
"""
if not suggestion.recommends_split:
# A single-child suggestion is the no-split verdict; applying it as a
# one-way split is degenerate. The CLI routes this to single-transaction
# classification instead — never here.
raise TransactionValidationError(
"this proposal is a no-split verdict (one line); classify the transaction instead of splitting it",
context={"transaction_id": suggestion.transaction_id, "child_count": len(suggestion.children)},
)
repository = _transaction_repository(bucket_id=bucket_id, repository=transaction_repository)
parent = repository.load().get(suggestion.transaction_id)
if parent is None:
raise TransactionNotFoundError(
translated_message="application.ledger.errors.transaction_not_found",
context={"transaction_id": suggestion.transaction_id},
)
commands = tuple(
SplitChildCommand(amount=child.amount, description=child.description) for child in suggestion.children
)
split_result = split_transaction(
bucket_id=bucket_id,
transaction_id=suggestion.transaction_id,
children=commands,
actor=actor,
source_command=source_command,
reason=suggestion.reason,
transaction_repository=repository,
bucket_event_repository=bucket_event_repository,
occurred_at=occurred_at,
)
evidence_link: dict[str, object] = {}
if parent.purchase_invoice_evidence_id is not None:
evidence_link["purchase_invoice_evidence_id"] = parent.purchase_invoice_evidence_id
elif parent.attachment_ids:
evidence_link["attachment_ids"] = parent.attachment_ids
classified = 0
for child_txn, child in zip(split_result.child_transactions, suggestion.children, strict=True):
patch_fields: dict[str, object] = {
"business_classification": BusinessClassification.BUSINESS,
**evidence_link,
}
if child.category is not None:
patch_fields["category_id"] = child.category.value
if child.iva_category is not None:
patch_fields["iva_category"] = child.iva_category
if child.rate_derivable:
patch_fields["taxable_base"] = child.taxable_base
patch_fields["iva_rate"] = child.iva_rate
patch_fields["iva_amount"] = child.iva_amount
patch = ManualLedgerTransactionPatch.model_validate(patch_fields)
update_manual_transaction_fields(
bucket_id=bucket_id,
transaction_id=child_txn.transaction_id,
patch=patch,
actor=actor,
source_command=source_command,
classified_by_override=suggestion.provenance,
transaction_repository=repository,
bucket_event_repository=bucket_event_repository,
occurred_at=occurred_at,
)
classified += 1
_logger.info(
"llm split apply: parent=%s split_group=%s children=%d classified=%d classified_by=%s",
split_result.parent_transaction_id,
split_result.split_group_id,
len(split_result.child_transaction_ids),
classified,
suggestion.provenance,
)
return LLMSplitApplyResult(
bucket_id=bucket_id,
parent_transaction_id=split_result.parent_transaction_id,
split_group_id=split_result.split_group_id,
child_transaction_ids=split_result.child_transaction_ids,
provenance=suggestion.provenance,
classified_child_count=classified,
)
[docs]
def apply_evidence_classification(
suggestion: LLMSplitSuggestion,
*,
bucket_id: str,
actor: str = "operator",
source_command: str = "aeat app ledger classify --read-evidence --auto-split --apply",
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None,
bucket_event_repository: BucketEventHistoryRepositoryProtocol | None = None,
occurred_at: datetime | None = None,
) -> ManualLedgerTransactionResult:
"""Apply a no-split (single-child) evidence suggestion in place on the parent.
The auto-split router uses one model call — the split proposer — to decide
whether to split. When the proposer returns a single child (the "no split
warranted" verdict), that child already carries the model-selected expense and
IVA categories and the registry-DERIVED ``taxable_base`` / ``iva_rate`` /
``iva_amount`` for the whole gross. This stamps them on the parent through the
single-writer :func:`update_manual_transaction_fields`, with the parent invoice's
evidence link and the ``llm:<model>`` provenance — exactly the per-child write
:func:`apply_evidence_split` performs, but without splitting. The model emits no
euro amount or regulated number (``llm-selects-system-derives-tax-numbers``).
Args:
suggestion: A no-split
:class:`~application.ledger._llm_suggestions.LLMSplitSuggestion`
(exactly one child).
bucket_id: Active profile bucket id.
actor: Operator identity for the audit event.
source_command: Source-command label recording the operator's verb.
transaction_repository: Injected catalogue repository.
bucket_event_repository: Injected audit-event repository.
occurred_at: Override clock for deterministic tests.
Returns:
The :class:`~application.ledger._models.ManualLedgerTransactionResult`
for the in-place classification.
Raises:
TransactionValidationError: When the suggestion recommends a split (use
:func:`apply_evidence_split`).
TransactionNotFoundError: When the transaction id is unknown.
"""
if suggestion.recommends_split:
raise TransactionValidationError(
"this proposal recommends a split; apply it with apply_evidence_split, not in place",
context={"transaction_id": suggestion.transaction_id, "child_count": len(suggestion.children)},
)
repository = _transaction_repository(bucket_id=bucket_id, repository=transaction_repository)
parent = repository.load().get(suggestion.transaction_id)
if parent is None:
raise TransactionNotFoundError(
translated_message="application.ledger.errors.transaction_not_found",
context={"transaction_id": suggestion.transaction_id},
)
child = suggestion.children[0]
patch_fields: dict[str, object] = {"business_classification": BusinessClassification.BUSINESS}
if parent.purchase_invoice_evidence_id is not None:
patch_fields["purchase_invoice_evidence_id"] = parent.purchase_invoice_evidence_id
elif parent.attachment_ids:
patch_fields["attachment_ids"] = parent.attachment_ids
if child.category is not None:
patch_fields["category_id"] = child.category.value
if child.iva_category is not None:
patch_fields["iva_category"] = child.iva_category
if child.rate_derivable:
patch_fields["taxable_base"] = child.taxable_base
patch_fields["iva_rate"] = child.iva_rate
patch_fields["iva_amount"] = child.iva_amount
patch = ManualLedgerTransactionPatch.model_validate(patch_fields)
result = update_manual_transaction_fields(
bucket_id=bucket_id,
transaction_id=parent.transaction_id,
patch=patch,
actor=actor,
source_command=source_command,
classified_by_override=suggestion.provenance,
transaction_repository=repository,
bucket_event_repository=bucket_event_repository,
occurred_at=occurred_at,
)
_logger.info(
"llm auto-classify (no split): transaction=%s classified_by=%s category=%s iva_category=%s",
parent.transaction_id,
suggestion.provenance,
child.category.value if child.category is not None else "",
child.iva_category.value if child.iva_category is not None else "",
)
return result
# ── reject: the fourth decision terminal (audit-trailed) ──────────
[docs]
def reject_llm_suggestion(
suggestion: LLMClassificationSuggestion | LLMSaturatedSuggestion | LLMSplitSuggestion,
*,
bucket_id: str,
reason: str = "",
actor: str = "operator",
source_command: str = "aeat app ledger classify --llm --reject",
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None,
bucket_event_repository: BucketEventHistoryRepositoryProtocol | None = None,
occurred_at: datetime | None = None,
) -> LLMSuggestionRejectionResult:
"""Record an explicit, audit-trailed rejection of an LLM suggestion.
This is the fourth decision terminal of the suggest -> review -> decide loop
(after approve = apply and update = manual override). It captures *what* the
model proposed and the operator's reason in a
``LEDGER_TRANSACTION_LLM_SUGGESTION_REJECTED`` bucket event, and **mutates
nothing** — the transaction's classification, numbers, and lifecycle are
untouched, so its review status stays ``pending`` (it is still unclassified).
No regulated number is written; the model emitted none and reject writes none.
Args:
suggestion: The captured proposal being rejected — a stage-1
classification, a saturated suggestion, or an evidence-driven split.
bucket_id: Active profile bucket id.
reason: The operator's free-text reason for rejecting (optional).
actor: Operator identity for the audit event.
source_command: Source-command label recording the operator's verb.
transaction_repository: Injected catalogue repository.
bucket_event_repository: Injected audit-event repository.
occurred_at: Override clock for deterministic tests.
Returns:
An
:class:`~application.ledger._llm_suggestions.LLMSuggestionRejectionResult`
naming the recorded event.
Raises:
TransactionNotFoundError: When the transaction id is unknown.
TransactionValidationError: When the transaction is not active.
"""
repository = _transaction_repository(bucket_id=bucket_id, repository=transaction_repository)
catalogue = repository.load()
transaction = catalogue.get(suggestion.transaction_id)
if transaction is None:
raise TransactionNotFoundError(
translated_message="application.ledger.errors.transaction_not_found",
context={"transaction_id": suggestion.transaction_id},
)
if transaction.lifecycle_state is not TransactionLifecycleState.ACTIVE:
raise TransactionValidationError(
"only active ledger transactions can carry an LLM rejection record",
context={
"transaction_id": suggestion.transaction_id,
"lifecycle_state": transaction.lifecycle_state.value,
},
)
occurred = coerce_utc_aware(occurred_at or now())
if isinstance(suggestion, LLMSplitSuggestion):
suggestion_kind = "split"
payload: dict[str, str] = {
"suggestion_kind": suggestion_kind,
"child_count": str(len(suggestion.children)),
"model_reason": suggestion.reason,
}
else:
suggestion_kind = "classification"
payload = {
"suggestion_kind": suggestion_kind,
"classification": suggestion.classification.value,
"category": suggestion.category.value if suggestion.category is not None else "",
"confidence": format(suggestion.confidence, "f"),
"model_reason": suggestion.reason,
}
if isinstance(suggestion, LLMSaturatedSuggestion) and suggestion.iva_category is not None:
payload["iva_category"] = suggestion.iva_category.value
payload["provider"] = suggestion.provider.value if suggestion.provider is not None else "local-vision"
payload["provenance"] = suggestion.provenance
payload["operator_reason"] = reason
payload["source_command"] = source_command
payload["mutation_kind"] = "llm_suggestion_rejected"
event = _build_bucket_event(
bucket_id=bucket_id,
event_type=BucketEventType.LEDGER_TRANSACTION_LLM_SUGGESTION_REJECTED,
occurred_at=occurred,
actor=actor,
object_type=BucketEventObjectType.LEDGER_TRANSACTION,
object_id=suggestion.transaction_id,
payload=payload,
)
# Persist the event through the transaction repository's secure-write batch
# (the unchanged catalogue rides along as a no-op), exactly as the apply path
# does — a bare BucketEventHistoryRepository().save() does not bind to the
# active bucket store in the CLI flow.
_event_repo_arg = bucket_event_repository or BucketEventHistoryRepository()
assert isinstance(_event_repo_arg, BucketEventHistoryRepository), (
"reject_llm_suggestion requires a concrete BucketEventHistoryRepository "
"(to_secure_object_write is not on the protocol)"
)
_save_transaction_catalogue_and_events(
transaction_repository=repository,
event_repository=_event_repo_arg,
catalogue=catalogue,
events=(event,),
)
_logger.info(
"llm reject: transaction=%s kind=%s provenance=%s",
suggestion.transaction_id,
suggestion_kind,
suggestion.provenance,
)
return LLMSuggestionRejectionResult(
bucket_id=bucket_id,
transaction_id=suggestion.transaction_id,
bucket_event_id=event.event_id,
suggestion_kind=suggestion_kind,
provenance=suggestion.provenance,
operator_reason=reason,
)
__all__ = [
"LLMClassificationSuggestion",
"LLMProvider",
"LLMProviderAvailability",
"LLMSaturatedSuggestion",
"LLMSplitApplyResult",
"LLMSplitChildSuggestion",
"LLMSplitSuggestion",
"LLMSuggestionRejectionResult",
"OperatorIvaDerivationResult",
"apply_evidence_classification",
"apply_evidence_split",
"apply_llm_classification",
"apply_saturated_llm_classification",
"available_llm_providers",
"derive_operator_iva_substrate",
"is_llm_provider_available",
"reject_llm_suggestion",
"saturate_llm_classification",
"suggest_evidence_split",
"suggest_llm_classification",
]