dodil/erp-brick-tax

DODIL ERP brick: tax

src/brick_tax/infrastructure/ai_steps.py

"""AI steps (erp-roadmap AI-step contract): an AI step is a PROVIDER of a capability (`ai.suggest`), with the person as its
fallback. Mode `off | suggest` (the default: the AI proposes, a person decides). `ai_suggest_code` is suggest-only
(`safe_for_auto: false`): tax decides what the customer pays and what is filed. `check_vat_number` is deferred to M2.

What the step does: mode `off` -> nothing runs. Otherwise `propose()` asks the provider the composition wired. None wired
is a first-class path, not an error: the answer is `manual` (the line stays refused until a person chooses a code). The
provider never applies anything; its suggestion is stored beside the line and a person accepts or rejects it.
Tests run with no provider (or a fake one), so they need no model call.
"""
from dataclasses import dataclass, field
from typing import Optional

from erp_platform.capabilities import CapabilitySeam, ManualTask

AI_STEPS = {
    "ai_suggest_code": {
        "capability": "ai.suggest", "mode": "suggest", "modes_allowed": ["off", "suggest"], "safe_for_auto": False,
        "fallback": "the line is refused until a person chooses a code",
        "benefit": "a free-text line or a new kind of supply gets a reasoned code instead of a guess; the reason is kept for the accountant",
        "inputs": ["the line description", "the customer's treatment", "the codes and their descriptions", "the rules"],
        "eval": "lines with an accountant's code; accuracy, and zero tolerance for suggesting a lower rate than the label",
    },
    "check_vat_number": {
        "capability": "ai.suggest", "mode": "suggest", "modes_allowed": ["off", "suggest"], "safe_for_auto": False,
        "fallback": "a person confirms the number",
        "benefit": "a reverse charge on a customer who is not registered is a VAT debt for the supplier; the HMRC check is a service "
                   "call, the AI part compares HMRC's name and address with the partner's",
        "inputs": ["the VAT number", "the HMRC check result", "the customer's name and address"],
        "eval": "numbers with known status and name matches; mismatch-flag precision",
    },
    "ai_review_vat_return": {
        "capability": "ai.suggest", "mode": "suggest", "modes_allowed": ["off", "suggest"], "safe_for_auto": False,
        "fallback": "the tax admin reviews without findings",
        "benefit": "a box that moved far from the last four periods, a document with an unusual code or a large one-off is looked at "
                   "before HMRC sees it",
        "inputs": ["the draft boxes", "the last four returns", "the entries above a size", "their codes"],
        "eval": "past returns with seeded errors (a doubled bill, a wrong code); recall, false findings per return",
    },
}
SEAMS = {"ai_suggest_code": CapabilitySeam("ai.suggest"), "check_vat_number": CapabilitySeam("ai.suggest"),
         "ai_review_vat_return": CapabilitySeam("ai.suggest")}


@dataclass
class Answer:
    """`kind` is off | done (a provider answered) | manual (a person does it) | pending (try again)."""
    step: str
    mode: str
    kind: str
    suggestion: dict = field(default_factory=dict)
    rationale: str = ""
    confidence: Optional[int] = None
    model: str = ""
    message: str = ""
    task: Optional[ManualTask] = None

    def as_dict(self) -> dict:
        return {"step": self.step, "mode": self.mode, "kind": self.kind, "suggestion": self.suggestion,
                "rationale": self.rationale, "confidence": self.confidence, "model": self.model, "message": self.message}


def propose(step: str, idempotency_key: str, input: dict, mode: str):
    spec = AI_STEPS[step]
    if mode == "off":
        return None
    if mode not in spec["modes_allowed"]:
        raise ValueError(f"{step}: mode {mode!r} is not allowed ({', '.join(spec['modes_allowed'])})")
    if mode == "auto" and not spec["safe_for_auto"]:
        raise ValueError(f"{step} is not safe for auto: a person approves it")
    if step not in SEAMS:
        return None
    return SEAMS[step].call(idempotency_key, input)


def ask(step: str, mode: str, subject_ref: str, version: int, context: dict, fields: tuple = ("tax_code",)) -> Answer:
    """One run of an AI step for one version of one record. `context` holds JSON SCALARS only."""
    key = f"{step}|{subject_ref}|{version}"
    got = propose(step, key, {"step": step, "subject_ref": subject_ref, "fields": list(fields), "context": context,
                              "instructions": None}, mode)
    if got is None:
        return Answer(step, mode, "off", message="this step is switched off in the settings")
    if isinstance(got, ManualTask):
        return Answer(step, mode, "manual", message=got.reason, task=got)
    if got.status == "done":
        out = got.output or {}
        return Answer(step, mode, "done", out.get("suggestion") or {}, out.get("rationale", ""), out.get("confidence"),
                      out.get("model", ""), got.message)
    return Answer(step, mode, "pending", message=got.message or got.status)