Business Rules System

Cross-cutting validation engine for Coveris. Evaluates operations that change capacity state — assignments, contracts, coverage — against configurable rules with three severity levels. Decision: [[adr-016-capacity-planning-patterns]] §1. Severity levels: [[adr-010-rule-severity-levels]].

Overview

The business rules system answers one question before any capacity-changing operation: “Should this be allowed?”

It does NOT compute hours, coverage, or balance — those computations already exist in the Hours Ledger and Coverage modules. Instead, it validates constraints: legal caps, duplicates, terminated employees, expiring contracts. It’s the gatekeeper between “the user clicked Assign” and “the assignment is created.”

Two-Part Architecture

The system has two components that live in apps/business_rules/:

Catalogue — A Django model (BusinessRule) storing what rules exist, their severity, thresholds, and whether they’re enabled. Exposed via a read-only REST endpoint. Configurable via Django admin.
Engine — A Python module (engine.py) that loads enabled rules, evaluates them against a context, and returns grouped results. Other Django apps import and call evaluate_rules() — they never touch the model directly for evaluation.

This separation means:

The catalogue can change at runtime (admin toggles enabled, changes severity) without touching code
The engine can add new evaluators without touching existing ones (Open/Closed)
Other apps have a clean dependency: they call one function and get results

The Catalogue

RuleSeverity Enum

Three levels per ADR-010. No ERROR level — that’s for technical failures, not business logic.

class RuleSeverity(models.TextChoices):
    BLOCKING = 'BLOCKING', 'Blocking'   # Prevents the operation
    WARNING = 'WARNING', 'Warning'      # Advisory — operation proceeds, violation flagged
    INFO = 'INFO', 'Info'               # Logged for audit trail only

BusinessRule Model

class BusinessRule(BaseModel):
    code = models.CharField(max_length=50, unique=True)
    name = models.CharField(max_length=255)
    severity = models.CharField(max_length=10, choices=RuleSeverity.choices)
    threshold = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True)
    enabled = models.BooleanField(default=True)
    description = models.TextField(blank=True)

    class Meta:
        ordering = ['code']

Inherits from BaseModel (UUID pk, created_at, updated_at).

MVP Seed Rules (7)

Seeded via the Django seed management commands on first deploy (no separate data migration — rules are created via get_or_create in seed commands):

Code	Severity	Threshold	Description	When Evaluated
`MAX_WEEKLY_HOURS`	BLOCKING	60.00	No exceder el tope legal de horas semanales	Assignment preview
`DUPLICATE_ASSIGNMENT`	BLOCKING	—	No duplicar asignación activa por empleado y puesto	Assignment preview
`EMPLOYEE_TERMINATED`	BLOCKING	—	No asignar empleados desvinculados	Assignment preview
`TAG_REQUIREMENT_MISMATCH`	BLOCKING	—	Empleado no cumple los requisitos de tags del puesto	Assignment preview
`MAX_CONSECUTIVE_SHIFTS`	WARNING	—	Empleado acercándose al límite de días consecutivos	Assignment preview
`COVERAGE_EXCEEDED`	WARNING	—	Puesto quedaría en excedente de horas	Assignment preview
`CONTRACT_NEAR_EXPIRY`	INFO	30.00	Contrato próximo a vencer	Employee detail, assignment preview

TAG_REQUIREMENT_MISMATCH introduced by [[adr-019-tag-based-requirements]]. Default BLOCKING; downgrade to WARNING via admin.

REST Endpoint

GET /api/v1/business-rules/ — Non-paginated, returns the full catalogue. Read-only. Authentication required. See API.md for the full contract.

Admin Panel

Registered with list_display, list_filter (severity, enabled), search_fields (code, name, description). Admins can toggle enabled and change severity/threshold without deployments.

The Engine

RuleContext

A dataclass that carries everything an evaluator might need:

@dataclass
class RuleContext:
    employee: Employee
    position_id: UUID | None = None
    effective_hours: Decimal = Decimal('0.00')
    current_assigned_hours: Decimal = Decimal('0.00')  # SUM of employee's active assignments

The context is built by the caller (e.g., assignment preview endpoint) from request data and existing DB state.

RuleResult and RuleResults

@dataclass
class RuleResult:
    rule_code: str
    severity: str       # BLOCKING, WARNING, INFO
    passed: bool
    message: str = ''

@dataclass
class RuleResults:
    blocking: list[RuleResult]
    warnings: list[RuleResult]
    info: list[RuleResult]

    @property
    def is_valid(self) -> bool:
        return len(self.blocking) == 0

evaluate_rules()

The single entry point. Other apps call this and nothing else:

def evaluate_rules(context: RuleContext) -> RuleResults:
    """
    Load all enabled rules, evaluate each against context,
    return results grouped by severity.
    """
    results = RuleResults(blocking=[], warnings=[], info=[])
    for rule in BusinessRule.objects.filter(enabled=True):
        evaluator = RULE_REGISTRY.get(rule.code)
        if evaluator is None:
            continue  # Rule exists in DB but no evaluator yet — skip silently
        result = evaluator(rule, context)
        if not result.passed:
            results.add(rule.severity, result)
    return results

Rule Registry

A dict mapping rule codes to evaluator functions:

RULE_REGISTRY: dict[str, Callable[[BusinessRule, RuleContext], RuleResult]] = {
    'MAX_WEEKLY_HOURS': evaluate_max_weekly_hours,
    'DUPLICATE_ASSIGNMENT': evaluate_duplicate_assignment,
    'EMPLOYEE_TERMINATED': evaluate_employee_terminated,
    'MAX_CONSECUTIVE_SHIFTS': evaluate_max_consecutive_shifts,
    'COVERAGE_EXCEEDED': evaluate_coverage_exceeded,
    'CONTRACT_NEAR_EXPIRY': evaluate_contract_near_expiry,
    'TAG_REQUIREMENT_MISMATCH': evaluate_tag_requirement_mismatch,
}

Evaluator Functions

Each evaluator is a pure function — no side effects, no DB writes:

def evaluate_max_weekly_hours(rule: BusinessRule, ctx: RuleContext) -> RuleResult:
    """BLOCKING if total assigned + proposed would exceed threshold."""
    projected = ctx.current_assigned_hours + ctx.effective_hours
    if projected > rule.threshold:
        return RuleResult(
            rule_code=rule.code,
            severity=rule.severity,
            passed=False,
            message=f"Total semanal sería {projected}h, excede el tope de {rule.threshold}h"
        )
    return RuleResult(rule_code=rule.code, severity=rule.severity, passed=True)

def evaluate_employee_terminated(rule: BusinessRule, ctx: RuleContext) -> RuleResult:
    """BLOCKING if employee status is TERMINATED."""
    if ctx.employee.status == 'TERMINATED':
        return RuleResult(
            rule_code=rule.code,
            severity=rule.severity,
            passed=False,
            message="No se puede asignar un empleado desvinculado"
        )
    return RuleResult(rule_code=rule.code, severity=rule.severity, passed=True)

def evaluate_duplicate_assignment(rule: BusinessRule, ctx: RuleContext) -> RuleResult:
    """BLOCKING if employee already has an active assignment for this position."""
    from apps.assignments.models import Assignment
    exists = Assignment.objects.filter(
        employee=ctx.employee,
        position_id=ctx.position_id,
        status='ACTIVE',
    ).exists()
    if exists:
        return RuleResult(
            rule_code=rule.code,
            severity=rule.severity,
            passed=False,
            message="Ya existe una asignación activa para este empleado y puesto"
        )
    return RuleResult(rule_code=rule.code, severity=rule.severity, passed=True)

Integration with Other Apps

Assignments (Primary Consumer)

The assignment preview endpoint is the main caller. When a manager wants to assign an employee to a position, the frontend calls:

POST /api/v1/assignments/preview/
{
  "employee": "uuid",
  "position_id": "uuid",
  "effective_hours": "12.00"
}

The backend:

Builds a RuleContext from the request + current DB state
Calls evaluate_rules(context)
Returns the result:

{
  "is_valid": true,
  "violations": {
    "blocking": [],
    "warnings": [
      {
        "rule_code": "COVERAGE_EXCEEDED",
        "message": "Puesto quedaría en excedente de horas"
      }
    ],
    "info": []
  }
}

If is_valid is false (any BLOCKING violation), the frontend prevents the assignment. If there are only WARNINGs, it shows them but allows the user to proceed.

Important: The preview does NOT recalculate hours. The employee’s balance (DEFICIT/BALANCED/SURPLUS) is already visible on the Roster detail card — the Hours Ledger pipeline computes it independently. The preview only checks constraints.

Hours Ledger (Not a Direct Consumer)

The Hours Ledger computes employee balance: effective_pool - assigned_hours = balance. This happens independently via GET /api/v1/offer/employees/{id}/balance/. The rules engine does NOT call the ledger service.

Rules like MAX_WEEKLY_HOURS check assigned hours by querying assignments directly — they don’t need the full balance computation. The balance is for display, not for validation.

Coverage (Read-Time, Not Rule-Time)

Coverage states (VACANT, PARTIAL, COVERED, OVER_COVERED) are computed at read time from active assignments. The COVERAGE_EXCEEDED rule checks whether a proposed assignment would push a position into OVER_COVERED — but this is a WARNING, never a BLOCKING.

Coverage is never blocked. Even if a position is already at 100%, you can assign more hours. The system warns, the manager decides.

Contracts

CONTRACT_NEAR_EXPIRY is an INFO rule. It evaluates the employee’s active contract against the threshold (default: 30 days). If the contract’s end_date is within the threshold, it returns an INFO result. This doesn’t prevent anything — it’s audit trail.

The Pipeline Already Answers Hours Questions

A critical design decision: the assignment preview does NOT simulate hours. The balance pipeline already runs and its result is displayed on the employee card. The manager sees:

Base: 36h | Ajustes: -12h | Pool: 24h | Asignado: 8h | Disponible: 16h | Estado: DEFICIT

This is visible BEFORE they open the assignment dialog. When they enter 12h for a position, they can see: 16h - 12h = 4h remaining. No preview endpoint needs to calculate this.

The preview only answers: “Do any BLOCKING rules fire?” That’s it.

Even SURPLUS employees (negative balance = overtime) can be assigned. Going into overtime is a WARNING from MAX_WEEKLY_HOURS if total exceeds 60h, but it’s not blocked by the balance state itself.

Adding a New Rule (Step by Step)

Example: Add a rule that warns when an employee has more than 3 active assignments.

Step 1: Data Migration

def seed(apps, schema_editor):
    BusinessRule = apps.get_model('business_rules', 'BusinessRule')
    BusinessRule.objects.get_or_create(
        code='MAX_ACTIVE_ASSIGNMENTS',
        defaults={
            'name': 'Máximo de asignaciones activas',
            'severity': 'WARNING',
            'threshold': Decimal('3.00'),
            'enabled': True,
            'description': 'Advertir cuando un empleado tiene más de N asignaciones activas',
        }
    )

Step 2: Evaluator Function

def evaluate_max_active_assignments(rule: BusinessRule, ctx: RuleContext) -> RuleResult:
    from apps.assignments.models import Assignment
    count = Assignment.objects.filter(
        employee=ctx.employee,
        status='ACTIVE',
    ).count()
    if count >= rule.threshold:
        return RuleResult(
            rule_code=rule.code,
            severity=rule.severity,
            passed=False,
            message=f"Empleado ya tiene {count} asignaciones activas (máximo: {int(rule.threshold)})"
        )
    return RuleResult(rule_code=rule.code, severity=rule.severity, passed=True)

Step 3: Register

RULE_REGISTRY['MAX_ACTIVE_ASSIGNMENTS'] = evaluate_max_active_assignments

Step 4: Done

No changes to the preview endpoint, the assignment model, or any existing evaluator. The new rule is picked up automatically because evaluate_rules() loads all enabled rules from the DB and looks them up in the registry.

Configuration (Runtime)

Via Django admin at /admin/business_rules/businessrule/:

Toggle enabled: Disable a rule without removing it. evaluate_rules() skips disabled rules.
Change severity: Promote WARNING → BLOCKING or demote BLOCKING → WARNING. Takes effect immediately.
Update threshold: Change MAX_WEEKLY_HOURS from 60 to 48. Takes effect immediately.

No deployment, no migration, no code change. The engine reads from the DB on every call.

File Structure

backend/apps/business_rules/
├── __init__.py
├── apps.py              # BusinessRulesConfig
├── enums.py             # RuleSeverity
├── models.py            # BusinessRule
├── engine.py            # evaluate_rules(), RuleContext, RuleResult, RuleResults, RULE_REGISTRY
├── evaluators.py        # One function per rule code
├── serializers.py       # BusinessRuleSerializer (read-only)
├── views.py             # BusinessRuleViewSet (ReadOnlyModelViewSet)
├── urls.py              # Router registration
├── admin.py             # BusinessRuleAdmin
└── migrations/
    ├── __init__.py
    └── 0001_initial.py           # CreateModel