"""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 (the default:
AI drafts, a person approves). All three steps here are suggest-only (`safe_for_auto: false` in the spec).
What a step does, in order:
1. mode `off` -> nothing runs; the process is the fallback.
2. a fixed BASELINE the brick computes itself, with no model (word overlap for duplicates, a percentage
threshold for prices): it is always shown, so the step is useful with no provider wired.
3. `propose()` asks the `ai.suggest` provider the composition wired. None wired (or none connected) is a
first-class path, not an error: the answer is a ManualTask, stored in prod_task for a person (rule 8).
The provider never applies anything; a person accepts, edits or dismisses (close_product_task).
Tests run with no provider, so they need no model call.
"""
from dataclasses import dataclass, field
from typing import Optional
from erp_platform.capabilities import CapabilitySeam, ManualTask
# 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_duplicate": {
"capability": "ai.suggest",
"mode": "suggest",
"modes_allowed": ["off", "suggest"],
"safe_for_auto": False,
"fallback": "no suggestion; the catalogue manager adds the product as typed",
"benefit": "a clean product list: duplicates are the reason two invoices for the same thing never reconcile",
"inputs": ["name", "description", "the existing products' names and descriptions (embeddings)"],
"eval": "a labelled set of product pairs (duplicate / not), precision and recall at the shown threshold",
},
"ai_suggest_details": {
"capability": "ai.suggest",
"mode": "suggest",
"modes_allowed": ["off", "suggest"],
"safe_for_auto": False,
"fallback": "the fields stay empty and activation asks for them",
"benefit": "adding a product takes one line: the category, the unit and the tax category are proposed from the name",
"inputs": ["name", "description", "the categories", "the units", "the tax categories and their descriptions"],
"eval": "products with known category, unit and tax category; accuracy per field",
},
"ai_price_check": {
"capability": "ai.suggest",
"mode": "suggest",
"modes_allowed": ["off", "suggest"],
"safe_for_auto": False,
"fallback": "the fixed threshold check in prod_settings alone",
"benefit": "a mistyped price (12.00 for 120.00) is caught before a customer is quoted it",
"inputs": ["the draft rows", "the base list", "prices of products in the same category"],
"eval": "price lists with seeded errors; recall of the seeded errors, false flags per 100 rows",
},
}
SEAMS = {
"ai_check_duplicate": CapabilitySeam("ai.suggest"),
"ai_suggest_details": CapabilitySeam("ai.suggest"),
"ai_price_check": 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 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)
@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)