The operational model: objects, links and actions

"""Halden's operational model: objects, links and actions over three sources that disagree.

Standard library only; runs offline. Column names and ID shapes match the Halden estate in
_lab/estates/halden (this is a small inline subset, so the file runs on its own).
"""
from __future__ import annotations

import difflib
import json
import random
import re
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone

UTC = timezone.utc
NIGHT = datetime(2026, 2, 12, 0, 0, tzinfo=UTC)  # the night of a Thursday cut-off at Santos


def at(hhmm: str, day: int = 0) -> datetime:
    h, m = map(int, hhmm.split(":"))
    return NIGHT + timedelta(days=day, hours=h, minutes=m)


# --- 1. Three sources, as they arrive ------------------------------------------------
rng = random.Random(2026)
SAP_VENDORS = {"0000400123": "Ostra Lines", "0000400131": "Kestrel Line", "0000400147": "Orca Shipping"}
OWNER = {"0000400123": "OSLU", "0000400131": "KSLU", "0000400147": "ORCU"}
PORTAL_NAMES = {"0000400123": ["OSTRA LINES", "Ostra Lns", "Ostra Lines S.A."],
                "0000400131": ["KESTREL LINE", "Kestrel Line Ltd"],
                "0000400147": ["Orca Shpg", "ORCA SHIPPING", "Orcas Shipping Co"]}
HS = ["0901.11", "6204.62", "0806.10", "4011.10"]

sap_shipments, portal_events, desk_queue, documents = [], [], [], []
for i in range(40):
    vendor = rng.choice(sorted(SAP_VENDORS))
    sid = f"40{i + 1:08d}"
    eta = (NIGHT + timedelta(days=17 + rng.randint(0, 2))).date().isoformat()
    sap_shipments.append({"SHIPMENT_ID": sid, "CARRIER_VENDOR_ID": vendor, "ORIGIN_PORT": "BRSSZ",
                          "DEST_PORT": "NLRTM", "CONTAINER_NO": f"{OWNER[vendor]}{rng.randint(10**6, 10**7 - 1)}",
                          "HS_CODE": rng.choice(HS), "ETA": eta, "INV_AMT": f"{rng.uniform(2e4, 9e4):.2f}",
                          "INV_CCY": "USD"})
    portal_events.append({"carrier_name": rng.choice(PORTAL_NAMES[vendor]), "shipment": sid,
                          "event_code": "LOADED", "eta": eta, "event_time": at("23:10", -1), "scraped_at": at("01:40")})
    if i % 3 == 0:
        documents.append({"doc_id": f"DOC-0{i + 1:06d}", "shipment_id": sid, "doc_type": "packing_list"})
# Shipment 17's vessel was reported late at 23:30, but the scraper only picked that up at 03:30.
v17 = sap_shipments[16]["CARRIER_VENDOR_ID"]
portal_events[16]["carrier_name"] = PORTAL_NAMES[v17][0]
portal_events.append({"carrier_name": PORTAL_NAMES[v17][1], "shipment": "4000000017", "event_code": "ETA_CHANGED",
                      "eta": "2026-03-05", "event_time": at("23:30", -1), "scraped_at": at("03:30")})
for n, sid in enumerate(["4000000017", "4000000004", "4000000009", "4000000023", "4000000031"]):
    desk_queue.append({"EXC_ID": f"EXC-{10482 + n}", "SHIPMENT_ID": sid, "DESK": "São Paulo",
                       "TYPE": ("WRONG_HS_CODE", "MISSING_DOCS", "INVOICE_MISMATCH")[n % 3], "STATUS": "OPEN"})


# --- 2. Entity resolution: one carrier, many spellings ----------------------------
ABBREV = {"lns": "lines", "ln": "line", "shpg": "shipping"}
SUFFIXES = {"sa", "ltd", "co", "gmbh", "pte"}


def norm(name: str) -> str:
    words = re.sub(r"[^a-z0-9 ]", "", name.lower().replace("s.a.", "sa")).split()
    return " ".join(ABBREV.get(w, w) for w in words if w not in SUFFIXES)


def resolve(name: str, master: dict[str, str]) -> tuple[str | None, str, float]:
    """(vendor id, match class, score). Probable matches go to a person, never straight in."""
    for vid, canon in master.items():
        if name == canon:
            return vid, "exact", 1.0
    score, vid = max((difflib.SequenceMatcher(None, norm(name), norm(c)).ratio(), v) for v, c in master.items())
    if score == 1.0:
        return vid, "normalised", 1.0
    return (vid, "probable", round(score, 2)) if score >= 0.9 else (None, "none", round(score, 2))


# --- 3. Objects, links, and a value history per property ----------------------------
@dataclass
class Value:
    value: object
    source: str  # SAP | portal | desk | action:<name>
    valid_from: datetime  # when it became true in the world
    recorded: datetime  # when the model learned it


@dataclass
class Obj:
    type: str
    key: str
    props: dict[str, list[Value]] = field(default_factory=dict)
    links: dict[str, set[str]] = field(default_factory=dict)


# Survivorship: which source wins a property, in order; the losers are kept, not deleted.
SURVIVOR = {"eta": ("portal", "SAP"), "status": ("action", "desk")}
DEFAULT_ORDER = ("action", "SAP", "portal", "desk")
# The mapping is versioned: a renamed column changes this, not every consumer.
MAPPING = {"version": 1, "Shipment": {"container": "CONTAINER_NO", "hs_code": "HS_CODE", "eta": "ETA",
                                      "invoice_amount": "INV_AMT", "invoice_currency": "INV_CCY"}}


class Model:
    def __init__(self) -> None:
        self.objects: dict[str, Obj] = {}
        self.audit: list[dict] = []
        self.ledger: dict[str, dict] = {}  # idempotency key -> first applied result
        self.review: list[tuple[str, str, str]] = []  # (name, why, shipment) for a person to decide

    def obj(self, type_: str, key: str) -> Obj:
        return self.objects.setdefault(key, Obj(type_, key))

    def put(self, key: str, prop: str, value, source: str, valid_from: datetime, recorded: datetime) -> None:
        self.objects[key].props.setdefault(prop, []).append(Value(value, source, valid_from, recorded))

    def link(self, a: str, name: str, b: str) -> None:
        self.objects[a].links.setdefault(name, set()).add(b)

    def get(self, key: str, prop: str, as_of: datetime | None = None, valid_at: datetime | None = None) -> Value | None:
        """The surviving value. as_of = what we knew then; valid_at = what was true then."""
        vals = [v for v in self.objects[key].props.get(prop, [])
                if (as_of is None or v.recorded <= as_of) and (valid_at is None or v.valid_from <= valid_at)]
        order = SURVIVOR.get(prop, DEFAULT_ORDER)
        rank = lambda v: (order.index(v.source.split(":")[0]), -v.valid_from.timestamp())  # noqa: E731
        return min(vals, key=rank) if vals else None

    def val(self, key: str, prop: str):
        v = self.get(key, prop)
        return v.value if v else None

    def linked(self, key: str, name: str) -> list[Obj]:
        return [self.objects[k] for k in sorted(self.objects[key].links.get(name, ()))]


def build() -> Model:
    m, extract = Model(), at("22:00", -1)  # the nightly SAP extract landed at 22:00
    for vid, name in SAP_VENDORS.items():
        m.obj("Carrier", vid)
        m.put(vid, "name", name, "SAP", extract, extract)
    m.obj("Lane", "BRSSZ-NLRTM")
    m.put("BRSSZ-NLRTM", "cutoff", "Thursday 15:00 local", "SAP", extract, extract)
    for row in sap_shipments:
        s = m.obj("Shipment", row["SHIPMENT_ID"])
        for prop, column in MAPPING["Shipment"].items():
            m.put(s.key, prop, row[column], "SAP", extract, extract)
        m.link(s.key, "carried_by", row["CARRIER_VENDOR_ID"])
        m.link(s.key, "on_lane", "BRSSZ-NLRTM")
        m.link("BRSSZ-NLRTM", "shipments", s.key)
    for ev in portal_events:
        vid, how, score = resolve(ev["carrier_name"], SAP_VENDORS)
        expected = next(iter(m.objects[ev["shipment"]].links["carried_by"]))
        if how == "probable" or vid != expected:  # a person decides before the event counts
            why = f"{how} match ({score})" if how == "probable" else f"resolves to {vid}, SAP says {expected}"
            m.review.append((ev["carrier_name"], why, ev["shipment"]))
            continue
        m.put(vid, "alias", ev["carrier_name"], "portal", ev["scraped_at"], ev["scraped_at"])
        m.put(ev["shipment"], "eta", ev["eta"], "portal", ev["event_time"], ev["scraped_at"])
    for row in desk_queue:
        e = m.obj("Exception", row["EXC_ID"])
        m.put(e.key, "type", row["TYPE"], "desk", at("01:05"), at("01:05"))
        m.put(e.key, "status", row["STATUS"], "desk", at("01:05"), at("01:05"))
        m.put(e.key, "desk", row["DESK"], "desk", at("01:05"), at("01:05"))
        m.link(e.key, "about", row["SHIPMENT_ID"])
        m.link(row["SHIPMENT_ID"], "exceptions", e.key)
    for d in documents:
        m.obj("Document", d["doc_id"])
        m.put(d["doc_id"], "doc_type", d["doc_type"], "SAP", extract, extract)
        m.link(d["doc_id"], "for", d["shipment_id"])
        m.link(d["shipment_id"], "documents", d["doc_id"])
    return m


# --- 4. Actions: the only way anything changes -------------------------------------
USERS = {"rafaela.souza": {"desk_lead", "approver"}, "camila.rocha": {"operator"},
         "joao.ferreira": {"operator", "approver"}, "exception-agent": {"agent"}}
OPERATORS = {"OP-SP-01": {"invoice", "customs", "reefer"}, "OP-SP-02": {"invoice", "documents"},
             "OP-SP-03": {"customs", "reefer"}}
SKILL = {"WRONG_HS_CODE": "customs", "MISSING_DOCS": "documents", "INVOICE_MISMATCH": "invoice"}
FITS = {"WRONG_HS_CODE": {"correct_hs_code"}, "MISSING_DOCS": {"request_document"},
        "INVOICE_MISMATCH": {"request_revised_invoice", "accept_sap_amount"}}
WRITES_SAP = {"correct_hs_code", "accept_sap_amount"}  # these change the system of record
JSON_TYPE = {str: "string", int: "integer"}


@dataclass(frozen=True)
class Param:
    name: str
    kind: type
    choices: tuple = ()
    description: str = ""


def exc_type(m: Model, a: dict):
    return m.val(a["exception_id"], "type")


def is_open(m: Model, a: dict) -> bool:
    return a["exception_id"] in m.objects and m.val(a["exception_id"], "status") == "OPEN"


@dataclass(frozen=True)
class ActionType:
    name: str
    description: str
    params: tuple[Param, ...]
    roles: frozenset[str]
    checks: tuple  # submission criteria: (what must be true, test)
    effect: object  # (model, args) -> [(object key, property, new value)]
    approvals: object = lambda a: 0  # how many approvers this call needs
    adapter: str = ""  # the system written to, if any

    def validate(self, args: dict) -> list[str]:
        out = [f"unknown parameter {k}" for k in args if k not in {p.name for p in self.params}]
        for p in self.params:
            if not isinstance(args.get(p.name), p.kind):
                out.append(f"{p.name} must be {JSON_TYPE[p.kind]}")
            elif p.choices and args[p.name] not in p.choices:
                out.append(f"{p.name} must be one of {list(p.choices)}")
        return out

    def tool(self) -> dict:
        """The agent's tool is generated from the action, so the two can't drift apart."""
        props = {p.name: {"type": JSON_TYPE[p.kind], **({"description": p.description} if p.description else {}),
                          **({"enum": list(p.choices)} if p.choices else {})} for p in self.params}
        return {"name": self.name, "description": self.description,
                "input_schema": {"type": "object", "properties": props,
                                 "required": [p.name for p in self.params], "additionalProperties": False}}


EXC = Param("exception_id", str, description="EXC- id of an open exception")
ACTIONS = {a.name: a for a in (
    ActionType("assign_exception", "Assign an open exception to an operator with the right skill.",
               (EXC, Param("operator_id", str, tuple(OPERATORS))), frozenset({"desk_lead"}),
               (("the exception is open", is_open),
                ("the operator has the skill this type needs",
                 lambda m, a: is_open(m, a) and SKILL[exc_type(m, a)] in OPERATORS[a["operator_id"]])),
               lambda m, a: [(a["exception_id"], "assigned_to", a["operator_id"])]),
    ActionType("request_document", "Ask the shipper for a missing document by a deadline.",
               (EXC, Param("doc_type", str, ("packing_list", "certificate_of_origin", "commercial_invoice")),
                Param("due_hours", int, description="1 to 72")),
               frozenset({"agent", "operator", "desk_lead"}),
               (("the exception is open", is_open),
                ("the deadline is 1 to 72 hours away", lambda m, a: 1 <= a["due_hours"] <= 72),
                ("the document is not already on file", lambda m, a: is_open(m, a) and a["doc_type"] not in {
                    m.val(d.key, "doc_type") for s in m.linked(a["exception_id"], "about")
                    for d in m.linked(s.key, "documents")})),
               lambda m, a: [(a["exception_id"], "waiting_for", a["doc_type"])], adapter="email"),
    ActionType("propose_resolution", "Resolve an open exception. Changes to SAP need one approver.",
               (EXC, Param("resolution", str, tuple(sorted(set().union(*FITS.values())))),
                Param("note", str, description="why, in one sentence")),
               frozenset({"agent", "operator", "desk_lead"}),
               (("the exception is open", is_open),
                ("the resolution fits the exception type",
                 lambda m, a: is_open(m, a) and a["resolution"] in FITS[exc_type(m, a)])),
               lambda m, a: [(a["exception_id"], "status", "RESOLVED"), (a["exception_id"], "resolution", a["resolution"])],
               approvals=lambda a: 1 if a["resolution"] in WRITES_SAP else 0, adapter="SAP"),
)}


def sap_dry_run(m: Model, args: dict) -> dict:
    """What the SAP adapter would send. A real one calls the customer's API with its own credentials."""
    s = m.linked(args["exception_id"], "about")[0]
    return {"system": "SAP S/4HANA (dry run)", "shipment": s.key, "change": args["resolution"],
            "hs_code_before": m.val(s.key, "hs_code")}


def submit(m: Model, name: str, args: dict, actor: str, key: str, now: datetime,
           approvers: tuple[str, ...] = ()) -> dict:
    if key in m.ledger:  # the same request again (a retry, a double click): same answer, no new change
        return {**m.ledger[key], "replayed": True}
    a = ACTIONS[name]
    reasons = a.validate(args)
    if not reasons and not USERS.get(actor, set()) & a.roles:
        reasons = [f"{actor} may not run {name}"]
    if not reasons:
        reasons = [f"failed: {what}" for what, test in a.checks if not test(m, args)]
    signed = sorted({u for u in approvers if u != actor and "approver" in USERS.get(u, set())})
    if reasons:
        result = {"status": "refused", "reasons": reasons}
    elif len(signed) < a.approvals(args):
        result = {"status": "pending_approval", "needs": a.approvals(args), "has": len(signed)}
    else:
        edits = a.effect(m, args)
        for k, prop, value in edits:
            m.put(k, prop, value, f"action:{name}", now, now)
        sent = sap_dry_run(m, args) if a.adapter == "SAP" and args["resolution"] in WRITES_SAP else None
        result = {"status": "applied", "edits": edits, **({"sent": sent} if sent else {})}
        m.ledger[key] = result
    m.audit.append({"at": now.isoformat(), "action": name, "actor": actor, "approvers": signed,
                    "args": args, "key": key, "status": result["status"]})
    return result


# --- 5. Run it ---------------------------------------------------------------------
def show(label: str, result: dict) -> None:
    print(f"  {label:<44} {json.dumps(result, ensure_ascii=False, default=str)}")


def main() -> None:
    print("1. Carrier resolution (portal name -> SAP vendor)")
    for name in sorted({ev["carrier_name"] for ev in portal_events}):
        vid, how, score = resolve(name, SAP_VENDORS)
        print(f"  {name:<20} {how:<11} {score:<5} {vid or '-'} {SAP_VENDORS.get(vid or '', '')}")
    m = build()
    kinds: dict[str, int] = {}
    for o in m.objects.values():
        kinds[o.type] = kinds.get(o.type, 0) + 1
    print(f"\n2. Model: {kinds}; mapping v{MAPPING['version']}; for review: {len(m.review)}")
    for name, why in sorted({(n, w) for n, w, _ in m.review}):
        held = sorted(s for n, w, s in m.review if (n, w) == (name, why))
        print(f"  review: {name!r} is a {why}; {len(held)} events held, e.g. {held[0]}")
    s = "4000000017"
    print(f"\n3. Provenance of {s}.eta (every value kept, in the order learned)")
    for v in sorted(m.objects[s].props["eta"], key=lambda v: v.recorded):
        print(f"  {v.value}  from {v.source:<6} valid {v.valid_from:%d %H:%M}  recorded {v.recorded:%d %H:%M}")
    print(f"  now:                      {m.get(s, 'eta').value}")
    print(f"  as we knew it at 02:00:   {m.get(s, 'eta', as_of=at('02:00')).value}"
          "   <- what the agent saw when it wrote to the consignee")
    print(f"  true at 02:00, known now: {m.get(s, 'eta', valid_at=at('02:00')).value}")
    print("\n4. Tools generated from actions")
    for a in ACTIONS.values():
        t = a.tool()
        print(f"  {t['name']}({', '.join(t['input_schema']['required'])})")
    print("  " + json.dumps(ACTIONS["request_document"].tool()["input_schema"]["properties"]["doc_type"]))
    print("\n5. Actions")
    assign = {"exception_id": "EXC-10482", "operator_id": "OP-SP-03"}
    show("agent tries to assign", submit(m, "assign_exception", assign, "exception-agent", "k1", at("02:05")))
    show("desk lead assigns", submit(m, "assign_exception", assign, "rafaela.souza", "k2", at("02:06")))
    show("same request again (retry)", submit(m, "assign_exception", assign, "rafaela.souza", "k2", at("02:06")))
    show("wrong skill", submit(m, "assign_exception", {"exception_id": "EXC-10483", "operator_id": "OP-SP-03"},
                               "rafaela.souza", "k3", at("02:07")))
    docs = {"exception_id": "EXC-10483", "doc_type": "packing_list", "due_hours": 24}
    show("agent asks for a packing list on file", submit(m, "request_document", docs, "exception-agent", "k4", at("02:08")))
    fix = {"exception_id": "EXC-10482", "resolution": "correct_hs_code", "note": "declared code differs from the invoice"}
    show("agent proposes an SAP change", submit(m, "propose_resolution", fix, "exception-agent", "k5", at("02:10")))
    show("...with one approver", submit(m, "propose_resolution", fix, "exception-agent", "k5", at("02:31"),
                                        approvers=("joao.ferreira",)))
    print(f"\n6. Audit log: {len(m.audit)} attempts, {sum(e['status'] == 'applied' for e in m.audit)} applied")
    for e in m.audit:
        print(f"  {e['at'][11:16]} {e['actor']:<16} {e['action']:<19} {e['status']:<17} approvers={e['approvers']}")
    assert m.val("EXC-10482", "status") == "RESOLVED" and m.val("EXC-10482", "assigned_to") == "OP-SP-03"
    assert len([e for e in m.audit if e["key"] == "k2"]) == 1, "a replay must not write twice"


if __name__ == "__main__":
    main()

1. Carrier resolution (portal name -> SAP vendor)
  Kestrel Line Ltd     normalised  1.0   0000400131 Kestrel Line
  Orcas Shipping Co    probable    0.96  0000400147 Orca Shipping
  Ostra Lns            normalised  1.0   0000400123 Ostra Lines

2. Model: {'Carrier': 3, 'Lane': 1, 'Shipment': 40, 'Exception': 5, 'Document': 14}; mapping v1; for review: 4
  review: 'Orcas Shipping Co' is a probable match (0.96); 4 events held, e.g. 4000000002

3. Provenance of 4000000017.eta (every value kept, in the order learned)
  2026-03-03  from SAP    valid 11 22:00  recorded 11 22:00
  2026-03-03  from portal valid 11 23:10  recorded 12 01:40
  2026-03-05  from portal valid 11 23:30  recorded 12 03:30
  now:                      2026-03-05
  as we knew it at 02:00:   2026-03-03   <- what the agent saw when it wrote to the consignee
  true at 02:00, known now: 2026-03-05

5. Actions
  agent tries to assign                        {"status": "refused", "reasons": ["exception-agent may not run assign_exception"]}
  same request again (retry)                   {"status": "applied", ..., "replayed": true}
  wrong skill                                  {"status": "refused", "reasons": ["failed: the operator has the skill this type needs"]}
  agent proposes an SAP change                 {"status": "pending_approval", "needs": 1, "has": 0}
  ...with one approver                         {"status": "applied", ..., "sent": {"system": "SAP S/4HANA (dry run)", ...}}

6. Audit log: 6 attempts, 2 applied