diff --git a/docs/doctoring/position-reporting-hierarchy-references.md b/docs/doctoring/position-reporting-hierarchy-references.md new file mode 100644 index 000000000..d442c8d1f --- /dev/null +++ b/docs/doctoring/position-reporting-hierarchy-references.md @@ -0,0 +1,17 @@ +# Position reporting hierarchy references + +Checked 2026-08-23. These sources inform the active PR #94 design without claiming that an external standard mandates Orgmetra's exact internal schema. + +## Primary and authoritative sources + +HR Open Standards Consortium. (2026). *About HR Open*. https://www.hropenstandards.org/about-hr-open + +- HR Open describes its specifications as voluntary consensus standards for human-resource-related data exchange and interoperability. Orgmetra therefore keeps the new reporting contract modular and does not bind internal Position reporting truth to a vendor-specific worker-manager payload. This source does **not** establish that a particular `reports_to` field or table is mandatory. + +Python Software Foundation. (2026). *datetime — Basic date and time types* (Python 3.14 documentation). https://docs.python.org/3.14/library/datetime.html + +- The official `datetime` contract states that timezone-aware behavior delegates to `tzinfo.utcoffset()` and that an unknown offset may be represented by `None`. The reporting snapshot boundary therefore resolves the caller-owned offset once, normalizes it to a built-in UTC instant, and fails closed when the offset is absent or resolution raises, instead of repeatedly executing caller-owned timezone behavior during bitemporal comparisons. + +## Internal design constraint + +Protected Orgmetra architecture is the authoritative source for the distinction among Job, Position and Assignment. PR #94 adds only the missing Position-to-Position solid-line relationship reconstruction. It does not infer a manager Person from an Assignment and does not reinterpret organization-unit parentage as supervisory authority. \ No newline at end of file diff --git a/docs/traceability/position-reporting-hierarchy.md b/docs/traceability/position-reporting-hierarchy.md new file mode 100644 index 000000000..a85f7ae8a --- /dev/null +++ b/docs/traceability/position-reporting-hierarchy.md @@ -0,0 +1,41 @@ +# Position reporting hierarchy traceability + +## Protected-main truth and parent adoption + +The feature was originally proved against `develop@9e3e4847510e1e612b48474ba42b177b8ed824df`, where Orgmetra separated Job, Position and Assignment and stored organization-unit parentage but had no position-to-position reporting fact. Current protected truth is `develop@eb9757f8649aaad026a9865508d9aad50c1a7a4f`; this branch adopts that protected revision non-destructively while preserving the position-reporting delta. Protected #161 consolidates repository workflows but does not add authoritative position-reporting persistence or mutation, so the buyer gap remains after parent adoption. + +## Active PR + +PR #94 adds an in-memory HRIS-kernel contract for bitemporal solid-line Position reporting. `PositionReportingRelationship` binds one subordinate Position to one manager Position with tenant, effective/business-time, and system-recorded-time scope. `build_position_reporting_snapshot(...)` reconstructs one tenant's hierarchy at an explicit coordinate and returns deterministic subordinate-to-manager pairs. + +The active contract fails closed when: + +- one subordinate resolves to two visible solid-line managers; +- either endpoint does not resolve to exactly one same-tenant `active` or `open` Position version; +- a position reports to itself or the visible graph contains a cycle; +- caller-defined relationship/position/date/datetime runtime subclasses attempt to control trust-bearing comparisons; +- the system knowledge timestamp is naive, has no concrete UTC offset, or its timezone implementation raises during offset resolution. + +Routine representations redact position-correlation UUIDs. The snapshot is descriptive organizational evidence, not a Person-manager link and not employment-decision authority. + +## RED and repair evidence + +- RED contract commit: `36f8f7d0605688c95ddebdc6d6f513eb81d4e144`. +- Foundation CI run `32616830004`, job `97138821309`, checked out that exact SHA and failed at `ModuleNotFoundError: No module named 'orgmetra_hris_kernel.position_reporting'` after the existing 171-test collection reached the missing owner boundary. +- Root implementation commit: `f75ef9a7d785229d2e8a11fe3a5257ce40a5e0c8`. +- The first implementation run then exposed an exact-coverage defect rather than a product-behavior failure: Workforce Intelligence run `32616862967`, job `97138902601`, passed all 181 tests but reported 95% coverage for the new module with missing trust-boundary branches. +- Follow-up adversarial regressions cover invalid UUID identity, unresolved/raising timezone offsets, and validation-bypassing PositionVersion runtime subtypes. The production timezone failure path is no longer excluded from coverage. + +Only exact-current-head terminal workflow results may be treated as final GREEN evidence. + +## Accepted architecture + +Reporting authority belongs to Position, not Person. Assignment remains the independent, potentially multiple-membership fact that connects a person/employment to a seat. Organization-unit parentage remains a separate structural hierarchy and must not be repurposed as a manager relationship. + +## Planned / not yet production-complete + +PR #94 deliberately does **not** claim authoritative persistence or mutation. A later bounded owner slice must provide a normalized tenant-qualified persistence model, immutable audit/outbox evidence for reporting-line changes, bitemporal write/correction semantics, purpose-bound authorization, and database-level integrity without duplicating Person or Assignment data. UI organization-chart work is also separate and must receive Product Design/Figma/Storybook/accessibility evidence when it enters scope. + +## Out of scope + +No Keyverse, Naruon, contextual-orchestrator, Semantic Data Portal, or other dedicated-writer repository is modified. No cross-service application-table SQL is introduced. \ No newline at end of file diff --git a/packages/hris-kernel/README.md b/packages/hris-kernel/README.md index 655f998df..f7f768fc3 100644 --- a/packages/hris-kernel/README.md +++ b/packages/hris-kernel/README.md @@ -12,9 +12,12 @@ Use this package to: 6. Build a CloudEvents 1.0-compatible audit/outbox envelope that carries tenant, actor, purpose, reason, evidence version, result, and accountable human-confirmation references without copying mutable HR payload fields into a shadow system of record. 7. Build a deterministic `JobAnalysisSnapshot` that links observable Tasks to explicit KSAO requirements, retains source/version/digest provenance, and optionally carries historical Functional Job Analysis Data/People/Things codes without confusing Job with Position or Assignment. 8. Build a deterministic `WorkforceCompositionSnapshot` for one tenant, effective day, and recorded-time cutoff with distinct-person headcount, reportable employment count, staffed assignment count/FTE, unassigned-person count, and status counts without serializing row-level person, employment, position, or assignment identifiers. +9. Build a deterministic solid-line `PositionReportingSnapshot` for one tenant, effective day, and system-knowledge cutoff without deriving supervision from Person, Assignment, or organization-unit parentage. Every historical reconstruction and portfolio/capacity decision requires an explicit `tenant_record_id`. A colliding durable identifier from another tenant is ignored rather than treated as local employment truth. +`PositionReportingRelationship` represents managerial structure between durable Position seats. Both endpoints must resolve to exactly one same-tenant `active` or `open` `PositionVersion` at the requested business/system coordinate. One subordinate can have only one visible solid-line manager, self-reporting and cycles fail closed, and caller-defined runtime subclasses cannot control identity or temporal comparisons. `build_position_reporting_snapshot(...)` returns deterministic subordinate-to-manager UUID pairs while its routine `repr` redacts those correlation identifiers. This is descriptive organizational evidence only: it neither identifies the worker occupying a seat nor grants employment-decision authority. Persistence and reporting-line mutation remain separate authoritative write-boundary work. + `WorkforceCompositionSnapshot` is descriptive reporting evidence, not an employment recommendation or decision. It derives `active` and `leave` workforce composition from the same authoritative bitemporal facts and assignment-integrity rules used by the HRIS kernel. Concurrent employments count one person once for headcount while employment count and staffed FTE retain the actual portfolio shape. Contradictory visible versions, duplicate assignment identities, invalid coverage, and over-allocation fail closed instead of becoming plausible-looking metrics. Canonical snapshot JSON carries only the opaque tenant identifier, report coordinate, aggregate metrics, and schema version; `content_digest()` addresses those exact UTF-8 bytes with SHA-256. `AuditOutboxEvent` fails closed on runtime type confusion, reserved nil UUID identities, ambiguous occurrence time, one-word/noncanonical source-service identifiers, malformed event types, free-text data placed in opaque-reference fields, noncanonical purpose/reason/result codes, whitespace-bearing evidence-version tokens, and missing confirmation for high-impact events. Source services use two-or-more-word `snake_case`; event types use the lower-case `orgmetra..` namespace; resource, actor, and confirmation identifiers use namespaced opaque references rather than human-readable payload text. @@ -25,4 +28,4 @@ Every historical reconstruction and portfolio/capacity decision requires an expl `AuditOutboxEvent.canonical_json()` is the exact deterministic JSON text that the owning service persists. `AuditOutboxEvent.content_digest()` is SHA-256 over the UTF-8 bytes of that exact text. Callers must not independently serialize `to_cloudevent()` with library defaults and then assume the digest still addresses the stored representation. The Orgmetra PostgreSQL persistence boundary in migration `0003_audit_outbox_persistence.sql` reparses and allowlists that envelope, verifies tenant/event identity and high-impact confirmation, recomputes the digest over the supplied bytes, writes immutable `audit_event_record` evidence, and creates separate `outbox_delivery_record` transport state. The owning service calls `record_audit_outbox_event(...)` inside the same transaction as its business mutation. -This kernel itself does not talk to PostgreSQL, Keyverse, O*NET, or any other service. Persistence, authorization, source retrieval, and UI stay at their adapter boundaries. Production dispatcher claiming, retry scheduling, lease-expiry recovery, retention/export, external delivery receipts, job-analysis persistence, SME workflow, workforce-report authorization/presentation, and selection-validity computation remain separately proven integration/operability work. +This kernel itself does not talk to PostgreSQL, Keyverse, O*NET, or any other service. Persistence, authorization, source retrieval, and UI stay at their adapter boundaries. Production dispatcher claiming, retry scheduling, lease-expiry recovery, retention/export, external delivery receipts, job-analysis persistence, SME workflow, workforce-report authorization/presentation, position-reporting persistence/mutation, and selection-validity computation remain separately proven integration/operability work. \ No newline at end of file diff --git a/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py b/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py index 5d4b720fa..b47c9128d 100644 --- a/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py +++ b/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py @@ -45,6 +45,12 @@ TaskKSAOLink, ) from orgmetra_hris_kernel.organization import validate_organization_hierarchy +from orgmetra_hris_kernel.position_reporting import ( + PositionReportingHierarchyError, + PositionReportingRelationship, + PositionReportingSnapshot, + build_position_reporting_snapshot, +) from orgmetra_hris_kernel.resolution import resolve_bitemporal_facts, resolve_single_valued_fact from orgmetra_hris_kernel.workforce import ( WorkforceCompositionSnapshot, @@ -70,6 +76,9 @@ "OrganizationHierarchyError", "OrganizationUnitVersion", "PositionCoverageError", + "PositionReportingHierarchyError", + "PositionReportingRelationship", + "PositionReportingSnapshot", "PositionSeatError", "PositionVersion", "RecordedInterval", @@ -77,6 +86,7 @@ "TaskEvidence", "TaskKSAOLink", "WorkforceCompositionSnapshot", + "build_position_reporting_snapshot", "build_workforce_composition_snapshot", "close_recorded_interval", "resolve_bitemporal_facts", diff --git a/packages/hris-kernel/src/orgmetra_hris_kernel/position_reporting.py b/packages/hris-kernel/src/orgmetra_hris_kernel/position_reporting.py new file mode 100644 index 000000000..d749d2a6d --- /dev/null +++ b/packages/hris-kernel/src/orgmetra_hris_kernel/position_reporting.py @@ -0,0 +1,254 @@ +"""Tenant-scoped bitemporal solid-line reporting relationships between positions. + +Reporting authority belongs to Position rather than Person. Assignments can change +without rewriting the position hierarchy, and the same historical chart can be +reconstructed at an explicit business date and system-knowledge cutoff. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone +from uuid import UUID + +from orgmetra_hris_kernel.errors import KernelError +from orgmetra_hris_kernel.facts import PositionVersion +from orgmetra_hris_kernel.intervals import DateInterval, RecordedInterval + +_STAFFABLE_POSITION_STATUSES = frozenset({"active", "open"}) +_SOLID_LINE = "solid_line" + + +class PositionReportingHierarchyError(KernelError): + """Position reporting evidence is ambiguous, cyclic, or outside staffable scope.""" + + +def _require_uuid(value: UUID, field_name: str) -> None: + """Require an exact non-sentinel operational UUID at the reporting boundary.""" + if type(value) is not UUID or value.int in (0, (1 << 128) - 1): + raise PositionReportingHierarchyError( + f"{field_name} must be an exact non-sentinel UUID.", + next_action="Resolve the tenant and position identities again, then rebuild the reporting chart.", + ) + + +def _freeze_known_at(value: datetime) -> datetime: + """Detach one caller-provided timezone offset into an exact built-in UTC instant.""" + if type(value) is not datetime: + raise PositionReportingHierarchyError( + "known_at must be an exact built-in datetime.", + next_action="Use the authoritative UTC system-knowledge timestamp, then rebuild the chart.", + ) + if value.tzinfo is None: + raise PositionReportingHierarchyError( + "known_at must be a timezone-aware datetime.", + next_action="Attach the authoritative timezone or convert the knowledge cutoff to UTC.", + ) + try: + offset = value.utcoffset() + except Exception as exc: + raise PositionReportingHierarchyError( + "known_at timezone could not be resolved safely.", + next_action="Convert the knowledge cutoff to a fixed UTC timestamp before rebuilding the chart.", + ) from exc + if type(offset) is not timedelta: + raise PositionReportingHierarchyError( + "known_at must have one concrete UTC offset.", + next_action="Convert the knowledge cutoff to a fixed UTC timestamp before rebuilding the chart.", + ) + wall_time = datetime( + value.year, + value.month, + value.day, + value.hour, + value.minute, + value.second, + value.microsecond, + tzinfo=timezone.utc, + fold=value.fold, + ) + try: + return wall_time - offset + except (OverflowError, ValueError) as exc: + raise PositionReportingHierarchyError( + "known_at cannot be represented as a UTC datetime.", + next_action=( + "Use a representable UTC knowledge cutoff, then rebuild the reporting chart." + ), + ) from exc + + +@dataclass(frozen=True, slots=True, repr=False) +class PositionReportingRelationship: + """One bitemporal solid-line relationship from a subordinate seat to a manager seat.""" + + tenant_record_id: UUID + position_reporting_relationship_id: UUID + subordinate_position_record_id: UUID + manager_position_record_id: UUID + relationship_type_code: str + effective: DateInterval + recorded: RecordedInterval + + def __post_init__(self) -> None: + """Reject malformed or self-referential reporting evidence immediately.""" + _require_uuid(self.tenant_record_id, "tenant_record_id") + _require_uuid( + self.position_reporting_relationship_id, + "position_reporting_relationship_id", + ) + _require_uuid(self.subordinate_position_record_id, "subordinate_position_record_id") + _require_uuid(self.manager_position_record_id, "manager_position_record_id") + if type(self.relationship_type_code) is not str or self.relationship_type_code != _SOLID_LINE: + raise PositionReportingHierarchyError( + "relationship_type_code must be the reviewed solid_line value.", + next_action="Choose the governed solid-line relationship type, then save again.", + ) + if type(self.effective) is not DateInterval or type(self.recorded) is not RecordedInterval: + raise PositionReportingHierarchyError( + "Position reporting requires an exact governed interval pair.", + next_action="Build the relationship from Orgmetra DateInterval and RecordedInterval values.", + ) + if self.subordinate_position_record_id == self.manager_position_record_id: + raise PositionReportingHierarchyError( + "A position cannot report to itself.", + next_action="Select a different manager position, then save the reporting relationship again.", + ) + + def __repr__(self) -> str: + """Keep opaque position-correlation identifiers out of routine logs.""" + return "" + + +@dataclass(frozen=True, slots=True, repr=False) +class PositionReportingSnapshot: + """Deterministic solid-line position hierarchy at one business/system coordinate.""" + + tenant_record_id: UUID + effective_on: date + known_at: datetime + manager_by_subordinate: tuple[tuple[UUID, UUID], ...] + + def __repr__(self) -> str: + """Redact reporting edges from routine logging and assertion output.""" + return "" + + +def _require_staffable_position( + position_versions: list[PositionVersion], + *, + tenant_record_id: UUID, + position_record_id: UUID, + effective_on: date, + known_at: datetime, +) -> None: + """Require exactly one visible active/open version for one reporting endpoint.""" + visible = [ + version + for version in position_versions + if version.tenant_record_id == tenant_record_id + and version.position_record_id == position_record_id + and version.effective.contains(effective_on) + and version.recorded.contains(known_at) + ] + if len(visible) != 1 or visible[0].position_status_code not in _STAFFABLE_POSITION_STATUSES: + raise PositionReportingHierarchyError( + "A visible reporting edge must reference exactly one staffable position version in this tenant.", + next_action="Open or correct both position seats at this business/system coordinate, then rebuild the chart.", + ) + + +def build_position_reporting_snapshot( + relationships: list[PositionReportingRelationship], + position_versions: list[PositionVersion], + *, + tenant_record_id: UUID, + effective_on: date, + known_at: datetime, +) -> PositionReportingSnapshot: + """Build one deterministic, cycle-free solid-line position hierarchy. + + Args: + relationships: Candidate reporting facts, including other tenants and history. + position_versions: Candidate position versions used to prove each visible endpoint exists. + tenant_record_id: Tenant whose reporting hierarchy is reconstructed. + effective_on: Business date represented by the hierarchy. + known_at: System-knowledge cutoff represented by the hierarchy. + + Returns: + A redacted snapshot whose ordered pairs are `(subordinate_position, manager_position)`. + + Raises: + PositionReportingHierarchyError: Reporting evidence is malformed, ambiguous, + cyclic, or references a non-staffable position at the requested coordinate. + """ + _require_uuid(tenant_record_id, "tenant_record_id") + if type(effective_on) is not date: + raise PositionReportingHierarchyError( + "effective_on must be an exact built-in date.", + next_action="Use the authoritative HR business date, then rebuild the reporting chart.", + ) + frozen_known_at = _freeze_known_at(known_at) + + for value in relationships: + if type(value) is not PositionReportingRelationship: + raise PositionReportingHierarchyError( + "Position reporting snapshots accept only the exact governed relationship runtime type.", + next_action="Reconstruct the relationship through the governed Orgmetra reporting boundary.", + ) + for version in position_versions: + if type(version) is not PositionVersion: + raise PositionReportingHierarchyError( + "Position reporting snapshots accept only exact PositionVersion evidence.", + next_action="Resolve authoritative position versions again, then rebuild the reporting chart.", + ) + + visible = [ + relationship + for relationship in relationships + if relationship.tenant_record_id == tenant_record_id + and relationship.effective.contains(effective_on) + and relationship.recorded.contains(frozen_known_at) + ] + + manager_by_subordinate: dict[UUID, UUID] = {} + verified_positions: set[UUID] = set() + for relationship in visible: + subordinate = relationship.subordinate_position_record_id + manager = relationship.manager_position_record_id + if subordinate in manager_by_subordinate: + raise PositionReportingHierarchyError( + "A position resolves to more than one solid-line manager at this coordinate.", + next_action="Close or correct the superseded reporting relationship, then rebuild the chart.", + ) + for position_record_id in (subordinate, manager): + if position_record_id not in verified_positions: + _require_staffable_position( + position_versions, + tenant_record_id=tenant_record_id, + position_record_id=position_record_id, + effective_on=effective_on, + known_at=frozen_known_at, + ) + verified_positions.add(position_record_id) + manager_by_subordinate[subordinate] = manager + + for start in manager_by_subordinate: + seen: set[UUID] = set() + current: UUID | None = start + while current is not None: + if current in seen: + raise PositionReportingHierarchyError( + "Visible solid-line position reporting relationships form a cycle in this tenant.", + next_action="Close or correct one reporting edge in the cycle, then rebuild the chart.", + ) + seen.add(current) + current = manager_by_subordinate.get(current) + + ordered = tuple(sorted(manager_by_subordinate.items(), key=lambda pair: pair[0].int)) + return PositionReportingSnapshot( + tenant_record_id=tenant_record_id, + effective_on=effective_on, + known_at=frozen_known_at, + manager_by_subordinate=ordered, + ) diff --git a/packages/hris-kernel/tests/test_position_reporting.py b/packages/hris-kernel/tests/test_position_reporting.py new file mode 100644 index 000000000..79d51e458 --- /dev/null +++ b/packages/hris-kernel/tests/test_position_reporting.py @@ -0,0 +1,286 @@ +"""Executable contract for tenant-scoped bitemporal position reporting.""" + +from datetime import date, datetime, timezone +from uuid import UUID + +import pytest + +from orgmetra_hris_kernel.facts import PositionVersion +from orgmetra_hris_kernel.intervals import DateInterval, RecordedInterval +from orgmetra_hris_kernel.position_reporting import ( + PositionReportingHierarchyError, + PositionReportingRelationship, + build_position_reporting_snapshot, +) + +TENANT_ALPHA = UUID("018f0d35-7b1a-7cc2-8d9c-111111111111") +TENANT_BETA = UUID("018f0d35-7b1a-7cc2-8d9c-222222222222") +POSITION_A = UUID("018f0d35-7b1a-7cc2-8d9c-aaaaaaaaaaa1") +POSITION_B = UUID("018f0d35-7b1a-7cc2-8d9c-aaaaaaaaaaa2") +POSITION_C = UUID("018f0d35-7b1a-7cc2-8d9c-aaaaaaaaaaa3") +POSITION_D = UUID("018f0d35-7b1a-7cc2-8d9c-aaaaaaaaaaa4") +RELATIONSHIP_A = UUID("018f0d35-7b1a-4cc2-8d9c-bbbbbbbbbbb1") +RELATIONSHIP_B = UUID("018f0d35-7b1a-4cc2-8d9c-bbbbbbbbbbb2") +RELATIONSHIP_C = UUID("018f0d35-7b1a-4cc2-8d9c-bbbbbbbbbbb3") +VERSION_A = UUID("018f0d35-7b1a-7cc2-8d9c-ccccccccccc1") +VERSION_B = UUID("018f0d35-7b1a-7cc2-8d9c-ccccccccccc2") +VERSION_C = UUID("018f0d35-7b1a-7cc2-8d9c-ccccccccccc3") +VERSION_D = UUID("018f0d35-7b1a-7cc2-8d9c-ccccccccccc4") +EFFECTIVE_ON = date(2026, 8, 23) +KNOWN_AT = datetime(2026, 8, 23, 3, 30, tzinfo=timezone.utc) + + +def position( + position_record_id: UUID, + position_record_version_id: UUID, + *, + tenant_record_id: UUID = TENANT_ALPHA, + status: str = "active", + effective: DateInterval | None = None, + recorded: RecordedInterval | None = None, +) -> PositionVersion: + """Build one visible position version for reporting-contract tests.""" + return PositionVersion( + tenant_record_id=tenant_record_id, + position_record_id=position_record_id, + position_record_version_id=position_record_version_id, + position_status_code=status, + effective=effective or DateInterval(date(2026, 1, 1)), + recorded=recorded + or RecordedInterval(datetime(2026, 1, 1, tzinfo=timezone.utc)), + ) + + +def relationship( + relationship_id: UUID, + subordinate: UUID, + manager: UUID, + *, + tenant_record_id: UUID = TENANT_ALPHA, + effective: DateInterval | None = None, + recorded: RecordedInterval | None = None, +) -> PositionReportingRelationship: + """Build one solid-line reporting relationship for tests.""" + return PositionReportingRelationship( + tenant_record_id=tenant_record_id, + position_reporting_relationship_id=relationship_id, + subordinate_position_record_id=subordinate, + manager_position_record_id=manager, + relationship_type_code="solid_line", + effective=effective or DateInterval(date(2026, 1, 1)), + recorded=recorded + or RecordedInterval(datetime(2026, 1, 1, tzinfo=timezone.utc)), + ) + + +def visible_positions() -> list[PositionVersion]: + """Return three staffable Alpha positions visible at the review coordinate.""" + return [ + position(POSITION_A, VERSION_A), + position(POSITION_B, VERSION_B), + position(POSITION_C, VERSION_C), + ] + + +def test_builds_deterministic_position_reporting_snapshot() -> None: + """Acyclic solid-line edges resolve to deterministic position-to-position evidence.""" + snapshot = build_position_reporting_snapshot( + [ + relationship(RELATIONSHIP_B, POSITION_C, POSITION_B), + relationship(RELATIONSHIP_A, POSITION_B, POSITION_A), + ], + visible_positions(), + tenant_record_id=TENANT_ALPHA, + effective_on=EFFECTIVE_ON, + known_at=KNOWN_AT, + ) + + assert snapshot.tenant_record_id == TENANT_ALPHA + assert snapshot.effective_on == EFFECTIVE_ON + assert snapshot.known_at == KNOWN_AT + assert snapshot.manager_by_subordinate == ( + (POSITION_B, POSITION_A), + (POSITION_C, POSITION_B), + ) + assert "aaaaaaaa" not in repr(snapshot) + + +def test_ignores_other_tenants_and_nonvisible_relationships() -> None: + """Foreign and not-yet-effective edges cannot enter one tenant's reporting chart.""" + snapshot = build_position_reporting_snapshot( + [ + relationship( + RELATIONSHIP_A, + POSITION_B, + POSITION_A, + tenant_record_id=TENANT_BETA, + ), + relationship( + RELATIONSHIP_B, + POSITION_C, + POSITION_B, + effective=DateInterval(date(2027, 1, 1)), + ), + ], + visible_positions(), + tenant_record_id=TENANT_ALPHA, + effective_on=EFFECTIVE_ON, + known_at=KNOWN_AT, + ) + + assert snapshot.manager_by_subordinate == () + + +def test_rejects_two_visible_managers_for_one_subordinate() -> None: + """One solid-line subordinate cannot resolve to two managers at one coordinate.""" + with pytest.raises(PositionReportingHierarchyError, match="more than one solid-line manager"): + build_position_reporting_snapshot( + [ + relationship(RELATIONSHIP_A, POSITION_C, POSITION_A), + relationship(RELATIONSHIP_B, POSITION_C, POSITION_B), + ], + visible_positions(), + tenant_record_id=TENANT_ALPHA, + effective_on=EFFECTIVE_ON, + known_at=KNOWN_AT, + ) + + +def test_rejects_visible_reporting_cycle() -> None: + """A position reporting chain must fail closed when it cycles.""" + with pytest.raises(PositionReportingHierarchyError, match="form a cycle"): + build_position_reporting_snapshot( + [ + relationship(RELATIONSHIP_A, POSITION_A, POSITION_B), + relationship(RELATIONSHIP_B, POSITION_B, POSITION_A), + ], + visible_positions(), + tenant_record_id=TENANT_ALPHA, + effective_on=EFFECTIVE_ON, + known_at=KNOWN_AT, + ) + + +def test_rejects_self_reporting_at_construction() -> None: + """A position cannot be its own direct solid-line manager.""" + with pytest.raises(PositionReportingHierarchyError, match="cannot report to itself"): + relationship(RELATIONSHIP_A, POSITION_A, POSITION_A) + + +def test_rejects_missing_or_nonstaffable_endpoint_position() -> None: + """Both ends of a visible reporting edge must resolve to staffable tenant seats.""" + with pytest.raises(PositionReportingHierarchyError, match="staffable position"): + build_position_reporting_snapshot( + [relationship(RELATIONSHIP_A, POSITION_B, POSITION_D)], + visible_positions(), + tenant_record_id=TENANT_ALPHA, + effective_on=EFFECTIVE_ON, + known_at=KNOWN_AT, + ) + + closed_positions = visible_positions() + [position(POSITION_D, VERSION_D, status="closed")] + with pytest.raises(PositionReportingHierarchyError, match="staffable position"): + build_position_reporting_snapshot( + [relationship(RELATIONSHIP_A, POSITION_B, POSITION_D)], + closed_positions, + tenant_record_id=TENANT_ALPHA, + effective_on=EFFECTIVE_ON, + known_at=KNOWN_AT, + ) + + +def test_rejects_invalid_relationship_primitives() -> None: + """Trust-bearing relationship primitives fail closed before graph traversal.""" + with pytest.raises(PositionReportingHierarchyError, match="relationship_type_code"): + PositionReportingRelationship( + tenant_record_id=TENANT_ALPHA, + position_reporting_relationship_id=RELATIONSHIP_A, + subordinate_position_record_id=POSITION_B, + manager_position_record_id=POSITION_A, + relationship_type_code="dotted_line", + effective=DateInterval(date(2026, 1, 1)), + recorded=RecordedInterval(datetime(2026, 1, 1, tzinfo=timezone.utc)), + ) + + with pytest.raises(PositionReportingHierarchyError, match="exact governed interval"): + PositionReportingRelationship( + tenant_record_id=TENANT_ALPHA, + position_reporting_relationship_id=RELATIONSHIP_A, + subordinate_position_record_id=POSITION_B, + manager_position_record_id=POSITION_A, + relationship_type_code="solid_line", + effective=object(), # type: ignore[arg-type] + recorded=RecordedInterval(datetime(2026, 1, 1, tzinfo=timezone.utc)), + ) + + +def test_rejects_untrusted_runtime_types_at_snapshot_boundary() -> None: + """Caller-defined runtime subclasses cannot control identity or temporal comparisons.""" + + class ForgedDate(date): + """Caller-controlled date subtype rejected before interval comparison.""" + + class ForgedRelationship(PositionReportingRelationship): + """Validation-bypassing relationship subtype rejected at the snapshot boundary.""" + + base = relationship(RELATIONSHIP_A, POSITION_B, POSITION_A) + forged = object.__new__(ForgedRelationship) + for field_name in ( + "tenant_record_id", + "position_reporting_relationship_id", + "subordinate_position_record_id", + "manager_position_record_id", + "relationship_type_code", + "effective", + "recorded", + ): + object.__setattr__(forged, field_name, getattr(base, field_name)) + + with pytest.raises(PositionReportingHierarchyError, match="exact governed relationship"): + build_position_reporting_snapshot( + [forged], + visible_positions(), + tenant_record_id=TENANT_ALPHA, + effective_on=EFFECTIVE_ON, + known_at=KNOWN_AT, + ) + + with pytest.raises(PositionReportingHierarchyError, match="exact built-in date"): + build_position_reporting_snapshot( + [], + visible_positions(), + tenant_record_id=TENANT_ALPHA, + effective_on=ForgedDate(2026, 8, 23), + known_at=KNOWN_AT, + ) + + +def test_rejects_naive_or_subclassed_system_time() -> None: + """System knowledge coordinates must be exact built-in timezone-aware datetimes.""" + + class ForgedDateTime(datetime): + """Caller-controlled datetime subtype rejected before recorded-time comparison.""" + + with pytest.raises(PositionReportingHierarchyError, match="timezone-aware datetime"): + build_position_reporting_snapshot( + [], + visible_positions(), + tenant_record_id=TENANT_ALPHA, + effective_on=EFFECTIVE_ON, + known_at=datetime(2026, 8, 23, 3, 30), + ) + + with pytest.raises(PositionReportingHierarchyError, match="exact built-in datetime"): + build_position_reporting_snapshot( + [], + visible_positions(), + tenant_record_id=TENANT_ALPHA, + effective_on=EFFECTIVE_ON, + known_at=ForgedDateTime(2026, 8, 23, 3, 30, tzinfo=timezone.utc), + ) + + +def test_relationship_repr_is_redacted() -> None: + """Routine logs do not expose position-correlation UUIDs.""" + value = relationship(RELATIONSHIP_A, POSITION_B, POSITION_A) + assert repr(value) == "" diff --git a/packages/hris-kernel/tests/test_position_reporting_hardening.py b/packages/hris-kernel/tests/test_position_reporting_hardening.py new file mode 100644 index 000000000..3fd4bffe2 --- /dev/null +++ b/packages/hris-kernel/tests/test_position_reporting_hardening.py @@ -0,0 +1,97 @@ +"""Adversarial runtime-type regressions for position-reporting evidence.""" + +from datetime import date, datetime, timezone, tzinfo +from uuid import UUID + +import pytest + +from orgmetra_hris_kernel.facts import PositionVersion +from orgmetra_hris_kernel.intervals import DateInterval, RecordedInterval +from orgmetra_hris_kernel.position_reporting import ( + PositionReportingHierarchyError, + PositionReportingRelationship, + build_position_reporting_snapshot, +) + +TENANT = UUID("018f0d35-7b1a-7cc2-8d9c-111111111111") +POSITION_A = UUID("018f0d35-7b1a-7cc2-8d9c-aaaaaaaaaaa1") +VERSION_A = UUID("018f0d35-7b1a-7cc2-8d9c-ccccccccccc1") +EFFECTIVE_ON = date(2026, 8, 23) +KNOWN_AT = datetime(2026, 8, 23, 3, 30, tzinfo=timezone.utc) + + +def test_rejects_non_uuid_identity_before_relationship_validation() -> None: + """Text that merely looks like an identifier cannot become governed UUID evidence.""" + with pytest.raises(PositionReportingHierarchyError, match="exact non-sentinel UUID"): + PositionReportingRelationship( + tenant_record_id="018f0d35-7b1a-7cc2-8d9c-111111111111", # type: ignore[arg-type] + position_reporting_relationship_id=UUID("018f0d35-7b1a-4cc2-8d9c-bbbbbbbbbbb1"), + subordinate_position_record_id=POSITION_A, + manager_position_record_id=UUID("018f0d35-7b1a-7cc2-8d9c-aaaaaaaaaaa2"), + relationship_type_code="solid_line", + effective=DateInterval(date(2026, 1, 1)), + recorded=RecordedInterval(datetime(2026, 1, 1, tzinfo=timezone.utc)), + ) + + +def test_rejects_timezone_without_concrete_utc_offset() -> None: + """A tzinfo object that cannot resolve one offset cannot drive bitemporal reconstruction.""" + + class OffsetlessTimezone(tzinfo): + """Timezone fixture whose offset contract deliberately resolves to None.""" + + def utcoffset(self, dt: datetime | None) -> None: + """Return no usable UTC offset.""" + return None + + def dst(self, dt: datetime | None) -> None: + """Return no daylight-saving offset.""" + return None + + def tzname(self, dt: datetime | None) -> str: + """Return a diagnostic fixture name.""" + return "offsetless" + + with pytest.raises(PositionReportingHierarchyError, match="concrete UTC offset"): + build_position_reporting_snapshot( + [], + [], + tenant_record_id=TENANT, + effective_on=EFFECTIVE_ON, + known_at=datetime(2026, 8, 23, 3, 30, tzinfo=OffsetlessTimezone()), + ) + + +def test_rejects_position_version_subclass_before_field_access() -> None: + """A validation-bypassing PositionVersion subtype cannot supply endpoint evidence.""" + + class ForgedPositionVersion(PositionVersion): + """Caller-defined position subtype rejected by the exact-type boundary.""" + + original = PositionVersion( + tenant_record_id=TENANT, + position_record_id=POSITION_A, + position_record_version_id=VERSION_A, + position_status_code="active", + effective=DateInterval(date(2026, 1, 1)), + recorded=RecordedInterval(datetime(2026, 1, 1, tzinfo=timezone.utc)), + ) + forged = object.__new__(ForgedPositionVersion) + for field_name in ( + "tenant_record_id", + "position_record_id", + "position_record_version_id", + "position_status_code", + "effective", + "recorded", + ): + object.__setattr__(forged, field_name, getattr(original, field_name)) + + with pytest.raises(PositionReportingHierarchyError, match="exact PositionVersion"): + build_position_reporting_snapshot( + [], + [forged], + tenant_record_id=TENANT, + effective_on=EFFECTIVE_ON, + known_at=KNOWN_AT, + ) \ No newline at end of file diff --git a/packages/hris-kernel/tests/test_position_reporting_public_api.py b/packages/hris-kernel/tests/test_position_reporting_public_api.py new file mode 100644 index 000000000..361200d49 --- /dev/null +++ b/packages/hris-kernel/tests/test_position_reporting_public_api.py @@ -0,0 +1,17 @@ +"""Executable contract for the public position-reporting package surface.""" + +import orgmetra_hris_kernel as kernel + + +def test_position_reporting_contract_is_exported_from_package_root() -> None: + """Position-reporting users can import the governed contract from the package root.""" + expected_exports = { + "PositionReportingHierarchyError", + "PositionReportingRelationship", + "PositionReportingSnapshot", + "build_position_reporting_snapshot", + } + + assert expected_exports <= set(kernel.__all__) + for export_name in expected_exports: + assert getattr(kernel, export_name).__module__ == "orgmetra_hris_kernel.position_reporting" diff --git a/packages/hris-kernel/tests/test_position_reporting_timezone_failure.py b/packages/hris-kernel/tests/test_position_reporting_timezone_failure.py new file mode 100644 index 000000000..972e42b5f --- /dev/null +++ b/packages/hris-kernel/tests/test_position_reporting_timezone_failure.py @@ -0,0 +1,54 @@ +"""Failure-path coverage for position-reporting system-time normalization.""" + +from datetime import date, datetime, timedelta, timezone, tzinfo +from uuid import UUID + +import pytest + +from orgmetra_hris_kernel.position_reporting import ( + PositionReportingHierarchyError, + build_position_reporting_snapshot, +) + + +class ExplodingTimezone(tzinfo): + """Timezone fixture that refuses to resolve an offset.""" + + def utcoffset(self, dt: datetime | None): + """Raise to emulate a hostile or broken caller-owned timezone object.""" + raise RuntimeError("untrusted timezone code must not escape") + + def dst(self, dt: datetime | None): + """Return no daylight-saving offset because it is not needed by the regression.""" + return None + + def tzname(self, dt: datetime | None) -> str: + """Return a bounded diagnostic fixture name.""" + return "exploding" + + +def test_timezone_exception_is_normalized_to_governed_error() -> None: + """Caller-owned timezone exceptions cannot escape the reporting boundary.""" + with pytest.raises(PositionReportingHierarchyError, match="could not be resolved safely"): + build_position_reporting_snapshot( + [], + [], + tenant_record_id=UUID("018f0d35-7b1a-7cc2-8d9c-111111111111"), + effective_on=date(2026, 8, 23), + known_at=datetime(2026, 8, 23, 3, 30, tzinfo=ExplodingTimezone()), + ) + + +def test_unrepresentable_utc_conversion_is_normalized_to_governed_error() -> None: + """UTC normalization overflow cannot escape the reporting boundary.""" + with pytest.raises( + PositionReportingHierarchyError, + match="represented as a UTC datetime", + ): + build_position_reporting_snapshot( + [], + [], + tenant_record_id=UUID("018f0d35-7b1a-7cc2-8d9c-111111111111"), + effective_on=date(2026, 8, 23), + known_at=datetime.max.replace(tzinfo=timezone(timedelta(hours=-14))), + )