diff --git a/docs/adr/0152-purpose-bound-position-history-read.md b/docs/adr/0152-purpose-bound-position-history-read.md new file mode 100644 index 000000000..1f2a124cd --- /dev/null +++ b/docs/adr/0152-purpose-bound-position-history-read.md @@ -0,0 +1,67 @@ +# ADR 0152: Purpose-bound bitemporal Position history read + +- **Status:** Accepted for active PR #152; not protected-main truth until integrated +- **Date:** 2026-08-30 +- **Owners:** Orgmetra People domain + +## Context + +Protected `develop` already models `position_record` separately from `job_profile`, `assignment_record`, and their time-varying facts. It also stores Position versions with business-effective (`effective_from`, `effective_to`) and system-recorded (`recorded_from`, `recorded_to`) intervals. A commercial HRIS still needs a buyer-visible way to inspect how one Position was understood over time without widening the read to Person, Assignment, compensation, candidate, or employment-decision data. + +A Position-history read is a high-value governance boundary because historical workforce interpretation is frequently used in reorganizations, audit, workforce planning, and downstream validity analysis. Returning persistence rows directly would make authorization order, tenant scope, bitemporal interpretation, and field minimization depend on adapter behavior instead of the domain contract. + +## Decision + +Orgmetra adds a read-only `position_history` application boundary in the People service. + +1. The caller supplies an exact operational tenant UUID, an exact operational Position UUID, an exact built-in UTC knowledge instant, a declared purpose, and an explicit requested-field set. +2. Purpose-bound authorization is evaluated **before** the injected read port may retrieve protected Position history. +3. The persistence adapter returns immutable `PositionHistoryRecord` values. Application code treats those values as untrusted evidence and revalidates exact row shape, primitive types, tenant and Position identity, system-time visibility, version uniqueness, and business-effective non-overlap. +4. System-recorded intervals are interpreted as half-open intervals: `recorded_from <= known_at < recorded_to`, with an absent `recorded_to` meaning open-ended visibility. +5. Business-effective intervals are also half-open. Two versions visible at the same knowledge instant may not claim overlapping business truth for the same Position. An absent business end is represented as **unbounded**, not by substituting a finite date sentinel such as `date.max`; this preserves overlap semantics even when a valid interval begins on Python's maximum representable date. +6. The response is deterministic and contains only fields explicitly authorized by the purpose-bound policy. Unknown fields and `str` subclasses fail closed rather than reaching reflection-based serialization. +7. `position_record_version_id`, organization lineage, Job lineage, status, business-effective dates, and system-recorded timestamps remain distinct concepts. The read does not collapse Job, Position, or Assignment. +8. The application boundary depends on an injected port. It does not query another service's application tables and does not introduce cross-service SQL. + +## Trust and time semantics + +The service accepts exact built-in UUID/date/datetime/timezone primitives at the trust boundary. Caller-controlled subclasses and timezone implementations are rejected. This prevents user-defined equality, hashing, formatting, or UTC-offset behavior from participating in authorization, chronology, or evidence serialization. + +`known_at` is system-recorded time, not business-effective time. A version may be visible at `known_at` while describing a past or future business-effective period. These dimensions must never be substituted for one another. + +Open-ended business time is a semantic infinity, not the largest finite date representable by one runtime. Overlap therefore uses direct endpoint-presence logic: a left interval is before a right end when the right end is absent or the left start is strictly earlier, and conversely for the right interval. This keeps half-open interval algebra correct at representational extremes and avoids treating `[date.max, ∞)` as empty. + +## Data-model boundary + +This ADR does not change protected-main storage. The existing schema remains authoritative: + +- `job_profile` describes the reusable Job/work content. +- `position_record` is the tenant-owned Position anchor in an organization and references the Job profile. +- `position_record_version` carries Position status and business-effective/system-recorded version evidence. +- `assignment_record` links a worker/employment relationship to a Position and remains a separate lifecycle fact. + +An adapter that materializes Position history must preserve those meanings and may not use the new view to imply that Assignment or Person history is part of a Position version. + +## Consequences + +### Positive + +- HR operators can inspect Position history without broad Person/Assignment disclosure. +- Authorization-before-retrieval is executable and testable. +- Bitemporal contradictions fail closed at the service boundary, including valid extreme-date intervals whose end is genuinely unbounded. +- The module is standalone and can be extracted behind a service/API boundary later without rewriting its authorization and evidence semantics. +- Exact owned statement/branch coverage can be enforced independently of a future database adapter. + +### Trade-offs + +- The read port must deliberately materialize data that the application can validate; adapters cannot return arbitrary ORM entities. +- A database adapter must provide a transactionally coherent snapshot. The application checks cannot replace MVCC/snapshot isolation where concurrent database writes are possible. +- This slice exposes no HTTP route or write mutation. Those are separate bounded decisions and must not be inferred from this ADR. + +## Verification + +PR #152 records a hosted test-first sequence. A test-only head failed because the production Position-history module did not exist. The smallest application implementation then satisfied the behavioral contract, after which a remaining 100%-coverage branch for malformed low-level row reconstruction was covered with an explicit fail-closed regression rather than by excluding code or weakening the gate. + +A later source sweep found that open-ended business intervals were approximated with `date.max` during overlap checks. Test-only head `af8d0b9b88c50f17c87eb8ecf1eea29918835dce` produced genuine hosted RED in People API Quality run `33267978859`, job `99141335635`: 157 existing tests passed, exact owned coverage remained 100%, but the new extreme-date regression failed because `[date.max, ∞)` was incorrectly treated as non-overlapping with an earlier open interval. Root repair `955956f838c467c06c25b63127b7c6e976dea812` removes the finite-infinity sentinel and compares optional interval ends directly. + +The PR remains Draft until the exact current head has fresh applicable local/central evidence and qualifying independent review. Evidence from predecessor heads is non-transferable. diff --git a/docs/doctoring/position-history-read-references.md b/docs/doctoring/position-history-read-references.md new file mode 100644 index 000000000..89597d96e --- /dev/null +++ b/docs/doctoring/position-history-read-references.md @@ -0,0 +1,33 @@ +# Position history read references + +**Scope:** Research and standards basis for active PR #152. This file does not claim certification or protected-main integration. + +The Position-history contract uses established temporal/database and security-control concepts rather than creating Orgmetra-specific substitutes for them. Implementation details remain constrained by the actual protected-main schema and executable tests. + +## APA 7 references + +Joint Task Force. (2020). *Security and privacy controls for information systems and organizations* (NIST Special Publication 800-53, Revision 5). National Institute of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-53r5 + +Klyne, G., & Newman, C. (2002). *Date and time on the Internet: Timestamps* (RFC 3339). RFC Editor. https://doi.org/10.17487/RFC3339 + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: 5.5. Constraints*. https://www.postgresql.org/docs/18/ddl-constraints.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: 8.17. Range types*. https://www.postgresql.org/docs/18/rangetypes.html + +## Decision relevance + +### PostgreSQL range and exclusion semantics + +The protected Orgmetra schema already uses database-level bitemporal constraints. PostgreSQL's range/exclusion facilities provide the primary technical basis for treating effective intervals as non-overlapping business truth where the schema requires it. The application read boundary does not replace those constraints; it independently rejects contradictory adapter output before buyer-visible serialization. + +### RFC 3339 timestamps + +System-recorded evidence is serialized in one UTC RFC 3339 representation (`Z`). This is an interoperability/canonicalization choice. The service separately validates that trust-bearing input is an exact built-in UTC datetime rather than accepting arbitrary caller-controlled timezone implementations that merely produce a zero offset. + +### NIST SP 800-53 Rev. 5 + +The read boundary is designed toward evidence-ready access-control, least-privilege, auditability, and system/information-integrity practices. The design does not claim NIST compliance, SOC 2 certification, CSAP certification, or any external attestation. Purpose-bound authorization and field minimization are product controls whose effectiveness must remain demonstrable through exact-current executable evidence. + +## Research classification + +These references inform accepted architecture for PR #152. They do not authorize scope expansion into Person, Assignment, compensation, candidate, performance, or employment-decision data, and they do not supersede dedicated-writer dependency contracts. diff --git a/docs/traceability/position-history-read.md b/docs/traceability/position-history-read.md new file mode 100644 index 000000000..70f3affe0 --- /dev/null +++ b/docs/traceability/position-history-read.md @@ -0,0 +1,69 @@ +# Position history read traceability + +**Lifecycle status:** Active PR #152. This document describes the PR contract, not protected-main truth until integration. + +## Buyer outcome + +An authorized HR operator can inspect the bitemporal history of one Position for a declared workforce purpose without receiving unrelated Person, Employment, Assignment, candidate, compensation, or decision data. + +## Protected-main prerequisites + +Protected `develop` already provides the authoritative data-model separation needed by this slice: + +| Concern | Protected-main truth used by #152 | +| --- | --- | +| Job | `job_profile` remains reusable Job/work content. | +| Position | `position_record` remains a tenant-owned Position anchor with organization and Job lineage. | +| Position version | `position_record_version` preserves business-effective and system-recorded time. | +| Assignment | `assignment_record` remains distinct from Position and links the worker/employment relationship to a Position. | +| Authorization | People service uses purpose-bound policy evaluation before protected reads. | + +#152 does not add a database migration, mutate those tables, or create cross-service application-table SQL. + +## Requirement-to-evidence matrix + +| Requirement | Implementation boundary | Executable evidence | +| --- | --- | --- | +| Authorize before retrieval | `read_position_history()` calls the purpose-bound authorization boundary before `PositionHistoryReadPort` | denied fields prove the port is never called | +| Tenant/context isolation | exact tenant and Position are rechecked on every returned row | wrong-tenant and wrong-Position rows fail closed | +| Bitemporal system truth | half-open `recorded_from`/`recorded_to` at exact UTC `known_at` | future and already-closed rows fail closed | +| Business-time consistency | visible half-open effective intervals may not overlap; absent ends remain semantically unbounded rather than mapped to `date.max` | ordinary-overlap and `date.max` open-interval regressions fail closed | +| Immutable evidence | exact tuple container and exact `PositionHistoryRecord`; runtime revalidation after low-level reconstruction | unsupported container/type, forged values, and short low-level row all fail closed | +| Opaque identifiers | exact operational UUIDs; nil/max protocol sentinels and subclasses rejected | invalid request/record regressions | +| Field minimization | explicit serializer whitelist over authorized fields only | one-field policy returns one field; unknown/subclass fields fail closed | +| Deterministic history | sort by effective start then version identity | reversed persistence order produces deterministic output | +| Job/Position/Assignment separation | view contains Position/Job lineage only; no worker/Assignment expansion | schema and response contract | +| Exact owned coverage | People API quality workflow | 100% statement and branch gate on exact current head required | + +## Test-first chain + +1. **Initial test-only head:** `d751f117e37e2169015004ab89fa728731b2a7ec`. +2. **Initial hosted RED:** People API Quality run `33267334677`, job `99139623454`, failed during collection because `orgmetra_people_api.position_history` did not exist. +3. **Root implementation:** `f633aa3d008d7832759bb83dead8d4e5a6977a8b` added the smallest Orgmetra-owned read boundary. +4. **Coverage gate held:** run `33267487363`, job `99140037925`, passed all 156 tests but correctly failed exact coverage because one deliberate malformed-row branch remained unexecuted. +5. **Regression strengthening:** `cbb343a40864694ac243946615aee5f91685beda` added a low-level short-row reconstruction regression. +6. **Exact GREEN at that predecessor:** People API Quality run `33267577477`, job `99140279359`: 157 tests; 1,543/1,543 statements; 504/504 branches; compile and clean-checkout GREEN. +7. **Extreme-date integrity RED:** source review found that `_business_intervals_overlap()` substituted finite `date.max` for an absent business end. Test-only head `af8d0b9b88c50f17c87eb8ecf1eea29918835dce` added a valid `[date.max, ∞)` overlap case. People API Quality run `33267978859`, job `99141335635`, checked out that exact SHA and failed exactly that regression: **1 failed / 157 passed** while owned production coverage remained **1,544/1,544 statements and 504/504 branches = 100.00%**. The service returned instead of raising `PositionHistoryIntegrityError`, proving a real business-time integrity defect rather than a coverage artifact. +8. **Extreme-date root repair:** `955956f838c467c06c25b63127b7c6e976dea812` removes the finite-infinity sentinel and compares optional interval endpoints directly. Open-ended intervals therefore remain unbounded even at the maximum representable finite date. + +Documentation commits after the root repair invalidate predecessor GREEN as merge evidence. The final exact PR head must receive its own fresh hosted evidence before advancement. + +## Security/privacy invariants + +- No PII is added to the Position-history response merely because it exists elsewhere in HRIS. +- No dynamic attribute access is used to serialize policy-controlled field names. +- Caller-controlled UUID/string/timezone subclasses do not participate in identity, authorization, chronology, or output canonicalization. +- Persistence is an injected boundary and its output is revalidated. +- Open-ended business-time semantics are represented explicitly; runtime maximum dates are never overloaded as infinity. +- Application checks do not claim to replace database snapshot/MVCC semantics for concurrent writes. + +## Out of scope / planned separately + +- Position-history HTTP presentation. +- Position mutation/correction workflow. +- Assignment or Employment history joins. +- Compensation, candidate, performance, or selection-decision expansion. +- Database-specific Position-history adapter and its transaction-isolation proof. +- Release/version/tag publication. + +Any later slice must keep these concerns bounded and must not infer protected-main availability from this active-PR traceability document. diff --git a/services/people-api/src/orgmetra_people_api/__init__.py b/services/people-api/src/orgmetra_people_api/__init__.py index b043bed33..3313e7f4f 100644 --- a/services/people-api/src/orgmetra_people_api/__init__.py +++ b/services/people-api/src/orgmetra_people_api/__init__.py @@ -40,6 +40,14 @@ WorkerPeopleRecord, read_worker_people_record, ) +from orgmetra_people_api.position_history import ( + AuthorizedPositionHistoryEntry, + AuthorizedPositionHistoryView, + PositionHistoryIntegrityError, + PositionHistoryReadPort, + PositionHistoryRecord, + read_position_history, +) from orgmetra_people_api.postgres import PostgresPeopleReadPort from orgmetra_people_api.postgres_hire import PostgresHireAcceptancePort from orgmetra_people_api.postgres_mutations import PostgresPeopleMutationPort @@ -47,6 +55,8 @@ __all__ = [ "AuthenticatedPrincipal", "AuthenticationFailed", + "AuthorizedPositionHistoryEntry", + "AuthorizedPositionHistoryView", "AuthorizedWorkerPeopleView", "HireAcceptanceCommand", "HireAcceptancePort", @@ -62,6 +72,9 @@ "PeopleReadPort", "PeopleRecordIntegrityError", "PeopleRecordNotFound", + "PositionHistoryIntegrityError", + "PositionHistoryReadPort", + "PositionHistoryRecord", "PositionMutationCommand", "PositionMutationResult", "PostgresHireAcceptancePort", @@ -79,5 +92,6 @@ "create_employment_record", "create_position_record", "extract_bearer_token", + "read_position_history", "read_worker_people_record", ] diff --git a/services/people-api/src/orgmetra_people_api/position_history.py b/services/people-api/src/orgmetra_people_api/position_history.py new file mode 100644 index 000000000..242c2b2fd --- /dev/null +++ b/services/people-api/src/orgmetra_people_api/position_history.py @@ -0,0 +1,331 @@ +"""Purpose-bound bitemporal Position-history reads. + +The service authorizes one exact Position before an injected persistence port may +retrieve protected facts. Persistence output is treated as untrusted evidence: +identity, system-time visibility, business-time consistency, row shape, and +field schema are revalidated before a minimized response is returned. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone +import re +from typing import Protocol, runtime_checkable +from uuid import UUID + +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy + +from orgmetra_people_api.auth import AuthenticatedPrincipal +from orgmetra_people_api.authorization import authorize_resource_fields + +_MAX_UUID_INT = (1 << 128) - 1 +_STATUS_CODE_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:_[a-z0-9]+)*$") + + +class PositionHistoryIntegrityError(RuntimeError): + """Indicate that Position-history persistence violated the authorized contract.""" + + +def _validate_operational_uuid(field_name: str, value: object) -> None: + """Require an exact UUID that is not an Orgmetra protocol sentinel.""" + if type(value) is not UUID or value.int in (0, _MAX_UUID_INT): + raise ValueError(f"{field_name} must be an operational UUID.") + + +def _validate_utc_instant(field_name: str, value: object) -> None: + """Require an exact datetime using Python's built-in zero-offset timezone.""" + if ( + type(value) is not datetime + or type(value.tzinfo) is not timezone + or value.utcoffset() != timedelta(0) + ): + raise ValueError(f"{field_name} must be a timezone-aware UTC datetime.") + + +def _validate_position_status(value: object) -> None: + """Require a canonical built-in lower snake-case Position status code.""" + if type(value) is not str or _STATUS_CODE_PATTERN.fullmatch(value) is None: + raise ValueError("position_status_code must be a canonical lower snake-case string.") + + +def _validate_record_values(values: tuple[object, ...]) -> None: + """Validate one structural Position-history row at construction and read time.""" + if len(values) != 10: + raise ValueError("Position history must contain exactly ten fields.") + for field_name, value in zip( + ( + "tenant_record_id", + "position_record_id", + "position_record_version_id", + "organization_unit_id", + "job_profile_id", + ), + values[:5], + strict=True, + ): + _validate_operational_uuid(field_name, value) + _validate_position_status(values[5]) + effective_from = values[6] + effective_to = values[7] + recorded_from = values[8] + recorded_to = values[9] + if type(effective_from) is not date: + raise ValueError("effective_from must be a business date.") + if effective_to is not None and ( + type(effective_to) is not date or effective_to <= effective_from + ): + raise ValueError("effective_to must be later than effective_from when present.") + _validate_utc_instant("recorded_from", recorded_from) + if recorded_to is not None: + _validate_utc_instant("recorded_to", recorded_to) + if recorded_to <= recorded_from: + raise ValueError("recorded_to must be later than recorded_from when present.") + + +class PositionHistoryRecord(tuple): + """Structurally immutable Position version returned by the persistence boundary. + + Tuple storage deliberately prevents low-level attribute mutation after the row + crosses into the service. ``effective_*`` is business time and ``recorded_*`` + is the half-open system-recorded interval for the version evidence. + """ + + __slots__ = () + + def __new__( + cls, + *, + tenant_record_id: UUID, + position_record_id: UUID, + position_record_version_id: UUID, + organization_unit_id: UUID, + job_profile_id: UUID, + position_status_code: str, + effective_from: date, + effective_to: date | None, + recorded_from: datetime, + recorded_to: datetime | None, + ) -> PositionHistoryRecord: + values: tuple[object, ...] = ( + tenant_record_id, + position_record_id, + position_record_version_id, + organization_unit_id, + job_profile_id, + position_status_code, + effective_from, + effective_to, + recorded_from, + recorded_to, + ) + _validate_record_values(values) + return tuple.__new__(cls, values) + + @property + def tenant_record_id(self) -> UUID: + """Return the tenant that owns this Position version.""" + return tuple.__getitem__(self, 0) + + @property + def position_record_id(self) -> UUID: + """Return the stable Position anchor identity.""" + return tuple.__getitem__(self, 1) + + @property + def position_record_version_id(self) -> UUID: + """Return the immutable Position-version identity.""" + return tuple.__getitem__(self, 2) + + @property + def organization_unit_id(self) -> UUID: + """Return the organization owning the Position anchor.""" + return tuple.__getitem__(self, 3) + + @property + def job_profile_id(self) -> UUID: + """Return the Job profile bound to the Position anchor.""" + return tuple.__getitem__(self, 4) + + @property + def position_status_code(self) -> str: + """Return the canonical Position status code.""" + return tuple.__getitem__(self, 5) + + @property + def effective_from(self) -> date: + """Return the first business date for this Position version.""" + return tuple.__getitem__(self, 6) + + @property + def effective_to(self) -> date | None: + """Return the exclusive business end date when one exists.""" + return tuple.__getitem__(self, 7) + + @property + def recorded_from(self) -> datetime: + """Return the system time from which this version was recorded.""" + return tuple.__getitem__(self, 8) + + @property + def recorded_to(self) -> datetime | None: + """Return the exclusive system-recorded end instant when one exists.""" + return tuple.__getitem__(self, 9) + + def assert_runtime_integrity(self) -> None: + """Revalidate a row reconstructed through low-level tuple mechanisms.""" + _validate_record_values(tuple(self)) + + +@runtime_checkable +class PositionHistoryReadPort(Protocol): + """Read tenant/Position-scoped versions at one system knowledge cutoff.""" + + def read_position_history( + self, + *, + tenant_record_id: UUID, + position_record_id: UUID, + known_at: datetime, + ) -> tuple[PositionHistoryRecord, ...]: + """Return Position versions visible to persistence at ``known_at``.""" + + +@dataclass(frozen=True, slots=True) +class AuthorizedPositionHistoryEntry: + """One Position version containing only purpose-authorized fields.""" + + field_values: tuple[tuple[str, str | None], ...] + + +@dataclass(frozen=True, slots=True) +class AuthorizedPositionHistoryView: + """Purpose-bound bitemporal Position history response.""" + + resource_reference: str + entries: tuple[AuthorizedPositionHistoryEntry, ...] + + +def _instant_text(value: datetime) -> str: + """Render a validated UTC instant in one RFC 3339 representation.""" + return value.isoformat().replace("+00:00", "Z") + + +def _authorized_field_value(record: PositionHistoryRecord, field_name: str) -> str | None: + """Serialize one explicitly supported field without reflective access.""" + if type(field_name) is not str: + raise PositionHistoryIntegrityError("authorization returned an unsupported Position-history field") + if field_name == "effective_from": + return record.effective_from.isoformat() + if field_name == "effective_to": + return None if record.effective_to is None else record.effective_to.isoformat() + if field_name == "job_profile_id": + return str(record.job_profile_id) + if field_name == "organization_unit_id": + return str(record.organization_unit_id) + if field_name == "position_record_version_id": + return str(record.position_record_version_id) + if field_name == "position_status_code": + return record.position_status_code + if field_name == "recorded_from": + return _instant_text(record.recorded_from) + if field_name == "recorded_to": + return None if record.recorded_to is None else _instant_text(record.recorded_to) + raise PositionHistoryIntegrityError("authorization returned an unsupported Position-history field") + + +def _is_recorded_visible(record: PositionHistoryRecord, known_at: datetime) -> bool: + """Return whether ``known_at`` lies inside the half-open system interval.""" + return record.recorded_from <= known_at and ( + record.recorded_to is None or known_at < record.recorded_to + ) + + +def _business_intervals_overlap(left: PositionHistoryRecord, right: PositionHistoryRecord) -> bool: + """Return whether two half-open business intervals overlap without finite infinity sentinels.""" + return ( + (right.effective_to is None or left.effective_from < right.effective_to) + and (left.effective_to is None or right.effective_from < left.effective_to) + ) + + +def read_position_history( + *, + principal: AuthenticatedPrincipal, + tenant_record_id: UUID, + position_record_id: UUID, + known_at: datetime, + purpose_code: str, + requested_fields: frozenset[str], + policy: PurposeBoundAccessPolicy, + read_port: PositionHistoryReadPort, +) -> AuthorizedPositionHistoryView: + """Authorize, validate, minimize, and return one Position's bitemporal history. + + Authorization happens before protected retrieval. A persistence row from a + different tenant or Position, outside the requested system-time view, with a + malformed runtime shape, duplicated version identity, or contradictory + business-effective truth fails closed before any row is returned. + """ + _validate_operational_uuid("tenant_record_id", tenant_record_id) + _validate_operational_uuid("position_record_id", position_record_id) + _validate_utc_instant("known_at", known_at) + + resource_reference = f"position_history:{position_record_id.hex}" + decision = authorize_resource_fields( + principal=principal, + tenant_record_id=tenant_record_id, + resource_tenant_record_id=tenant_record_id, + resource_reference=resource_reference, + purpose_code=purpose_code, + operation_code="read_record", + resource_kind="position_history", + requested_fields=requested_fields, + policy=policy, + ) + + records = read_port.read_position_history( + tenant_record_id=tenant_record_id, + position_record_id=position_record_id, + known_at=known_at, + ) + if type(records) is not tuple: + raise PositionHistoryIntegrityError("Position-history persistence must return an immutable tuple") + + seen_version_ids: set[UUID] = set() + verified: list[PositionHistoryRecord] = [] + for record in records: + if type(record) is not PositionHistoryRecord: + raise PositionHistoryIntegrityError("Position-history persistence returned an unsupported row type") + try: + record.assert_runtime_integrity() + except ValueError as exc: + raise PositionHistoryIntegrityError("Position-history row failed runtime integrity") from exc + if record.tenant_record_id != tenant_record_id or record.position_record_id != position_record_id: + raise PositionHistoryIntegrityError("Position-history row does not match the authorized target") + if not _is_recorded_visible(record, known_at): + raise PositionHistoryIntegrityError("Position-history row is not visible at the requested knowledge cutoff") + if record.position_record_version_id in seen_version_ids: + raise PositionHistoryIntegrityError("duplicate visible Position version") + if any(_business_intervals_overlap(record, existing) for existing in verified): + raise PositionHistoryIntegrityError("overlapping visible Position truth") + seen_version_ids.add(record.position_record_version_id) + verified.append(record) + + authorized_fields = tuple(sorted(decision.authorized_fields)) + entries = tuple( + AuthorizedPositionHistoryEntry( + field_values=tuple( + (field_name, _authorized_field_value(record, field_name)) + for field_name in authorized_fields + ) + ) + for record in sorted( + verified, + key=lambda item: (item.effective_from, item.position_record_version_id.int), + ) + ) + return AuthorizedPositionHistoryView( + resource_reference=decision.resource_reference, + entries=entries, + ) diff --git a/services/people-api/tests/test_position_history_extreme_dates.py b/services/people-api/tests/test_position_history_extreme_dates.py new file mode 100644 index 000000000..bde6f23d9 --- /dev/null +++ b/services/people-api/tests/test_position_history_extreme_dates.py @@ -0,0 +1,93 @@ +"""Regression for open Position business intervals at Python's maximum date.""" + +from __future__ import annotations + +from datetime import date, datetime, timezone +from uuid import UUID + +import pytest + +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy +from orgmetra_people_api.auth import AuthenticatedPrincipal +from orgmetra_people_api.position_history import ( + PositionHistoryIntegrityError, + PositionHistoryRecord, + read_position_history, +) + +TENANT = UUID("0198a413-7000-7000-8000-000000000001") +POSITION = UUID("0198a413-7000-7000-8000-000000000010") +VERSION_A = UUID("0198a413-7000-7000-8000-000000000020") +VERSION_B = UUID("0198a413-7000-7000-8000-000000000021") +ORGANIZATION = UUID("0198a413-7000-7000-8000-000000000030") +JOB = UUID("0198a413-7000-7000-8000-000000000040") +KNOWN_AT = datetime(2026, 8, 30, 2, 0, tzinfo=timezone.utc) +RECORDED_FROM = datetime(2026, 8, 20, 0, 0, tzinfo=timezone.utc) + + +class ExtremeDatePort: + """Return two exact Position rows whose open business intervals overlap.""" + + def read_position_history( + self, + *, + tenant_record_id: UUID, + position_record_id: UUID, + known_at: datetime, + ) -> tuple[PositionHistoryRecord, ...]: + return ( + PositionHistoryRecord( + tenant_record_id=TENANT, + position_record_id=POSITION, + position_record_version_id=VERSION_A, + organization_unit_id=ORGANIZATION, + job_profile_id=JOB, + position_status_code="active", + effective_from=date(9999, 1, 1), + effective_to=None, + recorded_from=RECORDED_FROM, + recorded_to=None, + ), + PositionHistoryRecord( + tenant_record_id=TENANT, + position_record_id=POSITION, + position_record_version_id=VERSION_B, + organization_unit_id=ORGANIZATION, + job_profile_id=JOB, + position_status_code="frozen", + effective_from=date.max, + effective_to=None, + recorded_from=RECORDED_FROM, + recorded_to=None, + ), + ) + + +def test_open_interval_starting_at_date_max_still_overlaps_prior_open_truth() -> None: + """Open-endedness must not be approximated with ``date.max`` as an exclusive end.""" + principal = AuthenticatedPrincipal( + tenant_record_id=TENANT, + actor_reference="keyverse:hr-operator", + granted_scope_codes=frozenset({"orgmetra.people.position_history.read"}), + ) + policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="position-history-extreme-date-v1", + resource_kind="position_history", + purpose_code="workforce_position_review", + operation_code="read_record", + required_scope_code="orgmetra.people.position_history.read", + permitted_fields=frozenset({"position_status_code"}), + ) + + with pytest.raises(PositionHistoryIntegrityError, match="overlapping visible Position truth"): + read_position_history( + principal=principal, + tenant_record_id=TENANT, + position_record_id=POSITION, + known_at=KNOWN_AT, + purpose_code="workforce_position_review", + requested_fields=frozenset({"position_status_code"}), + policy=policy, + read_port=ExtremeDatePort(), + ) diff --git a/services/people-api/tests/test_position_history_read.py b/services/people-api/tests/test_position_history_read.py new file mode 100644 index 000000000..5ffd4b17e --- /dev/null +++ b/services/people-api/tests/test_position_history_read.py @@ -0,0 +1,328 @@ +"""Executable contracts for purpose-bound bitemporal Position history reads.""" + +from __future__ import annotations + +from datetime import date, datetime, timedelta, timezone, tzinfo +import unittest +from uuid import UUID + +from orgmetra_keyverse_adapter import AuthorizationDeniedError, PurposeBoundAccessPolicy +from orgmetra_people_api.auth import AuthenticatedPrincipal +from orgmetra_people_api.position_history import ( + PositionHistoryIntegrityError, + PositionHistoryRecord, + read_position_history, +) + +TENANT = UUID("0198a413-7000-7000-8000-000000000001") +OTHER_TENANT = UUID("0198a413-7000-7000-8000-000000000002") +POSITION = UUID("0198a413-7000-7000-8000-000000000010") +OTHER_POSITION = UUID("0198a413-7000-7000-8000-000000000011") +VERSION_A = UUID("0198a413-7000-7000-8000-000000000020") +VERSION_B = UUID("0198a413-7000-7000-8000-000000000021") +ORGANIZATION = UUID("0198a413-7000-7000-8000-000000000030") +JOB = UUID("0198a413-7000-7000-8000-000000000040") +KNOWN_AT = datetime(2026, 8, 30, 2, 0, tzinfo=timezone.utc) +RECORDED_FROM = datetime(2026, 8, 20, 0, 0, tzinfo=timezone.utc) + + +class ForgedUUID(UUID): + """Prove trust-bearing identity validators reject UUID subclasses.""" + + +class ForgedField(str): + """Prove authorization output cannot smuggle string subclass behavior.""" + + +class ZeroOffsetTimezone(tzinfo): + """Caller-controlled timezone that merely looks like UTC.""" + + def utcoffset(self, dt: datetime | None) -> timedelta: + return timedelta(0) + + def dst(self, dt: datetime | None) -> timedelta: + return timedelta(0) + + +class FakePositionHistoryPort: + """Capture protected reads so authorization-before-retrieval is observable.""" + + def __init__(self, records: object) -> None: + self.records = records + self.calls: list[tuple[UUID, UUID, datetime]] = [] + + def read_position_history( + self, + *, + tenant_record_id: UUID, + position_record_id: UUID, + known_at: datetime, + ) -> object: + self.calls.append((tenant_record_id, position_record_id, known_at)) + return self.records + + +def position_record( + *, + tenant_record_id: UUID = TENANT, + position_record_id: UUID = POSITION, + position_record_version_id: UUID = VERSION_A, + organization_unit_id: UUID = ORGANIZATION, + job_profile_id: UUID = JOB, + position_status_code: str = "active", + effective_from: date = date(2026, 1, 1), + effective_to: date | None = date(2026, 7, 1), + recorded_from: datetime = RECORDED_FROM, + recorded_to: datetime | None = None, +) -> PositionHistoryRecord: + """Build one Position history row crossing the injected persistence boundary.""" + return PositionHistoryRecord( + tenant_record_id=tenant_record_id, + position_record_id=position_record_id, + position_record_version_id=position_record_version_id, + organization_unit_id=organization_unit_id, + job_profile_id=job_profile_id, + position_status_code=position_status_code, + effective_from=effective_from, + effective_to=effective_to, + recorded_from=recorded_from, + recorded_to=recorded_to, + ) + + +class PositionHistoryReadTests(unittest.TestCase): + """Prove Position history stays purpose-bound, bitemporal, and fail-closed.""" + + def setUp(self) -> None: + self.principal = AuthenticatedPrincipal( + tenant_record_id=TENANT, + actor_reference="keyverse:hr-operator", + granted_scope_codes=frozenset({"orgmetra.people.position_history.read"}), + ) + self.policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="position-history-v1", + resource_kind="position_history", + purpose_code="workforce_position_review", + operation_code="read_record", + required_scope_code="orgmetra.people.position_history.read", + permitted_fields=frozenset( + { + "position_record_version_id", + "organization_unit_id", + "job_profile_id", + "position_status_code", + "effective_from", + "effective_to", + "recorded_from", + "recorded_to", + } + ), + ) + + def read(self, records: object, requested_fields: frozenset[str] | None = None): + port = FakePositionHistoryPort(records) + view = read_position_history( + principal=self.principal, + tenant_record_id=TENANT, + position_record_id=POSITION, + known_at=KNOWN_AT, + purpose_code="workforce_position_review", + requested_fields=self.policy.permitted_fields if requested_fields is None else requested_fields, + policy=self.policy, + read_port=port, + ) + return view, port + + def test_returns_authorized_versions_in_deterministic_effective_order(self) -> None: + later = position_record( + position_record_version_id=VERSION_B, + position_status_code="frozen", + effective_from=date(2026, 7, 1), + effective_to=None, + ) + earlier = position_record(recorded_to=datetime(2026, 8, 31, 0, 0, tzinfo=timezone.utc)) + + view, port = self.read((later, earlier)) + + self.assertEqual(view.resource_reference, f"position_history:{POSITION.hex}") + self.assertEqual(port.calls, [(TENANT, POSITION, KNOWN_AT)]) + rows = tuple(dict(entry.field_values) for entry in view.entries) + self.assertEqual(tuple(row["position_record_version_id"] for row in rows), (str(VERSION_A), str(VERSION_B))) + self.assertEqual(rows[0]["organization_unit_id"], str(ORGANIZATION)) + self.assertEqual(rows[0]["job_profile_id"], str(JOB)) + self.assertEqual(rows[0]["position_status_code"], "active") + self.assertEqual(rows[0]["effective_from"], "2026-01-01") + self.assertEqual(rows[0]["effective_to"], "2026-07-01") + self.assertEqual(rows[0]["recorded_from"], "2026-08-20T00:00:00Z") + self.assertEqual(rows[0]["recorded_to"], "2026-08-31T00:00:00Z") + self.assertIsNone(rows[1]["effective_to"]) + self.assertIsNone(rows[1]["recorded_to"]) + + def test_field_minimization_never_adds_position_identity(self) -> None: + limited = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="position-history-minimal-v1", + resource_kind="position_history", + purpose_code="workforce_position_review", + operation_code="read_record", + required_scope_code="orgmetra.people.position_history.read", + permitted_fields=frozenset({"position_status_code"}), + ) + port = FakePositionHistoryPort((position_record(),)) + + view = read_position_history( + principal=self.principal, + tenant_record_id=TENANT, + position_record_id=POSITION, + known_at=KNOWN_AT, + purpose_code="workforce_position_review", + requested_fields=frozenset({"position_status_code"}), + policy=limited, + read_port=port, + ) + + self.assertEqual(view.entries[0].field_values, (("position_status_code", "active"),)) + + def test_denied_field_never_reaches_protected_repository(self) -> None: + limited = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="position-history-limited-v1", + resource_kind="position_history", + purpose_code="workforce_position_review", + operation_code="read_record", + required_scope_code="orgmetra.people.position_history.read", + permitted_fields=frozenset({"position_status_code"}), + ) + port = FakePositionHistoryPort((position_record(),)) + + with self.assertRaises(AuthorizationDeniedError): + read_position_history( + principal=self.principal, + tenant_record_id=TENANT, + position_record_id=POSITION, + known_at=KNOWN_AT, + purpose_code="workforce_position_review", + requested_fields=frozenset({"job_profile_id"}), + policy=limited, + read_port=port, + ) + self.assertEqual(port.calls, []) + + def test_scope_or_system_visibility_mismatch_fails_closed(self) -> None: + cases = ( + position_record(tenant_record_id=OTHER_TENANT), + position_record(position_record_id=OTHER_POSITION), + position_record(recorded_from=datetime(2026, 8, 31, 0, 0, tzinfo=timezone.utc)), + position_record(recorded_to=KNOWN_AT), + ) + for record in cases: + with self.subTest(record=record), self.assertRaises(PositionHistoryIntegrityError): + self.read((record,)) + + def test_container_row_and_low_level_forgery_fail_closed(self) -> None: + valid = position_record() + forged_values = tuple(valid) + forged_values = forged_values[:5] + ("NOT_CANONICAL",) + forged_values[6:] + forged = tuple.__new__(PositionHistoryRecord, forged_values) + cases = ( + ([valid], "immutable tuple"), + ((object(),), "unsupported row type"), + ((forged,), "runtime integrity"), + ) + for records, message in cases: + with self.subTest(message=message), self.assertRaisesRegex(PositionHistoryIntegrityError, message): + self.read(records) + + def test_record_is_structurally_immutable(self) -> None: + record = position_record() + with self.assertRaises(AttributeError): + object.__setattr__(record, "position_status_code", "closed") + self.assertEqual(record.position_status_code, "active") + + def test_duplicate_version_identity_or_overlapping_business_truth_fails_closed(self) -> None: + duplicate = position_record( + effective_from=date(2026, 7, 1), + effective_to=None, + position_status_code="frozen", + ) + with self.assertRaisesRegex(PositionHistoryIntegrityError, "duplicate visible Position version"): + self.read((position_record(), duplicate)) + + overlapping = position_record( + position_record_version_id=VERSION_B, + effective_from=date(2026, 6, 1), + effective_to=None, + position_status_code="frozen", + ) + with self.assertRaisesRegex(PositionHistoryIntegrityError, "overlapping visible Position truth"): + self.read((position_record(), overlapping)) + + def test_policy_schema_or_field_subclass_drift_fails_closed(self) -> None: + for field in ("future_sensitive_field", ForgedField("position_status_code")): + policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="position-history-drift-v1", + resource_kind="position_history", + purpose_code="workforce_position_review", + operation_code="read_record", + required_scope_code="orgmetra.people.position_history.read", + permitted_fields=frozenset({field}), + ) + with self.subTest(field=field), self.assertRaisesRegex(PositionHistoryIntegrityError, "unsupported Position-history field"): + read_position_history( + principal=self.principal, + tenant_record_id=TENANT, + position_record_id=POSITION, + known_at=KNOWN_AT, + purpose_code="workforce_position_review", + requested_fields=frozenset({field}), + policy=policy, + read_port=FakePositionHistoryPort((position_record(),)), + ) + + def test_invalid_request_shape_fails_before_repository_access(self) -> None: + port = FakePositionHistoryPort(()) + invalid = ( + ("tenant_record_id", UUID(int=0)), + ("tenant_record_id", ForgedUUID(str(TENANT))), + ("position_record_id", UUID(int=(1 << 128) - 1)), + ("known_at", datetime(2026, 8, 30, 2, 0)), + ("known_at", datetime(2026, 8, 30, 2, 0, tzinfo=ZeroOffsetTimezone())), + ) + for field_name, value in invalid: + kwargs = { + "principal": self.principal, + "tenant_record_id": TENANT, + "position_record_id": POSITION, + "known_at": KNOWN_AT, + "purpose_code": "workforce_position_review", + "requested_fields": frozenset({"position_status_code"}), + "policy": self.policy, + "read_port": port, + } + kwargs[field_name] = value + with self.subTest(field_name=field_name, value_type=type(value).__name__), self.assertRaises(ValueError): + read_position_history(**kwargs) + self.assertEqual(port.calls, []) + + def test_record_rejects_noncanonical_identity_status_and_time_values(self) -> None: + invalid_overrides = ( + {"position_record_version_id": UUID(int=0)}, + {"organization_unit_id": ForgedUUID(str(ORGANIZATION))}, + {"position_status_code": "NOT_CANONICAL"}, + {"position_status_code": ForgedField("active")}, + {"effective_from": datetime(2026, 1, 1, tzinfo=timezone.utc)}, + {"effective_to": date(2026, 1, 1)}, + {"recorded_from": datetime(2026, 8, 20, 0, 0)}, + {"recorded_from": datetime(2026, 8, 20, 0, 0, tzinfo=ZeroOffsetTimezone())}, + {"recorded_to": datetime(2026, 8, 21, 0, 0)}, + {"recorded_to": RECORDED_FROM}, + ) + for override in invalid_overrides: + with self.subTest(override=override), self.assertRaises(ValueError): + position_record(**override) + + +if __name__ == "__main__": + unittest.main() diff --git a/services/people-api/tests/test_position_history_row_shape.py b/services/people-api/tests/test_position_history_row_shape.py new file mode 100644 index 000000000..92da0999e --- /dev/null +++ b/services/people-api/tests/test_position_history_row_shape.py @@ -0,0 +1,82 @@ +"""Regression for malformed low-level Position-history row reconstruction.""" + +from __future__ import annotations + +from datetime import date, datetime, timezone +from uuid import UUID + +import pytest + +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy +from orgmetra_people_api.auth import AuthenticatedPrincipal +from orgmetra_people_api.position_history import ( + PositionHistoryIntegrityError, + PositionHistoryRecord, + read_position_history, +) + +TENANT = UUID("0198a413-7000-7000-8000-000000000001") +POSITION = UUID("0198a413-7000-7000-8000-000000000010") +VERSION = UUID("0198a413-7000-7000-8000-000000000020") +ORGANIZATION = UUID("0198a413-7000-7000-8000-000000000030") +JOB = UUID("0198a413-7000-7000-8000-000000000040") +KNOWN_AT = datetime(2026, 8, 30, 2, 0, tzinfo=timezone.utc) + + +class ShortRowPort: + """Return a malformed Position row reconstructed below the public constructor.""" + + def __init__(self, row: PositionHistoryRecord) -> None: + self.row = row + + def read_position_history( + self, + *, + tenant_record_id: UUID, + position_record_id: UUID, + known_at: datetime, + ) -> tuple[PositionHistoryRecord, ...]: + return (self.row,) + + +def test_short_low_level_row_fails_closed_at_runtime_boundary() -> None: + """A tuple-level reconstruction missing a field must never reach serialization.""" + valid = PositionHistoryRecord( + tenant_record_id=TENANT, + position_record_id=POSITION, + position_record_version_id=VERSION, + organization_unit_id=ORGANIZATION, + job_profile_id=JOB, + position_status_code="active", + effective_from=date(2026, 1, 1), + effective_to=None, + recorded_from=datetime(2026, 8, 20, 0, 0, tzinfo=timezone.utc), + recorded_to=None, + ) + short_row = tuple.__new__(PositionHistoryRecord, tuple(valid)[:-1]) + principal = AuthenticatedPrincipal( + tenant_record_id=TENANT, + actor_reference="keyverse:hr-operator", + granted_scope_codes=frozenset({"orgmetra.people.position_history.read"}), + ) + policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="position-history-v1", + resource_kind="position_history", + purpose_code="workforce_position_review", + operation_code="read_record", + required_scope_code="orgmetra.people.position_history.read", + permitted_fields=frozenset({"position_status_code"}), + ) + + with pytest.raises(PositionHistoryIntegrityError, match="runtime integrity"): + read_position_history( + principal=principal, + tenant_record_id=TENANT, + position_record_id=POSITION, + known_at=KNOWN_AT, + purpose_code="workforce_position_review", + requested_fields=frozenset({"position_status_code"}), + policy=policy, + read_port=ShortRowPort(short_row), + )