"""Anti-corruption layer: the outside world, as capabilities (capabilities.yaml 0.4), never vendors.
hr speaks capability names through erp_platform.capabilities.CapabilitySeam and translates what comes back —
the platform's Result or ManualTask — into the domain's Outcome. Rule 8 is first-class: with no provider wired,
the call returns a ManualTask carrying the capability and the EXACT validated input it would have sent; the step
becomes a manual task with the plan in words. Wiring a provider later makes the same step automatic: no new step
kind, no playbook change. A provider answering connection_required becomes a manual task too, naming what is
missing. dry_run (preview) validates and plans, and claims and sends nothing.
"""
from typing import Callable, Optional
from erp_platform import capabilities as caps
from brick_hr.domain.bpmn import Capability
from brick_hr.domain.run import Outcome
REQUIRED = ("identity.provision_user", "identity.suspend_user", "identity.delete_user", "identity.invite_app_user",
"groups.add_member", "groups.remove_member", "chat.add_to_space", "chat.remove_from_space",
"drive.create_folder", "drive.transfer_ownership", "drive.transfer_status", "mail.send",
"mail.set_autoreply", "calendar.create_event", "contracts.draft", "contracts.send", "contracts.status",
"contracts.fetch_executed", "identity.revoke_app_user")
# identity.revoke_app_user (HR-54) was hr's own contract (capabilities.local.yaml) until erp-platform 0.9.7's
# capabilities.yaml 0.6 took it under the same name and shape; nothing in a playbook changed.
_FAKES: dict = {} # name -> fn(idempotency_key, data, dry_run) -> caps.Result (tests, tier 1)
class FakeableSeam(caps.CapabilitySeam):
"""The platform's CapabilitySeam — the same provider lookup, call and manual task — with a test's fake in front."""
def provider(self) -> Optional[str]:
return "fake" if self.name in _FAKES else super().provider()
def call(self, idempotency_key: str, input: dict, dry_run: bool = False, actor: str = ""):
if self.name in _FAKES:
return _FAKES[self.name](idempotency_key, self.capability.validate_input(input), dry_run)
return super().call(idempotency_key, input, dry_run=dry_run, actor=actor)
def fake_connector(name: str, fn: Optional[Callable]) -> None:
"""A neighbour's fake for one capability (None removes it) — tests and tier 1."""
if fn is None:
_FAKES.pop(name, None)
else:
_FAKES[name] = fn
SEAMS = {name: FakeableSeam(name) for name in REQUIRED}
def _cap(name: str) -> Optional[caps.Capability]:
return caps.contract().get(name) if name in caps.contract() else None
def known(name: str) -> bool:
return _cap(name) is not None
def is_destructive(name: str) -> bool:
c = _cap(name) if name else None
return bool(c and c.destructive)
def is_read(name: str) -> bool:
return name in ("drive.transfer_status", "contracts.status", "contracts.fetch_executed")
def described(names) -> dict:
"""What the contract says about each capability, as the domain's BPMN picture needs it (name, summary, the
input and output field names, destructive, read). An unknown name is left out."""
out = {}
for name in sorted({n for n in names if n}):
c = _cap(name)
if c is not None:
out[name] = Capability(name, c.summary, tuple(f.name for f in c.input), tuple(f.name for f in c.output),
bool(c.destructive), bool(c.read))
return out
def provider(name: str) -> Optional[str]:
s = SEAMS.get(name)
return s.provider() if s else None
def describe(name: str, data: dict) -> str:
try:
return caps.describe(_cap(name), data)
except Exception: # noqa: BLE001 — words for a person, never a failure
return f"{name}: {data}"
def call(name: str, key: str, data: dict, *, dry_run: bool = False, actor: str = "") -> Outcome:
seam = SEAMS.get(name) or caps.CapabilitySeam(name)
try:
r = seam.call(key, data, dry_run=dry_run, actor=actor)
except caps.InvalidCapabilityInput as e:
# Not sendable as it stands (a value is not known yet, or wrong): a person does it, with what we have.
return Outcome("manual", manual_input=dict(data), plan=describe(name, data),
manual_reason=f"cannot be sent to a provider as it stands: {str(e)[:300]}")
except caps.CapabilityRefused as e:
return Outcome("failed", retryable=False, message=f"{name} refused the call: {e.reason[:300]}")
if isinstance(r, caps.ManualTask):
return Outcome("manual", manual_input=r.input, plan=r.plan, manual_reason=r.reason)
plan = ""
if r.plan:
plan = "; ".join(f"{p.get('action', '')} {p.get('target', '')}".strip() if isinstance(p, dict) else str(p)
for p in r.plan)
return Outcome(r.status, retryable=r.retryable, provider=r.provider, provider_ref=r.provider_ref,
output=r.output or {}, message=r.message, plan=plan)
def complete_manual(name: str, key: str, data: dict, plan: str, reason: str, *, provider_ref: Optional[str] = None,
output: Optional[dict] = None) -> Outcome:
"""A person did the manual task: the same `done` a provider would have answered (ManualTask.complete)."""
cap = _cap(name)
task = caps.ManualTask(name, key, data, plan, reason, cap.destructive, cap.summary)
r = task.complete(provider_ref=provider_ref, output=output)
return Outcome("done", provider=r.provider, provider_ref=r.provider_ref, output=r.output or {}, message=r.message)
__all__ = ["REQUIRED", "fake_connector", "SEAMS", "call", "known", "described", "is_destructive", "is_read", "provider", "describe", "complete_manual"]