"""The neighbours this brick asks, behind one adapter: invoicing (AR-29) and sales orders (SO-23).
Both are DECLARED here and nowhere else, so the domain never knows how they are reached. Each answer is one of:
- done: the neighbour has it (invoicing's draft keyed acc:<item_id>; the order's delivery keyed by item);
- pending: the neighbour is not composed, or is down — the item keeps billing_status pending and a person
(or the next retry) finishes it. Accepting work is never undone because another module is restarting.
The permissions are the signed M1 specs' (erp-method 0.1.22): invoicing's raise_for_acceptance needs ar:invoice:raise,
a machine permission (AR-29); the order's record_delivery needs so:order:deliver (SO-23). The specs name no routes:
the paths below are this brick's expectation until those bricks exist — if they differ, this file is the only one
that changes.
"""
from dataclasses import dataclass
from erp_platform.seam import Seam
# The identity each neighbour sees is the platform's seam identity: in process the permission the neighbour checks,
# across apps the service client's token (SEAM_CLIENT_ID). Never a role name — the neighbour checks permissions.
INVOICING = Seam("ar", "AR_BASE_URL", roles=[], permissions=["ar:invoice:raise"])
ORDERS = Seam("so", "SO_BASE_URL", roles=[], permissions=["so:order:deliver"])
@dataclass(frozen=True)
class Asked:
done: bool
reason: str = "" # why it is still pending, in words a person reads
plan: str = "" # what to do by hand meanwhile
def raise_for_acceptance(item, actor: str) -> Asked:
"""PJ-22 → AR-29: one draft invoice for an accepted one-off item, keyed acc:<item_id> (a retry bills once)."""
plan = (f"Raise a draft invoice for '{item.title}' (customer {item.bp_id}, {item.source_kind} "
f"{item.source_id}, line {item.line_id}), reference acc:{item.item_id}.")
if not INVOICING.installed:
return Asked(False, "invoicing is not part of this installation", plan)
r = INVOICING.post("/invoices/for-acceptance", {
"source_key": f"acc:{item.item_id}", "item_id": item.item_id, "bp_id": str(item.bp_id),
"source_kind": item.source_kind, "source_id": item.source_id, "line_id": item.line_id,
"product_id": item.product_id, "description": item.title,
"accepted_at": item.approved_at.isoformat() if item.approved_at else None,
"approved_how": item.approved_how}, actor=actor)
return Asked(True) if r else Asked(False, f"invoicing did not take it: {r.reason or r.status}", plan)
def record_delivery(item, actor: str) -> Asked:
"""PJ-21 → SO-23: tell the order one unit of its line was delivered, keyed by the item (idempotent there)."""
plan = f"Record 1 delivered on order {item.source_id}, line {item.line_id}, reference {item.item_id}."
if item.source_kind != "sales_order" or item.source != "sold" or not item.line_id or item.state != "done":
return Asked(False, "only accepted work sold on an order is reported to it")
if not ORDERS.installed:
return Asked(False, "sales orders are not part of this installation", plan)
r = ORDERS.post(f"/orders/{item.source_id}/deliveries", {
"line_no": item.line_id, "quantity": 1, "delivery_ref": item.item_id,
"delivered_on": item.delivered_at.date().isoformat() if item.delivered_at else None}, actor=actor)
return Asked(True) if r else Asked(False, f"the order did not take it: {r.reason or r.status}", plan)