dodil/erp-brick-inventory

DODIL ERP brick: inventory

src/brick_inventory/infrastructure/ai_steps.py

"""AI steps (erp-roadmap #53, the AI-step contract): an AI step is a PROVIDER of a capability (`ai.suggest`), with the
ManualTask as its fallback and its reviewer. The mode is part of the signed BPMN: off | suggest (AI drafts, a person decides).
The one step here, ai_explain_variance, is suggest-only (`safe_for_auto: false`): an adjustment writes off value, so a person
decides. A suggestion lands in inv_suggestion and changes no quantity; nothing is adjusted or approved by it.

The mode is the owner's setting (`ai_explain_variance`, off by default; turning it on is recorded as the owner's consent).
With no provider wired the answer is a ManualTask and the approver reads the moves list unaided. Tests run with no provider
unless they wire 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

AUDIT = ("model", "inputs", "output", "confidence", "approver")

AI_STEPS = {
    "ai_explain_variance": {
        "capability": "ai.suggest",
        "mode": "suggest",
        "modes_allowed": ["off", "suggest"],
        "safe_for_auto": False,
        "fallback": "no suggestion; the approver sees the moves list",
        "benefit": "an approver checks the likely cause first (a mis-put transfer, a unit confusion) instead of writing off stock that is on the next shelf",
        "inputs": ["the count line", "the moves on that product in that warehouse since the last count", "pack conversions of the product"],
        "eval": "counts with seeded causes (a transfer to the wrong bin, a pack counted as a unit); share of lines whose suggestion names the seeded cause",
    },
}

SEAMS = {"ai_explain_variance": CapabilitySeam("ai.suggest")}


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 connected) the ManualTask with the exact input. 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
    return SEAMS[step].call(idempotency_key, input)


@dataclass
class Answer:
    """What a step produced: `kind` is off | done (a provider answered) | manual (a person does it) | pending
    (the provider could not answer yet; 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 ask(step: str, mode: str, subject_ref: str, version: int, fields: list, context: dict) -> Answer:
    """One run of an AI step for one version of one record. `context` holds JSON SCALARS only (the capability's
    `map of any`): the caller resolves ids to values, the provider reads nothing."""
    key = f"{step}|{subject_ref}|{version}"         # a retried step asks once per version of the record
    got = propose(step, key, {"step": step, "subject_ref": subject_ref, "fields": 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)