fix(core): protect canonical evidence runtime types

CHANGELOG.md

@@ -12,6 +12,7 @@ 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.

packages/hris-kernel/src/orgmetra_hris_kernel/audit.py

@@ -11,7 +11,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from hashlib import sha256
 import json
 import re
@@ -34,6 +34,29 @@
 _ALL_REQUIRED_TEXT_FIELDS = ("source_service", "event_type", *_REQUIRED_TEXT_FIELDS)
 
 
+def _freeze_timestamp(value: datetime) -> datetime:
+    """Detach caller-controlled timezone behavior 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.")
+    try:
+        offset = value.utcoffset()
+    except Exception as exc:  # noqa: BLE001 - normalize provider behavior at trust boundary.
+        raise ValueError("occurred_at must resolve to a UTC offset.") from exc
+    if offset is None or type(offset) is not timedelta:
+        raise ValueError("occurred_at must resolve to a UTC offset.")
+    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
+
+
+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")
+
+
 @dataclass(frozen=True, slots=True)
 class AuditOutboxEvent:
     """One immutable governance envelope for an Orgmetra domain mutation.
@@ -61,9 +84,9 @@ class AuditOutboxEvent:
 
     def __post_init__(self) -> None:
         """Reject envelopes that cannot provide accountable, portable audit evidence."""
-        if not isinstance(self.event_id, UUID):
+        if type(self.event_id) is not UUID:
             raise ValueError("event_id must be a UUID.")
-        if not isinstance(self.tenant_record_id, UUID):
+        if type(self.tenant_record_id) is not 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.")
@@ -73,19 +96,16 @@ def __post_init__(self) -> None:
             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):
+        if type(self.occurred_at) is not 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):
+            if type(getattr(self, field_name)) is not str:
                 raise ValueError(f"{field_name} must be a string.")
-        if self.confirmation_reference is not None and not isinstance(self.confirmation_reference, str):
+        if self.confirmation_reference is not None and type(self.confirmation_reference) is not 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.")
+        object.__setattr__(self, "occurred_at", _freeze_timestamp(self.occurred_at))
         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:
@@ -118,14 +138,13 @@ 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)
         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"),
+            "time": _canonical_timestamp(self.occurred_at),
             "datacontenttype": "application/json",
             "orgmetratenant": str(self.tenant_record_id),
             "orgmetraactor": self.actor_reference,

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,27 +92,36 @@ 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")
     return value
 
 
 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,16 +307,17 @@ 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 item in (*self.tasks, *self.ksao_requirements, self.fja_profile):
@@ -331,7 +345,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 +365,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):