dodil/erp-brick-ar

DODIL ERP brick: accounts receivable

src/brick_ar/infrastructure/ai_steps.py

"""AI steps (erp-roadmap #53, the AI-step contract): an AI step is a PROVIDER of a capability, with the
ManualTask as its fallback and its reviewer. The mode is part of the signed BPMN: off | suggest (the
default: AI drafts, a person approves) | auto-with-review | auto (only a step marked safe_for_auto; never
destructive or money-moving without approval).

TODO(ai): erp-platform has no step-mode API yet. Until it lands, `suggest` is this brick's job: the use
case stores the provider's draft as a ManualTask for a person to approve, and never acts on it alone.
Generated by `erp-compose scaffold` from module.yaml.
"""
from erp_platform.capabilities import CapabilitySeam

# What every AI action leaves behind (the AI-step contract): the provider's action log holds the model,
# the inputs, the output and the confidence; the brick records the approver — who accepted the ManualTask,
# or reviewed an auto-with-review result.
AUDIT = ("model", "inputs", "output", "confidence", "approver")

AI_STEPS = {
    "ai_check_draft": {
        "capability": "ai.suggest",
        "mode": "suggest",
        "modes_allowed": ["off", "suggest"],
        "safe_for_auto": False,
        "fallback": "the biller reviews without findings",
        "benefit": "the invoice that is double last month's, misses the customer's PO number or carries an odd tax code is caught before the customer sees it",
        "inputs": ["the draft", "the customer's last invoices", "the source order or schedule", "the customer's order policy"],
        "eval": "drafts with seeded anomalies and clean drafts; recall of seeded anomalies, false findings per 100 drafts",
    },
    "billing_inquiry_reply": {
        "capability": "ai.draft_message",
        "mode": "suggest",
        "modes_allowed": ["off", "suggest"],
        "safe_for_auto": False,
        "fallback": "the biller answers",
        "benefit": "a customer asking 'what is this charge?' gets an answer drafted from the invoice, its order and its schedule (APQC 10798)",
        "inputs": ["the customer's message", "the invoice", "its source documents", "the payments on it"],
        "eval": "past inquiries with the replies sent; graded for factual accuracy against the documents",
    },
}

SEAMS = {
    "ai_check_draft": CapabilitySeam("ai.suggest"),
    "billing_inquiry_reply": CapabilitySeam("ai.draft_message"),
}


def propose(step: str, idempotency_key: str, input: dict, mode: str):
    """Ask the step's provider for a proposal. `off` → None (the process is the fallback). Otherwise the
    provider's Result, or — no provider, or none wired — the ManualTask with the exact input (rule 8). What
    happens next is the caller's, by the mode: suggest → a ManualTask a person accepts, edits or rejects;
    auto-with-review → apply it and open a review task; auto → apply it (safe_for_auto steps only).
    The provider never applies anything."""
    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                     # no capability yet: the fallback, a person (the note above)
    return SEAMS[step].call(idempotency_key, input)


# ── ai_check_draft: the fixed BASELINE this brick computes itself, with no model ──────────────────────────────────────
# It is always available, so the step is useful with no provider wired. Every finding is a SUGGESTION written `open`; a
# person accepts or dismisses it (AR-32); it never edits the draft, and issue never waits on it.
from decimal import Decimal  # noqa: E402
from statistics import median  # noqa: E402

BASELINE_MODEL = "baseline-rules/1"


def baseline_findings(total: Decimal, po_ref: str, recent: list, today) -> list:
    """[(finding, evidence)] for a draft, given the customer's recent issued invoices as (total, issue_date, po_ref)."""
    out = []
    totals = [t for t, _d, _p in recent[:6] if t > 0]
    if len(totals) >= 3 and total > 0:
        m = Decimal(str(median(totals)))
        if m > 0 and total >= 2 * m:
            out.append(("an unusual amount", f"total {total} is {total / m:.1f}x the median {m} of the customer's last {len(totals)} invoices"))
    if recent and any(p for _t, _d, p in recent) and not po_ref:
        out.append(("a missing PO number", "earlier invoices to this customer carried a PO reference; this draft has none"))
    for t, d, _p in recent:
        if d and total > 0 and t == total and 0 <= (today - d).days <= 45:
            out.append(("a duplicate of a recent invoice", f"an invoice of exactly {total} was issued {(today - d).days} days ago"))
            break
    return out