diff --git a/.github/workflows/foundation-ci.yml b/.github/workflows/foundation-ci.yml index 6b475d6f2..9a2906434 100644 --- a/.github/workflows/foundation-ci.yml +++ b/.github/workflows/foundation-ci.yml @@ -56,6 +56,45 @@ jobs: run: | python -m pip install --require-hashes --no-deps --only-binary=:all: -r .github/requirements/foundation-test.txt python -m pip check + - name: Run candidate offer response installed-artifact contract + run: | + printf '%s\n' 'setuptools==84.0.0 --hash=sha256:51a52592b3b99e102b609654876bd65f19f999935166d1352678931132b0c670' > /tmp/orgmetra-candidate-offer-response-build.txt + python -m pip install --require-hashes --no-deps --only-binary=:all: -r /tmp/orgmetra-candidate-offer-response-build.txt + rm -rf /tmp/orgmetra-candidate-offer-response-build /tmp/orgmetra-candidate-offer-response-dist /tmp/orgmetra-candidate-offer-response-venv + cp -a packages/candidate-offer-response /tmp/orgmetra-candidate-offer-response-build + mkdir -p /tmp/orgmetra-candidate-offer-response-dist + python -m pip wheel --no-deps --no-build-isolation --wheel-dir /tmp/orgmetra-candidate-offer-response-dist /tmp/orgmetra-candidate-offer-response-build + test "$(find /tmp/orgmetra-candidate-offer-response-dist -maxdepth 1 -type f -name '*.whl' | wc -l)" -eq 1 + python -m venv /tmp/orgmetra-candidate-offer-response-venv + /tmp/orgmetra-candidate-offer-response-venv/bin/python -m pip install --require-hashes --no-deps --only-binary=:all: -r "$GITHUB_WORKSPACE/.github/requirements/foundation-test.txt" + wheel_path="$(find /tmp/orgmetra-candidate-offer-response-dist -maxdepth 1 -type f -name '*.whl' -print -quit)" + wheel_sha="$(sha256sum "$wheel_path" | awk '{print $1}')" + printf 'orgmetra-candidate-offer-response[test] @ file://%s --hash=sha256:%s\n' "$wheel_path" "$wheel_sha" > /tmp/orgmetra-candidate-offer-response-install.txt + /tmp/orgmetra-candidate-offer-response-venv/bin/python -m pip install --require-hashes --no-deps -r /tmp/orgmetra-candidate-offer-response-install.txt + /tmp/orgmetra-candidate-offer-response-venv/bin/python -m pip check + /tmp/orgmetra-candidate-offer-response-venv/bin/python - <<'PY' + from importlib.metadata import metadata + from pathlib import Path + import coverage + import pytest + import pytest_cov + import orgmetra_candidate_offer_response + + venv_root = Path("/tmp/orgmetra-candidate-offer-response-venv").resolve() + module_path = Path(orgmetra_candidate_offer_response.__file__).resolve() + if not module_path.is_relative_to(venv_root): + raise SystemExit(f"package imported outside isolated environment: {module_path}") + for module in (coverage, pytest, pytest_cov): + dependency_path = Path(module.__file__).resolve() + if not dependency_path.is_relative_to(venv_root): + raise SystemExit( + f"test dependency imported outside isolated environment: {dependency_path}" + ) + if "test" not in (metadata("orgmetra-candidate-offer-response").get_all("Provides-Extra") or []): + raise SystemExit("built distribution does not expose the reviewed test extra") + PY + cd /tmp + COVERAGE_FILE=/tmp/orgmetra-candidate-offer-response.coverage /tmp/orgmetra-candidate-offer-response-venv/bin/python -m pytest -c "$GITHUB_WORKSPACE/packages/candidate-offer-response/pyproject.toml" "$GITHUB_WORKSPACE/packages/candidate-offer-response/tests" - name: Run owned unit and service contracts once run: | PYTHONPATH=packages/candidate-evidence/src COVERAGE_FILE=/tmp/orgmetra-candidate-evidence.coverage python -m pytest -c packages/candidate-evidence/pyproject.toml packages/candidate-evidence/tests diff --git a/docs/doctoring/candidate-offer-response-references.md b/docs/doctoring/candidate-offer-response-references.md new file mode 100644 index 000000000..55814fa9a --- /dev/null +++ b/docs/doctoring/candidate-offer-response-references.md @@ -0,0 +1,33 @@ +# Candidate Offer Response — Standards and Research Notes + +## Evidence status + +These references inform the candidate-response trust boundary. They do **not** claim that Orgmetra, an identity provider, or a customer deployment is NIST-certified, NIST-conformant, SOC 2 certified, or compliant with any employment law merely because the design cites them. + +## Design implications + +1. **Identity must be evidence, not a caller assertion.** NIST SP 800-63 Revision 4 is the current final Digital Identity Guidelines suite (July 2025) and treats identity proofing, authentication, federation, security, privacy, and customer experience as risk-managed digital identity functions. Orgmetra therefore records only an opaque candidate actor plus identity-resolution evidence and requires authoritative re-resolution before relying on the response. +2. **Minimize candidate data at the response boundary.** NIST Privacy Framework 1.0 is a risk- and outcome-based enterprise privacy framework. The candidate-response packet excludes candidate PII, compensation values and free-form decline reasons because those values are unnecessary to prove the response event itself. +3. **Do not invent a UUID version for an external identity.** OpenID Connect Core defines `sub` as a case-sensitive, locally unique, never-reassigned subject string of at most 255 ASCII characters and relies on the `(iss, sub)` pair for stable cross-issuer identity. Keyverse protected-main product requirements likewise make exact `(identity_provider, subject)` the strongest matching evidence and require RPs to validate issuer and subject; they do not publish a UUIDv4-only subject contract. Orgmetra's protected Keyverse adapter therefore accepts a namespaced opaque `actor_reference`. Candidate-offer-response preserves that owner boundary: its `candidate_actor_reference` is bounded namespaced opaque text and is re-resolved authoritatively before consequential use. +4. **Packet-owned references keep their explicit identifier contract.** RFC 9562 is the current standards-track UUID specification and obsoletes RFC 4122. Orgmetra-owned packet/evidence correlation references in this slice use canonical non-sentinel UUIDv4 suffixes; the Orgmetra tenant identifier remains an authoritative operational UUID and may use the repository's UUIDv7 convention. The UUIDv4 rule is not projected onto the externally owned candidate actor. +5. **Acceptance is not employment authority.** Digital identity evidence establishes who acted; it does not establish that an approved offer is still eligible, unique, unsuperseded, or sufficient to create employment. Those facts stay at their authoritative Orgmetra boundaries and must be re-resolved before consequential mutation. + +## APA 7 references + +Davis, K., Peabody, B., & Leach, P. (2024). *Universally unique IDentifiers (UUIDs)* (RFC 9562). Internet Engineering Task Force. https://doi.org/10.17487/RFC9562 + +National Institute of Standards and Technology. (2020). *NIST Privacy Framework: A tool for improving privacy through enterprise risk management, version 1.0* (NIST CSWP 10). U.S. Department of Commerce. https://doi.org/10.6028/NIST.CSWP.10 + +Temoshok, D., Proud-Madruga, D., Choong, Y.-Y., Galluzzo, R., Gupta, S., LaSalle, C., Lefkovitz, N., & Regenscheid, A. (2025). *Digital identity guidelines* (NIST Special Publication 800-63-4). National Institute of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-63-4 + +OpenID Foundation. (2014). *OpenID Connect Core 1.0 incorporating errata set 2*. https://openid.net/specs/openid-connect-core-1_0.html + +## Primary-source verification + +- CSRC records SP 800-63-4 with document date **July 2025** (`Date Published: July 2025`, document-history final entry `07/31/25`); NIST's public announcement of the final suite followed on August 1, 2025. The two dates refer to different events and are both retained here. +- Official CSRC author order for SP 800-63-4: David Temoshok, Diana Proud-Madruga, Yee-Yin Choong, Ryan Galluzzo, Sarbari Gupta, Connie LaSalle, Naomi Lefkovitz, Andrew Regenscheid. The APA entry preserves this exact order. +- NIST published final SP 800-63 Revision 4 in July 2025; it supersedes SP 800-63-3. +- NIST Privacy Framework 1.0 was published January 16, 2020 and remains the final 1.0 publication while newer Privacy Framework work is developed separately. +- RFC 9562 was published May 2024 as an IETF Standards Track RFC and obsoletes RFC 4122. +- OpenID Connect Core specifies `sub` as a case-sensitive string, not a UUID, and makes `(iss, sub)` the stable identity pair available to the relying party. +- Keyverse protected `main` documents exact `(identity_provider, subject)` matching and issuer/subject validation without a UUIDv4-only subject guarantee; Orgmetra does not mutate Keyverse to change that contract. diff --git a/docs/traceability/candidate-offer-response.md b/docs/traceability/candidate-offer-response.md new file mode 100644 index 000000000..b1439a016 --- /dev/null +++ b/docs/traceability/candidate-offer-response.md @@ -0,0 +1,43 @@ +# Candidate Offer Response Traceability + +## State legend + +- **Protected-main truth**: already present on protected `develop` when this lane was cut. +- **Active PR**: implemented only on the candidate-offer-response branch until merged. +- **Dependency contract**: read-only interface owned elsewhere. +- **Out of scope**: intentionally not claimed by this slice. + +## Requirement mapping + +| Requirement | State | Evidence | +|---|---|---| +| Human offer approval is separate from candidate response | Protected-main truth | `packages/offer-approval`; ADR 0017 | +| Candidate-to-worker/confirmed-hire materialization is separately governed | Protected-main truth | `database/migrations/0009_candidate_worker_conversion_governance.sql`; People mutation boundary | +| Candidate response binds exact approved-offer and offer-terms digests | Active PR | `packages/candidate-offer-response/src/orgmetra_candidate_offer_response/response.py` | +| Acceptance and decline are both candidate-originated, closed-vocabulary evidence | Active PR | `response_code` allow-list plus adversarial tests | +| Employer-side shadow rejection is contractually barred, and the bar is enforced only by mandatory host-side re-resolution | Active PR | `candidate_actor_reference` and `identity_resolution_reference` are mandatory and no employer actor vocabulary exists, but the packet does not itself authenticate the candidate; fixed `scope_verification_state=requires_authoritative_resolution` makes consequential use without authoritative re-resolution a contract violation for downstream callers | +| Candidate actor correlation follows the published identity-owner boundary rather than an invented UUID version | Active PR + dependency contract | protected-main `packages/keyverse-adapter` accepts namespaced opaque actor references; `test_external_identity_reference_contract.py` proves a non-UUID Keyverse-compatible actor reference remains valid | +| Candidate response never directly authorizes hire or employment mutation | Active PR | fixed `employment_effect=not_authorized_to_hire`; governed `next_action` | +| Candidate identity is re-resolved before consequential downstream use | Active PR + dependency contract | fixed `scope_verification_state=requires_authoritative_resolution`; Keyverse remains read-only | +| Candidate PII, compensation values and free-form decline reasons are excluded | Active PR | fixed false sensitivity flags and canonical payload tests | +| Evidence preserves candidate response time and system-recorded time | Active PR | detached UTC `responded_at` / `recorded_at`; chronology regression | +| Caller-defined scalar/time subclasses cannot forge canonical evidence | Active PR | exact runtime type checks and hostile-subclass regressions | +| Post-construction rewriting invalidates evidence | Active PR | creation-time canonical digest seal plus mutation regressions | +| Canonical export emits the same snapshot that passed integrity validation | Active PR | `_assert_integrity()` returns the checked canonical bytes; `test_checked_snapshot_integrity.py` reproduces an interleaving valid-value rewrite and requires the previously checked snapshot to be emitted | +| Exact 100% owned statement/branch coverage | Active PR | canonical `.github/workflows/foundation-ci.yml` builds the candidate-offer-response wheel, installs it by exact SHA-256 into an isolated venv, and executes `packages/candidate-offer-response/tests`; `test_artifact_execution.py` keeps the retired leaf workflow from returning | +| Keyverse credentials or source state are never persisted here | Dependency contract | existing `packages/keyverse-adapter`; candidate-response packet stores opaque identity-resolution evidence only | +| Actual identity proofing/authentication assurance selection | Out of scope | authoritative identity owner / relying-party risk assessment | +| Offer eligibility, expiry, supersession and authoritative uniqueness | Out of scope for packet; required next step | owning talent-acquisition/offer workflow must re-resolve before action | +| Employment creation, assignment creation, compensation execution, offer delivery | Out of scope | existing owning HRIS boundaries | + +## Architecture alignment + +This slice does not introduce a new cross-service persistence path or a new architecture decision. It implements the existing Orgmetra principles in ADR 0001 (authoritative HRIS record), ADR 0006 (governed immutable audit/outbox evidence), ADR 0008 (purpose-bound PII authorization), and ADR 0017 (governed offer approval). It therefore adds no competing numbered ADR and does not edit the active canonical ADR index. + +Keyverse remains read-only. The candidate actor is validated as a bounded namespaced opaque reference compatible with Orgmetra's protected-main Keyverse adapter; the response packet does not infer, rewrite, or constrain Keyverse's underlying OIDC `sub` to UUIDv4. `identity_resolution_reference` remains an Orgmetra-owned correlation reference with its explicit UUIDv4 contract and digest. + +The candidate-response canonicalizer validates one payload snapshot against the process-local issuance seal and returns that same snapshot. It does not validate one read and then serialize the mutable object again. This preserves checked-versus-emitted audit integrity even if a same-process caller uses low-level mutation between those two phases; any later export from the changed packet still fails closed against the original issuance seal. + +## Buyer outcome + +A recruiter can no longer treat an approved offer as implicitly accepted, and an employer-side caller has no legitimate contract path to manufacture a decline through the candidate response: the packet carries opaque candidate identity evidence but never authenticates the candidate itself, so every consequential use must first re-resolve the authoritative identity boundary. The next actionable state is explicit: re-resolve candidate identity and exact offer scope, then use the owning employment boundary if and only if the response is verified as authoritative and eligible. diff --git a/manifest.json b/manifest.json index f7b6cf55e..b1af91573 100644 --- a/manifest.json +++ b/manifest.json @@ -5,9 +5,9 @@ "files": [ { "path": ".github/workflows/foundation-ci.yml", - "sha256": "b6a4365936b66803a8112f034c77d53d33301a7a798ed4f68746a4f2d8b081d7", - "bytes": 6651, - "lines": 125 + "sha256": "f68fecb02c7ebbd6fb4b9d4befa852645e89d32cee8ff7f767dbb97c70f9a4ea", + "bytes": 9936, + "lines": 164 }, { "path": ".gitignore", diff --git a/packages/candidate-offer-response/CHANGELOG.md b/packages/candidate-offer-response/CHANGELOG.md new file mode 100644 index 000000000..ad5a777f3 --- /dev/null +++ b/packages/candidate-offer-response/CHANGELOG.md @@ -0,0 +1,14 @@ +# Changelog + +## 0.1.0 - Unreleased + +- Add candidate-originated `offer_accepted` / `offer_declined` evidence bound to exact offer approval and offer-terms digests. +- Require candidate actor and identity-resolution provenance while keeping Keyverse read-only. +- Preserve the published identity-owner contract by accepting a bounded namespaced opaque `candidate_actor_reference` instead of imposing an Orgmetra-invented UUIDv4 requirement on the external actor identity. +- Keep every response explicitly non-authorizing for hire, employment, compensation execution, or candidate-to-worker conversion. +- Exclude candidate PII, compensation values, free-form decline reasons, credentials, and model output from the evidence packet. +- Normalize recorded/responded instants to detached built-in UTC values, reject trust-bearing runtime subclasses, redact `repr`, and detect post-construction evidence rewriting. +- Bind canonical export to the exact snapshot that passed issuance-seal validation so an interleaving valid-value rewrite cannot become emitted audit evidence after the integrity check. +- Add exact 100% statement/branch coverage and exact-head CI for the owned package. +- Build a wheel and execute the quality suite against the SHA-256-bound installed artifact in a fully isolated virtual environment; install the reviewed hash-pinned pytest/coverage toolchain inside that environment and fail closed if package or test-tool imports resolve outside it. +- Retire the package-specific quality workflow after protected repository-workflow consolidation; preserve the same SHA-256-bound installed-wheel and isolated-toolchain contract inside the canonical one-job Foundation CI lane. diff --git a/packages/candidate-offer-response/README.md b/packages/candidate-offer-response/README.md new file mode 100644 index 000000000..3ff5e04ba --- /dev/null +++ b/packages/candidate-offer-response/README.md @@ -0,0 +1,57 @@ +# Orgmetra Candidate Offer Response + +This package records a **candidate-originated response to one exact reviewed offer** without turning that response into employment authority. + +## What it does + +`build_candidate_offer_response(...)` produces immutable, value-minimized evidence that binds: + +- one Orgmetra tenant; +- one candidate profile; +- one exact human-reviewed offer approval and SHA-256 digest; +- one exact offer-terms reference and SHA-256 digest; +- one authenticated candidate actor plus identity-resolution evidence; +- the closed response code `offer_accepted` or `offer_declined`; +- the candidate response instant and the later/equal system-recorded instant; and +- one bounded evidence version. + +The packet normalizes caller timestamps to built-in UTC values at construction, rejects caller-defined subclasses at trust-bearing scalar boundaries, redacts routine `repr`, and seals its canonical construction digest so later field rewriting is rejected before serialization. + +Orgmetra-owned packet/evidence references keep their reviewed canonical UUIDv4 suffix contract. `candidate_actor_reference` is different: it is an opaque actor correlation supplied by the approved identity boundary, so this package validates the exact `candidate:` namespace and a bounded opaque token without inventing a UUID version requirement. The authoritative identity boundary must still re-resolve the actor before consequential use. + +## What it deliberately does not do + +An accepted response is **not** authorization to hire, create employment, create an assignment, convert a candidate to a worker, send another offer, execute compensation, or mutate Keyverse. The packet therefore always carries `employment_effect=not_authorized_to_hire` and `scope_verification_state=requires_authoritative_resolution`. + +The packet also contains no candidate PII, compensation values, free-form decline reason, credentials, tokens, or LLM output. A decline is candidate-originated evidence; it is not an employer-side rejection shortcut. + +## Required next action + +Before any consequential action, the authoritative workflow must re-resolve the candidate actor through the approved identity boundary, re-resolve the exact offer approval and offer-terms digests, verify that the offer was eligible for response at `responded_at`, and establish that this is the authoritative candidate response. Confirmed-hire materialization remains the responsibility of the existing People/candidate-to-worker boundary. + +Keyverse is a read-only dependency from this package's perspective. Orgmetra stores only the opaque identity-resolution evidence needed to correlate the candidate action; it stores no Keyverse credentials. + +## Example + +```python +from datetime import datetime, timezone +from orgmetra_candidate_offer_response import build_candidate_offer_response + +packet = build_candidate_offer_response( + tenant_record_id="018f6e2a-4f7c-7a1b-9c20-1f3a7d8e5b60", + offer_response_reference="candidate_offer_response:6ba7b810-9dad-4b11-80b4-00c04fd430c8", + candidate_profile_reference="candidate_profile:6ba7b811-9dad-4b11-80b4-00c04fd430c8", + offer_approval_reference="offer_approval:6ba7b812-9dad-4b11-80b4-00c04fd430c8", + offer_approval_digest="a" * 64, + offer_terms_reference="offer_terms:6ba7b813-9dad-4b11-80b4-00c04fd430c8", + offer_terms_digest="b" * 64, + candidate_actor_reference="candidate:AItOawmwtWwcT0k51BayewNvutrJUqsvl6qs7A4", + identity_resolution_reference="identity_resolution:6ba7b815-9dad-4b11-80b4-00c04fd430c8", + identity_resolution_digest="c" * 64, + response_code="offer_accepted", + responded_at=datetime(2026, 8, 22, 9, 30, tzinfo=timezone.utc), + recorded_at=datetime(2026, 8, 22, 9, 30, 1, tzinfo=timezone.utc), +) +``` + +Persist or publish `packet.canonical_json()` only through the owning Orgmetra audit/outbox boundary after authoritative scope resolution. Never treat `packet.sha256_digest()` alone as identity, approval, or hire authority. diff --git a/packages/candidate-offer-response/pyproject.toml b/packages/candidate-offer-response/pyproject.toml new file mode 100644 index 000000000..6b603fc8b --- /dev/null +++ b/packages/candidate-offer-response/pyproject.toml @@ -0,0 +1,24 @@ +[build-system] +requires = ["setuptools>=69"] +build-backend = "setuptools.build_meta" + +[project] +name = "orgmetra-candidate-offer-response" +version = "0.1.0" +description = "Governed candidate-originated offer-response evidence for Orgmetra." +requires-python = ">=3.12" + +[project.optional-dependencies] +test = ["pytest>=8.3", "pytest-cov>=5.0"] + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = [ + "--cov=orgmetra_candidate_offer_response", + "--cov-branch", + "--cov-report=term-missing", + "--cov-fail-under=100", +] diff --git a/packages/candidate-offer-response/src/orgmetra_candidate_offer_response/__init__.py b/packages/candidate-offer-response/src/orgmetra_candidate_offer_response/__init__.py new file mode 100644 index 000000000..349e2775e --- /dev/null +++ b/packages/candidate-offer-response/src/orgmetra_candidate_offer_response/__init__.py @@ -0,0 +1,5 @@ +"""Public candidate offer-response evidence contract for Orgmetra.""" + +from .response import CandidateOfferResponsePacket, build_candidate_offer_response + +__all__ = ["CandidateOfferResponsePacket", "build_candidate_offer_response"] diff --git a/packages/candidate-offer-response/src/orgmetra_candidate_offer_response/response.py b/packages/candidate-offer-response/src/orgmetra_candidate_offer_response/response.py new file mode 100644 index 000000000..e0033b471 --- /dev/null +++ b/packages/candidate-offer-response/src/orgmetra_candidate_offer_response/response.py @@ -0,0 +1,340 @@ +"""Governed, value-minimized candidate offer-response evidence. + +This module records what an authenticated candidate said about one exact reviewed offer. +It deliberately does not authorize employment creation, offer delivery, compensation +execution, or candidate-to-worker conversion. Authoritative services must re-resolve the +candidate identity and exact offer scope before taking any consequential action. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from hashlib import sha256 +import json +import re +from threading import RLock +from uuid import UUID +from weakref import finalize + +_DIGEST_PATTERN = re.compile(r"^[0-9a-f]{64}$") +_REFERENCE_PATTERN = re.compile(r"^[a-z][a-z0-9_]*:[A-Za-z0-9][A-Za-z0-9._~-]*$") +_ALLOWED_RESPONSE_CODES = frozenset({"offer_accepted", "offer_declined"}) +_DECISION_AUTHORITY = "candidate_response_only" +_SCOPE_VERIFICATION_STATE = "requires_authoritative_resolution" +_EMPLOYMENT_EFFECT = "not_authorized_to_hire" +_NEXT_ACTION = ( + "Within tenant_record_id, re-resolve candidate_actor_reference through the approved " + "identity boundary and re-resolve the exact offer approval and offer terms digests; " + "verify the offer was eligible for response at responded_at and that this evidence is " + "the authoritative candidate response before any communication, hire, employment, or " + "candidate-to-worker conversion action." +) +_CREATION_EVIDENCE_SEALS: dict[int, str] = {} +_CREATION_EVIDENCE_SEALS_LOCK = RLock() + + +def _discard_creation_evidence_seal(packet_id: int) -> None: + """Discard the process-local issuance seal when its packet is collected.""" + with _CREATION_EVIDENCE_SEALS_LOCK: + _CREATION_EVIDENCE_SEALS.pop(packet_id, None) + + +def _register_creation_evidence_seal(packet: object, digest: str) -> None: + """Bind one packet identity to its creation-time evidence outside writable slots.""" + packet_id = id(packet) + with _CREATION_EVIDENCE_SEALS_LOCK: + _CREATION_EVIDENCE_SEALS[packet_id] = digest + finalize(packet, _discard_creation_evidence_seal, packet_id) + + +def _creation_evidence_seal(packet: object) -> str: + """Return the authoritative process-local seal for a live governed packet.""" + with _CREATION_EVIDENCE_SEALS_LOCK: + seal = _CREATION_EVIDENCE_SEALS.get(id(packet)) + if seal is None: + raise ValueError("candidate offer response evidence has no issuance seal") + return seal + + +def _validate_operational_uuid(value: str, field_name: str) -> None: + """Require exact canonical non-sentinel UUID text for an HRIS-owned identifier.""" + if type(value) is not str: + raise ValueError(f"{field_name} must be canonical UUID text") + try: + parsed = UUID(value) + except (ValueError, AttributeError, TypeError) as exc: + raise ValueError(f"{field_name} must be canonical UUID text") from exc + if str(parsed) != value: + raise ValueError(f"{field_name} must be canonical UUID text") + if parsed.int in (0, (1 << 128) - 1): + raise ValueError(f"{field_name} must be a canonical operational UUID") + + +def _validate_reference( + value: str, + prefix: str, + field_name: str, + *, + require_uuid4: bool = True, +) -> None: + """Require one bounded namespace and UUIDv4 only when Orgmetra owns that contract.""" + max_length = 160 if require_uuid4 else 288 + reference_kind = "UUIDv4" if require_uuid4 else "opaque" + error_message = f"{field_name} must be a bounded {prefix}: {reference_kind} reference" + if ( + type(value) is not str + or len(value) > max_length + or not _REFERENCE_PATTERN.fullmatch(value) + or not value.startswith(f"{prefix}:") + ): + raise ValueError(error_message) + if not require_uuid4: + return + suffix = value.split(":", 1)[1] + try: + parsed = UUID(suffix) + except (ValueError, AttributeError, TypeError) as exc: + raise ValueError(error_message) from exc + if str(parsed) != suffix or parsed.version != 4 or parsed.int in (0, (1 << 128) - 1): + raise ValueError(error_message) + + +def _validate_digest(value: str, field_name: str) -> None: + """Require exact built-in lowercase SHA-256 hexadecimal evidence.""" + if type(value) is not str or not _DIGEST_PATTERN.fullmatch(value): + raise ValueError(f"{field_name} must be lowercase SHA-256 hex") + + +def _freeze_timestamp(value: datetime, field_name: str) -> datetime: + """Detach caller-controlled timezone behavior and return one exact UTC instant.""" + if type(value) is not datetime or value.tzinfo is None: + raise ValueError(f"{field_name} must be an exact timezone-aware datetime") + try: + offset = value.utcoffset() + except Exception: + raise ValueError(f"{field_name} must have a valid timezone offset") from None + if offset is None: + raise ValueError(f"{field_name} must have a valid timezone offset") + try: + utc_naive = value.replace(tzinfo=None) - offset + except OverflowError: + raise ValueError(f"{field_name} must have a valid timezone offset") from None + return utc_naive.replace(tzinfo=timezone.utc) + + +def _validate_canonical_timestamp(value: datetime, field_name: str) -> None: + """Require the already-detached built-in UTC timestamp used by canonical evidence.""" + if type(value) is not datetime or value.tzinfo is not timezone.utc: + raise ValueError(f"{field_name} must remain a canonical UTC datetime") + + +def _canonical_timestamp(value: datetime) -> str: + """Render a previously validated UTC instant as precision-preserving RFC 3339 text.""" + return value.isoformat().replace("+00:00", "Z") + + +def _validate_evidence_version(value: int) -> None: + """Require one bounded built-in integer evidence version.""" + if type(value) is not int or value < 1 or value > 2_147_483_647: + raise ValueError("evidence_version must be an integer from 1 through 2147483647") + + +def _validate_fixed_text(value: str, expected: str, field_name: str) -> None: + """Prevent runtime-polymorphic text from forging fixed governance evidence.""" + if type(value) is not str or value != expected: + raise ValueError(f"{field_name} must remain {expected}") + + +@dataclass(frozen=True, slots=True, repr=False, weakref_slot=True) +class CandidateOfferResponsePacket: + """Immutable candidate-originated offer response awaiting authoritative re-resolution.""" + + tenant_record_id: str + offer_response_reference: str + candidate_profile_reference: str + offer_approval_reference: str + offer_approval_digest: str + offer_terms_reference: str + offer_terms_digest: str + candidate_actor_reference: str + identity_resolution_reference: str + identity_resolution_digest: str + response_code: str + responded_at: datetime + recorded_at: datetime + evidence_version: int = 1 + contains_candidate_pii: bool = False + contains_compensation_values: bool = False + contains_free_form_reason: bool = False + candidate_confirmation_required: bool = True + decision_authority: str = _DECISION_AUTHORITY + scope_verification_state: str = _SCOPE_VERIFICATION_STATE + employment_effect: str = _EMPLOYMENT_EFFECT + next_action: str = _NEXT_ACTION + _creation_evidence_digest: str = field(init=False, repr=False, compare=False) + + def __init_subclass__(cls, **kwargs: object) -> None: + """Keep the trust-bearing packet runtime-final rather than polymorphic.""" + raise TypeError("CandidateOfferResponsePacket is final") + + def __repr__(self) -> str: + """Return a representation that emits no candidate or offer correlation evidence.""" + return "CandidateOfferResponsePacket()" + + def __post_init__(self) -> None: + """Validate inputs, detach timezone behavior, and seal canonical construction evidence.""" + object.__setattr__(self, "responded_at", _freeze_timestamp(self.responded_at, "responded_at")) + object.__setattr__(self, "recorded_at", _freeze_timestamp(self.recorded_at, "recorded_at")) + self._validate_live() + creation_digest = self._raw_sha256_digest() + object.__setattr__(self, "_creation_evidence_digest", creation_digest) + _register_creation_evidence_seal(self, creation_digest) + + def _validate_live(self) -> None: + """Fail closed if direct construction or later rewriting drifts from the contract.""" + _validate_operational_uuid(self.tenant_record_id, "tenant_record_id") + _validate_reference( + self.offer_response_reference, + "candidate_offer_response", + "offer_response_reference", + ) + _validate_reference( + self.candidate_profile_reference, + "candidate_profile", + "candidate_profile_reference", + ) + _validate_reference( + self.offer_approval_reference, + "offer_approval", + "offer_approval_reference", + ) + _validate_digest(self.offer_approval_digest, "offer_approval_digest") + _validate_reference(self.offer_terms_reference, "offer_terms", "offer_terms_reference") + _validate_digest(self.offer_terms_digest, "offer_terms_digest") + _validate_reference( + self.candidate_actor_reference, + "candidate", + "candidate_actor_reference", + require_uuid4=False, + ) + _validate_reference( + self.identity_resolution_reference, + "identity_resolution", + "identity_resolution_reference", + ) + _validate_digest(self.identity_resolution_digest, "identity_resolution_digest") + if type(self.response_code) is not str or self.response_code not in _ALLOWED_RESPONSE_CODES: + raise ValueError("response_code must be offer_accepted or offer_declined") + _validate_canonical_timestamp(self.responded_at, "responded_at") + _validate_canonical_timestamp(self.recorded_at, "recorded_at") + if self.recorded_at < self.responded_at: + raise ValueError("recorded_at must not precede responded_at") + _validate_evidence_version(self.evidence_version) + if self.contains_candidate_pii is not False: + raise ValueError("candidate offer response evidence must not contain candidate PII") + if self.contains_compensation_values is not False: + raise ValueError("candidate offer response evidence must not contain compensation values") + if self.contains_free_form_reason is not False: + raise ValueError("candidate offer response evidence must not contain a free-form reason") + if self.candidate_confirmation_required is not True: + raise ValueError("candidate confirmation is mandatory for candidate offer response evidence") + _validate_fixed_text(self.decision_authority, _DECISION_AUTHORITY, "decision_authority") + _validate_fixed_text( + self.scope_verification_state, + _SCOPE_VERIFICATION_STATE, + "scope_verification_state", + ) + _validate_fixed_text(self.employment_effect, _EMPLOYMENT_EFFECT, "employment_effect") + _validate_fixed_text(self.next_action, _NEXT_ACTION, "next_action") + + def _payload(self) -> dict[str, object]: + """Return the exact value-minimized payload used for audit correlation.""" + return { + "candidate_actor_reference": self.candidate_actor_reference, + "candidate_confirmation_required": self.candidate_confirmation_required, + "candidate_profile_reference": self.candidate_profile_reference, + "contains_candidate_pii": self.contains_candidate_pii, + "contains_compensation_values": self.contains_compensation_values, + "contains_free_form_reason": self.contains_free_form_reason, + "decision_authority": self.decision_authority, + "employment_effect": self.employment_effect, + "evidence_version": self.evidence_version, + "identity_resolution_digest": self.identity_resolution_digest, + "identity_resolution_reference": self.identity_resolution_reference, + "next_action": self.next_action, + "offer_approval_digest": self.offer_approval_digest, + "offer_approval_reference": self.offer_approval_reference, + "offer_response_reference": self.offer_response_reference, + "offer_terms_digest": self.offer_terms_digest, + "offer_terms_reference": self.offer_terms_reference, + "recorded_at": _canonical_timestamp(self.recorded_at), + "responded_at": _canonical_timestamp(self.responded_at), + "response_code": self.response_code, + "scope_verification_state": self.scope_verification_state, + "tenant_record_id": self.tenant_record_id, + } + + def _raw_canonical_json(self) -> str: + """Serialize live fields without recursively invoking the integrity check.""" + return json.dumps(self._payload(), sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + def _raw_sha256_digest(self) -> str: + """Hash live canonical bytes without recursively invoking the integrity check.""" + return sha256(self._raw_canonical_json().encode("utf-8")).hexdigest() + + def _assert_integrity(self) -> str: + """Validate and return the exact canonical snapshot that passed the issuance check.""" + self._validate_live() + canonical_json = self._raw_canonical_json() + authoritative_seal = _creation_evidence_seal(self) + if ( + sha256(canonical_json.encode("utf-8")).hexdigest() != authoritative_seal + or self._creation_evidence_digest != authoritative_seal + ): + raise ValueError("candidate offer response evidence changed after construction") + return canonical_json + + def canonical_json(self) -> str: + """Return the exact deterministic snapshot that passed creation-time integrity.""" + return self._assert_integrity() + + def sha256_digest(self) -> str: + """Return SHA-256 over exact canonical UTF-8 offer-response evidence.""" + return sha256(self.canonical_json().encode("utf-8")).hexdigest() + + +def build_candidate_offer_response( + *, + tenant_record_id: str, + offer_response_reference: str, + candidate_profile_reference: str, + offer_approval_reference: str, + offer_approval_digest: str, + offer_terms_reference: str, + offer_terms_digest: str, + candidate_actor_reference: str, + identity_resolution_reference: str, + identity_resolution_digest: str, + response_code: str, + responded_at: datetime, + recorded_at: datetime, + evidence_version: int = 1, +) -> CandidateOfferResponsePacket: + """Build candidate-originated response evidence without granting downstream authority.""" + return CandidateOfferResponsePacket( + tenant_record_id=tenant_record_id, + offer_response_reference=offer_response_reference, + candidate_profile_reference=candidate_profile_reference, + offer_approval_reference=offer_approval_reference, + offer_approval_digest=offer_approval_digest, + offer_terms_reference=offer_terms_reference, + offer_terms_digest=offer_terms_digest, + candidate_actor_reference=candidate_actor_reference, + identity_resolution_reference=identity_resolution_reference, + identity_resolution_digest=identity_resolution_digest, + response_code=response_code, + responded_at=responded_at, + recorded_at=recorded_at, + evidence_version=evidence_version, + ) diff --git a/packages/candidate-offer-response/tests/test_artifact_execution.py b/packages/candidate-offer-response/tests/test_artifact_execution.py new file mode 100644 index 000000000..c0eb3048e --- /dev/null +++ b/packages/candidate-offer-response/tests/test_artifact_execution.py @@ -0,0 +1,33 @@ +"""Regression contract for consolidated exact installed-wheel quality execution.""" + +from pathlib import Path + + +_REPOSITORY_ROOT = Path(__file__).resolve().parents[3] +_FOUNDATION_WORKFLOW = _REPOSITORY_ROOT / ".github/workflows/foundation-ci.yml" +_RETIRED_LEAF_WORKFLOW = ( + _REPOSITORY_ROOT / ".github/workflows/candidate-offer-response-quality.yml" +) +_VENV_PATH = "/tmp/orgmetra-candidate-offer-response-venv" + + +def test_foundation_executes_the_hash_bound_installed_wheel() -> None: + """Keep artifact parity inside the canonical one-job Foundation lane.""" + workflow = _FOUNDATION_WORKFLOW.read_text(encoding="utf-8") + + assert not _RETIRED_LEAF_WORKFLOW.exists() + assert "Run candidate offer response installed-artifact contract" in workflow + assert "PYTHONPATH=packages/candidate-offer-response/src" not in workflow + assert f"python -m venv {_VENV_PATH}" in workflow + assert ( + f'{_VENV_PATH}/bin/python -m pip install --require-hashes --no-deps ' + f'--only-binary=:all: -r "$GITHUB_WORKSPACE/.github/requirements/foundation-test.txt"' + in workflow + ) + assert 'wheel_sha="$(sha256sum "$wheel_path" | awk \'{print $1}\')"' in workflow + assert "for module in (coverage, pytest, pytest_cov):" in workflow + assert ( + f"{_VENV_PATH}/bin/python -m pytest " + '-c "$GITHUB_WORKSPACE/packages/candidate-offer-response/pyproject.toml"' + in workflow + ) diff --git a/packages/candidate-offer-response/tests/test_checked_snapshot_integrity.py b/packages/candidate-offer-response/tests/test_checked_snapshot_integrity.py new file mode 100644 index 000000000..c29434b07 --- /dev/null +++ b/packages/candidate-offer-response/tests/test_checked_snapshot_integrity.py @@ -0,0 +1,51 @@ +"""Regression for checked-versus-emitted candidate offer-response evidence.""" + +from __future__ import annotations + +from datetime import datetime, timezone +import json + +import pytest + +from orgmetra_candidate_offer_response.response import ( + CandidateOfferResponsePacket, + build_candidate_offer_response, +) + + +def _packet() -> CandidateOfferResponsePacket: + """Build one valid accepted-offer packet for snapshot-integrity testing.""" + return build_candidate_offer_response( + tenant_record_id="018f6e2a-4f7c-7a1b-9c20-1f3a7d8e5b60", + offer_response_reference="candidate_offer_response:6ba7b810-9dad-4b11-80b4-00c04fd430c8", + candidate_profile_reference="candidate_profile:6ba7b811-9dad-4b11-80b4-00c04fd430c8", + offer_approval_reference="offer_approval:6ba7b812-9dad-4b11-80b4-00c04fd430c8", + offer_approval_digest="a" * 64, + offer_terms_reference="offer_terms:6ba7b813-9dad-4b11-80b4-00c04fd430c8", + offer_terms_digest="b" * 64, + candidate_actor_reference="candidate:6ba7b814-9dad-4b11-80b4-00c04fd430c8", + identity_resolution_reference="identity_resolution:6ba7b815-9dad-4b11-80b4-00c04fd430c8", + identity_resolution_digest="c" * 64, + response_code="offer_accepted", + responded_at=datetime(2026, 8, 22, 9, 30, 15, tzinfo=timezone.utc), + recorded_at=datetime(2026, 8, 22, 9, 30, 16, tzinfo=timezone.utc), + ) + + +def test_canonical_json_emits_the_same_snapshot_that_passed_integrity_check( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A valid rewrite after checking must not become the emitted canonical truth.""" + packet = _packet() + original_assert_integrity = CandidateOfferResponsePacket._assert_integrity + + def rewrite_after_check(self: CandidateOfferResponsePacket) -> object: + """Simulate an interleaving rewrite immediately after the integrity check.""" + checked_snapshot = original_assert_integrity(self) + object.__setattr__(self, "response_code", "offer_declined") + return checked_snapshot + + monkeypatch.setattr(CandidateOfferResponsePacket, "_assert_integrity", rewrite_after_check) + + document = json.loads(packet.canonical_json()) + assert document["response_code"] == "offer_accepted" diff --git a/packages/candidate-offer-response/tests/test_creation_seal_integrity.py b/packages/candidate-offer-response/tests/test_creation_seal_integrity.py new file mode 100644 index 000000000..911818b4d --- /dev/null +++ b/packages/candidate-offer-response/tests/test_creation_seal_integrity.py @@ -0,0 +1,59 @@ +"""Regression for creation-seal tamper resistance in candidate offer responses.""" + +from copy import copy, deepcopy +from datetime import datetime, timezone +import pickle + +import pytest + +from orgmetra_candidate_offer_response.response import build_candidate_offer_response + + +def _issued_packet(): + """Return one freshly issued governed candidate offer response.""" + return build_candidate_offer_response( + tenant_record_id="018f6e2a-4f7c-7a1b-9c20-1f3a7d8e5b60", + offer_response_reference="candidate_offer_response:6ba7b810-9dad-4b11-80b4-00c04fd430c8", + candidate_profile_reference="candidate_profile:6ba7b811-9dad-4b11-80b4-00c04fd430c8", + offer_approval_reference="offer_approval:6ba7b812-9dad-4b11-80b4-00c04fd430c8", + offer_approval_digest="a" * 64, + offer_terms_reference="offer_terms:6ba7b813-9dad-4b11-80b4-00c04fd430c8", + offer_terms_digest="b" * 64, + candidate_actor_reference="candidate:6ba7b814-9dad-4b11-80b4-00c04fd430c8", + identity_resolution_reference="identity_resolution:6ba7b815-9dad-4b11-80b4-00c04fd430c8", + identity_resolution_digest="c" * 64, + response_code="offer_accepted", + responded_at=datetime(2026, 8, 22, 9, 30, 15, tzinfo=timezone.utc), + recorded_at=datetime(2026, 8, 22, 9, 30, 16, tzinfo=timezone.utc), + ) + + +def test_creation_seal_cannot_be_rewritten_with_payload() -> None: + """A caller must not turn post-issuance rewrites into freshly valid evidence.""" + packet = _issued_packet() + + object.__setattr__(packet, "response_code", "offer_declined") + forged_live_digest = packet._raw_sha256_digest() # noqa: SLF001 - adversarial regression + object.__setattr__(packet, "_creation_evidence_digest", forged_live_digest) + + with pytest.raises(ValueError, match="offer response evidence changed after construction"): + packet.canonical_json() + + +@pytest.mark.parametrize( + "clone_factory", + [ + copy, + deepcopy, + lambda packet: pickle.loads(pickle.dumps(packet)), + ], + ids=["copy", "deepcopy", "pickle_round_trip"], +) +def test_cloned_packets_fail_closed_without_issuance_seal(clone_factory) -> None: + """Copies bypass issuance, so they must fail closed with an explicit error.""" + packet = _issued_packet() + clone = clone_factory(packet) + + assert isinstance(clone, type(packet)) + with pytest.raises(ValueError, match="candidate offer response evidence has no issuance seal"): + clone.canonical_json() diff --git a/packages/candidate-offer-response/tests/test_external_identity_reference_contract.py b/packages/candidate-offer-response/tests/test_external_identity_reference_contract.py new file mode 100644 index 000000000..c1ad33284 --- /dev/null +++ b/packages/candidate-offer-response/tests/test_external_identity_reference_contract.py @@ -0,0 +1,28 @@ +"""Regression for the read-only Keyverse opaque actor-reference contract.""" + +from datetime import datetime, timezone +import json + +from orgmetra_candidate_offer_response import build_candidate_offer_response + + +def test_accepts_non_uuid_keyverse_candidate_actor_reference() -> None: + """Do not invent a UUIDv4 requirement for the externally owned actor identity.""" + packet = build_candidate_offer_response( + tenant_record_id="018f6e2a-4f7c-7a1b-9c20-1f3a7d8e5b60", + offer_response_reference="candidate_offer_response:6ba7b810-9dad-4b11-80b4-00c04fd430c8", + candidate_profile_reference="candidate_profile:6ba7b811-9dad-4b11-80b4-00c04fd430c8", + offer_approval_reference="offer_approval:6ba7b812-9dad-4b11-80b4-00c04fd430c8", + offer_approval_digest="a" * 64, + offer_terms_reference="offer_terms:6ba7b813-9dad-4b11-80b4-00c04fd430c8", + offer_terms_digest="b" * 64, + candidate_actor_reference="candidate:AItOawmwtWwcT0k51BayewNvutrJUqsvl6qs7A4", + identity_resolution_reference="identity_resolution:6ba7b815-9dad-4b11-80b4-00c04fd430c8", + identity_resolution_digest="c" * 64, + response_code="offer_accepted", + responded_at=datetime(2026, 8, 22, 9, 30, tzinfo=timezone.utc), + recorded_at=datetime(2026, 8, 22, 9, 30, 1, tzinfo=timezone.utc), + ) + + evidence = json.loads(packet.canonical_json()) + assert evidence["candidate_actor_reference"] == "candidate:AItOawmwtWwcT0k51BayewNvutrJUqsvl6qs7A4" diff --git a/packages/candidate-offer-response/tests/test_response.py b/packages/candidate-offer-response/tests/test_response.py new file mode 100644 index 000000000..3ed49f60f --- /dev/null +++ b/packages/candidate-offer-response/tests/test_response.py @@ -0,0 +1,332 @@ +"""Executable contract for candidate-originated offer response evidence.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone, tzinfo +import json +from uuid import UUID + +import pytest + +from orgmetra_candidate_offer_response.response import ( + CandidateOfferResponsePacket, + build_candidate_offer_response, +) + +TENANT_ID = "018f6e2a-4f7c-7a1b-9c20-1f3a7d8e5b60" +OFFER_RESPONSE = "candidate_offer_response:6ba7b810-9dad-4b11-80b4-00c04fd430c8" +CANDIDATE = "candidate_profile:6ba7b811-9dad-4b11-80b4-00c04fd430c8" +OFFER_APPROVAL = "offer_approval:6ba7b812-9dad-4b11-80b4-00c04fd430c8" +OFFER_TERMS = "offer_terms:6ba7b813-9dad-4b11-80b4-00c04fd430c8" +CANDIDATE_ACTOR = "candidate:6ba7b814-9dad-4b11-80b4-00c04fd430c8" +IDENTITY_RESOLUTION = "identity_resolution:6ba7b815-9dad-4b11-80b4-00c04fd430c8" +DIGEST_A = "a" * 64 +DIGEST_B = "b" * 64 +DIGEST_C = "c" * 64 +RESPONDED_AT = datetime(2026, 8, 22, 9, 30, 15, 123456, tzinfo=timezone.utc) +RECORDED_AT = datetime(2026, 8, 22, 9, 30, 16, 123456, tzinfo=timezone.utc) + + +def _kwargs() -> dict[str, object]: + """Return one valid, value-minimized accepted-offer evidence fixture.""" + return { + "tenant_record_id": TENANT_ID, + "offer_response_reference": OFFER_RESPONSE, + "candidate_profile_reference": CANDIDATE, + "offer_approval_reference": OFFER_APPROVAL, + "offer_approval_digest": DIGEST_A, + "offer_terms_reference": OFFER_TERMS, + "offer_terms_digest": DIGEST_B, + "candidate_actor_reference": CANDIDATE_ACTOR, + "identity_resolution_reference": IDENTITY_RESOLUTION, + "identity_resolution_digest": DIGEST_C, + "response_code": "offer_accepted", + "responded_at": RESPONDED_AT, + "recorded_at": RECORDED_AT, + "evidence_version": 1, + } + + +def _build(**overrides: object) -> CandidateOfferResponsePacket: + """Build a packet while allowing one test to replace selected inputs.""" + values = _kwargs() + values.update(overrides) + return build_candidate_offer_response(**values) # type: ignore[arg-type] + + +def test_candidate_acceptance_is_value_minimized_and_non_authorizing() -> None: + packet = _build() + document = json.loads(packet.canonical_json()) + + assert document["response_code"] == "offer_accepted" + assert document["candidate_confirmation_required"] is True + assert document["scope_verification_state"] == "requires_authoritative_resolution" + assert document["employment_effect"] == "not_authorized_to_hire" + assert document["decision_authority"] == "candidate_response_only" + assert document["contains_candidate_pii"] is False + assert document["contains_compensation_values"] is False + assert document["contains_free_form_reason"] is False + assert document["responded_at"] == "2026-08-22T09:30:15.123456Z" + assert document["recorded_at"] == "2026-08-22T09:30:16.123456Z" + assert "re-resolve" in document["next_action"] + assert "offer" in document["next_action"] + assert "hire" in document["next_action"] + assert len(packet.sha256_digest()) == 64 + assert packet.sha256_digest() == packet.sha256_digest() + assert repr(packet) == "CandidateOfferResponsePacket()" + + +def test_candidate_decline_uses_the_same_candidate_originated_boundary() -> None: + packet = _build(response_code="offer_declined") + assert json.loads(packet.canonical_json())["response_code"] == "offer_declined" + assert packet.employment_effect == "not_authorized_to_hire" + + +@pytest.mark.parametrize( + ("field_name", "value", "message"), + [ + ("tenant_record_id", "not-a-uuid", "tenant_record_id"), + ("tenant_record_id", "00000000-0000-0000-0000-000000000000", "operational UUID"), + ("tenant_record_id", "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF", "canonical"), + ("offer_response_reference", "candidate_offer_response:not-uuid", "offer_response_reference"), + ("candidate_profile_reference", "candidate_profile:not-uuid", "candidate_profile_reference"), + ("offer_approval_reference", "offer_approval:not-uuid", "offer_approval_reference"), + ("offer_terms_reference", "offer_terms:not-uuid", "offer_terms_reference"), + ("candidate_actor_reference", "staff:6ba7b814-9dad-4b11-80b4-00c04fd430c8", "candidate_actor_reference"), + ("identity_resolution_reference", "identity_resolution:not-uuid", "identity_resolution_reference"), + ("offer_approval_digest", "A" * 64, "offer_approval_digest"), + ("offer_terms_digest", "b" * 63, "offer_terms_digest"), + ("identity_resolution_digest", "not-a-digest", "identity_resolution_digest"), + ("response_code", "offer_pending", "response_code"), + ("evidence_version", 0, "evidence_version"), + ("evidence_version", 2_147_483_648, "evidence_version"), + ], +) +def test_invalid_trust_evidence_fails_closed(field_name: str, value: object, message: str) -> None: + with pytest.raises(ValueError, match=message): + _build(**{field_name: value}) + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("offer_response_reference", "offer_response:6ba7b810-9dad-4b11-80b4-00c04fd430c8"), + ("candidate_profile_reference", "candidate:6ba7b811-9dad-4b11-80b4-00c04fd430c8"), + ("offer_approval_reference", "offer_terms:6ba7b812-9dad-4b11-80b4-00c04fd430c8"), + ("offer_terms_reference", "offer_approval:6ba7b813-9dad-4b11-80b4-00c04fd430c8"), + ("identity_resolution_reference", "candidate:6ba7b815-9dad-4b11-80b4-00c04fd430c8"), + ], +) +def test_reference_namespaces_are_not_interchangeable(field_name: str, value: str) -> None: + with pytest.raises(ValueError): + _build(**{field_name: value}) + + +def test_packet_owned_references_require_uuid4_suffixes() -> None: + version_one = "candidate_offer_response:6ba7b810-9dad-1b11-80b4-00c04fd430c8" + with pytest.raises(ValueError, match="offer_response_reference"): + _build(offer_response_reference=version_one) + + +class _ForgedText(str): + def __eq__(self, other: object) -> bool: + return True + + def __hash__(self) -> int: + return hash("offer_accepted") + + def __len__(self) -> int: + return 1 + + +class _ForgedInt(int): + def __lt__(self, other: object) -> bool: + return False + + def __le__(self, other: object) -> bool: + return True + + def __gt__(self, other: object) -> bool: + return False + + def __ge__(self, other: object) -> bool: + return True + + +class _ForgedDatetime(datetime): + pass + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("tenant_record_id", _ForgedText(TENANT_ID)), + ("offer_response_reference", _ForgedText(OFFER_RESPONSE)), + ("candidate_profile_reference", _ForgedText(CANDIDATE)), + ("offer_approval_reference", _ForgedText(OFFER_APPROVAL)), + ("offer_approval_digest", _ForgedText(DIGEST_A)), + ("offer_terms_reference", _ForgedText(OFFER_TERMS)), + ("offer_terms_digest", _ForgedText(DIGEST_B)), + ("candidate_actor_reference", _ForgedText(CANDIDATE_ACTOR)), + ("identity_resolution_reference", _ForgedText(IDENTITY_RESOLUTION)), + ("identity_resolution_digest", _ForgedText(DIGEST_C)), + ("response_code", _ForgedText("shadow_rejection")), + ("evidence_version", _ForgedInt(0)), + ("responded_at", _ForgedDatetime(2026, 8, 22, tzinfo=timezone.utc)), + ("recorded_at", _ForgedDatetime(2026, 8, 22, tzinfo=timezone.utc)), + ], +) +def test_caller_controlled_runtime_subclasses_are_rejected(field_name: str, value: object) -> None: + with pytest.raises(ValueError): + _build(**{field_name: value}) + + +def test_recorded_time_cannot_precede_candidate_response() -> None: + with pytest.raises(ValueError, match="recorded_at must not precede responded_at"): + _build(recorded_at=RESPONDED_AT - timedelta(microseconds=1)) + + +class _MutableTimezone(tzinfo): + def __init__(self, offset: timedelta) -> None: + self.offset = offset + + def utcoffset(self, dt: datetime | None) -> timedelta: + return self.offset + + def dst(self, dt: datetime | None) -> timedelta: + return timedelta(0) + + def tzname(self, dt: datetime | None) -> str: + return "mutable" + + +class _NoOffsetTimezone(tzinfo): + def utcoffset(self, dt: datetime | None) -> None: + return None + + def dst(self, dt: datetime | None) -> None: + return None + + def tzname(self, dt: datetime | None) -> str: + return "none" + + +class _BrokenTimezone(tzinfo): + def utcoffset(self, dt: datetime | None) -> timedelta: + raise RuntimeError("provider secret") + + def dst(self, dt: datetime | None) -> timedelta: + return timedelta(0) + + def tzname(self, dt: datetime | None) -> str: + return "broken" + + +def test_timezone_behavior_is_detached_at_construction() -> None: + tz = _MutableTimezone(timedelta(hours=9)) + responded = datetime(2026, 8, 22, 18, 30, 15, 123456, tzinfo=tz) + recorded = datetime(2026, 8, 22, 18, 30, 16, 123456, tzinfo=tz) + packet = _build(responded_at=responded, recorded_at=recorded) + before = packet.canonical_json() + + tz.offset = timedelta(hours=-7) + + assert packet.canonical_json() == before + assert packet.responded_at.tzinfo is timezone.utc + assert packet.recorded_at.tzinfo is timezone.utc + assert json.loads(before)["responded_at"] == "2026-08-22T09:30:15.123456Z" + + +@pytest.mark.parametrize("field_name", ["responded_at", "recorded_at"]) +def test_offsetless_timestamp_fails_closed(field_name: str) -> None: + bad = datetime(2026, 8, 22, 9, 30, tzinfo=_NoOffsetTimezone()) + with pytest.raises(ValueError, match=field_name): + _build(**{field_name: bad}) + + +@pytest.mark.parametrize("field_name", ["responded_at", "recorded_at"]) +def test_timezone_provider_exception_is_normalized(field_name: str) -> None: + bad = datetime(2026, 8, 22, 9, 30, tzinfo=_BrokenTimezone()) + with pytest.raises(ValueError, match=field_name) as exc_info: + _build(**{field_name: bad}) + assert "provider secret" not in str(exc_info.value) + + +def test_valid_value_replacement_after_issuance_invalidates_evidence() -> None: + packet = _build() + object.__setattr__(packet, "response_code", "offer_declined") + with pytest.raises(ValueError, match="offer response evidence changed after construction"): + packet.canonical_json() + + +def test_invalid_value_replacement_after_issuance_fails_validation() -> None: + packet = _build() + object.__setattr__(packet, "offer_approval_digest", "A" * 64) + with pytest.raises(ValueError, match="offer_approval_digest"): + packet.sha256_digest() + + +def test_canonicalization_rejects_post_construction_timestamp_reinjection() -> None: + packet = _build() + object.__setattr__(packet, "recorded_at", datetime(2026, 8, 22, 9, 30, 16, tzinfo=timezone(timedelta(hours=1)))) + with pytest.raises(ValueError, match="recorded_at"): + packet.canonical_json() + + +def test_packet_runtime_is_final() -> None: + with pytest.raises(TypeError, match="final"): + type("ForgedPacket", (CandidateOfferResponsePacket,), {}) + + +def test_direct_construction_cannot_weaken_fixed_governance() -> None: + values = _kwargs() + values["candidate_confirmation_required"] = False + with pytest.raises(ValueError, match="candidate confirmation"): + CandidateOfferResponsePacket(**values) # type: ignore[arg-type] + + values = _kwargs() + values["employment_effect"] = "authorized_to_hire" + with pytest.raises(ValueError, match="employment_effect"): + CandidateOfferResponsePacket(**values) # type: ignore[arg-type] + + values = _kwargs() + values["scope_verification_state"] = "verified" + with pytest.raises(ValueError, match="scope_verification_state"): + CandidateOfferResponsePacket(**values) # type: ignore[arg-type] + + +def test_direct_construction_cannot_claim_sensitive_payloads_or_model_authority() -> None: + for field_name in ( + "contains_candidate_pii", + "contains_compensation_values", + "contains_free_form_reason", + ): + values = _kwargs() + values[field_name] = True + with pytest.raises(ValueError): + CandidateOfferResponsePacket(**values) # type: ignore[arg-type] + + values = _kwargs() + values["decision_authority"] = "model_decided" + with pytest.raises(ValueError, match="decision_authority"): + CandidateOfferResponsePacket(**values) # type: ignore[arg-type] + + +def test_direct_construction_cannot_rewrite_next_action() -> None: + values = _kwargs() + values["next_action"] = "Hire immediately" + with pytest.raises(ValueError, match="next_action"): + CandidateOfferResponsePacket(**values) # type: ignore[arg-type] + + +def test_operational_tenant_accepts_uuid7() -> None: + packet = _build() + assert UUID(packet.tenant_record_id).version == 7 + + +@pytest.mark.parametrize("field_name", ["responded_at", "recorded_at"]) +def test_offset_overflow_normalizes_to_governed_value_error(field_name: str) -> None: + """Normalize range overflow during UTC detachment to the governed error.""" + extreme = datetime(1, 1, 1, 0, 0, tzinfo=_MutableTimezone(timedelta(hours=23, minutes=59))) + with pytest.raises(ValueError, match=f"{field_name} must have a valid timezone offset"): + _build(**{field_name: extreme})