diff --git a/.github/workflows/position-history-postgres-read-quality.yml b/.github/workflows/position-history-postgres-read-quality.yml new file mode 100644 index 000000000..48d5d2f0d --- /dev/null +++ b/.github/workflows/position-history-postgres-read-quality.yml @@ -0,0 +1,60 @@ +name: Position History PostgreSQL Read Quality + +on: + pull_request: + branches: + - develop + - feat/position-history-read + paths: + - "services/people-api/**" + - "packages/hris-kernel/**" + - "packages/keyverse-adapter/**" + - ".github/requirements/foundation-test.txt" + - ".github/workflows/position-history-postgres-read-quality.yml" + - "docs/adr/0153-postgres-position-history-read.md" + - "docs/doctoring/postgres-position-history-read-references.md" + - "docs/traceability/postgres-position-history-read.md" + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: position-history-postgres-read-quality-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + unit: + name: PostgreSQL Position-history read contract + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - name: Checkout exact candidate + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Prove exact candidate checkout + env: + ORGMETRA_EXPECTED_HEAD_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: test "$(git rev-parse HEAD)" = "$ORGMETRA_EXPECTED_HEAD_SHA" + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.14" + check-latest: false + - name: Install reviewed test toolchain + run: | + python -m pip install --require-hashes --no-deps --only-binary=:all: -r .github/requirements/foundation-test.txt + python -m pip check + - name: Compile People API boundary + run: python -m compileall -q services/people-api/src packages/hris-kernel/src packages/keyverse-adapter/src services/people-api/tests + - name: Test governed People contracts with exact statement and branch coverage + env: + PYTHONPATH: services/people-api/src:packages/hris-kernel/src:packages/keyverse-adapter/src + COVERAGE_FILE: /tmp/orgmetra-position-history-postgres-read.coverage + run: python -m pytest -c services/people-api/pyproject.toml services/people-api/tests + - name: Require clean checkout + run: | + git diff --exit-code + test -z "$(git status --porcelain)" diff --git a/docs/adr/0153-postgres-position-history-read.md b/docs/adr/0153-postgres-position-history-read.md new file mode 100644 index 000000000..93c488c06 --- /dev/null +++ b/docs/adr/0153-postgres-position-history-read.md @@ -0,0 +1,48 @@ +# ADR 0153: Read Position history from canonical PostgreSQL truth + +- **Status:** Proposed on active stacked PR #153; not protected-main truth until integrated +- **Date:** 2026-08-30 +- **Owners:** Orgmetra People API / HRIS persistence +- **Extends:** ADR 0003 (bitemporal HRIS data), ADR 0008 (purpose-bound PII authorization), ADR 0152 (Position-history read contract) + +## Context + +PR #152 defines the buyer-facing, purpose-bound Position-history read but intentionally injects its persistence port. An integrated application still needs a canonical adapter for normalized `position_record` and `position_record_version` truth; otherwise each deployment would supply bespoke persistence code and could silently widen the read. + +The adapter is not a second authorization engine or source of truth. The parent People service authorizes before calling it and revalidates its typed output before disclosure. The existing schema already owns Position/Job lineage, bitemporal version facts, tenant RLS, and immutable-history guards. + +## Decision + +Add `PostgresPositionHistoryReadPort` as the canonical PostgreSQL implementation of the `PositionHistoryReadPort` protocol introduced by PR #152. + +The adapter: + +1. validates exact operational tenant/Position UUIDs and an exact built-in UTC `known_at` before acquiring a connection; +2. opens one `READ COMMITTED, READ ONLY` transaction and sets the transaction-local tenant context before the protected query; +3. joins only `public.position_record_version` to its Orgmetra-owned `public.position_record` anchor, preserving Job and organization lineage without Person, Employment, Assignment, compensation, candidate, performance, credential, or decision joins; +4. applies explicit tenant, Position, parent-recorded, and version-recorded half-open predicates; +5. projects recorded timestamps with `AT TIME ZONE 'UTC'`, accepts only exact naive UTC DB projections, and attaches built-in UTC after validation; +6. treats DB-API output as untrusted by checking the default list collection, exact tuple row shape, parent-record integrity, requested target identity, and knowledge-cutoff visibility before returning an immutable tuple. + +Purpose-bound field authorization remains in the parent service. This adapter performs no mutation, audit/outbox write, foreign-service call, decision, or disclosure. + +## Consequences + +### Positive + +- The Position-history application contract can use canonical normalized PostgreSQL truth without host-specific persistence code. +- Read-only transaction mode, explicit predicates, and forced-RLS tenant context provide layered database scope controls. +- Position, Job, and Assignment remain separate concepts, and business-effective history remains distinct from system-recorded visibility. +- Exact DB timestamp validation prevents driver/session timezone behavior from silently changing evidence meaning. + +### Trade-offs + +- The adapter is PostgreSQL/DB-API specific and intentionally requires the default tuple-row contract. +- Database RLS and bitemporal constraints still require independent PostgreSQL tests; this adapter does not claim that SQL predicates replace authorization or schema constraints. +- The parent service must be integrated first and must continue to revalidate rows before serialization. + +## Verification + +The contract-first child test head `bf93924e` fails during collection while the adapter module is absent. The final child must show exact-current-head full People API coverage, invalid-input zero-connection behavior, transaction ordering, explicit SQL scope, UTC projection, malformed-row rejection, target/visibility rechecks, immutable results, and a clean checkout. Parent #152 evidence and reviews do not transfer. + +The implementation follows PostgreSQL 18 transaction access-mode guidance and the existing protected Orgmetra RLS contract. These controls are defense in depth and do not authorize a merge or protected-main representation while this PR is Draft or central gates lack authoritative verdicts. diff --git a/docs/doctoring/postgres-position-history-read-references.md b/docs/doctoring/postgres-position-history-read-references.md new file mode 100644 index 000000000..cd8de0cdb --- /dev/null +++ b/docs/doctoring/postgres-position-history-read-references.md @@ -0,0 +1,25 @@ +# PostgreSQL Position-history read references + +**Scope:** Standards and research basis for active PR #153. This file does not claim certification or protected-main integration. + +## 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: SET TRANSACTION*. https://www.postgresql.org/docs/18/sql-set-transaction.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Row security policies*. https://www.postgresql.org/docs/18/ddl-rowsecurity.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Range types*. https://www.postgresql.org/docs/18/rangetypes.html + +## Decision relevance + +PostgreSQL's transaction access mode and isolation level support the adapter's explicit `READ COMMITTED, READ ONLY` boundary. Row security remains database defense in depth, while the application still binds tenant context and checks exact returned identity. PostgreSQL range/exclusion semantics remain the schema-level basis for bitemporal non-overlap; this read adapter does not replace those constraints. + +RFC 3339 and explicit UTC projection support one interoperable representation for system-recorded evidence. NIST SP 800-53 Rev. 5 informs least privilege, access control, and information-integrity evidence readiness; no compliance or certification claim follows from this PR. + +## Research classification + +These references constrain the accepted adapter architecture for PR #153. They do not authorize scope expansion into worker data, Assignment joins, compensation, candidate, performance, or employment-decision automation. diff --git a/docs/traceability/postgres-position-history-read.md b/docs/traceability/postgres-position-history-read.md new file mode 100644 index 000000000..c31b1b726 --- /dev/null +++ b/docs/traceability/postgres-position-history-read.md @@ -0,0 +1,41 @@ +# PostgreSQL Position-history read traceability + +**Lifecycle status:** Active stacked PR #153 only. This document does not claim protected-`develop` integration. + +## Buyer problem + +PR #152 defines an authorized Position-history read but leaves persistence injected. Without a canonical adapter, an Orgmetra deployment cannot obtain that bounded history from normalized `position_record` and `position_record_version` truth without bespoke host code. + +## Requirement-to-evidence matrix + +| Requirement | Production boundary | Regression | +| --- | --- | --- | +| No DB access on invalid input | exact tenant/Position UUID and built-in UTC `known_at` validation before `connection_factory()` | invalid UUID/time cases assert zero connection calls | +| Database cannot mutate HR truth | `SET TRANSACTION ISOLATION LEVEL READ COMMITTED, READ ONLY` | SQL execution-order assertion | +| Tenant defense in depth | transaction-local `pg_catalog.set_config('orgmetra.tenant_record_id', ..., true)` before SELECT | exact SQL and parameter assertion | +| Explicit Position scope | fully qualified join between `public.position_record_version` and `public.position_record` with tenant/Position predicates | SQL contract assertions | +| Preserve system knowledge | half-open parent/version `recorded_from`/`recorded_to` predicates at `known_at` | future and closed-at-cutoff rows fail closed | +| Preserve business history | no effective-date filter; deterministic effective start/version ordering | returned typed dates and SQL ordering assertion | +| Canonical UTC | `AT TIME ZONE 'UTC'` projection and exact naive DB timestamp validation | string/aware/non-datetime timestamp regressions | +| Untrusted DB-API boundary | exact list result, exact tuple row shape, parent record reconstruction | malformed collection/row/value regressions | +| Immutable typed result | tuple of `PositionHistoryRecord` values | empty and non-empty result regressions | +| Parent authority remains single owner | adapter accepts no purpose or authorization input | PR #152 performs authorization and service revalidation | + +## Test-first chain + +1. **Contract-only child head:** `bf93924e` adds the adapter regressions while `orgmetra_people_api.postgres_position_history` is absent. +2. **Expected RED:** local and exact hosted collection must fail with `ModuleNotFoundError` at that owning module boundary; predecessor or parent failures are not relabeled as adapter evidence. +3. **Implementation:** add the smallest adapter and package-root export, then rerun the full People API suite with exact statement and branch coverage. +4. **Hosted evidence rule:** only the final exact current child head's dedicated workflow and applicable central checks may be used for advancement. Parent #152 evidence does not transfer. + +## Security and data boundary + +The adapter reads only Position anchor lineage and Position-version fields. It does not join Person, Employment, Assignment, compensation, candidate, performance, credential, prompt, or model-output data. Purpose-bound authorization-before-retrieval remains in the parent service; the adapter performs no mutation, audit/outbox write, or high-impact employment decision. + +## Out of scope + +- Position-history HTTP/presentation integration. +- Position mutation or correction workflows. +- Assignment/Employment history joins. +- Database migrations; the protected schema already owns these relations and RLS policies. +- Release, tag, publication, or protected-default-branch authority. diff --git a/services/people-api/src/orgmetra_people_api/__init__.py b/services/people-api/src/orgmetra_people_api/__init__.py index 3313e7f4f..20c9908e6 100644 --- a/services/people-api/src/orgmetra_people_api/__init__.py +++ b/services/people-api/src/orgmetra_people_api/__init__.py @@ -51,6 +51,7 @@ from orgmetra_people_api.postgres import PostgresPeopleReadPort from orgmetra_people_api.postgres_hire import PostgresHireAcceptancePort from orgmetra_people_api.postgres_mutations import PostgresPeopleMutationPort +from orgmetra_people_api.postgres_position_history import PostgresPositionHistoryReadPort __all__ = [ "AuthenticatedPrincipal", @@ -80,6 +81,7 @@ "PostgresHireAcceptancePort", "PostgresPeopleMutationPort", "PostgresPeopleReadPort", + "PostgresPositionHistoryReadPort", "AssignmentMutationCommand", "AssignmentMutationResult", "EmploymentMutationCommand", diff --git a/services/people-api/src/orgmetra_people_api/postgres_position_history.py b/services/people-api/src/orgmetra_people_api/postgres_position_history.py new file mode 100644 index 000000000..cd5bc7bd0 --- /dev/null +++ b/services/people-api/src/orgmetra_people_api/postgres_position_history.py @@ -0,0 +1,177 @@ +"""PostgreSQL adapter for purpose-bound Position-history reads. + +The parent People service owns purpose-bound authorization. This adapter owns +only a read-only, tenant-scoped projection of canonical Orgmetra Position and +Position-version facts, returning typed rows for the parent service to +revalidate before disclosure. +""" + +from __future__ import annotations + +from contextlib import AbstractContextManager +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any, Callable +from uuid import UUID + +from orgmetra_people_api.position_history import ( + PositionHistoryIntegrityError, + PositionHistoryRecord, +) + +PostgresConnectionFactory = Callable[[], AbstractContextManager[Any]] + +_READ_ONLY_SQL = "SET TRANSACTION ISOLATION LEVEL READ COMMITTED, READ ONLY" +_TENANT_CONTEXT_SQL = "SELECT pg_catalog.set_config('orgmetra.tenant_record_id', %s, true)" +_POSITION_HISTORY_SQL = """ +SELECT + position_version.tenant_record_id, + position_version.position_record_id, + position_version.position_record_version_id, + position_anchor.organization_unit_id, + position_anchor.job_profile_id, + position_version.position_status_code, + position_version.effective_from, + position_version.effective_to, + position_version.recorded_from AT TIME ZONE 'UTC' AS recorded_from_utc, + position_version.recorded_to AT TIME ZONE 'UTC' AS recorded_to_utc +FROM public.position_record_version AS position_version +JOIN public.position_record AS position_anchor + ON position_anchor.tenant_record_id = position_version.tenant_record_id + AND position_anchor.position_record_id = position_version.position_record_id +WHERE position_version.tenant_record_id = %s + AND position_version.position_record_id = %s + AND position_anchor.recorded_from <= %s + AND (position_anchor.recorded_to IS NULL OR %s < position_anchor.recorded_to) + AND position_version.recorded_from <= %s + AND (position_version.recorded_to IS NULL OR %s < position_version.recorded_to) +ORDER BY position_version.effective_from, position_version.position_record_version_id +""".strip() +_MAX_UUID_INT = (1 << 128) - 1 + + +def _require_operational_uuid(field_name: str, value: object) -> None: + """Require an exact non-sentinel UUID before any database access.""" + if type(value) is not UUID: + raise ValueError(f"{field_name} must be an operational UUID.") + if value.int in (0, _MAX_UUID_INT): + raise ValueError(f"{field_name} must be an operational UUID.") + + +def _require_utc_instant(field_name: str, value: object) -> None: + """Require exact built-in UTC time before using it as a history cutoff.""" + if type(value) is not datetime: + raise ValueError(f"{field_name} must be a timezone-aware UTC datetime.") + if type(value.tzinfo) is not timezone: + raise ValueError(f"{field_name} must be a timezone-aware UTC datetime.") + if value.utcoffset() != timedelta(0): + raise ValueError(f"{field_name} must be a timezone-aware UTC datetime.") + + +def _db_utc_instant(value: object) -> datetime: + """Attach built-in UTC only to PostgreSQL's explicit naive UTC projection.""" + if type(value) is not datetime or value.tzinfo is not None: + raise PositionHistoryIntegrityError( + "database recorded time must be a naive UTC projection" + ) + return value.replace(tzinfo=timezone.utc) + + +def _record_from_row(row: object) -> PositionHistoryRecord: + """Convert one untrusted DB-API row into the parent governed record type.""" + if type(row) is not tuple or len(row) != 10: + raise PositionHistoryIntegrityError("database Position-history row has an invalid shape") + ( + 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, + ) = row + try: + 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=_db_utc_instant(recorded_from), + recorded_to=None if recorded_to is None else _db_utc_instant(recorded_to), + ) + except ValueError as exc: + raise PositionHistoryIntegrityError( + "database Position-history row failed integrity" + ) from exc + + +@dataclass(frozen=True, slots=True) +class PostgresPositionHistoryReadPort: + """Read canonical Position history through a tenant-scoped read-only transaction.""" + + connection_factory: PostgresConnectionFactory + + def __post_init__(self) -> None: + """Reject an unusable connection factory before a protected read can start.""" + if not callable(self.connection_factory): + raise TypeError("connection_factory must be callable") + + def read_position_history( + self, + *, + tenant_record_id: UUID, + position_record_id: UUID, + known_at: datetime, + ) -> tuple[PositionHistoryRecord, ...]: + """Return Position versions visible at ``known_at`` without authorizing disclosure.""" + _require_operational_uuid("tenant_record_id", tenant_record_id) + _require_operational_uuid("position_record_id", position_record_id) + _require_utc_instant("known_at", known_at) + + with self.connection_factory() as connection: + with connection.cursor() as cursor: + cursor.execute(_READ_ONLY_SQL) + cursor.execute(_TENANT_CONTEXT_SQL, (str(tenant_record_id),)) + cursor.execute( + _POSITION_HISTORY_SQL, + ( + tenant_record_id, + position_record_id, + known_at, + known_at, + known_at, + known_at, + ), + ) + rows = cursor.fetchall() + + if type(rows) is not list: + raise PositionHistoryIntegrityError( + "database Position-history read must return the default list row collection" + ) + + records: list[PositionHistoryRecord] = [] + for row in rows: + record = _record_from_row(row) + if ( + record.tenant_record_id != tenant_record_id + or record.position_record_id != position_record_id + ): + raise PositionHistoryIntegrityError( + "database Position-history row does not match the requested target" + ) + if record.recorded_from > known_at or ( + record.recorded_to is not None and known_at >= record.recorded_to + ): + raise PositionHistoryIntegrityError( + "database Position-history row is not visible at the requested knowledge cutoff" + ) + records.append(record) + return tuple(records) diff --git a/services/people-api/tests/test_postgres_position_history.py b/services/people-api/tests/test_postgres_position_history.py new file mode 100644 index 000000000..1f3f97adb --- /dev/null +++ b/services/people-api/tests/test_postgres_position_history.py @@ -0,0 +1,327 @@ +"""Regression contract for the PostgreSQL Position-history read adapter.""" + +from __future__ import annotations + +from contextlib import AbstractContextManager +from datetime import date, datetime, timedelta, timezone, tzinfo +from uuid import UUID + +import pytest + +from orgmetra_people_api.position_history import PositionHistoryIntegrityError +from orgmetra_people_api.postgres_position_history import PostgresPositionHistoryReadPort + + +TENANT_ID = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c1") +POSITION_ID = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c2") +VERSION_ID = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c3") +ORGANIZATION_ID = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c4") +JOB_ID = UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9c5") +KNOWN_AT = datetime(2026, 8, 30, 0, 0, tzinfo=timezone.utc) + + +class ForgedUUID(UUID): + """Prove caller-controlled UUID subclasses cannot cross the adapter boundary.""" + + +class ZeroOffsetProvider(tzinfo): + """Prove a caller-controlled zero-offset timezone is not canonical UTC.""" + + def utcoffset(self, dt: datetime | None) -> timedelta: + return timedelta(0) + + def dst(self, dt: datetime | None) -> timedelta: + return timedelta(0) + + +def position_row( + *, + tenant_record_id: object = TENANT_ID, + position_record_id: object = POSITION_ID, + position_record_version_id: object = VERSION_ID, + organization_unit_id: object = ORGANIZATION_ID, + job_profile_id: object = JOB_ID, + position_status_code: object = "active", + effective_from: object = date(2026, 1, 1), + effective_to: object = date(2026, 7, 1), + recorded_from: object = datetime(2026, 8, 1), + recorded_to: object = None, +) -> tuple[object, ...]: + """Return one default DB row projected by the governed SQL query.""" + return ( + 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, + ) + + +class FakeCursor(AbstractContextManager["FakeCursor"]): + """Minimal DB-API cursor that records SQL and returns configured rows.""" + + def __init__(self, rows: object) -> None: + self.rows = rows + self.executions: list[tuple[str, object | None]] = [] + + def __enter__(self) -> "FakeCursor": + return self + + def __exit__(self, exc_type: object, exc: object, traceback: object) -> None: + return None + + def execute(self, statement: str, parameters: object | None = None) -> None: + self.executions.append((statement, parameters)) + + def fetchall(self) -> object: + return self.rows + + +class FakeConnection(AbstractContextManager["FakeConnection"]): + """Minimal connection exposing one stable cursor.""" + + def __init__(self, cursor: FakeCursor) -> None: + self._cursor = cursor + + def __enter__(self) -> "FakeConnection": + return self + + def __exit__(self, exc_type: object, exc: object, traceback: object) -> None: + return None + + def cursor(self) -> FakeCursor: + return self._cursor + + +class ConnectionFactory: + """Count connection acquisition so invalid inputs prove zero database access.""" + + def __init__(self, rows: object) -> None: + self.calls = 0 + self.cursor = FakeCursor(rows) + + def __call__(self) -> FakeConnection: + self.calls += 1 + return FakeConnection(self.cursor) + + +@pytest.mark.parametrize("invalid_factory", [None, 7, "connection"]) +def test_rejects_non_callable_connection_factory(invalid_factory: object) -> None: + with pytest.raises(TypeError, match="connection_factory must be callable"): + PostgresPositionHistoryReadPort(invalid_factory) # type: ignore[arg-type] + + +def test_read_is_tenant_scoped_read_only_bitemporal_and_typed() -> None: + factory = ConnectionFactory( + [position_row(recorded_to=datetime(2026, 9, 1))] + ) + port = PostgresPositionHistoryReadPort(factory) + + records = port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + assert len(records) == 1 + record = records[0] + assert record.tenant_record_id == TENANT_ID + assert record.position_record_id == POSITION_ID + assert record.position_record_version_id == VERSION_ID + assert record.organization_unit_id == ORGANIZATION_ID + assert record.job_profile_id == JOB_ID + assert record.position_status_code == "active" + assert record.effective_from == date(2026, 1, 1) + assert record.effective_to == date(2026, 7, 1) + assert record.recorded_from == datetime(2026, 8, 1, tzinfo=timezone.utc) + assert record.recorded_to == datetime(2026, 9, 1, tzinfo=timezone.utc) + assert factory.calls == 1 + + assert len(factory.cursor.executions) == 3 + transaction_sql, transaction_parameters = factory.cursor.executions[0] + tenant_sql, tenant_parameters = factory.cursor.executions[1] + history_sql, history_parameters = factory.cursor.executions[2] + assert transaction_sql == "SET TRANSACTION ISOLATION LEVEL READ COMMITTED, READ ONLY" + assert transaction_parameters is None + assert "pg_catalog.set_config('orgmetra.tenant_record_id', %s, true)" in tenant_sql + assert tenant_parameters == (str(TENANT_ID),) + assert "FROM public.position_record_version AS position_version" in history_sql + assert "JOIN public.position_record AS position_anchor" in history_sql + assert "position_version.tenant_record_id = %s" in history_sql + assert "position_version.position_record_id = %s" in history_sql + assert "position_anchor.recorded_from <= %s" in history_sql + assert "%s < position_anchor.recorded_to" in history_sql + assert "position_version.recorded_from <= %s" in history_sql + assert "%s < position_version.recorded_to" in history_sql + assert "AT TIME ZONE 'UTC'" in history_sql + assert "ORDER BY position_version.effective_from, position_version.position_record_version_id" in history_sql + assert "SELECT *" not in history_sql.upper() + assert history_parameters == ( + TENANT_ID, + POSITION_ID, + KNOWN_AT, + KNOWN_AT, + KNOWN_AT, + KNOWN_AT, + ) + + +def test_empty_database_result_returns_immutable_empty_tuple() -> None: + factory = ConnectionFactory([]) + port = PostgresPositionHistoryReadPort(factory) + + assert port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) == () + + +@pytest.mark.parametrize( + ("tenant_record_id", "position_record_id", "known_at"), + [ + ("not-a-uuid", POSITION_ID, KNOWN_AT), + (ForgedUUID(str(TENANT_ID)), POSITION_ID, KNOWN_AT), + (UUID(int=0), POSITION_ID, KNOWN_AT), + (TENANT_ID, ForgedUUID(str(POSITION_ID)), KNOWN_AT), + (TENANT_ID, UUID(int=(1 << 128) - 1), KNOWN_AT), + (TENANT_ID, POSITION_ID, "2026-08-30"), + (TENANT_ID, POSITION_ID, datetime(2026, 8, 30)), + (TENANT_ID, POSITION_ID, datetime(2026, 8, 30, tzinfo=timezone(timedelta(hours=9)))), + (TENANT_ID, POSITION_ID, datetime(2026, 8, 30, tzinfo=ZeroOffsetProvider())), + ], +) +def test_invalid_request_identity_or_time_fails_before_database_access( + tenant_record_id: object, + position_record_id: object, + known_at: object, +) -> None: + factory = ConnectionFactory([]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(ValueError): + port.read_position_history( # type: ignore[arg-type] + tenant_record_id=tenant_record_id, + position_record_id=position_record_id, + known_at=known_at, + ) + + assert factory.calls == 0 + + +def test_rejects_non_default_fetchall_collection() -> None: + factory = ConnectionFactory((position_row(),)) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="default list row collection"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +@pytest.mark.parametrize("row", [[1] * 10, (1, 2)]) +def test_rejects_unsupported_row_container_or_shape(row: object) -> None: + factory = ConnectionFactory([row]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="row has an invalid shape"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +@pytest.mark.parametrize( + ("recorded_from", "recorded_to"), + [ + ("2026-08-01", None), + (datetime(2026, 8, 1, tzinfo=timezone.utc), None), + (datetime(2026, 8, 1), "2026-09-01"), + (datetime(2026, 8, 1), datetime(2026, 9, 1, tzinfo=timezone.utc)), + ], +) +def test_rejects_noncanonical_database_timestamp_projection( + recorded_from: object, + recorded_to: object, +) -> None: + factory = ConnectionFactory([position_row(recorded_from=recorded_from, recorded_to=recorded_to)]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="database recorded time must be a naive UTC projection"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +def test_rejects_database_row_that_fails_position_record_integrity() -> None: + factory = ConnectionFactory([position_row(position_status_code="NOT_CANONICAL")]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="database Position-history row failed integrity"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +@pytest.mark.parametrize( + "row", + [ + position_row(tenant_record_id=UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9d1")), + position_row(position_record_id=UUID("018d51d2-9ab1-7ac0-8eb1-0a5dc487b9d2")), + ], +) +def test_rejects_row_outside_requested_tenant_or_position(row: tuple[object, ...]) -> None: + factory = ConnectionFactory([row]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="does not match the requested target"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +@pytest.mark.parametrize( + "row", + [ + position_row(recorded_from=datetime(2026, 8, 31)), + position_row(recorded_to=datetime(2026, 8, 30)), + ], +) +def test_rejects_row_outside_requested_system_knowledge_cutoff(row: tuple[object, ...]) -> None: + factory = ConnectionFactory([row]) + port = PostgresPositionHistoryReadPort(factory) + + with pytest.raises(PositionHistoryIntegrityError, match="not visible at the requested knowledge cutoff"): + port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + +def test_open_recorded_interval_is_visible_at_known_at() -> None: + factory = ConnectionFactory([position_row()]) + port = PostgresPositionHistoryReadPort(factory) + + records = port.read_position_history( + tenant_record_id=TENANT_ID, + position_record_id=POSITION_ID, + known_at=KNOWN_AT, + ) + + assert records[0].recorded_to is None + assert isinstance(records, tuple)