Add operations layer: SaaS+cost, equipment, inventory, vendors, safety

Makes LabSuite the centralized ops source of truth beyond identity/access:

- operations.Operations: SaaS licences (who has what + monthly/annual spend,
  orphaned-seat detection), equipment (maintenance/calibration due+overdue),
  inventory (low-stock reorder), vendors (upcoming renewals + annual cost),
  facility safety (open issues)
- onboarding grants baseline + role SaaS seats; offboarding reclaims them same-day
- engine: ops wired in, ops_summary(); state persisted
- cli: ops, saas, assets, inventory, vendors, safety; onboard/offboard show SaaS
- api: GET /ops /saas /assets /inventory /vendors /safety
- seed: realistic Merge-lab equipment/reagents/vendors/safety with live flags
- 6 operations tests (61 total, ruff-clean)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01H9BehfGmiCrz3J15W18rc9

Author: sanjaydoc

src/labsuite/api.py

@@ -151,6 +151,37 @@ def expire_training(username: str = Body(...), training: str = Body(...)) -> dic
         control.expire_training(username, training)
         return {"username": username, "training": training, "status": "expired"}
 
+    # --------------------------------------------------------------- #
+    # Operations (SaaS / equipment / inventory / vendors / safety)
+    # --------------------------------------------------------------- #
+    @app.get("/ops")
+    def ops() -> dict:
+        return control.ops_summary()
+
+    @app.get("/saas")
+    def saas() -> dict:
+        return {
+            "monthly_spend": control.ops.monthly_spend(),
+            "annual_cost": control.ops.annual_saas_cost(),
+            "apps": [a.to_dict() for a in control.ops.saas.values()],
+        }
+
+    @app.get("/assets")
+    def assets() -> dict:
+        return {"equipment": [e.to_dict() for e in control.ops.equipment]}
+
+    @app.get("/inventory")
+    def inventory() -> dict:
+        return {"inventory": [i.to_dict() for i in control.ops.inventory]}
+
+    @app.get("/vendors")
+    def vendors() -> dict:
+        return {"vendors": [v.to_dict() for v in control.ops.vendors]}
+
+    @app.get("/safety")
+    def safety() -> dict:
+        return {"safety": [s.to_dict() for s in control.ops.safety]}
+
     # --------------------------------------------------------------- #
     # Web GUI + marketing site
     # --------------------------------------------------------------- #

src/labsuite/cli.py

@@ -111,6 +111,9 @@ def cmd_onboard(args: argparse.Namespace) -> int:
         print(f"  required : {', '.join(result.trainings)} {DIM}(pending){RESET}")
         for share, missing in sorted(result.gated.items()):
             print(f"  {_c('gated', RED)} {share:18} until {', '.join(missing)} complete")
+    if result.saas:
+        _header("SaaS provisioned")
+        print(f"  {', '.join(result.saas)}")
     _save(cp, args.state)
     return 0
 
@@ -132,6 +135,8 @@ def cmd_offboard(args: argparse.Namespace) -> int:
     if result.device:
         d = result.device
         print(f"  device              : {d['asset_tag']} ({d['model']}) -> {_c('wipe & return', RED)}")
+    if result.saas_revoked:
+        print(f"  SaaS seats reclaimed: {', '.join(result.saas_revoked)}")
     _save(cp, args.state)
     return 0 if result.clean else 1
 
@@ -246,6 +251,86 @@ def cmd_devices(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_ops(args: argparse.Namespace) -> int:
+    cp = _load_or_build(args.state)
+    s = cp.ops_summary()
+    _header("Operations dashboard")
+    spend = f"${s['monthly_saas_spend']:,.0f}/mo"
+    print(f"  SaaS spend       : {_c(spend, CYAN)} across {s['saas_apps']} apps")
+    print(f"  Equipment        : {s['equipment']} assets")
+
+    def _flags(title, items, colour=RED):
+        mark = _c("!", colour)
+        if items:
+            print(f"  {title}:")
+            for item in items:
+                print(f"    {mark} {item}")
+
+    _flags("Orphaned SaaS seats (inactive users)", s["orphaned_seats"])
+    _flags("Overdue maintenance", s["overdue_equipment"])
+    _flags("Maintenance due soon", s["due_equipment"], CYAN)
+    _flags("Low stock", s["low_stock"])
+    _flags("Upcoming renewals (<=60d)", s["upcoming_renewals"], CYAN)
+    _flags("Open safety issues", s["open_safety"])
+    clean = not any(s[k] for k in ("orphaned_seats", "overdue_equipment", "low_stock", "open_safety"))
+    if clean:
+        print(_c("  all clear", GREEN))
+    return 0
+
+
+def cmd_saas(args: argparse.Namespace) -> int:
+    cp = _load_or_build(args.state)
+    _header("SaaS licences")
+    for app in sorted(cp.ops.saas.values(), key=lambda a: a.name):
+        seats = len(app.assignees)
+        monthly = app.monthly_cost_per_seat * seats
+        print(f"  {app.name:18} {seats:2} seats  ${app.monthly_cost_per_seat:>5.1f}/seat  = ${monthly:>7.0f}/mo")
+    print(_c(f"  total: ${cp.ops.monthly_spend():,.0f}/mo  (${cp.ops.annual_saas_cost():,.0f}/yr)", CYAN))
+    return 0
+
+
+def cmd_assets(args: argparse.Namespace) -> int:
+    cp = _load_or_build(args.state)
+    _header("Equipment / assets")
+    for e in cp.ops.equipment:
+        if e.maintenance_in_days < 0:
+            m = _c(f"overdue {-e.maintenance_in_days}d", RED)
+        elif e.maintenance_in_days <= 14:
+            m = _c(f"due {e.maintenance_in_days}d", CYAN)
+        else:
+            m = f"{e.maintenance_in_days}d"
+        print(f"  {e.asset_tag}  {e.name:30} {e.location:14} maint {m}")
+    return 0
+
+
+def cmd_inventory(args: argparse.Namespace) -> int:
+    cp = _load_or_build(args.state)
+    _header("Inventory")
+    for i in cp.ops.inventory:
+        flag = _c("LOW", RED) if i.low else "ok"
+        print(f"  {i.sku:8} {i.name:22} {i.qty:3} {i.unit:8} (reorder {i.reorder_point})  {flag}")
+    return 0
+
+
+def cmd_vendors(args: argparse.Namespace) -> int:
+    cp = _load_or_build(args.state)
+    _header("Vendors / contracts")
+    for v in sorted(cp.ops.vendors, key=lambda x: x.renewal_in_days):
+        soon = _c("renews soon", CYAN) if v.renewal_in_days <= 60 else ""
+        print(f"  {v.name:16} {v.category:22} renews {v.renewal_in_days:3}d  ${v.annual_cost:>8,.0f}/yr  {soon}")
+    return 0
+
+
+def cmd_safety(args: argparse.Namespace) -> int:
+    cp = _load_or_build(args.state)
+    _header("Facility safety checks")
+    for s in cp.ops.safety:
+        mark = _c("OPEN", RED) if s.status == "open" else _c("pass", GREEN)
+        note = f"{DIM} — {s.note}{RESET}" if s.note else ""
+        print(f"  {mark:5} {s.area:16} {s.check}{note}")
+    return 0
+
+
 def cmd_sync(args: argparse.Namespace) -> int:
     cp = _load_or_build(args.state)
     report = cp.sync()
@@ -397,6 +482,12 @@ def build_parser() -> argparse.ArgumentParser:
     p.add_argument("--expire", action="store_true", help="mark the training lapsed instead of complete")
     p.set_defaults(func=cmd_train)
 
+    sub.add_parser("ops", help="operations dashboard (SaaS spend + flags)").set_defaults(func=cmd_ops)
+    sub.add_parser("saas", help="SaaS licences + cost").set_defaults(func=cmd_saas)
+    sub.add_parser("assets", help="equipment + maintenance").set_defaults(func=cmd_assets)
+    sub.add_parser("inventory", help="reagent / consumable inventory").set_defaults(func=cmd_inventory)
+    sub.add_parser("vendors", help="vendor contracts + renewals").set_defaults(func=cmd_vendors)
+    sub.add_parser("safety", help="facility safety checks").set_defaults(func=cmd_safety)
     sub.add_parser("sync", help="run the SCIM reconcile now").set_defaults(func=cmd_sync)
 
     p = sub.add_parser("audit", help="print the audit trail")

src/labsuite/engine.py

@@ -35,6 +35,7 @@
 from labsuite.endpoints import DeviceFleet
 from labsuite.models import AccessDecision, AccessLevel, Department, ProxmoxRole
 from labsuite.okta import OktaDirectory
+from labsuite.operations import Operations
 from labsuite.policy import image_for, onboarding_groups, required_trainings
 from labsuite.proxmox import Proxmox
 from labsuite.scim import ScimSync, SyncReport
@@ -60,6 +61,7 @@ class OnboardResult:
     device: dict | None = None
     trainings: dict[str, str] = field(default_factory=dict)
     gated: dict[str, list[str]] = field(default_factory=dict)
+    saas: list[str] = field(default_factory=list)
 
     def to_dict(self) -> dict:
         return {
@@ -73,6 +75,7 @@ def to_dict(self) -> dict:
             "device": self.device,
             "trainings": self.trainings,
             "gated": self.gated,
+            "saas": self.saas,
         }
 
 
@@ -84,6 +87,7 @@ class OffboardResult:
     residual_truenas: dict[str, str] = field(default_factory=dict)
     residual_proxmox: dict[int, str] = field(default_factory=dict)
     device: dict | None = None
+    saas_revoked: list[str] = field(default_factory=list)
 
     @property
     def clean(self) -> bool:
@@ -98,6 +102,7 @@ def to_dict(self) -> dict:
             "residual_proxmox": self.residual_proxmox,
             "clean": self.clean,
             "device": self.device,
+            "saas_revoked": self.saas_revoked,
         }
 
 
@@ -136,6 +141,7 @@ def __init__(
         proxmox: ComputeProvider | None = None,
         endpoints: EndpointProvider | None = None,
         compliance: ComplianceRegistry | None = None,
+        operations: Operations | None = None,
         audit: AuditLog | None = None,
     ) -> None:
         # Any of these may be a real-system adapter; the control plane only ever
@@ -146,6 +152,7 @@ def __init__(
         self.proxmox: ComputeProvider = proxmox or Proxmox()
         self.endpoints: EndpointProvider = endpoints or DeviceFleet()
         self.compliance = compliance or ComplianceRegistry()
+        self.ops = operations or Operations()
         self.audit = audit or AuditLog()
         self.scim = ScimSync(self.okta, self.ad, self.audit)
 
@@ -214,6 +221,10 @@ def onboard(
             detail=f"pending: {', '.join(trainings)}",
         )
 
+        # SaaS half: grant the baseline + role app seats.
+        saas = self.ops.provision_role_apps(username, role)
+        self.audit.record(actor, "saas.provision", username, "saas", detail=", ".join(saas))
+
         access = self.resolve_access(username)
         return OnboardResult(
             username=username,
@@ -226,6 +237,7 @@ def onboard(
             device=device.to_dict(),
             trainings=access.trainings,
             gated=access.blocked,
+            saas=saas,
         )
 
     # ------------------------------------------------------------------ #
@@ -249,6 +261,11 @@ def offboard(self, username: str, *, actor: str = "it-admin") -> OffboardResult:
                 actor, "device.wipe", device.asset_tag, "endpoint", detail="wipe & return",
             )
 
+        # SaaS half: reclaim all licence seats (same-day).
+        saas_revoked = self.ops.revoke_saas_all(username)
+        if saas_revoked:
+            self.audit.record(actor, "saas.revoke", username, "saas", detail=", ".join(saas_revoked))
+
         # Verification: after deprovisioning, effective access must be empty.
         groups = self.ad.effective_groups(username)
         residual_nas = {k: str(v) for k, v in self.truenas.shares_for(groups).items()}
@@ -272,6 +289,7 @@ def offboard(self, username: str, *, actor: str = "it-admin") -> OffboardResult:
             residual_truenas=residual_nas,
             residual_proxmox={int(p.split("/")[-1]): r for p, r in residual_pve.items()},
             device=device.to_dict() if device is not None else None,
+            saas_revoked=saas_revoked,
         )
 
     # ------------------------------------------------------------------ #
@@ -426,6 +444,17 @@ def access_review(self, *, actor: str = "it-admin") -> dict:
     def sync(self, *, actor: str = "it-admin") -> SyncReport:
         return self.scim.reconcile(actor=actor)
 
+    # ------------------------------------------------------------------ #
+    # Operations (SaaS / equipment / inventory / vendors / safety)
+    # ------------------------------------------------------------------ #
+    def _is_active(self, username: str) -> bool:
+        user = self.okta.get_user(username)
+        return bool(user and user.active)
+
+    def ops_summary(self) -> dict:
+        """SaaS spend + flags across equipment, inventory, vendors, and safety."""
+        return self.ops.summary(self._is_active)
+
     # ------------------------------------------------------------------ #
     # Serialisation (persist state between CLI invocations)
     # ------------------------------------------------------------------ #
@@ -437,6 +466,7 @@ def to_dict(self) -> dict:
             "proxmox": self.proxmox.to_dict(),
             "endpoints": self.endpoints.to_dict(),
             "compliance": self.compliance.to_dict(),
+            "operations": self.ops.to_dict(),
             "audit": self.audit.to_list(),
         }
 
@@ -451,6 +481,7 @@ def from_dict(cls, data: dict) -> ControlPlane:
             proxmox=Proxmox.from_dict(data["proxmox"]),
             endpoints=DeviceFleet.from_dict(data.get("endpoints", {})),
             compliance=ComplianceRegistry.from_dict(data.get("compliance", {})),
+            operations=Operations.from_dict(data.get("operations", {})),
             audit=audit,
         )
         return cp