ACA Tracking Logic

The Affordable Care Act (ACA) Employer Shared Responsibility Provisions (ESRP) require deterministic tracking of Hours of Service (HOS) and Full-Time Equivalent (FTE) status across rolling measurement periods. This module operates as a stateful compliance engine that ingests raw timekeeping records, normalizes them against jurisdictional thresholds, and outputs audit-ready status flags aligned with IRS Form 1094-C/1095-C reporting requirements. Implementation must enforce idempotent processing, explicit validation gates, and immutable audit trails to withstand IRS scrutiny and internal compliance audits.

Pipeline Architecture & Compliance Scope

ACA tracking logic functions as a downstream consumer of normalized payroll events. It relies on the foundational Core Architecture & Compliance Mapping for Payroll Systems to resolve entity relationships, tax jurisdictions, and reporting hierarchies before applying statutory calculations. The pipeline must maintain strict referential integrity between employee lifecycle events, compensation boundaries, and statutory look-back methodologies.

State management requires a rolling 12-month measurement window paired with a 13-month stability period. All calculations must be deterministic: identical input payloads must yield identical HOS totals and FTE classifications regardless of execution order or retry attempts.

Ingestion & Boundary Enforcement

Raw time and attendance data enters the pipeline in heterogeneous formats: biometric clock-ins, HRIS leave records, payroll batch adjustments, and third-party scheduling exports. ACA tracking logic requires strict normalization of these inputs into a canonical HoursOfService schema. The IRS defines HOS as each hour for which an employee is paid or entitled to payment, plus hours for vacation, holiday, illness, incapacity, jury duty, military duty, and leave of absence. Unpaid leave and non-compensated breaks must be explicitly excluded.

Normalization must enforce Data Boundary Definitions to prevent cross-contamination between exempt/non-exempt classifications, contractor records, and seasonal worker pools. Boundary enforcement occurs at the ingestion layer through schema validation, timezone alignment to the employer’s primary worksite, and explicit rejection of malformed or duplicate time punches.

from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from enum import Enum
from typing import Optional, List, Dict, Tuple
import logging
import hashlib
import pytz

logger = logging.getLogger(__name__)

class HOSCategory(str, Enum):
    PAID_WORK = "paid_work"
    PAID_LEAVE = "paid_leave"
    UNPAID_LEAVE = "unpaid_leave"
    EXCLUDED = "excluded"

class ProcessingStatus(str, Enum):
    ACCEPTED = "accepted"
    QUARANTINED = "quarantined"
    REJECTED = "rejected"

@dataclass(frozen=True)
class TimeRecord:
    employee_id: str
    record_date: date
    hours: float
    category: HOSCategory
    source_system: str
    timezone: str = "UTC"

    def __post_init__(self):
        if self.hours < 0:
            raise ValueError(f"Negative hours detected for {self.employee_id} on {self.record_date}")
        if self.hours > 24:
            raise ValueError(f"Hours exceed daily cap for {self.employee_id} on {self.record_date}")

    @property
    def record_hash(self) -> str:
        payload = f"{self.employee_id}|{self.record_date}|{self.hours}|{self.category.value}"
        return hashlib.sha256(payload.encode()).hexdigest()

@dataclass
class AuditEvent:
    record_id: str
    timestamp: datetime
    status: ProcessingStatus
    reason: Optional[str] = None
    metadata: Dict = field(default_factory=dict)

HOS Normalization Engine

The normalization engine maps raw inputs to IRS-compliant HOS totals. It applies category weighting, filters excluded records, and routes anomalies to quarantine queues. The engine operates statelessly per record but aggregates deterministically at the employee-period level.

class HOSNormalizer:
    VALID_CATEGORIES = {HOSCategory.PAID_WORK, HOSCategory.PAID_LEAVE}
    MAX_DAILY_HOS = 24.0

    def __init__(self, primary_worksite_tz: str = "America/New_York"):
        self.tz = pytz.timezone(primary_worksite_tz)
        self.audit_trail: List[AuditEvent] = []

    def validate_and_normalize(self, record: TimeRecord) -> Tuple[Optional[float], AuditEvent]:
        """Returns normalized HOS value and audit event. Routes invalid records to quarantine."""
        try:
            if record.category not in self.VALID_CATEGORIES:
                return self._quarantine(record, f"Non-HOS category: {record.category.value}")

            if record.category == HOSCategory.PAID_LEAVE:
                # IRS permits up to 8 hours/day for unpaid leave, but paid leave counts fully
                normalized = min(record.hours, self.MAX_DAILY_HOS)
            else:
                normalized = min(record.hours, self.MAX_DAILY_HOS)

            event = AuditEvent(
                record_id=record.record_hash,
                timestamp=datetime.now(self.tz),
                status=ProcessingStatus.ACCEPTED,
                metadata={"original_category": record.category.value, "normalized_hours": normalized}
            )
            self.audit_trail.append(event)
            return normalized, event

        except Exception as e:
            return self._quarantine(record, str(e))

    def _quarantine(self, record: TimeRecord, reason: str) -> Tuple[None, AuditEvent]:
        event = AuditEvent(
            record_id=record.record_hash,
            timestamp=datetime.now(self.tz),
            status=ProcessingStatus.QUARANTINED,
            reason=reason,
            metadata={"source": record.source_system, "category": record.category.value}
        )
        self.audit_trail.append(event)
        logger.warning(f"Record quarantined: {record.record_hash} | Reason: {reason}")
        return None, event

FTE Aggregation & Measurement Periods

FTE status derivation requires aggregating normalized HOS across a defined measurement window. The IRS threshold for full-time status is 130 hours per month (or 30 hours per week). Classification logic must align with FLSA Threshold Mapping to ensure consistent worker categorization across payroll and compliance modules.

The aggregation engine calculates rolling monthly totals, applies the 130-hour threshold, and flags stability period transitions. Detailed implementation of the rolling window mechanics and look-back period alignment is covered in Automating ACA full-time equivalent tracking.

@dataclass(frozen=True)
class MeasurementPeriod:
    employee_id: str
    start_date: date
    end_date: date
    is_lookback: bool = True

class FTEAggregator:
    FULL_TIME_MONTHLY_THRESHOLD = 130.0
    FULL_TIME_WEEKLY_THRESHOLD = 30.0

    def __init__(self):
        self.employee_monthly_totals: Dict[str, Dict[date, float]] = {}

    def aggregate_period(self, records: List[TimeRecord], period: MeasurementPeriod) -> Dict[str, str]:
        """Aggregates HOS for a measurement period and returns FTE status flags."""
        period_totals: Dict[str, float] = {}

        for rec in records:
            if not (period.start_date <= rec.record_date <= period.end_date):
                continue

            period_totals.setdefault(rec.employee_id, 0.0)
            period_totals[rec.employee_id] += rec.hours

        statuses: Dict[str, str] = {}
        for emp_id, total_hours in period_totals.items():
            if total_hours >= self.FULL_TIME_MONTHLY_THRESHOLD:
                statuses[emp_id] = "FULL_TIME"
            elif total_hours >= (self.FULL_TIME_MONTHLY_THRESHOLD * 0.8):
                statuses[emp_id] = "NEAR_THRESHOLD"
            else:
                statuses[emp_id] = "PART_TIME"

        return statuses

Compliance Verification & Audit Routing

Before exporting to IRS reporting formats, the pipeline must execute deterministic verification steps. These checks validate mathematical consistency, enforce statutory caps, and generate cryptographic audit trails.

1. HOS Reconciliation: Sum of daily normalized hours must equal monthly FTE calculation inputs. Discrepancies > 0.01 hours trigger a hard stop.
2. Threshold Boundary Validation: Employees crossing the 130-hour threshold must have explicit stability period assignments recorded.
3. Audit Immutability: All AuditEvent records must be serialized to a write-once ledger or immutable S3 bucket with SHA-256 checksums.

class ComplianceVerifier:
    def verify_fte_output(self, period: MeasurementPeriod, statuses: Dict[str, str], raw_hours: Dict[str, float]) -> bool:
        """Validates FTE output against raw inputs and IRS thresholds."""
        for emp_id, status in statuses.items():
            hours = raw_hours.get(emp_id, 0.0)
            if status == "FULL_TIME" and hours < FTEAggregator.FULL_TIME_MONTHLY_THRESHOLD:
                logger.error(f"Compliance violation: {emp_id} marked FULL_TIME with {hours} hours")
                return False
            if status == "PART_TIME" and hours >= FTEAggregator.FULL_TIME_MONTHLY_THRESHOLD:
                logger.error(f"Compliance violation: {emp_id} marked PART_TIME with {hours} hours")
                return False
        return True

    def generate_audit_manifest(self, events: List[AuditEvent]) -> str:
        """Creates a deterministic audit manifest for IRS submission readiness."""
        manifest_lines = []
        for evt in sorted(events, key=lambda x: x.timestamp):
            line = f"{evt.timestamp.isoformat()}|{evt.record_id}|{evt.status.value}|{evt.reason or 'OK'}"
            manifest_lines.append(line)
        manifest = "\n".join(manifest_lines)
        return hashlib.sha256(manifest.encode()).hexdigest()

Fallback & Emergency Pause Workflows

Production payroll pipelines must handle upstream data degradation without corrupting compliance state. Implement explicit fallback routing that isolates failures and triggers emergency pause workflows when validation thresholds are breached.

• Quarantine Routing: Records failing schema validation or category mapping route to a dedicated aca_quarantine queue. They do not block batch processing.
• Threshold Breach Pause: If >5% of records in a single payroll run are quarantined, the pipeline triggers an EMERGENCY_PAUSE state. Processing halts until manual review clears the backlog.
• Idempotent Retries: All normalization and aggregation functions must accept a retry_token parameter. Duplicate payloads are deduplicated via record_hash before execution.

class PipelineOrchestrator:
    QUARANTINE_THRESHOLD = 0.05  # 5% failure rate triggers pause

    def process_batch(self, records: List[TimeRecord], normalizer: HOSNormalizer) -> bool:
        if not records:
            return False

        accepted_count = 0
        for rec in records:
            _, audit = normalizer.validate_and_normalize(rec)
            if audit.status == ProcessingStatus.ACCEPTED:
                accepted_count += 1

        failure_rate = 1 - (accepted_count / len(records))
        if failure_rate >= self.QUARANTINE_THRESHOLD:
            logger.critical(f"EMERGENCY_PAUSE triggered. Failure rate: {failure_rate:.2%}")
            self._trigger_pause_workflow(records)
            return False
        return True

    def _trigger_pause_workflow(self, failed_records: List[TimeRecord]):
        # Integration point for alerting systems and manual review queues
        logger.info("Routing failed batch to compliance review queue.")
        # Implementation depends on internal alerting infrastructure (PagerDuty, Slack, etc.)

External Compliance References

For statutory definitions, threshold updates, and reporting form specifications, consult:

• IRS Employer Shared Responsibility Provisions
• IRS Publication 15-B: Employer’s Tax Guide to Fringe Benefits (HOS Definitions)