diff --git a/CHANGELOG.md b/CHANGELOG.md index 16454da3d..06553f5eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,13 +12,14 @@ All notable changes to Orgmetra will be documented in this file. - Active performance-criterion scope hardening: `criterion_observation_scope_guard` rejects criterion outcomes for a Job the worker did not effectively hold at the observation date, observations before the relevant assignment, and observations outside the referenced performance cycle while preserving valid multiple-assignment cases and existing bitemporal correction semantics. The guard evaluates current-recorded facts, derives the date coordinate from `observed_at` in UTC so session `TimeZone` cannot alter the result, uses a trusted function search path, and adds no PII or automated employment decision authority. The Foundation PostgreSQL contract also rejects a closed `recorded_to` on each time-coordinate lookup and proves UTC midnight plus non-UTC session `TimeZone` boundaries. - Bitemporal tenant-scoped organization hierarchy validation that rejects visible indirect parent cycles and reuses single-valued recorded-time reconstruction before graph traversal. - Stacked governed job-analysis evidence contract via `JobAnalysisSnapshot`, `TaskEvidence`, `KSAORequirement`, `TaskKSAOLink`, `FunctionalJobAnalysisProfile`, and `EvidenceSource`: tenant/Job-scoped observable tasks, explicit Task-to-KSAO linkage, importance/difficulty/proficiency ratings, source/version/retrieval/SHA-256 provenance, deterministic canonical snapshot bytes, current O*NET evidence support, and historical DOT Data/People/Things compatibility. Validated snapshots require accountable human review and complete non-LLM evidence; LLM-origin material remains `analysis_draft`, and the snapshot is evidence input rather than a hiring, promotion, termination, compensation, or other high-impact employment decision. +- Active-PR core evidence hardening now rejects caller-controlled built-in-type subclasses at audit and job-analysis trust boundaries and detaches accepted timestamps from mutable timezone providers before canonical serialization. - Stacked governed audit/outbox slice via `AuditOutboxEvent`, `audit_event_record`, `outbox_delivery_record`, and `outbox_delivery_escalation_record`: CloudEvents 1.0-compatible PII-minimized metadata, exact canonical JSON bytes, database-verified SHA-256 digests, mandatory human confirmation for high-impact events, immutable audit evidence, tenant RLS, atomic audit/outbox insertion, guarded pending/leased/delivered/dead-lettered delivery state, tenant-safe `claim_outbox_delivery(...)` with deterministic due-work ordering, `FOR UPDATE ... SKIP LOCKED`, opaque worker identity, bounded future leases, immutable envelope return, and atomic takeover of genuinely expired leases only while retry attempts remain; owner-bound `complete_outbox_delivery(...)` and `retry_outbox_delivery(...)`; database-budget-governed `dead_letter_outbox_delivery(...)`; and a separately privileged `operator_dead_letter_expired_outbox_delivery(...)` recovery path for an exhausted final lease whose recorded worker identity is permanently unavailable. `maximum_attempt_count` is persisted on the delivery row, defaults to 5, is constrained to 1 through 100, and cannot be lowered by a dispatcher during finalization. Migration 0007 prevents retry or expired-lease takeover from creating attempt N+1; migration 0008 adds TRUNCATE guards, trusted function search paths, a concurrently built due-work partial index, session-independent immutable envelope validation, and operator recovery backed by separate NOLOGIN/NOBYPASSRLS owner/capability roles so the externally assignable operator role can invoke recovery without receiving direct transport-table read/write rights. Migration 0008 also rejects pre-existing reserved recovery-role names before project DDL, atomically contains the temporary schema-creation privilege used for function ownership handoff, and forces deferred escalation binding while the narrow SECURITY DEFINER owner is still active. Exponential/backoff policy selection, policy-specific producer configuration, and external delivery receipts remain subsequent work. - `orgmetra_hris_kernel` 0.4.0 with exclusive-versus-concurrent employment, staffable position coverage, exclusive-seat capacity, and `validate_assignment_write` at 100% statement and branch coverage. - `POST /v1/employment-records`, `POST /v1/position-records`, and `POST /v1/assignment-records` with the same Keyverse mutation context, confirmation, and versioned evidence composition as other high-impact commands. - `employment_record_version.employment_concurrency_code` constrained to `exclusive` or `concurrent`. - ADR 0005 for exclusive employment and staffable seats. - `orgmetra_hris_kernel` 0.3.0 with identity-scoped bitemporal resolution, assignment-employment coverage, allocation-portfolio checks, and a Memorial Hospital RN correction case at 100% statement and branch coverage. -- `employment_record_version` and `position_record_version` so employment and position identity stay stable across retroactive corrections. +- `employment_record_version` and `position_record_version` so corrections no longer mint a new employment or position identifier. - `assignment_record.employment_record_id` bound to the same person as the covering employment. - `orgmetra_keyverse_adapter` that binds an opaque Keyverse subject to a person and rejects passwords, passkeys, and tokens. - Design tokens for the repeating HR actions: approve, review, correct, request evidence, compare, export, and escalate. diff --git a/manifest.json b/manifest.json index f7b6cf55e..a529372db 100644 --- a/manifest.json +++ b/manifest.json @@ -29,9 +29,9 @@ }, { "path": "CHANGELOG.md", - "sha256": "f2d2e0b488c0440533effa821808f2f17e37d92f8fb586174c2fdb594f760ca5", - "bytes": 17539, - "lines": 77 + "sha256": "791d0d29d9b27223e86331d91dd0743b48761818b76db0468caa7e90b812ec10", + "bytes": 17761, + "lines": 78 }, { "path": "CLAUDE.md", @@ -341,15 +341,15 @@ }, { "path": "packages/hris-kernel/src/orgmetra_hris_kernel/audit.py", - "sha256": "3e5b7190cf857dc8c1fc7e898cef303060f34aabee6c27a9034d4d9650e33190", - "bytes": 7707, - "lines": 160 + "sha256": "dcc7f78fbbe9cf0cf0207e3d18c6173bb1ad3ed133aeb7b578ef3657af49a192", + "bytes": 12593, + "lines": 307 }, { "path": "packages/hris-kernel/tests/test_audit_outbox.py", - "sha256": "5928dd7b97fe38d6b7472ce62966437e339058a59c3b301a93a7b5c05432b40c", - "bytes": 7556, - "lines": 200 + "sha256": "6dd86c98e3b4667e47d99cb009f8aa3fc410a6c1c6e59fee7f198dd52331b412", + "bytes": 9228, + "lines": 253 }, { "path": "schemas/openapi.yaml", diff --git a/packages/hris-kernel/src/orgmetra_hris_kernel/audit.py b/packages/hris-kernel/src/orgmetra_hris_kernel/audit.py index 30e3f002f..965f124c3 100644 --- a/packages/hris-kernel/src/orgmetra_hris_kernel/audit.py +++ b/packages/hris-kernel/src/orgmetra_hris_kernel/audit.py @@ -10,12 +10,14 @@ from __future__ import annotations -from dataclasses import dataclass -from datetime import datetime, timezone +from collections import namedtuple +from datetime import datetime, timedelta, timezone from hashlib import sha256 import json import re +from typing import cast from uuid import UUID +from zoneinfo import ZoneInfo _SOURCE_SERVICE_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:_[a-z0-9]+)+$") _EVENT_TYPE_PATTERN = re.compile(r"^orgmetra(?:\.[a-z][a-z0-9_]*){2,}$") @@ -32,83 +34,198 @@ "result_code", ) _ALL_REQUIRED_TEXT_FIELDS = ("source_service", "event_type", *_REQUIRED_TEXT_FIELDS) +_AUDIT_EVENT_FIELDS = ( + "event_id", + "tenant_record_id", + "source_service", + "event_type", + "resource_reference", + "actor_reference", + "purpose_code", + "reason_code", + "evidence_version_code", + "result_code", + "occurred_at", + "high_impact", + "confirmation_reference", +) + + +def _freeze_timestamp(value: datetime) -> datetime: + """Detach only standard-library timezone evidence as one immutable UTC instant.""" + if type(value) is not datetime or value.tzinfo is None: + raise ValueError("occurred_at must be an exact timezone-aware datetime.") + zone = value.tzinfo + if type(zone) not in (timezone, ZoneInfo): + raise ValueError("occurred_at timezone must be exact datetime.timezone or zoneinfo.ZoneInfo.") + offset = cast(timedelta, value.utcoffset()) + try: + return (value.replace(tzinfo=None) - offset).replace(tzinfo=timezone.utc) + except OverflowError as exc: + raise ValueError("occurred_at must be a representable timezone-aware datetime.") from exc -@dataclass(frozen=True, slots=True) -class AuditOutboxEvent: - """One immutable governance envelope for an Orgmetra domain mutation. +def _canonical_timestamp(value: datetime) -> str: + """Render only a previously detached built-in UTC instant as RFC 3339 text.""" + if type(value) is not datetime or value.tzinfo is not timezone.utc: + raise ValueError("occurred_at must be an exact timezone-aware datetime.") + return value.isoformat().replace("+00:00", "Z") + + +def _validate_event_snapshot( + *, + event_id: object, + tenant_record_id: object, + source_service: object, + event_type: object, + resource_reference: object, + actor_reference: object, + purpose_code: object, + reason_code: object, + evidence_version_code: object, + result_code: object, + occurred_at: object, + high_impact: object, + confirmation_reference: object, +) -> None: + """Validate one captured audit value set without executing untrusted coercions.""" + if type(event_id) is not UUID: + raise ValueError("event_id must be a UUID.") + if type(tenant_record_id) is not UUID: + raise ValueError("tenant_record_id must be a UUID.") + if event_id.int == 0: + raise ValueError("event_id must not be the reserved nil UUID.") + if tenant_record_id.int == 0: + raise ValueError("tenant_record_id must not be the reserved nil UUID.") + if event_id.int == _MAX_UUID_INT: + raise ValueError("event_id must not be the reserved max UUID.") + if tenant_record_id.int == _MAX_UUID_INT: + raise ValueError("tenant_record_id must not be the reserved max UUID.") + if type(occurred_at) is not datetime: + raise ValueError("occurred_at must be a datetime.") + if type(high_impact) is not bool: + raise ValueError("high_impact must be a boolean.") + text_values = { + "source_service": source_service, + "event_type": event_type, + "resource_reference": resource_reference, + "actor_reference": actor_reference, + "purpose_code": purpose_code, + "reason_code": reason_code, + "evidence_version_code": evidence_version_code, + "result_code": result_code, + } + for field_name in _ALL_REQUIRED_TEXT_FIELDS: + if type(text_values[field_name]) is not str: + raise ValueError(f"{field_name} must be a string.") + if confirmation_reference is not None and type(confirmation_reference) is not str: + raise ValueError("confirmation_reference must be a string when supplied.") + + if _SOURCE_SERVICE_PATTERN.fullmatch(source_service) is None: + raise ValueError("source_service must contain two or more lower snake_case words.") + if _EVENT_TYPE_PATTERN.fullmatch(event_type) is None: + raise ValueError( + "event_type must use a canonical lower-case orgmetra.. namespace." + ) + for field_name in _REQUIRED_TEXT_FIELDS: + value = text_values[field_name] + if not value.strip(): + raise ValueError(f"{field_name} must not be blank.") + for field_name in ("resource_reference", "actor_reference"): + if _OPAQUE_REFERENCE_PATTERN.fullmatch(text_values[field_name]) is None: + raise ValueError(f"{field_name} must be a namespaced opaque reference.") + for field_name in ("purpose_code", "reason_code", "result_code"): + if _CODE_PATTERN.fullmatch(text_values[field_name]) is None: + raise ValueError(f"{field_name} must be lower snake_case code data.") + if _VERSION_CODE_PATTERN.fullmatch(evidence_version_code) is None: + raise ValueError("evidence_version_code must be a whitespace-free version token.") + if confirmation_reference is not None: + if not confirmation_reference.strip(): + raise ValueError("confirmation_reference must not be blank when supplied.") + if _OPAQUE_REFERENCE_PATTERN.fullmatch(confirmation_reference) is None: + raise ValueError("confirmation_reference must be a namespaced opaque reference.") + if high_impact and confirmation_reference is None: + raise ValueError("high-impact events require confirmation_reference.") + + +_AuditOutboxEventTuple = namedtuple( + "_AuditOutboxEventTuple", + _AUDIT_EVENT_FIELDS, + module=__name__, +) + + +class AuditOutboxEvent(_AuditOutboxEventTuple): + """One structurally immutable governance envelope for an Orgmetra mutation. High-impact employment decisions require ``confirmation_reference``. The emitted CloudEvent intentionally excludes mutable HR payload fields so the audit/outbox record can be retained and shared without becoming a shadow system of record for names, compensation, or other necessary PII. Reserved Nil and Max UUID sentinels are rejected before persistence. + + Canonical evidence is held directly by the tuple value object. There is no + process-local mutable issuance registry to reset or overwrite. Persistence + authorization remains a separate service/port responsibility. """ - event_id: UUID - tenant_record_id: UUID - source_service: str - event_type: str - resource_reference: str - actor_reference: str - purpose_code: str - reason_code: str - evidence_version_code: str - result_code: str - occurred_at: datetime - high_impact: bool - confirmation_reference: str | None = None + __slots__ = () + + def __new__( + cls, + event_id: UUID, + tenant_record_id: UUID, + source_service: str, + event_type: str, + resource_reference: str, + actor_reference: str, + purpose_code: str, + reason_code: str, + evidence_version_code: str, + result_code: str, + occurred_at: datetime, + high_impact: bool, + confirmation_reference: str | None = None, + ) -> AuditOutboxEvent: + """Validate, detach time-provider behavior, and construct one immutable value.""" + if cls is not AuditOutboxEvent: + raise TypeError("AuditOutboxEvent must be constructed as the exact canonical type.") + _validate_event_snapshot( + event_id=event_id, + tenant_record_id=tenant_record_id, + source_service=source_service, + event_type=event_type, + resource_reference=resource_reference, + actor_reference=actor_reference, + purpose_code=purpose_code, + reason_code=reason_code, + evidence_version_code=evidence_version_code, + result_code=result_code, + occurred_at=occurred_at, + high_impact=high_impact, + confirmation_reference=confirmation_reference, + ) + frozen_occurred_at = _freeze_timestamp(occurred_at) + return super().__new__( + cls, + event_id, + tenant_record_id, + source_service, + event_type, + resource_reference, + actor_reference, + purpose_code, + reason_code, + evidence_version_code, + result_code, + frozen_occurred_at, + high_impact, + confirmation_reference, + ) def __post_init__(self) -> None: - """Reject envelopes that cannot provide accountable, portable audit evidence.""" - if not isinstance(self.event_id, UUID): - raise ValueError("event_id must be a UUID.") - if not isinstance(self.tenant_record_id, UUID): - raise ValueError("tenant_record_id must be a UUID.") - if self.event_id.int == 0: - raise ValueError("event_id must not be the reserved nil UUID.") - if self.tenant_record_id.int == 0: - raise ValueError("tenant_record_id must not be the reserved nil UUID.") - if self.event_id.int == _MAX_UUID_INT: - raise ValueError("event_id must not be the reserved max UUID.") - if self.tenant_record_id.int == _MAX_UUID_INT: - raise ValueError("tenant_record_id must not be the reserved max UUID.") - if not isinstance(self.occurred_at, datetime): - raise ValueError("occurred_at must be a datetime.") - if type(self.high_impact) is not bool: - raise ValueError("high_impact must be a boolean.") - for field_name in _ALL_REQUIRED_TEXT_FIELDS: - if not isinstance(getattr(self, field_name), str): - raise ValueError(f"{field_name} must be a string.") - if self.confirmation_reference is not None and not isinstance(self.confirmation_reference, str): - raise ValueError("confirmation_reference must be a string when supplied.") - if self.occurred_at.tzinfo is None: - raise ValueError("occurred_at must be timezone-aware.") - if self.occurred_at.utcoffset() is None: - raise ValueError("occurred_at must resolve to a UTC offset.") - if _SOURCE_SERVICE_PATTERN.fullmatch(self.source_service) is None: - raise ValueError("source_service must contain two or more lower snake_case words.") - if _EVENT_TYPE_PATTERN.fullmatch(self.event_type) is None: - raise ValueError("event_type must use a canonical lower-case orgmetra.. namespace.") - for field_name in _REQUIRED_TEXT_FIELDS: - value = getattr(self, field_name) - if not value.strip(): - raise ValueError(f"{field_name} must not be blank.") - for field_name in ("resource_reference", "actor_reference"): - if _OPAQUE_REFERENCE_PATTERN.fullmatch(getattr(self, field_name)) is None: - raise ValueError(f"{field_name} must be a namespaced opaque reference.") - for field_name in ("purpose_code", "reason_code", "result_code"): - if _CODE_PATTERN.fullmatch(getattr(self, field_name)) is None: - raise ValueError(f"{field_name} must be lower snake_case code data.") - if _VERSION_CODE_PATTERN.fullmatch(self.evidence_version_code) is None: - raise ValueError("evidence_version_code must be a whitespace-free version token.") - if self.confirmation_reference is not None: - if not self.confirmation_reference.strip(): - raise ValueError("confirmation_reference must not be blank when supplied.") - if _OPAQUE_REFERENCE_PATTERN.fullmatch(self.confirmation_reference) is None: - raise ValueError("confirmation_reference must be a namespaced opaque reference.") - if self.high_impact and self.confirmation_reference is None: - raise ValueError("high-impact events require confirmation_reference.") + """Reject compatibility-style re-entry; tuple construction already issued evidence.""" + raise ValueError("audit event identity has already issued canonical evidence.") def to_cloudevent(self) -> dict[str, object]: """Return the canonical structured CloudEvent 1.0 envelope. @@ -118,27 +235,57 @@ def to_cloudevent(self) -> dict[str, object]: PII-minimized result body. Persist this mapping atomically with the owning business write before asynchronous delivery. """ - occurred_utc = self.occurred_at.astimezone(timezone.utc) + if type(self) is not AuditOutboxEvent: + raise ValueError("audit event must be the exact canonical type.") + event_id = self.event_id + tenant_record_id = self.tenant_record_id + source_service = self.source_service + event_type = self.event_type + resource_reference = self.resource_reference + actor_reference = self.actor_reference + purpose_code = self.purpose_code + reason_code = self.reason_code + evidence_version_code = self.evidence_version_code + result_code = self.result_code + occurred_at = self.occurred_at + high_impact = self.high_impact + confirmation_reference = self.confirmation_reference + _validate_event_snapshot( + event_id=event_id, + tenant_record_id=tenant_record_id, + source_service=source_service, + event_type=event_type, + resource_reference=resource_reference, + actor_reference=actor_reference, + purpose_code=purpose_code, + reason_code=reason_code, + evidence_version_code=evidence_version_code, + result_code=result_code, + occurred_at=occurred_at, + high_impact=high_impact, + confirmation_reference=confirmation_reference, + ) + canonical_time = _canonical_timestamp(occurred_at) envelope: dict[str, object] = { "specversion": "1.0", - "id": str(self.event_id), - "source": f"urn:orgmetra:{self.source_service}", - "type": self.event_type, - "subject": self.resource_reference, - "time": occurred_utc.isoformat().replace("+00:00", "Z"), + "id": str(event_id), + "source": f"urn:orgmetra:{source_service}", + "type": event_type, + "subject": resource_reference, + "time": canonical_time, "datacontenttype": "application/json", - "orgmetratenant": str(self.tenant_record_id), - "orgmetraactor": self.actor_reference, - "orgmetrapurpose": self.purpose_code, - "orgmetrareason": self.reason_code, - "orgmetraevidence": self.evidence_version_code, + "orgmetratenant": str(tenant_record_id), + "orgmetraactor": actor_reference, + "orgmetrapurpose": purpose_code, + "orgmetrareason": reason_code, + "orgmetraevidence": evidence_version_code, "data": { - "result_code": self.result_code, - "high_impact": self.high_impact, + "result_code": result_code, + "high_impact": high_impact, }, } - if self.confirmation_reference is not None: - envelope["orgmetraconfirmation"] = self.confirmation_reference + if confirmation_reference is not None: + envelope["orgmetraconfirmation"] = confirmation_reference return envelope def canonical_json(self) -> str: diff --git a/packages/hris-kernel/src/orgmetra_hris_kernel/job_analysis.py b/packages/hris-kernel/src/orgmetra_hris_kernel/job_analysis.py index 9bb710dd1..181359fb6 100644 --- a/packages/hris-kernel/src/orgmetra_hris_kernel/job_analysis.py +++ b/packages/hris-kernel/src/orgmetra_hris_kernel/job_analysis.py @@ -10,7 +10,7 @@ from __future__ import annotations from dataclasses import dataclass -from datetime import date, datetime, timezone +from datetime import date, datetime, timedelta, timezone from hashlib import sha256 import json import re @@ -44,7 +44,7 @@ def _validate_uuid(value: object, field_name: str) -> UUID: """Return a durable UUID or reject type-confused and sentinel identities.""" - if not isinstance(value, UUID): + if type(value) is not UUID: raise ValueError(f"{field_name} must be a UUID") if value.int == 0: raise ValueError(f"{field_name} must not be the nil UUID") @@ -55,7 +55,7 @@ def _validate_uuid(value: object, field_name: str) -> UUID: def _validate_code(value: object, field_name: str) -> str: """Return a two-or-more-word lower snake_case contract code.""" - if not isinstance(value, str): + if type(value) is not str: raise ValueError(f"{field_name} must be a string") if not _CODE_PATTERN.fullmatch(value): raise ValueError(f"{field_name} must be a two-or-more-word snake_case code") @@ -64,7 +64,7 @@ def _validate_code(value: object, field_name: str) -> str: def _validate_reference(value: object, field_name: str) -> str: """Return a namespaced opaque reference instead of human-readable identity data.""" - if not isinstance(value, str): + if type(value) is not str: raise ValueError(f"{field_name} must be a string") if not _REFERENCE_PATTERN.fullmatch(value): raise ValueError(f"{field_name} must be a namespaced opaque reference") @@ -73,7 +73,7 @@ def _validate_reference(value: object, field_name: str) -> str: def _validate_version(value: object, field_name: str) -> str: """Return a compact immutable version token.""" - if not isinstance(value, str): + if type(value) is not str: raise ValueError(f"{field_name} must be a string") if not _VERSION_PATTERN.fullmatch(value): raise ValueError(f"{field_name} must be a compact version token") @@ -82,7 +82,7 @@ def _validate_version(value: object, field_name: str) -> str: def _validate_text(value: object, field_name: str, *, minimum: int = 1) -> str: """Return normalized nonblank explanatory text without changing its meaning.""" - if not isinstance(value, str): + if type(value) is not str: raise ValueError(f"{field_name} must be a string") normalized = " ".join(value.split()) if len(normalized) < minimum: @@ -92,7 +92,7 @@ def _validate_text(value: object, field_name: str, *, minimum: int = 1) -> str: def _validate_level(value: object, field_name: str) -> int: """Return an ordinal 1..5 job-analysis rating.""" - if isinstance(value, bool) or not isinstance(value, int): + if type(value) is not int: raise ValueError(f"{field_name} must be an integer") if not 1 <= value <= 5: raise ValueError(f"{field_name} must be between 1 and 5") @@ -100,19 +100,28 @@ def _validate_level(value: object, field_name: str) -> int: def _validate_aware_datetime(value: object, field_name: str) -> datetime: - """Return an offset-aware instant suitable for evidence ordering.""" - if not isinstance(value, datetime): + """Detach one exact offset-aware instant as immutable UTC evidence.""" + if type(value) is not datetime: raise ValueError(f"{field_name} must be a datetime") if value.tzinfo is None: raise ValueError(f"{field_name} must be timezone-aware") - if value.utcoffset() is None: + try: + offset = value.utcoffset() + except Exception as exc: # noqa: BLE001 - normalize provider behavior at trust boundary. + raise ValueError(f"{field_name} must resolve to a UTC offset") from exc + if offset is None or type(offset) is not timedelta: raise ValueError(f"{field_name} must resolve to a UTC offset") - return value + try: + return (value.replace(tzinfo=None) - offset).replace(tzinfo=timezone.utc) + except OverflowError as exc: + raise ValueError(f"{field_name} must be a representable timezone-aware datetime") from exc def _utc_text(value: datetime) -> str: - """Serialize an already-validated instant as canonical UTC text.""" - return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + """Serialize a previously detached built-in UTC instant as canonical text.""" + if type(value) is not datetime or value.tzinfo is not timezone.utc: + raise ValueError("datetime must be an exact timezone-aware datetime") + return value.isoformat().replace("+00:00", "Z") @dataclass(frozen=True, slots=True) @@ -128,7 +137,7 @@ class EvidenceSource: def __post_init__(self) -> None: """Reject ambiguous, credential-bearing, mutable, or untyped provenance.""" - if not isinstance(self.source_uri, str): + if type(self.source_uri) is not str: raise ValueError("source_uri must be a string") parsed = urlsplit(self.source_uri) if parsed.scheme != "https" or not parsed.hostname: @@ -141,8 +150,12 @@ def __post_init__(self) -> None: _validate_text(self.source_title, "source_title", minimum=3), ) _validate_version(self.source_version_code, "source_version_code") - _validate_aware_datetime(self.retrieved_at, "retrieved_at") - if not isinstance(self.content_digest_sha256, str): + object.__setattr__( + self, + "retrieved_at", + _validate_aware_datetime(self.retrieved_at, "retrieved_at"), + ) + if type(self.content_digest_sha256) is not str: raise ValueError("content_digest_sha256 must be a string") if not _SHA256_PATTERN.fullmatch(self.content_digest_sha256): raise ValueError("content_digest_sha256 must be 64 lowercase hexadecimal characters") @@ -175,7 +188,7 @@ def __post_init__(self) -> None: ) _validate_level(self.importance_level, "importance_level") _validate_level(self.difficulty_level, "difficulty_level") - if not isinstance(self.source, EvidenceSource): + if type(self.source) is not EvidenceSource: raise ValueError("source must be EvidenceSource") @@ -207,7 +220,7 @@ def __post_init__(self) -> None: ) _validate_level(self.importance_level, "importance_level") _validate_level(self.proficiency_level, "proficiency_level") - if not isinstance(self.source, EvidenceSource): + if type(self.source) is not EvidenceSource: raise ValueError("source must be EvidenceSource") @@ -236,11 +249,11 @@ def __post_init__(self) -> None: (self.people_function_code, "people_function_code", 8), (self.things_function_code, "things_function_code", 7), ): - if isinstance(value, bool) or not isinstance(value, int): + if type(value) is not int: raise ValueError(f"{field_name} must be an integer") if not 0 <= value <= maximum: raise ValueError(f"{field_name} must be between 0 and {maximum}") - if not isinstance(self.source, EvidenceSource): + if type(self.source) is not EvidenceSource: raise ValueError("source must be EvidenceSource") @@ -258,7 +271,7 @@ def __post_init__(self) -> None: _validate_uuid(self.task_record_id, "task_record_id") _validate_uuid(self.ksao_record_id, "ksao_record_id") _validate_level(self.relationship_strength, "relationship_strength") - if not isinstance(self.essential_for_task, bool): + if type(self.essential_for_task) is not bool: raise ValueError("essential_for_task must be a bool") @@ -294,18 +307,26 @@ def __post_init__(self) -> None: _validate_code(self.status_code, "status_code") if self.status_code not in _ALLOWED_STATUS_CODES: raise ValueError("status_code is not an allowed analysis status") - if not isinstance(self.effective_from, date) or isinstance(self.effective_from, datetime): + if type(self.effective_from) is not date: raise ValueError("effective_from must be a date") recorded_at = _validate_aware_datetime(self.recorded_at, "recorded_at") - if not isinstance(self.tasks, tuple) or not self.tasks: + object.__setattr__(self, "recorded_at", recorded_at) + if type(self.tasks) is not tuple or not self.tasks: raise ValueError("tasks must be a non-empty tuple") - if not isinstance(self.ksao_requirements, tuple) or not self.ksao_requirements: + if type(self.ksao_requirements) is not tuple or not self.ksao_requirements: raise ValueError("ksao_requirements must be a non-empty tuple") - if not isinstance(self.task_ksao_links, tuple) or not self.task_ksao_links: + if type(self.task_ksao_links) is not tuple or not self.task_ksao_links: raise ValueError("task_ksao_links must be a non-empty tuple") - if not isinstance(self.fja_profile, FunctionalJobAnalysisProfile): + if type(self.fja_profile) is not FunctionalJobAnalysisProfile: raise ValueError("fja_profile must be FunctionalJobAnalysisProfile") + for task in self.tasks: + if type(task) is not TaskEvidence: + raise ValueError("tasks must contain TaskEvidence values") + for item in self.ksao_requirements: + if type(item) is not KSAORequirement: + raise ValueError("ksao_requirements must contain KSAORequirement values") + for item in (*self.tasks, *self.ksao_requirements, self.fja_profile): if item.tenant_record_id != self.tenant_record_id: raise ValueError("all job-analysis evidence must share tenant_record_id") @@ -331,7 +352,7 @@ def __post_init__(self) -> None: ksao_id_set = set(ksao_ids) link_pairs: set[tuple[UUID, UUID]] = set() for link in self.task_ksao_links: - if not isinstance(link, TaskKSAOLink): + if type(link) is not TaskKSAOLink: raise ValueError("task_ksao_links must contain TaskKSAOLink values") if link.task_record_id not in task_id_set: raise ValueError("task_ksao_links contains an unknown task_record_id") @@ -351,6 +372,7 @@ def __post_init__(self) -> None: if self.reviewed_by_reference is not None: _validate_reference(self.reviewed_by_reference, "reviewed_by_reference") reviewed_at = _validate_aware_datetime(self.reviewed_at, "reviewed_at") + object.__setattr__(self, "reviewed_at", reviewed_at) if reviewed_at > recorded_at: raise ValueError("reviewed_at must not be later than recorded_at") if any(source.retrieved_at > reviewed_at for source in sources): diff --git a/packages/hris-kernel/tests/test_audit_creation_identity_integrity.py b/packages/hris-kernel/tests/test_audit_creation_identity_integrity.py new file mode 100644 index 000000000..574a3de88 --- /dev/null +++ b/packages/hris-kernel/tests/test_audit_creation_identity_integrity.py @@ -0,0 +1,83 @@ +"""Regression coverage for structurally immutable audit canonical evidence.""" + +from datetime import datetime, timezone +from uuid import UUID + +import pytest + +from orgmetra_hris_kernel.audit import AuditOutboxEvent + + +def _event_values() -> dict[str, object]: + """Return one valid high-impact audit event payload with stable opaque evidence.""" + return { + "event_id": UUID("00000000-0000-4000-8000-000000000002"), + "tenant_record_id": UUID("00000000-0000-4000-8000-000000000001"), + "source_service": "people_core", + "event_type": "orgmetra.people.assignment.recorded", + "resource_reference": "assignment_record:01JTESTOPAQUE", + "actor_reference": "keyverse_subject:01JACTOROPAQUE", + "purpose_code": "workforce_administration", + "reason_code": "hire_completion", + "evidence_version_code": "employment-offer:v3", + "result_code": "recorded", + "occurred_at": datetime(2026, 8, 17, 1, 30, tzinfo=timezone.utc), + "high_impact": True, + "confirmation_reference": "confirmation:01JCONFIRMOPAQUE", + } + + +def _event() -> AuditOutboxEvent: + """Build one valid high-impact audit event with stable opaque evidence.""" + return AuditOutboxEvent(**_event_values()) # type: ignore[arg-type] + + +@pytest.mark.parametrize( + ("field_name", "replacement"), + [ + ("actor_reference", "keyverse_subject:01JOTHERACTOR"), + ("reason_code", "manager_transfer"), + ("result_code", "updated"), + ("confirmation_reference", "confirmation:01JOTHERCONFIRM"), + ], +) +def test_canonical_fields_cannot_be_replaced_after_construction( + field_name: str, + replacement: str, +) -> None: + """Canonical evidence is structurally immutable rather than guarded by resettable state.""" + event = _event() + original = event.canonical_json() + + with pytest.raises(AttributeError): + object.__setattr__(event, field_name, replacement) + + assert event.canonical_json() == original + assert replacement not in original + + +def test_post_init_reentry_cannot_reissue_canonical_evidence() -> None: + """The compatibility re-entry hook is rejection-only after immutable construction.""" + event = _event() + + with pytest.raises(ValueError, match="already issued"): + event.__post_init__() + + +def test_low_level_tuple_forgery_is_revalidated_before_export() -> None: + """Bypassing the public constructor cannot bypass export-time contract validation.""" + values = list(_event()) + values[7] = "Manager Transfer" + forged = tuple.__new__(AuditOutboxEvent, values) + + with pytest.raises(ValueError, match="reason_code"): + forged.canonical_json() + + +def test_event_has_no_mutable_instance_slot_for_creation_seal() -> None: + """The immutable value object has no writable per-instance creation seal.""" + event = _event() + + assert not hasattr(event, "_creation_snapshot") + with pytest.raises(AttributeError): + object.__setattr__(event, "_creation_snapshot", ()) diff --git a/packages/hris-kernel/tests/test_audit_module_state_integrity.py b/packages/hris-kernel/tests/test_audit_module_state_integrity.py new file mode 100644 index 000000000..34f02cede --- /dev/null +++ b/packages/hris-kernel/tests/test_audit_module_state_integrity.py @@ -0,0 +1,16 @@ +"""Regressions for module-level audit issuance authority storage.""" + +from orgmetra_hris_kernel import audit as audit_module + + +def test_audit_issuance_backing_registries_are_not_module_mutation_capabilities() -> None: + """Importing the audit module must not expose mutable canonical-issuance storage.""" + assert not hasattr(audit_module, "_AUDIT_LIVE_ISSUANCES") + assert not hasattr(audit_module, "_AUDIT_CREATION_SNAPSHOTS") + + +def test_audit_module_does_not_export_closure_backed_issuance_mutators() -> None: + """Consumers must not recover authority dictionaries through exported closure cells.""" + assert not hasattr(audit_module, "_claim_audit_issuance") + assert not hasattr(audit_module, "_record_audit_creation_snapshot") + assert not hasattr(audit_module, "_lookup_audit_creation_snapshot") diff --git a/packages/hris-kernel/tests/test_audit_outbox.py b/packages/hris-kernel/tests/test_audit_outbox.py index 9353e50ec..1db1e0e47 100644 --- a/packages/hris-kernel/tests/test_audit_outbox.py +++ b/packages/hris-kernel/tests/test_audit_outbox.py @@ -173,19 +173,27 @@ def test_digest_changes_when_governance_context_changes(): assert original != changed -def test_event_rejects_timezone_object_without_resolved_offset(): - """A tzinfo object that cannot resolve an offset is not auditable time evidence.""" +def test_event_rejects_custom_timezone_before_provider_callback(): + """Caller-defined timezone behavior is rejected before any provider hook executes.""" from datetime import tzinfo - class UnresolvedTimezone(tzinfo): - """Minimal tzinfo fixture with intentionally unresolved UTC offset.""" + class ExecutableTimezone(tzinfo): + """Record any forbidden UTC-offset callback at the audit trust boundary.""" + + def __init__(self): + self.calls = 0 def utcoffset(self, dt): - """Return no offset so the contract must reject this timestamp.""" - return None + """Expose a tripwire if the boundary executes this provider.""" + del dt + self.calls += 1 + return timedelta(0) + + provider = ExecutableTimezone() + with pytest.raises(ValueError, match="datetime.timezone or zoneinfo.ZoneInfo"): + _event(occurred_at=datetime(2026, 8, 17, 1, 30, tzinfo=provider)) - with pytest.raises(ValueError, match="resolve to a UTC offset"): - _event(occurred_at=datetime(2026, 8, 17, 1, 30, tzinfo=UnresolvedTimezone())) + assert provider.calls == 0 def test_event_rejects_blank_optional_confirmation_reference(): @@ -198,3 +206,48 @@ def test_event_rejects_nonopaque_optional_confirmation_reference(): """Free-text confirmation data cannot enter an opaque-reference field.""" with pytest.raises(ValueError, match="opaque reference"): _event(high_impact=False, confirmation_reference="approved by Ada") + + +def test_canonical_event_prevents_actor_replacement_after_construction(): + """Canonical actor evidence is structurally immutable after validation.""" + event = _event() + original = event.canonical_json() + + with pytest.raises(AttributeError): + object.__setattr__(event, "actor_reference", "Ada Lovelace") + + assert event.canonical_json() == original + + +def test_canonical_event_prevents_confirmation_removal_after_construction(): + """A validated high-impact confirmation cannot be removed from the immutable event.""" + event = _event() + original = event.canonical_json() + + with pytest.raises(AttributeError): + object.__setattr__(event, "confirmation_reference", None) + + assert event.canonical_json() == original + + +def test_canonical_event_prevents_identity_replacement_before_stringification(): + """Untrusted replacement identities cannot be installed into the canonical value object.""" + + class ExecutableIdentifier: + """Fail if any rejected replacement is unexpectedly stringified.""" + + def __init__(self) -> None: + self.calls = 0 + + def __str__(self) -> str: + self.calls += 1 + raise AssertionError("untrusted identity stringification executed") + + event = _event() + replacement = ExecutableIdentifier() + + with pytest.raises(AttributeError): + object.__setattr__(event, "event_id", replacement) + + assert replacement.calls == 0 + assert event.event_id == EVENT_ID diff --git a/packages/hris-kernel/tests/test_audit_runtime_type_integrity.py b/packages/hris-kernel/tests/test_audit_runtime_type_integrity.py new file mode 100644 index 000000000..76b772e13 --- /dev/null +++ b/packages/hris-kernel/tests/test_audit_runtime_type_integrity.py @@ -0,0 +1,192 @@ +"""Runtime-type integrity regressions for immutable audit/outbox evidence.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone, tzinfo +from uuid import UUID +from zoneinfo import ZoneInfo + +import pytest + +from orgmetra_hris_kernel.audit import AuditOutboxEvent + + +class _ForgedUUID(UUID): + """Attempt to rewrite an immutable identity during canonical serialization.""" + + def __str__(self) -> str: + """Render an identity different from the underlying UUID value.""" + return "00000000-0000-4000-8000-ffffffffffff" + + +class _ForgedOccurredAt(datetime): + """Attempt to rewrite an immutable event timestamp during serialization.""" + + def astimezone(self, tz=None): # noqa: ANN001 + """Preserve hostile runtime behavior through UTC normalization.""" + return self + + def isoformat(self, sep="T", timespec="auto") -> str: # noqa: ARG002 + """Render a different instant from the underlying timestamp.""" + return "2099-01-01T00:00:00+00:00" + + +class _OpaqueText(str): + """Represent valid audit text through an untrusted runtime subclass.""" + + +class _ForgedAuditOutboxEvent(AuditOutboxEvent): + """Represent a non-canonical audit tuple created through a low-level bypass.""" + + __slots__ = () + + +class _TripwireOffset(tzinfo): + """Fail if caller-defined timezone behavior executes at an audit trust boundary.""" + + def __init__(self) -> None: + self.calls = 0 + + def utcoffset(self, value): # type: ignore[no-untyped-def] + """Record and reject any callback if the trust boundary invokes this provider.""" + del value + self.calls += 1 + raise AssertionError("caller-defined timezone callback executed before rejection") + + def dst(self, value): # type: ignore[no-untyped-def] + """Keep daylight saving fixed if unexpectedly queried.""" + del value + return timedelta(0) + + +def _event(**overrides: object) -> AuditOutboxEvent: + """Build one valid high-impact audit event with focused overrides.""" + values: dict[str, object] = { + "event_id": UUID("00000000-0000-4000-8000-000000000002"), + "tenant_record_id": UUID("00000000-0000-4000-8000-000000000001"), + "source_service": "people_core", + "event_type": "orgmetra.people.assignment.recorded", + "resource_reference": "assignment_record:01JTESTOPAQUE", + "actor_reference": "keyverse_subject:01JACTOROPAQUE", + "purpose_code": "workforce_administration", + "reason_code": "hire_completion", + "evidence_version_code": "employment-offer:v3", + "result_code": "recorded", + "occurred_at": datetime(2026, 8, 21, 5, 20, tzinfo=timezone.utc), + "high_impact": True, + "confirmation_reference": "confirmation:01JCONFIRMOPAQUE", + } + values.update(overrides) + return AuditOutboxEvent(**values) # type: ignore[arg-type] + + +@pytest.mark.parametrize("field_name", ["event_id", "tenant_record_id"]) +def test_audit_event_rejects_uuid_subclasses_before_canonicalization(field_name: str) -> None: + """Caller-controlled UUID rendering cannot alter durable audit identity evidence.""" + forged = _ForgedUUID("00000000-0000-4000-8000-000000000123") + with pytest.raises(ValueError, match=f"{field_name} must be a UUID"): + _event(**{field_name: forged}) + + +def test_audit_event_rejects_datetime_subclasses_before_canonicalization() -> None: + """Caller-controlled timestamp rendering cannot alter durable audit chronology.""" + forged = _ForgedOccurredAt(2026, 8, 21, 5, 20, tzinfo=timezone.utc) + with pytest.raises(ValueError, match="occurred_at must be a datetime"): + _event(occurred_at=forged) + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("source_service", _OpaqueText("people_core")), + ("event_type", _OpaqueText("orgmetra.people.assignment.recorded")), + ("resource_reference", _OpaqueText("assignment_record:01JTESTOPAQUE")), + ("actor_reference", _OpaqueText("keyverse_subject:01JACTOROPAQUE")), + ("purpose_code", _OpaqueText("workforce_administration")), + ("reason_code", _OpaqueText("hire_completion")), + ("evidence_version_code", _OpaqueText("employment-offer:v3")), + ("result_code", _OpaqueText("recorded")), + ("confirmation_reference", _OpaqueText("confirmation:01JCONFIRMOPAQUE")), + ], +) +def test_audit_event_rejects_string_subclasses_before_canonicalization( + field_name: str, value: str +) -> None: + """Reject caller-controlled runtime behavior in every audit text field.""" + with pytest.raises(ValueError, match="must be a string"): + _event(**{field_name: value}) + + +def test_audit_event_accepts_standard_zoneinfo_and_detaches_to_utc() -> None: + """Psycopg-style standard-library ZoneInfo timestamps remain supported.""" + event = _event( + high_impact=False, + confirmation_reference=None, + occurred_at=datetime(2026, 8, 21, 14, 20, tzinfo=ZoneInfo("Asia/Seoul")), + ) + + assert event.occurred_at == datetime(2026, 8, 21, 5, 20, tzinfo=timezone.utc) + assert type(event.occurred_at) is datetime + assert event.occurred_at.tzinfo is timezone.utc + + +def test_audit_event_rejects_custom_timezone_before_provider_callback() -> None: + """Caller-defined tzinfo must fail closed before any executable provider hook runs.""" + tripwire = _TripwireOffset() + occurred_at = datetime(2026, 8, 21, 5, 20, tzinfo=tripwire) + + with pytest.raises(ValueError, match="datetime.timezone or zoneinfo.ZoneInfo"): + _event(occurred_at=occurred_at) + + assert tripwire.calls == 0 + + +def test_audit_event_normalizes_offset_overflow_to_value_error() -> None: + """Fail closed when a standard-library UTC detachment exceeds representable values.""" + occurred_at = datetime( + 1, + 1, + 1, + 0, + 0, + tzinfo=timezone(timedelta(hours=23, minutes=59)), + ) + with pytest.raises(ValueError, match="occurred_at must be a representable"): + _event(occurred_at=occurred_at) + + +def test_audit_event_prevents_reintroduced_timezone_behavior_by_structure() -> None: + """Post-construction mutation fails before caller-controlled timezone callbacks can exist.""" + event = _event() + tripwire = _TripwireOffset() + replacement = datetime(2026, 8, 21, 5, 20, tzinfo=tripwire) + + with pytest.raises(AttributeError): + object.__setattr__(event, "occurred_at", replacement) + + assert event.occurred_at.tzinfo is timezone.utc + assert tripwire.calls == 0 + + +def test_audit_event_rejects_subclass_construction_before_snapshot_validation() -> None: + """Canonical construction rejects a tuple subtype before accepting any evidence.""" + with pytest.raises(TypeError, match="exact canonical type"): + _ForgedAuditOutboxEvent(*_event()) + + +def test_audit_event_rejects_low_level_subclass_before_export() -> None: + """A low-level tuple subtype cannot cross the canonical export boundary.""" + forged = tuple.__new__(_ForgedAuditOutboxEvent, tuple(_event())) + + with pytest.raises(ValueError, match="exact canonical type"): + forged.to_cloudevent() + + +def test_audit_event_rejects_low_level_non_utc_datetime_before_rendering() -> None: + """A low-level exact tuple with non-canonical time still fails closed on export.""" + values = list(_event()) + values[10] = datetime(2026, 8, 21, 5, 20) + forged = tuple.__new__(AuditOutboxEvent, values) + + with pytest.raises(ValueError, match="exact timezone-aware datetime"): + forged.to_cloudevent() diff --git a/packages/hris-kernel/tests/test_job_analysis_temporal_type_integrity.py b/packages/hris-kernel/tests/test_job_analysis_temporal_type_integrity.py new file mode 100644 index 000000000..37e502cbc --- /dev/null +++ b/packages/hris-kernel/tests/test_job_analysis_temporal_type_integrity.py @@ -0,0 +1,420 @@ +"""Runtime-type integrity for canonical job-analysis evidence.""" + +from __future__ import annotations + +from dataclasses import replace +from datetime import date, datetime, timedelta, timezone, tzinfo +from uuid import UUID + +import pytest + +from orgmetra_hris_kernel.job_analysis import ( + EvidenceSource, + FunctionalJobAnalysisProfile, + JobAnalysisSnapshot, + KSAORequirement, + TaskEvidence, + TaskKSAOLink, +) + +TENANT_ID = UUID("00000000-0000-4000-8000-000000002001") +JOB_ID = UUID("00000000-0000-4000-8000-000000002002") +ANALYSIS_ID = UUID("00000000-0000-4000-8000-000000002003") +TASK_ID = UUID("00000000-0000-4000-8000-000000002004") +KSAO_ID = UUID("00000000-0000-4000-8000-000000002005") +RECORDED_AT = datetime(2026, 8, 21, 5, 15, tzinfo=timezone.utc) + + +class _ForgedUUID(UUID): + """Attempt to forge an identity written into canonical job-analysis evidence.""" + + def __str__(self) -> str: + """Render a different identity from the underlying UUID value.""" + return "00000000-0000-4000-8000-ffffffffffff" + + +class _ForgedDate(date): + """Attempt to forge the business date written into canonical evidence.""" + + def isoformat(self) -> str: + """Render a different business date from the underlying value.""" + return "2099-01-01" + + +class _ForgedDateTime(datetime): + """Attempt to forge a recorded instant written into canonical evidence.""" + + def astimezone(self, tz=None): # noqa: ANN001 + """Preserve the hostile runtime type through UTC normalization.""" + return self + + def isoformat(self, sep="T", timespec="auto") -> str: # noqa: ARG002 + """Render a different recorded instant from the underlying value.""" + return "2099-01-01T00:00:00+00:00" + + +class _ForgedStatusCode(str): + """Masquerade an ungoverned status as an allow-listed draft status.""" + + def __hash__(self) -> int: + """Route membership lookup to the allowed draft-status bucket.""" + return hash("analysis_draft") + + def __eq__(self, other: object) -> bool: + """Claim equality with the allowed draft status despite different text.""" + if other == "analysis_draft": + return True + return str.__eq__(self, other) + + +class _ForgedOriginCode(str): + """Masquerade ungoverned provenance as an allow-listed evidence origin.""" + + def __hash__(self) -> int: + """Route membership lookup to the authoritative-origin bucket.""" + return hash("authoritative_occupation_source") + + def __eq__(self, other: object) -> bool: + """Claim equality with an authoritative origin despite different text.""" + if other == "authoritative_occupation_source": + return True + return str.__eq__(self, other) + + +class _ForgedLevel(int): + """Masquerade an out-of-range ordinal as an allowed job-analysis level.""" + + def __ge__(self, other: object) -> bool: + """Forge the lower-bound comparison used by the level validator.""" + return True + + def __le__(self, other: object) -> bool: + """Forge the upper-bound comparison used by the level validator.""" + return True + + def __lt__(self, other: object) -> bool: + """Keep adversarial ordering behavior internally consistent.""" + return True + + def __gt__(self, other: object) -> bool: + """Keep adversarial ordering behavior internally consistent.""" + return True + + +class _OpaqueText(str): + """Represent valid evidence text through an untrusted runtime subclass.""" + + +class _TaskEvidenceSubclass(TaskEvidence): + """Represent a valid task through an untrusted runtime subclass.""" + + +class _KSAORequirementSubclass(KSAORequirement): + """Represent a valid KSAO requirement through an untrusted runtime subclass.""" + + +class _MutableOffset(tzinfo): + """Expose timezone state that can change after evidence construction.""" + + def __init__(self) -> None: + """Start with a UTC offset.""" + self.offset = timedelta(0) + + def utcoffset(self, value): # type: ignore[no-untyped-def] + """Return the currently configured offset.""" + del value + return self.offset + + def dst(self, value): # type: ignore[no-untyped-def] + """Keep daylight saving fixed.""" + del value + return timedelta(0) + + +class _ExplodingOffset(tzinfo): + """Raise arbitrary provider behavior while an evidence instant is resolved.""" + + def utcoffset(self, value): # type: ignore[no-untyped-def] + """Force the trust boundary to normalize provider failures.""" + del value + raise RuntimeError("provider details must not escape") + + def dst(self, value): # type: ignore[no-untyped-def] + """Keep daylight saving fixed if queried.""" + del value + return timedelta(0) + + +class _OversizedOffset(tzinfo): + """Return an extreme offset that cannot be detached from year one.""" + + def utcoffset(self, value): # type: ignore[no-untyped-def] + """Force UTC detachment outside representable datetime values.""" + del value + return timedelta(hours=23, minutes=59) + + def dst(self, value): # type: ignore[no-untyped-def] + """Keep daylight saving fixed.""" + del value + return timedelta(0) + + +def _source(retrieved_at: datetime) -> EvidenceSource: + """Build one governed source record.""" + return EvidenceSource( + source_uri="https://www.onetcenter.org/database.html", + source_title="O*NET Database", + source_version_code="onet:30.3", + retrieved_at=retrieved_at, + content_digest_sha256="b" * 64, + origin_code="authoritative_occupation_source", + ) + + +def _snapshot(*, effective_from: date, recorded_at: datetime, reviewed_at: datetime) -> JobAnalysisSnapshot: + """Build one otherwise-valid validated snapshot around supplied temporal evidence.""" + source = _source(datetime(2026, 8, 21, 5, 0, tzinfo=timezone.utc)) + return JobAnalysisSnapshot( + analysis_record_id=ANALYSIS_ID, + tenant_record_id=TENANT_ID, + job_record_id=JOB_ID, + analysis_version_code="analysis:v1", + status_code="analysis_validated", + effective_from=effective_from, + recorded_at=recorded_at, + tasks=( + TaskEvidence( + tenant_record_id=TENANT_ID, + job_record_id=JOB_ID, + task_record_id=TASK_ID, + task_statement="Analyze governed workforce evidence and document findings.", + importance_level=5, + difficulty_level=4, + source=source, + ), + ), + ksao_requirements=( + KSAORequirement( + tenant_record_id=TENANT_ID, + job_record_id=JOB_ID, + ksao_record_id=KSAO_ID, + category_code="knowledge_requirement", + requirement_statement="Knowledge of governed workforce evidence and traceability.", + importance_level=5, + proficiency_level=4, + source=source, + ), + ), + task_ksao_links=( + TaskKSAOLink( + task_record_id=TASK_ID, + ksao_record_id=KSAO_ID, + relationship_strength=5, + essential_for_task=True, + ), + ), + fja_profile=FunctionalJobAnalysisProfile( + tenant_record_id=TENANT_ID, + job_record_id=JOB_ID, + data_function_code=2, + people_function_code=1, + things_function_code=7, + source=source, + ), + reviewed_by_reference="keyverse_subject:01JIOPSYCH", + reviewed_at=reviewed_at, + ) + + +def test_job_analysis_rejects_uuid_subclasses_before_identity_canonicalization() -> None: + """Caller-controlled UUID rendering cannot rewrite task/link evidence identity.""" + forged = _ForgedUUID("00000000-0000-4000-8000-000000002004") + with pytest.raises(ValueError, match="task_record_id must be a UUID"): + TaskKSAOLink( + task_record_id=forged, + ksao_record_id=KSAO_ID, + relationship_strength=5, + essential_for_task=True, + ) + + +def test_evidence_source_rejects_datetime_subclass_before_provenance_canonicalization() -> None: + """Reject caller-controlled timestamp rendering at the source-provenance boundary.""" + forged = _ForgedDateTime(2026, 8, 21, 5, 0, tzinfo=timezone.utc) + with pytest.raises(ValueError, match="retrieved_at must be a datetime"): + _source(forged) + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("source_uri", _OpaqueText("https://www.onetcenter.org/database.html")), + ("source_title", _OpaqueText("O*NET Database")), + ("source_version_code", _OpaqueText("onet:30.3")), + ("content_digest_sha256", _OpaqueText("b" * 64)), + ], +) +def test_evidence_source_rejects_string_subclasses_before_canonicalization( + field_name: str, value: str +) -> None: + """Reject caller-controlled string behavior in source provenance fields.""" + source = _source(datetime(2026, 8, 21, 5, 0, tzinfo=timezone.utc)) + with pytest.raises(ValueError, match="must be"): + replace(source, **{field_name: value}) + + +def test_snapshot_rejects_reference_string_subclasses_before_canonicalization() -> None: + """Reject caller-controlled runtime behavior in accountable review references.""" + snapshot = _snapshot( + effective_from=date(2026, 8, 1), + recorded_at=RECORDED_AT, + reviewed_at=RECORDED_AT, + ) + with pytest.raises(ValueError, match="reviewed_by_reference must be a string"): + replace(snapshot, reviewed_by_reference=_OpaqueText("keyverse_subject:01JIOPSYCH")) + + +def test_snapshot_rejects_task_and_ksao_subclasses_before_canonicalization() -> None: + """Reject nested evidence subclasses before their fields reach canonical serialization.""" + snapshot = _snapshot( + effective_from=date(2026, 8, 1), + recorded_at=RECORDED_AT, + reviewed_at=RECORDED_AT, + ) + task = snapshot.tasks[0] + forged_task = _TaskEvidenceSubclass( + tenant_record_id=task.tenant_record_id, + job_record_id=task.job_record_id, + task_record_id=task.task_record_id, + task_statement=task.task_statement, + importance_level=task.importance_level, + difficulty_level=task.difficulty_level, + source=task.source, + ) + with pytest.raises(ValueError, match="tasks must contain TaskEvidence"): + replace(snapshot, tasks=(forged_task,)) + + ksao = snapshot.ksao_requirements[0] + forged_ksao = _KSAORequirementSubclass( + tenant_record_id=ksao.tenant_record_id, + job_record_id=ksao.job_record_id, + ksao_record_id=ksao.ksao_record_id, + category_code=ksao.category_code, + requirement_statement=ksao.requirement_statement, + importance_level=ksao.importance_level, + proficiency_level=ksao.proficiency_level, + source=ksao.source, + ) + with pytest.raises(ValueError, match="ksao_requirements must contain KSAORequirement"): + replace(snapshot, ksao_requirements=(forged_ksao,)) + + +@pytest.mark.parametrize( + ("effective_from", "recorded_at", "reviewed_at"), + [ + (_ForgedDate(2026, 8, 1), RECORDED_AT, RECORDED_AT), + (date(2026, 8, 1), _ForgedDateTime(2026, 8, 21, 5, 15, tzinfo=timezone.utc), RECORDED_AT), + (date(2026, 8, 1), RECORDED_AT, _ForgedDateTime(2026, 8, 21, 5, 15, tzinfo=timezone.utc)), + ], +) +def test_snapshot_rejects_temporal_subclasses_before_canonicalization( + effective_from: date, + recorded_at: datetime, + reviewed_at: datetime, +) -> None: + """Reject business/recorded-time objects whose methods can rewrite immutable evidence.""" + with pytest.raises(ValueError): + _snapshot( + effective_from=effective_from, + recorded_at=recorded_at, + reviewed_at=reviewed_at, + ) + + +def test_snapshot_rejects_status_subclass_that_forges_allow_list_membership() -> None: + """An ungoverned status cannot masquerade as draft while serializing different text.""" + snapshot = _snapshot( + effective_from=date(2026, 8, 1), + recorded_at=RECORDED_AT, + reviewed_at=RECORDED_AT, + ) + with pytest.raises(ValueError, match="status_code must be a string"): + replace(snapshot, status_code=_ForgedStatusCode("shadow_state")) + + +def test_evidence_source_rejects_origin_subclass_that_forges_allow_list_membership() -> None: + """Provenance classification cannot pass as authoritative under different serialized text.""" + source = _source(datetime(2026, 8, 21, 5, 0, tzinfo=timezone.utc)) + with pytest.raises(ValueError, match="origin_code must be a string"): + replace(source, origin_code=_ForgedOriginCode("shadow_origin")) + + +def test_evidence_source_detaches_mutable_timezone_state() -> None: + """Keep source provenance chronology stable after timezone state mutates.""" + zone = _MutableOffset() + source = _source(datetime(2026, 8, 21, 5, 0, tzinfo=zone)) + first = source.retrieved_at + + zone.offset = timedelta(hours=9) + + assert source.retrieved_at == first + assert source.retrieved_at.tzinfo is timezone.utc + + +def test_snapshot_detaches_mutable_timezone_state() -> None: + """Keep snapshot chronology stable after caller-owned timezone state mutates.""" + zone = _MutableOffset() + snapshot = _snapshot( + effective_from=date(2026, 8, 1), + recorded_at=datetime(2026, 8, 21, 5, 15, tzinfo=zone), + reviewed_at=RECORDED_AT, + ) + first = snapshot.to_snapshot() + + zone.offset = timedelta(hours=9) + + assert snapshot.recorded_at.tzinfo is timezone.utc + assert snapshot.to_snapshot() == first + + +def test_evidence_source_normalizes_timezone_provider_exceptions() -> None: + """Do not leak arbitrary timezone-provider exceptions from source construction.""" + with pytest.raises(ValueError, match="retrieved_at must resolve to a UTC offset"): + _source(datetime(2026, 8, 21, 5, 0, tzinfo=_ExplodingOffset())) + + +def test_evidence_source_normalizes_offset_overflow_to_value_error() -> None: + """Fail closed when UTC detachment exceeds representable datetime values.""" + with pytest.raises(ValueError, match="retrieved_at must be a representable"): + _source(datetime(1, 1, 1, 0, 0, tzinfo=_OversizedOffset())) + + +def test_snapshot_canonicalization_rejects_reintroduced_timezone_behavior() -> None: + """Fail closed if low-level mutation reintroduces executable timezone behavior.""" + snapshot = _snapshot( + effective_from=date(2026, 8, 1), + recorded_at=RECORDED_AT, + reviewed_at=RECORDED_AT, + ) + object.__setattr__( + snapshot, + "recorded_at", + datetime(2026, 8, 21, 5, 15, tzinfo=_MutableOffset()), + ) + with pytest.raises(ValueError, match="exact timezone-aware datetime"): + snapshot.to_snapshot() + + +def test_task_rejects_integer_subclass_that_forges_level_bounds() -> None: + """Out-of-range ordinal evidence cannot override comparisons and serialize as valid.""" + source = _source(datetime(2026, 8, 21, 5, 0, tzinfo=timezone.utc)) + with pytest.raises(ValueError, match="importance_level must be an integer"): + TaskEvidence( + tenant_record_id=TENANT_ID, + job_record_id=JOB_ID, + task_record_id=TASK_ID, + task_statement="Analyze governed workforce evidence and document findings.", + importance_level=_ForgedLevel(99), + difficulty_level=4, + source=source, + )