diff --git a/.github/workflows/assignment-correction-quality.yml b/.github/workflows/assignment-correction-quality.yml new file mode 100644 index 000000000..9dbcd3f09 --- /dev/null +++ b/.github/workflows/assignment-correction-quality.yml @@ -0,0 +1,140 @@ +name: Assignment Correction Quality + +on: + pull_request: + branches: + - feat/explicit-assignment-category + - bootstrap + - develop + - main + paths: + - "database/migrations/0018_assignment_category_supersession.sql" + - "database/migrations/0019_assignment_correction_idempotency_route.sql" + - "docs/traceability/assignment-category-correction-provenance.md" + - "packages/hris-kernel/**" + - "services/people-api/**" + - "tests/test_assignment_category_correction_postgres.sh" + - "tests/test_assignment_correction_idempotency_postgres.sh" + - ".github/workflows/assignment-correction-quality.yml" + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: assignment-correction-quality-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + kernel: + name: Assignment correction application and 100% People coverage + runs-on: ubuntu-24.04 + timeout-minutes: 10 + env: + PYTHONPATH: packages/hris-kernel/src:packages/keyverse-adapter/src:services/people-api/src + steps: + - name: Checkout exact candidate + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Prove exact candidate checkout + env: + ORGMETRA_EXPECTED_HEAD_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: test "$(git rev-parse HEAD)" = "$ORGMETRA_EXPECTED_HEAD_SHA" + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.14" + check-latest: false + - name: Install locked test dependencies + run: | + python -m pip install --require-hashes --no-deps --only-binary=:all: -r .github/requirements/foundation-test.txt + python -m pip check + - name: Prove HRIS correction domain contract + run: >- + python -m pytest + packages/hris-kernel/tests/test_assignment_category_correction.py + packages/hris-kernel/tests/test_assignment_category_correction_recorded_time_integrity.py + - name: Prove the full People API at exact statement and branch coverage + env: + COVERAGE_FILE: /tmp/orgmetra-assignment-correction.coverage + run: python -m pytest -c services/people-api/pyproject.toml services/people-api/tests + - name: Require clean checkout + run: | + git diff --exit-code + test -z "$(git status --porcelain)" + + postgres: + name: Assignment correction PostgreSQL contract + runs-on: ubuntu-24.04 + timeout-minutes: 10 + services: + postgres: + image: postgres:16.14@sha256:33f923b05f64ca54ac4401c01126a6b92afe839a0aa0a52bc5aeb5cc958e5f20 + env: + POSTGRES_USER: orgmetra + POSTGRES_PASSWORD: orgmetra + POSTGRES_DB: orgmetra + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U orgmetra -d orgmetra" + --health-interval 5s + --health-timeout 5s + --health-retries 10 + env: + DATABASE_URL: postgresql://orgmetra:orgmetra@localhost:5432/orgmetra + steps: + - name: Checkout exact candidate + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Prove exact candidate checkout + env: + ORGMETRA_EXPECTED_HEAD_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: test "$(git rev-parse HEAD)" = "$ORGMETRA_EXPECTED_HEAD_SHA" + - name: Prove normalized supersession persistence + run: bash tests/test_assignment_category_correction_postgres.sh + - name: Require clean checkout + run: | + git diff --exit-code + test -z "$(git status --porcelain)" + + idempotency: + name: Assignment correction idempotency contract + runs-on: ubuntu-24.04 + timeout-minutes: 10 + services: + postgres: + image: postgres:16.14@sha256:33f923b05f64ca54ac4401c01126a6b92afe839a0aa0a52bc5aeb5cc958e5f20 + env: + POSTGRES_USER: orgmetra + POSTGRES_PASSWORD: orgmetra + POSTGRES_DB: orgmetra + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U orgmetra -d orgmetra" + --health-interval 5s + --health-timeout 5s + --health-retries 10 + env: + DATABASE_URL: postgresql://orgmetra:orgmetra@localhost:5432/orgmetra + steps: + - name: Checkout exact candidate + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Prove exact candidate checkout + env: + ORGMETRA_EXPECTED_HEAD_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: test "$(git rev-parse HEAD)" = "$ORGMETRA_EXPECTED_HEAD_SHA" + - name: Prove correction replay route is durable and closed + run: bash tests/test_assignment_correction_idempotency_postgres.sh + - name: Require clean checkout + run: | + git diff --exit-code + test -z "$(git status --porcelain)" diff --git a/database/migrations/0018_assignment_category_supersession.sql b/database/migrations/0018_assignment_category_supersession.sql new file mode 100644 index 000000000..4bdd30ad4 --- /dev/null +++ b/database/migrations/0018_assignment_category_supersession.sql @@ -0,0 +1,154 @@ +-- Persist immutable Assignment category-correction lineage as Orgmetra HRIS truth. +-- +-- An Assignment correction is a system-time replacement, never an in-place +-- business rewrite. The predecessor must already be closed at the correction +-- timestamp, the replacement must start at that same timestamp, all business +-- truth except category must be identical, and the two explicit categories must +-- differ. The normalized one-to-one edge prevents unlinked duplicates and forks. + +BEGIN; + +SET LOCAL search_path = public, pg_catalog; + +CREATE TABLE public.assignment_supersession_record ( + tenant_record_id uuid NOT NULL REFERENCES public.tenant_record(tenant_record_id), + assignment_supersession_record_id uuid PRIMARY KEY, + predecessor_assignment_record_id uuid NOT NULL, + replacement_assignment_record_id uuid NOT NULL, + recorded_at timestamptz NOT NULL, + CONSTRAINT assignment_supersession_record_id_operational_check + CHECK (public.is_operational_uuid(assignment_supersession_record_id)), + CONSTRAINT assignment_supersession_predecessor_id_operational_check + CHECK (public.is_operational_uuid(predecessor_assignment_record_id)), + CONSTRAINT assignment_supersession_replacement_id_operational_check + CHECK (public.is_operational_uuid(replacement_assignment_record_id)), + CONSTRAINT assignment_supersession_distinct_assignment_check + CHECK (predecessor_assignment_record_id <> replacement_assignment_record_id), + CONSTRAINT assignment_supersession_predecessor_tenant_fk + FOREIGN KEY (tenant_record_id, predecessor_assignment_record_id) + REFERENCES public.assignment_record(tenant_record_id, assignment_record_id), + CONSTRAINT assignment_supersession_replacement_tenant_fk + FOREIGN KEY (tenant_record_id, replacement_assignment_record_id) + REFERENCES public.assignment_record(tenant_record_id, assignment_record_id), + CONSTRAINT assignment_supersession_tenant_identity_unique + UNIQUE (tenant_record_id, assignment_supersession_record_id), + CONSTRAINT assignment_supersession_predecessor_unique + UNIQUE (tenant_record_id, predecessor_assignment_record_id), + CONSTRAINT assignment_supersession_replacement_unique + UNIQUE (tenant_record_id, replacement_assignment_record_id) +); + +CREATE FUNCTION public.enforce_assignment_supersession_link() +RETURNS trigger +LANGUAGE plpgsql +SET search_path = pg_catalog, public +AS $$ +DECLARE + predecessor_record public.assignment_record%ROWTYPE; + replacement_record public.assignment_record%ROWTYPE; +BEGIN + SELECT assignment.* + INTO predecessor_record + FROM public.assignment_record AS assignment + WHERE assignment.tenant_record_id = NEW.tenant_record_id + AND assignment.assignment_record_id = NEW.predecessor_assignment_record_id + FOR SHARE; + + IF NOT FOUND THEN + RAISE EXCEPTION 'assignment supersession predecessor does not exist in tenant scope' + USING ERRCODE = 'foreign_key_violation', + CONSTRAINT = 'assignment_supersession_predecessor_tenant_fk', + TABLE = 'assignment_supersession_record', + SCHEMA = 'public'; + END IF; + + SELECT assignment.* + INTO replacement_record + FROM public.assignment_record AS assignment + WHERE assignment.tenant_record_id = NEW.tenant_record_id + AND assignment.assignment_record_id = NEW.replacement_assignment_record_id + FOR SHARE; + + IF NOT FOUND THEN + RAISE EXCEPTION 'assignment supersession replacement does not exist in tenant scope' + USING ERRCODE = 'foreign_key_violation', + CONSTRAINT = 'assignment_supersession_replacement_tenant_fk', + TABLE = 'assignment_supersession_record', + SCHEMA = 'public'; + END IF; + + IF predecessor_record.recorded_to IS DISTINCT FROM NEW.recorded_at + OR replacement_record.recorded_from IS DISTINCT FROM NEW.recorded_at + OR replacement_record.recorded_to IS NOT NULL THEN + RAISE EXCEPTION 'assignment supersession recorded timestamp must equal predecessor close and replacement start' + USING ERRCODE = 'check_violation', + CONSTRAINT = 'assignment_supersession_recorded_time_check', + TABLE = 'assignment_supersession_record', + SCHEMA = 'public'; + END IF; + + IF predecessor_record.employment_record_id IS DISTINCT FROM replacement_record.employment_record_id + OR predecessor_record.person_record_id IS DISTINCT FROM replacement_record.person_record_id + OR predecessor_record.position_record_id IS DISTINCT FROM replacement_record.position_record_id + OR predecessor_record.allocation_ratio IS DISTINCT FROM replacement_record.allocation_ratio + OR predecessor_record.effective_from IS DISTINCT FROM replacement_record.effective_from + OR predecessor_record.effective_to IS DISTINCT FROM replacement_record.effective_to THEN + RAISE EXCEPTION 'assignment supersession replacement changed business truth outside category' + USING ERRCODE = 'check_violation', + CONSTRAINT = 'assignment_supersession_business_truth_check', + TABLE = 'assignment_supersession_record', + SCHEMA = 'public'; + END IF; + + IF predecessor_record.assignment_category_code NOT IN ('primary', 'concurrent_secondary') + OR replacement_record.assignment_category_code NOT IN ('primary', 'concurrent_secondary') + OR predecessor_record.assignment_category_code = replacement_record.assignment_category_code THEN + RAISE EXCEPTION 'assignment supersession must change one explicit assignment category to the other' + USING ERRCODE = 'check_violation', + CONSTRAINT = 'assignment_supersession_category_change_check', + TABLE = 'assignment_supersession_record', + SCHEMA = 'public'; + END IF; + + RETURN NEW; +END; +$$; + +COMMENT ON FUNCTION public.enforce_assignment_supersession_link() IS + 'Validates close-to-replacement Assignment category lineage against locked tenant-local HRIS facts.'; + +CREATE TRIGGER assignment_supersession_link_guard +BEFORE INSERT ON public.assignment_supersession_record +FOR EACH ROW +EXECUTE FUNCTION public.enforce_assignment_supersession_link(); + +CREATE TRIGGER assignment_supersession_append_only_guard +BEFORE UPDATE OR DELETE ON public.assignment_supersession_record +FOR EACH ROW +EXECUTE FUNCTION public.reject_append_only_mutation(); + +CREATE FUNCTION public.reject_assignment_supersession_truncate() +RETURNS trigger +LANGUAGE plpgsql +SET search_path = pg_catalog, public, pg_temp +AS $$ +BEGIN + RAISE EXCEPTION 'assignment supersession records cannot be truncated' + USING ERRCODE = '55000'; +END; +$$; + +CREATE TRIGGER assignment_supersession_truncate_guard +BEFORE TRUNCATE ON public.assignment_supersession_record +FOR EACH STATEMENT +EXECUTE FUNCTION public.reject_assignment_supersession_truncate(); + +REVOKE TRUNCATE ON public.assignment_supersession_record FROM PUBLIC; + +ALTER TABLE public.assignment_supersession_record ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.assignment_supersession_record FORCE ROW LEVEL SECURITY; +CREATE POLICY assignment_supersession_scope_policy ON public.assignment_supersession_record +USING (tenant_record_id = public.current_tenant_record_id()) +WITH CHECK (tenant_record_id = public.current_tenant_record_id()); + +COMMIT; diff --git a/database/migrations/0019_assignment_correction_idempotency_route.sql b/database/migrations/0019_assignment_correction_idempotency_route.sql new file mode 100644 index 000000000..17957b579 --- /dev/null +++ b/database/migrations/0019_assignment_correction_idempotency_route.sql @@ -0,0 +1,29 @@ +-- Extend the durable People mutation replay vocabulary for Assignment category corrections. +-- +-- Correction retries reuse the existing tenant-scoped idempotency ledger. The +-- semantic digest is bound to the predecessor and reviewed correction meaning, +-- while created_record_id stores the first committed replacement Assignment. + +BEGIN; + +SET LOCAL search_path = public, pg_catalog; + +ALTER TABLE public.people_mutation_idempotency_record + DROP CONSTRAINT people_mutation_idempotency_route_check; + +ALTER TABLE public.people_mutation_idempotency_record + ADD CONSTRAINT people_mutation_idempotency_route_check + CHECK ( + command_route IN ( + 'candidate-worker-conversions', + 'employment-records', + 'position-records', + 'assignment-records', + 'assignment-category-corrections' + ) + ) NOT VALID; + +ALTER TABLE public.people_mutation_idempotency_record + VALIDATE CONSTRAINT people_mutation_idempotency_route_check; + +COMMIT; diff --git a/docs/traceability/assignment-category-correction-provenance.md b/docs/traceability/assignment-category-correction-provenance.md new file mode 100644 index 000000000..bbd672537 --- /dev/null +++ b/docs/traceability/assignment-category-correction-provenance.md @@ -0,0 +1,66 @@ +# Assignment category correction provenance traceability + +Status: `implemented_on_active_pr` on Orgmetra PR #165. This document does not describe protected `develop` as shipped correction support. + +## Decision boundary + +Orgmetra owns Assignment category correction because Assignment, Employment, Person, Position, allocation, effective time, and system-recorded time are HRIS truth in the People/Organization–Job–Position–Assignment boundary. A correction is not an in-place category update. It closes one recorded-open explicit Assignment fact, creates a replacement with a new Assignment identity, and records a normalized predecessor→replacement provenance edge at the same system-recorded timestamp. + +The replacement must preserve tenant, Employment, Person, Position, allocation, and effective interval. Only `assignment_category_code` changes between the two explicit values `primary` and `concurrent_secondary`. Historical `legacy_unspecified` rows remain outside this correction contract; classifying them requires a separately governed workflow rather than inference from allocation, ordering, or topology. + +## Executable evidence + +| Concern | Active-PR evidence | Required behavior | +|---|---|---| +| Domain replacement semantics | `packages/hris-kernel/src/orgmetra_hris_kernel/assignment_correction.py`; `packages/hris-kernel/tests/test_assignment_category_correction.py` | Close predecessor recorded time, create a new identity, preserve other Assignment truth, and link the two facts. | +| Runtime identity integrity | same kernel module/tests; `database/migrations/0002_sealed_evidence_digest.sql` | Correction-owned UUIDs are exact built-in UUID values and reject RFC 9562 Nil/Max sentinels before equality or provenance construction. | +| Runtime recorded-time integrity | same kernel module/tests; `packages/hris-kernel/tests/test_assignment_category_correction_recorded_time_integrity.py` | Correction provenance accepts only an exact built-in, offset-aware `datetime`; resolves the UTC offset once, rejects non-exact/invalid offset evidence, detaches accepted caller-owned `tzinfo` behavior onto a built-in fixed-offset timezone before storing or comparing the timestamp, and normalizes timezone-provider failure to `CorrectionError`. Executable datetime subtypes and offsetless values fail closed. | +| Purpose-bound correction command | `services/people-api/src/orgmetra_people_api/assignment_correction_mutations.py`; `services/people-api/tests/test_assignment_correction_mutations.py`; `services/people-api/tests/test_assignment_correction_adapter_runtime_integrity.py`; `services/people-api/tests/test_assignment_correction_result_runtime_integrity.py` | Authorize exactly the predecessor Assignment's category field for `correct_record`; accept only the exact governed command and exact Keyverse `AuthorizationDecision` runtime types at semantic-digest/persistence trust boundaries; require the exact governed `AssignmentCorrectionMutationResult` before its replacement/supersession identities cross back to HTTP; require human confirmation/evidence version/idempotency; cap confirmation references at 300 characters and evidence-version tokens at 200 characters; bind semantic replay to predecessor/category/evidence while excluding retry-generated record IDs. | +| Buyer HTTP/OpenAPI boundary | `services/people-api/src/orgmetra_people_api/assignment_correction_http.py`; `services/people-api/assignment-correction.openapi.yaml`; `services/people-api/tests/test_assignment_correction_http.py`; `services/people-api/tests/test_assignment_correction_openapi.py` | Publish one POST-only predecessor-scoped correction route; require Keyverse bearer authentication plus tenant/actor/purpose/idempotency bindings; expose only the explicit target category, bounded confirmation, and bounded evidence version; return replacement and supersession identities without in-place mutation; keep the closed OpenAPI error object identical to the shared People mutation error envelope. | +| Full People API coverage | `.github/workflows/assignment-correction-quality.yml`; `services/people-api/pyproject.toml` | Run the complete People service test suite on the exact child head under the existing 100% owned statement and branch coverage threshold; a focused happy-path test is not accepted as coverage evidence. | +| Atomic People persistence | `services/people-api/src/orgmetra_people_api/postgres_assignment_corrections.py`; `services/people-api/tests/test_postgres_assignment_corrections.py`; `services/people-api/tests/test_assignment_correction_adapter_runtime_integrity.py` | Reject caller-defined command or authorization-decision subtypes before opening a database connection. In one tenant transaction, serialize the replay key, probe the recorded-open predecessor only to locate immutable Employment/Position scope, lock Employment then Position, then lock the affected Assignment portfolio in `assignment_record_id` order and re-resolve the predecessor from that locked portfolio. Only then take the database timestamp. This avoids both conflicting predecessor/position lock cycles and system-time backdating while waiting for the final authoritative lock. Re-run portfolio/seat-capacity validation, close the predecessor, insert the replacement and supersession edge, then persist audit/outbox and replay evidence. | +| Durable replay vocabulary | `database/migrations/0019_assignment_correction_idempotency_route.sql`; `tests/test_assignment_correction_idempotency_postgres.sh` | `assignment-category-corrections` is a first-class closed route in the existing People mutation idempotency ledger; unknown routes remain rejected. Matching retries resolve the first replacement plus normalized supersession rather than creating new HRIS or audit facts. | +| Normalized persistence | `database/migrations/0018_assignment_category_supersession.sql` | One tenant-scoped append-only edge links exactly one predecessor and one replacement; forks and replacement reuse are rejected while later correction chains remain possible. | +| Database linkage and recovery | `tests/test_assignment_category_correction_postgres.sh` | Migration late-failure rollback is atomic; predecessor close time equals edge time; replacement start equals edge time; non-category business truth is unchanged; explicit category truth changes; append-only and one-to-one lineage fail closed. | +| Tenant/privacy boundary | migration 0018 RLS policy/composite tenant FKs plus the PostgreSQL regression | A NOBYPASSRLS reader sees no provenance without tenant context, sees its own tenant, and cannot see another tenant's provenance. | +| Hosted exact-head proof | `.github/workflows/assignment-correction-quality.yml` | Exact checkout runs both HRIS correction suites, full People API coverage, supersession, and replay-route contracts on the current candidate head. Absence, queueing, cancellation, or predecessor results are not GREEN evidence. | + +## DDD and context-map mapping + +- Bounded context: People / Organization–Job–Position–Assignment. +- Aggregate/entity: immutable `assignment_record` fact identified by `assignment_record_id`. +- Value object: explicit `assignment_category_code`. +- Domain service: `correct_assignment_category` produces the closed predecessor, replacement, and supersession fact; portfolio/capacity invariants remain authoritative validation prerequisites before persistence. +- Application service: `correct_assignment_record_category` owns the purpose-bound authorization boundary for the exact predecessor category field before the write port is called, and accepts only the exact governed mutation result before its identities are returned to an adapter. +- HTTP adapter: `AssignmentCorrectionAsgiApp` owns request parsing and client-safe errors but delegates identity to Keyverse, authorization to the application service, and HRIS truth to the correction port. +- Repository/persistence boundary: `PostgresAssignmentCorrectionMutationPort` owns the transaction that writes `assignment_record`, `assignment_supersession_record`, audit/outbox evidence, and the existing People idempotency ledger. It consumes no external service database. +- Context map: Keyverse is an identity/authorization peer consumed through the released adapter contract; Orgmetra remains upstream owner of HR category and supersession truth. No shared HR vocabulary is copied into Keyverse and no external service database is queried. +- Invariants: tenant consistency, operational identities, strict system-time succession, detached fixed-offset correction time, preserved non-category business truth, explicit category change, one-to-one predecessor/replacement edge, append-only provenance, tenant RLS, semantic replay consistency, exact correction-command, authorization-decision, and correction-result runtime types, bounded high-impact evidence metadata, deterministic correction lock order, and post-lock revalidation of Employment/Position/Assignment truth. + +No shared kernel or cross-service SQL is introduced. Keyverse evaluates the purpose-bound access request but does not author Assignment truth. + +## Security, operability, and recovery handoff + +The active child already enforces the behavior that the canonical release documents must describe after the prerequisite stack integrates: + +- Security/threat model: a caller cannot submit a generic Assignment patch, extend the governed application command through a caller-defined subtype, inject a caller-defined `AuthorizationDecision` subtype into semantic replay/persistence, return a caller-defined `AssignmentCorrectionMutationResult` subtype for the HTTP layer to consume, or retain executable caller-owned timezone behavior in accepted correction provenance. The route names one predecessor, accepts one governed category field plus confirmation/evidence metadata, authenticates through Keyverse, binds tenant and actor to the authenticated principal, and authorizes only `assignment_category_code` under `correct_record`. Confirmation references are bounded to 300 characters and evidence-version tokens to 200 characters, matching the existing People high-impact write boundary instead of allowing an unbounded audit/idempotency payload. All responses are `no-store`/`Vary: Authorization`; backend details and bearer values stay out of client errors. +- ERD/data model: `assignment_record` remains the immutable business fact. `assignment_supersession_record` is a tenant-scoped normalized edge with one predecessor and one replacement, and the replacement preserves Employment, Person, Position, allocation, and effective interval while system-recorded time advances. +- UML/sequence: parse route and governed headers → authenticate → bind tenant/actor → parse bounded JSON → authorize exact predecessor category → serialize idempotency → probe predecessor scope without a row lock → lock Employment → lock Position → lock affected Assignments in UUID order and re-resolve predecessor → take database time → detach the accepted recorded instant from caller/runtime timezone behavior → validate portfolio/capacity → close predecessor → insert replacement → insert supersession → audit/outbox → durable replay record → commit → require exact governed result → return opaque identities. +- Operability: matching replay is normal operation and must return the first committed replacement/supersession pair. Changed semantics under one key are a conflict. Missing or stale recorded-open truth, invariant failure, malformed persisted reconstruction, non-exact runtime command/authorization/result evidence, unusable UTC offset evidence, or timezone-provider failure fails closed before those values can cross the next trust boundary. Unexpected dependency failures expose a non-sensitive support reference. +- Recovery: migration 0018 and 0019 regressions require transactional rollback on migration failure. Runtime writes use one database transaction, so predecessor closure cannot be committed without its replacement, supersession, audit/outbox, and replay evidence. Restore/replay checks must preserve the predecessor close time, replacement start time, and supersession time as one recorded coordinate. + +`ARCHITECTURE.md`, `docs/ERD.md`, `docs/UML.md`, `docs/SECURITY.md`, `docs/THREAT_MODEL.md`, `docs/OPERABILITY.md`, `docs/TEST_STRATEGY.md`, and the deterministic repository inventory/`manifest.json` are canonical foundation artifacts. They are intentionally not edited piecemeal on this dependent child because the current inventory/manifest is single-writer-sensitive and exact digest/byte/line validation would make an isolated documentation edit an invalid provenance state. The release handoff must update those artifacts and reseal the inventory atomically after #163 integrates and #165 is non-force restacked onto fresh protected truth. + +## Remaining active-PR gap + +PR #165 remains Draft. The domain, purpose-bound command, buyer HTTP/OpenAPI route, PostgreSQL correction adapter, durable replay route, service documentation, and exact-head 100% People coverage job now exist on the active child, but they are not protected-branch shipment and current-head hosted jobs must execute before they count as GREEN evidence. The remaining documentation work is the atomic canonical foundation handoff above, not a second competing source of truth. Any finding exposed by exact-head regression or independent review remains a repair finding. Parent PR #163 must integrate first; the child must then be non-force restacked/retargeted and reacquire exact-head workflows and independent review. + +The general recorded-interval trust boundary remains owned by its canonical repair lane; this feature owns only its correction-specific `recorded_at` ingress normalization and does not copy the general interval implementation. + +## References + +Allen, J. F. (1983). Maintaining knowledge about temporal intervals. *Communications of the ACM, 26*(11), 832–843. https://doi.org/10.1145/182.358434 + +Internet Engineering Task Force. (2024). *Universally unique IDentifiers (UUIDs)* (RFC 9562). https://doi.org/10.17487/RFC9562 + +Jensen, C. S., & Snodgrass, R. T. (1999). Temporal data management. *IEEE Transactions on Knowledge and Data Engineering, 11*(1), 36–44. https://doi.org/10.1109/69.755613 diff --git a/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py b/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py index 5d4b720fa..43f046f8e 100644 --- a/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py +++ b/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py @@ -13,6 +13,10 @@ validate_assignment_write, validate_position_seat_capacity, ) +from orgmetra_hris_kernel.assignment_correction import ( + AssignmentSupersessionFact, + correct_assignment_category, +) from orgmetra_hris_kernel.audit import AuditOutboxEvent from orgmetra_hris_kernel.correction import close_recorded_interval from orgmetra_hris_kernel.employment import validate_person_employment_exclusivity @@ -54,6 +58,7 @@ __all__ = [ "AssignmentFact", "AssignmentPortfolioError", + "AssignmentSupersessionFact", "AuditOutboxEvent", "CorrectionError", "DateInterval", @@ -79,6 +84,7 @@ "WorkforceCompositionSnapshot", "build_workforce_composition_snapshot", "close_recorded_interval", + "correct_assignment_category", "resolve_bitemporal_facts", "resolve_single_valued_fact", "validate_assignment_employment_coverage", diff --git a/packages/hris-kernel/src/orgmetra_hris_kernel/assignment_correction.py b/packages/hris-kernel/src/orgmetra_hris_kernel/assignment_correction.py new file mode 100644 index 000000000..72c74eaa2 --- /dev/null +++ b/packages/hris-kernel/src/orgmetra_hris_kernel/assignment_correction.py @@ -0,0 +1,175 @@ +"""Build immutable Assignment category corrections and supersession provenance.""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from datetime import datetime, timedelta, timezone +from uuid import UUID + +from orgmetra_hris_kernel.correction import close_recorded_interval +from orgmetra_hris_kernel.errors import CorrectionError +from orgmetra_hris_kernel.facts import AssignmentFact +from orgmetra_hris_kernel.intervals import RecordedInterval + +_EXPLICIT_ASSIGNMENT_CATEGORY_CODES = frozenset({"primary", "concurrent_secondary"}) +_MAX_UUID_INT = (1 << 128) - 1 + + +def _require_operational_uuid(value: object, field_name: str) -> UUID: + """Require exact runtime UUID identity and reject protocol sentinel values.""" + if type(value) is not UUID: + raise CorrectionError( + f"{field_name} must be an exact UUID.", + next_action="Use the authoritative operational UUID assigned to this correction record.", + ) + if value.int in (0, _MAX_UUID_INT): + raise CorrectionError( + f"{field_name} must be an operational UUID, not a reserved sentinel.", + next_action="Allocate a non-reserved operational UUID and retry the correction.", + ) + return value + + +def _require_recorded_at(value: object) -> datetime: + """Detach one exact system timestamp from caller-controlled timezone behavior.""" + if type(value) is not datetime or value.tzinfo is None: + raise CorrectionError( + "recorded_at must be an exact timezone-aware datetime.", + next_action="Use the database-owned correction timestamp with an explicit UTC offset.", + ) + try: + offset = value.utcoffset() + except Exception as exc: + raise CorrectionError( + "recorded_at must expose a stable UTC offset.", + next_action="Use the database-owned correction timestamp with an explicit UTC offset.", + ) from exc + if type(offset) is not timedelta: + raise CorrectionError( + "recorded_at must expose a stable UTC offset.", + next_action="Use the database-owned correction timestamp with an explicit UTC offset.", + ) + return value.replace(tzinfo=timezone(offset)) + + +@dataclass(frozen=True, slots=True) +class AssignmentSupersessionFact: + """Link one superseded Assignment fact to its immutable replacement.""" + + tenant_record_id: UUID + assignment_supersession_record_id: UUID + predecessor_assignment_record_id: UUID + replacement_assignment_record_id: UUID + recorded_at: datetime + + def __post_init__(self) -> None: + """Reject malformed provenance identities and detach its recorded timestamp.""" + for field_name in ( + "tenant_record_id", + "assignment_supersession_record_id", + "predecessor_assignment_record_id", + "replacement_assignment_record_id", + ): + _require_operational_uuid(getattr(self, field_name), field_name) + object.__setattr__(self, "recorded_at", _require_recorded_at(self.recorded_at)) + if self.predecessor_assignment_record_id == self.replacement_assignment_record_id: + raise CorrectionError( + "Supersession provenance requires distinct Assignment identities.", + next_action="Allocate a new replacement Assignment record ID and retry the correction.", + ) + + +def correct_assignment_category( + predecessor: AssignmentFact, + *, + replacement_assignment_record_id: UUID, + assignment_supersession_record_id: UUID, + corrected_category_code: str, + recorded_at: datetime, +) -> tuple[AssignmentFact, AssignmentFact, AssignmentSupersessionFact]: + """Close one explicit Assignment fact and create a linked category replacement. + + The replacement preserves tenant, Employment, Person, Position, allocation, + and effective-time truth. It receives a new Assignment identity and a new + open recorded interval beginning exactly when the predecessor closes. This + operation corrects a committed explicit category; classifying historical + ``legacy_unspecified`` rows remains outside this contract. + + Callers must re-run the Assignment portfolio and Position-capacity invariants + against locked authoritative state before persisting the three returned facts + in one transaction. + + Args: + predecessor: Recorded-open, explicitly classified Assignment being corrected. + replacement_assignment_record_id: New operational identity for the replacement. + assignment_supersession_record_id: Identity of the normalized provenance edge. + corrected_category_code: Exact explicit category chosen by the reviewer. + recorded_at: System-recorded time shared by closure, replacement, and edge. + + Returns: + The closed predecessor, open replacement, and normalized supersession fact. + + Raises: + CorrectionError: The predecessor is not explicitly classified, the + correction is malformed or a no-op, the identity is reused, or the + predecessor history cannot be closed. + """ + if ( + type(predecessor.assignment_category_code) is not str + or predecessor.assignment_category_code not in _EXPLICIT_ASSIGNMENT_CATEGORY_CODES + ): + raise CorrectionError( + "Predecessor Assignment must have an explicit governed category.", + next_action=( + "Use the separately governed historical-classification workflow for " + "legacy or malformed Assignment facts." + ), + ) + if ( + type(corrected_category_code) is not str + or corrected_category_code not in _EXPLICIT_ASSIGNMENT_CATEGORY_CODES + ): + raise CorrectionError( + "The corrected category must be primary or concurrent_secondary.", + next_action="Choose the reviewed explicit Assignment category, then save again.", + ) + if corrected_category_code == predecessor.assignment_category_code: + raise CorrectionError( + "Assignment category correction must select a different category.", + next_action="Keep the existing Assignment when its category is already correct.", + ) + + predecessor_assignment_record_id = _require_operational_uuid( + predecessor.assignment_record_id, + "predecessor_assignment_record_id", + ) + replacement_assignment_record_id = _require_operational_uuid( + replacement_assignment_record_id, + "replacement_assignment_record_id", + ) + assignment_supersession_record_id = _require_operational_uuid( + assignment_supersession_record_id, + "assignment_supersession_record_id", + ) + recorded_at = _require_recorded_at(recorded_at) + if replacement_assignment_record_id == predecessor_assignment_record_id: + raise CorrectionError( + "A category correction requires a new replacement Assignment identity.", + next_action="Allocate a new Assignment record ID and retry the correction.", + ) + + closed = close_recorded_interval(predecessor, recorded_to=recorded_at) + replacement = replace( + predecessor, + assignment_record_id=replacement_assignment_record_id, + assignment_category_code=corrected_category_code, + recorded=RecordedInterval(start=recorded_at), + ) + supersession = AssignmentSupersessionFact( + tenant_record_id=predecessor.tenant_record_id, + assignment_supersession_record_id=assignment_supersession_record_id, + predecessor_assignment_record_id=predecessor_assignment_record_id, + replacement_assignment_record_id=replacement_assignment_record_id, + recorded_at=recorded_at, + ) + return closed, replacement, supersession diff --git a/packages/hris-kernel/tests/test_assignment_category_correction.py b/packages/hris-kernel/tests/test_assignment_category_correction.py new file mode 100644 index 000000000..ca2ced4ef --- /dev/null +++ b/packages/hris-kernel/tests/test_assignment_category_correction.py @@ -0,0 +1,282 @@ +"""Assignment category correction and supersession regressions.""" + +from dataclasses import replace +from datetime import datetime +from uuid import UUID + +import pytest + +from orgmetra_hris_kernel import ( + AssignmentSupersessionFact, + CorrectionError, + RecordedInterval, + correct_assignment_category, +) + +from .conftest import recorded, utc + +SUPERSESSION = UUID("10000000-0000-7000-8000-000000000390") +REPLACEMENT = UUID("10000000-0000-7000-8000-000000000391") +MAX_UUID = UUID("ffffffff-ffff-ffff-ffff-ffffffffffff") +NIL_UUID = UUID(int=0) + + +class ForgedCategory(str): + """Represent caller-controlled string behavior at the correction boundary.""" + + +class ForgedUUID(UUID): + """Represent caller-controlled UUID equality at the correction boundary.""" + + def __eq__(self, other: object) -> bool: + """Lie about identity equality while retaining different UUID bytes.""" + return False + + __hash__ = UUID.__hash__ + + +class ForgedDateTime(datetime): + """Represent executable datetime behavior at the correction boundary.""" + + +def test_category_correction_closes_and_links_an_immutable_replacement( + jordan_icu_assignment, +) -> None: + """Correction preserves Assignment semantics while replacing category truth.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + corrected_at = utc(2024, 6, 1, 12) + + closed, replacement, supersession = correct_assignment_category( + predecessor, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="concurrent_secondary", + recorded_at=corrected_at, + ) + + assert closed.assignment_record_id == predecessor.assignment_record_id + assert closed.recorded.start == predecessor.recorded.start + assert closed.recorded.end == corrected_at + assert replacement.assignment_record_id == REPLACEMENT + assert replacement.tenant_record_id == predecessor.tenant_record_id + assert replacement.employment_record_id == predecessor.employment_record_id + assert replacement.person_record_id == predecessor.person_record_id + assert replacement.position_record_id == predecessor.position_record_id + assert replacement.allocation_ratio == predecessor.allocation_ratio + assert replacement.effective == predecessor.effective + assert replacement.recorded == RecordedInterval(start=corrected_at) + assert replacement.assignment_category_code == "concurrent_secondary" + assert supersession == AssignmentSupersessionFact( + tenant_record_id=predecessor.tenant_record_id, + assignment_supersession_record_id=SUPERSESSION, + predecessor_assignment_record_id=predecessor.assignment_record_id, + replacement_assignment_record_id=REPLACEMENT, + recorded_at=corrected_at, + ) + + +@pytest.mark.parametrize( + "predecessor_category", + ["legacy_unspecified", "secondary", ForgedCategory("primary")], +) +def test_category_correction_rejects_non_explicit_predecessor_categories( + jordan_icu_assignment, + predecessor_category, +) -> None: + """This correction contract starts only from exact committed explicit category truth.""" + predecessor = replace( + jordan_icu_assignment, + assignment_category_code=predecessor_category, + ) + + with pytest.raises(CorrectionError, match="explicit governed category"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="primary", + recorded_at=utc(2024, 6, 1, 12), + ) + + +@pytest.mark.parametrize( + "corrected_category_code", + ["legacy_unspecified", "secondary", ForgedCategory("concurrent_secondary")], +) +def test_category_correction_rejects_non_operational_target_categories( + jordan_icu_assignment, + corrected_category_code, +) -> None: + """A correction target is an exact explicit category, never a sentinel or alias.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + + with pytest.raises(CorrectionError, match="corrected category"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code=corrected_category_code, + recorded_at=utc(2024, 6, 1, 12), + ) + + +def test_category_correction_rejects_noop_and_identity_reuse(jordan_icu_assignment) -> None: + """A correction must change category truth and allocate a new Assignment identity.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + + with pytest.raises(CorrectionError, match="different category"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="primary", + recorded_at=utc(2024, 6, 1, 12), + ) + with pytest.raises(CorrectionError, match="replacement Assignment identity"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=predecessor.assignment_record_id, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="concurrent_secondary", + recorded_at=utc(2024, 6, 1, 12), + ) + + +def test_category_correction_rejects_forged_reused_assignment_identity( + jordan_icu_assignment, +) -> None: + """A UUID subtype cannot lie about equality to reuse the predecessor identity.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + forged_reuse = ForgedUUID(str(predecessor.assignment_record_id)) + + with pytest.raises(CorrectionError, match="replacement_assignment_record_id"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=forged_reuse, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="concurrent_secondary", + recorded_at=utc(2024, 6, 1, 12), + ) + + +@pytest.mark.parametrize("invalid_id", [NIL_UUID, MAX_UUID, "not-a-uuid"]) +def test_category_correction_rejects_non_operational_new_identities( + jordan_icu_assignment, + invalid_id, +) -> None: + """New correction identities must match the PostgreSQL operational UUID contract.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + + with pytest.raises(CorrectionError, match="replacement_assignment_record_id"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=invalid_id, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="concurrent_secondary", + recorded_at=utc(2024, 6, 1, 12), + ) + with pytest.raises(CorrectionError, match="assignment_supersession_record_id"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=invalid_id, + corrected_category_code="concurrent_secondary", + recorded_at=utc(2024, 6, 1, 12), + ) + + +def test_supersession_fact_rejects_direct_identity_drift(jordan_icu_assignment) -> None: + """The exported provenance fact cannot be directly constructed with invalid identity truth.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + valid = AssignmentSupersessionFact( + tenant_record_id=predecessor.tenant_record_id, + assignment_supersession_record_id=SUPERSESSION, + predecessor_assignment_record_id=predecessor.assignment_record_id, + replacement_assignment_record_id=REPLACEMENT, + recorded_at=utc(2024, 6, 1, 12), + ) + + for field_name in ( + "tenant_record_id", + "assignment_supersession_record_id", + "predecessor_assignment_record_id", + "replacement_assignment_record_id", + ): + with pytest.raises(CorrectionError, match=field_name): + replace(valid, **{field_name: ForgedUUID(str(REPLACEMENT))}) + with pytest.raises(CorrectionError, match=field_name): + replace(valid, **{field_name: NIL_UUID}) + + with pytest.raises(CorrectionError, match="distinct Assignment identities"): + replace(valid, replacement_assignment_record_id=valid.predecessor_assignment_record_id) + + +@pytest.mark.parametrize( + "invalid_recorded_at", + [ + "2024-06-01T12:00:00Z", + datetime(2024, 6, 1, 12), + ForgedDateTime.fromtimestamp(1717243200, tz=utc(2024, 6, 1, 12).tzinfo), + ], +) +def test_category_correction_rejects_malformed_recorded_time( + jordan_icu_assignment, + invalid_recorded_at, +) -> None: + """Correction provenance requires one exact offset-aware system timestamp.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + + with pytest.raises(CorrectionError, match="recorded_at"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="concurrent_secondary", + recorded_at=invalid_recorded_at, + ) + + +def test_supersession_fact_rejects_direct_recorded_time_drift(jordan_icu_assignment) -> None: + """Direct provenance construction cannot bypass recorded-time runtime integrity.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + + with pytest.raises(CorrectionError, match="recorded_at"): + AssignmentSupersessionFact( + tenant_record_id=predecessor.tenant_record_id, + assignment_supersession_record_id=SUPERSESSION, + predecessor_assignment_record_id=predecessor.assignment_record_id, + replacement_assignment_record_id=REPLACEMENT, + recorded_at=datetime(2024, 6, 1, 12), + ) + + +def test_category_correction_rejects_already_closed_predecessor(jordan_icu_assignment) -> None: + """Only the currently recorded-open fact can be superseded by this operation.""" + predecessor = replace( + jordan_icu_assignment, + assignment_category_code="primary", + recorded=recorded(utc(2024, 3, 1, 16), utc(2024, 5, 1, 16)), + ) + + with pytest.raises(CorrectionError, match="already closed"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="concurrent_secondary", + recorded_at=utc(2024, 6, 1, 12), + ) + + +def test_category_correction_rejects_non_forward_recorded_time(jordan_icu_assignment) -> None: + """Supersession cannot close history at or before the predecessor recorded start.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + + with pytest.raises(CorrectionError, match="strictly later"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="concurrent_secondary", + recorded_at=predecessor.recorded.start, + ) diff --git a/packages/hris-kernel/tests/test_assignment_category_correction_recorded_time_integrity.py b/packages/hris-kernel/tests/test_assignment_category_correction_recorded_time_integrity.py new file mode 100644 index 000000000..14dbc1f2c --- /dev/null +++ b/packages/hris-kernel/tests/test_assignment_category_correction_recorded_time_integrity.py @@ -0,0 +1,130 @@ +"""Recorded-time integrity regressions for Assignment category corrections.""" + +from dataclasses import replace +from datetime import datetime, timedelta, timezone, tzinfo +from uuid import UUID + +import pytest + +from orgmetra_hris_kernel import ( + AssignmentSupersessionFact, + CorrectionError, + correct_assignment_category, +) + +SUPERSESSION = UUID("10000000-0000-7000-8000-000000000390") +REPLACEMENT = UUID("10000000-0000-7000-8000-000000000391") + + +class FixedCallerTimezone(tzinfo): + """Expose caller-owned timezone behavior behind an exact built-in datetime.""" + + def utcoffset(self, dt: datetime | None) -> timedelta: + """Return one valid offset while remaining caller-controlled code.""" + return timedelta(hours=9) + + def dst(self, dt: datetime | None) -> timedelta: + """Provide a stable daylight-saving offset for datetime compatibility.""" + return timedelta(0) + + def tzname(self, dt: datetime | None) -> str: + """Return a deterministic display name that must not survive detachment.""" + return "CALLER" + + +class ExplodingCallerTimezone(FixedCallerTimezone): + """Raise from offset resolution to verify stable domain error normalization.""" + + def utcoffset(self, dt: datetime | None) -> timedelta: + """Simulate an untrusted timezone provider failure.""" + raise RuntimeError("caller-controlled timezone failure") + + +class OffsetlessCallerTimezone(FixedCallerTimezone): + """Return no usable offset despite carrying a non-null tzinfo object.""" + + def utcoffset(self, dt: datetime | None) -> None: + """Expose the offsetless custom-timezone case explicitly.""" + return None + + +class ForgedTimedelta(timedelta): + """Represent caller-defined executable offset evidence.""" + + +class ForgedOffsetCallerTimezone(FixedCallerTimezone): + """Return a timedelta subtype rather than an exact trusted offset value.""" + + def utcoffset(self, dt: datetime | None) -> timedelta: + """Preserve valid numeric offset semantics while changing runtime identity.""" + return ForgedTimedelta(hours=9) + + +def _caller_recorded_at(zone: tzinfo) -> datetime: + """Build an exact datetime whose timezone implementation remains caller-owned.""" + return datetime(2024, 6, 1, 12, 0, tzinfo=zone) + + +def test_category_correction_detaches_caller_timezone_before_returning_provenance( + jordan_icu_assignment, +) -> None: + """Accepted recorded time keeps the instant without executable caller timezone state.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + caller_time = _caller_recorded_at(FixedCallerTimezone()) + + closed, replacement, supersession = correct_assignment_category( + predecessor, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="concurrent_secondary", + recorded_at=caller_time, + ) + + for stored in ( + closed.recorded.end, + replacement.recorded.start, + supersession.recorded_at, + ): + assert type(stored) is datetime + assert type(stored.tzinfo) is timezone + assert stored.utcoffset() == timedelta(hours=9) + + +def test_direct_supersession_construction_detaches_caller_timezone( + jordan_icu_assignment, +) -> None: + """Direct provenance construction applies the same fixed-offset detachment boundary.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + + supersession = AssignmentSupersessionFact( + tenant_record_id=predecessor.tenant_record_id, + assignment_supersession_record_id=SUPERSESSION, + predecessor_assignment_record_id=predecessor.assignment_record_id, + replacement_assignment_record_id=REPLACEMENT, + recorded_at=_caller_recorded_at(FixedCallerTimezone()), + ) + + assert type(supersession.recorded_at) is datetime + assert type(supersession.recorded_at.tzinfo) is timezone + assert supersession.recorded_at.utcoffset() == timedelta(hours=9) + + +@pytest.mark.parametrize( + "caller_timezone", + [ExplodingCallerTimezone(), OffsetlessCallerTimezone(), ForgedOffsetCallerTimezone()], +) +def test_category_correction_rejects_untrusted_timezone_offset_evidence( + jordan_icu_assignment, + caller_timezone, +) -> None: + """Reject provider failures, missing offsets, and executable offset subtypes.""" + predecessor = replace(jordan_icu_assignment, assignment_category_code="primary") + + with pytest.raises(CorrectionError, match="recorded_at"): + correct_assignment_category( + predecessor, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + corrected_category_code="concurrent_secondary", + recorded_at=_caller_recorded_at(caller_timezone), + ) diff --git a/services/people-api/README.md b/services/people-api/README.md index 548a83446..ab198cebf 100644 --- a/services/people-api/README.md +++ b/services/people-api/README.md @@ -16,4 +16,6 @@ The People API quality workflow is part of this contract and must run for pull r `PeopleMutationAsgiApp` exposes the governed People mutation API as `POST /v1/employment-records`, `POST /v1/position-records`, and `POST /v1/assignment-records`. Each command requires an idempotency key, tenant/actor/purpose headers, a non-blank accountable decision reason, human confirmation, and versioned evidence. The HTTP boundary enforces the exact OpenAPI evidence-object shape and cardinality, rejects additional fields and duplicate evidence items, and canonicalizes the complete reference/version set independent of array order. It first derives a PII-minimized `evidence_set_v1:` identity and then binds that identity together with the exact validated decision reason into `governance_evidence_v1:`. The free-text reason and raw evidence references are not copied into the portable audit envelope, but any reason/reference/version drift changes the governance binding, the immutable audit correlation evidence, and the durable idempotency command digest. A caller therefore cannot reuse the same key after silently changing the high-impact rationale and receive an incorrect replay. The validated `Idempotency-Key` is copied onto the application command and into `PostgresPeopleMutationPort`. Employment and assignment writes require a current `candidate_worker_conversion_record` (`recorded_to IS NULL`) and reuse `orgmetra_hris_kernel` exclusivity and assignment-coverage checks before the port inserts the authoritative fact, calls `record_audit_outbox_event`, and stores `people_mutation_idempotency_record` in the same transaction. A matching retry returns the first committed identity without a second HRIS, audit, or outbox fact. Successful responses contain only opaque record identifiers. -The superseded persistence model must not be restored, and the service must not use direct cross-service application-table SQL. +`AssignmentCorrectionAsgiApp` exposes the active correction slice as `POST /v1/assignment-records/{assignment_record_id}/category-corrections`. The route accepts no generic Assignment update body: it binds the predecessor identity in the path and accepts only `corrected_category_code`, a namespaced human `confirmation_reference`, and `evidence_version_code`, together with the same bearer identity and tenant/actor/purpose/idempotency headers used by governed People writes. Authorization is field-scoped to `assignment_category_code` with operation `correct_record`. `PostgresAssignmentCorrectionMutationPort` then closes the recorded-open predecessor, creates a new Assignment identity with unchanged Employment/Person/Position/allocation/effective truth, persists the normalized predecessor→replacement supersession edge, audit/outbox evidence, and replay binding in one transaction. Exact retries return the first replacement and supersession identities; changed semantics under the same key fail closed. The additive service contract is `assignment-correction.openapi.yaml`; it does not silently broaden the foundation OpenAPI or create an in-place category mutation endpoint. + +The superseded persistence model must not be restored, and the service must not use direct cross-service application-table SQL. \ No newline at end of file diff --git a/services/people-api/assignment-correction.openapi.yaml b/services/people-api/assignment-correction.openapi.yaml new file mode 100644 index 000000000..07b8191af --- /dev/null +++ b/services/people-api/assignment-correction.openapi.yaml @@ -0,0 +1,183 @@ +openapi: 3.2.0 +info: + title: Orgmetra People API - Assignment category correction + version: 0.1.0 + summary: Purpose-bound correction of committed Assignment category facts +servers: + - url: https://api.orgmetra.example/v1 +paths: + /assignment-records/{assignment_record_id}/category-corrections: + post: + operationId: correctAssignmentRecordCategory + summary: Replace one committed Assignment with a linked category correction + security: + - keyverse_oidc: + - orgmetra.people.write + parameters: + - name: assignment_record_id + in: path + required: true + schema: + type: string + format: uuid + - name: Idempotency-Key + in: header + required: true + schema: + type: string + minLength: 16 + maxLength: 200 + - name: X-Tenant-Reference + in: header + required: true + schema: + type: string + format: uuid + - name: X-Actor-Reference + in: header + required: true + schema: + type: string + minLength: 1 + maxLength: 200 + - name: X-Purpose-Code + in: header + required: true + schema: + type: string + pattern: '^[a-z][a-z0-9_]{2,63}$' + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/AssignmentCategoryCorrectionCommand' + responses: + '201': + description: Correction committed as a linked replacement fact. + headers: + Location: + description: Canonical URI for the replacement Assignment record. + schema: + type: string + format: uri-reference + content: + application/json: + schema: + $ref: '#/components/schemas/AssignmentCategoryCorrectionResult' + '400': + $ref: '#/components/responses/InvalidRequest' + '401': + $ref: '#/components/responses/Unauthenticated' + '403': + $ref: '#/components/responses/Forbidden' + '404': + $ref: '#/components/responses/RouteNotFound' + '405': + $ref: '#/components/responses/MethodNotAllowed' + '409': + $ref: '#/components/responses/IntegrityConflict' + '413': + $ref: '#/components/responses/PayloadTooLarge' + '415': + $ref: '#/components/responses/UnsupportedMediaType' + '500': + $ref: '#/components/responses/InternalError' +components: + securitySchemes: + keyverse_oidc: + type: openIdConnect + description: Keyverse access token carrying the least-privilege People write scope. + openIdConnectUrl: https://identity.orgmetra.example/.well-known/openid-configuration + schemas: + AssignmentCategoryCorrectionCommand: + type: object + additionalProperties: false + required: + - corrected_category_code + - confirmation_reference + - evidence_version_code + properties: + corrected_category_code: + type: string + enum: [primary, concurrent_secondary] + confirmation_reference: + type: string + maxLength: 300 + pattern: '^[a-z][a-z0-9_]*:[A-Za-z0-9][A-Za-z0-9._~-]*$' + evidence_version_code: + type: string + maxLength: 200 + pattern: '^[A-Za-z0-9][A-Za-z0-9._:-]*$' + AssignmentCategoryCorrectionResult: + type: object + additionalProperties: false + required: + - replacement_assignment_record_id + - assignment_supersession_record_id + properties: + replacement_assignment_record_id: + type: string + format: uuid + assignment_supersession_record_id: + type: string + format: uuid + ErrorResponse: + type: object + additionalProperties: false + required: [error_code, message, next_action, support_reference] + properties: + error_code: + type: string + message: + type: string + next_action: + type: string + support_reference: + type: string + responses: + InvalidRequest: + description: The governed command or required headers are invalid. + content: + application/json: + schema: {$ref: '#/components/schemas/ErrorResponse'} + Unauthenticated: + description: Bearer authentication is missing or invalid. + content: + application/json: + schema: {$ref: '#/components/schemas/ErrorResponse'} + Forbidden: + description: Tenant, actor, purpose, scope, or field authorization is denied. + content: + application/json: + schema: {$ref: '#/components/schemas/ErrorResponse'} + RouteNotFound: + description: The request does not address the owned correction route. + content: + application/json: + schema: {$ref: '#/components/schemas/ErrorResponse'} + MethodNotAllowed: + description: The correction route accepts POST only. + content: + application/json: + schema: {$ref: '#/components/schemas/ErrorResponse'} + IntegrityConflict: + description: Current Assignment truth or idempotency evidence conflicts with the requested correction. + content: + application/json: + schema: {$ref: '#/components/schemas/ErrorResponse'} + PayloadTooLarge: + description: The JSON command exceeds the bounded request size. + content: + application/json: + schema: {$ref: '#/components/schemas/ErrorResponse'} + UnsupportedMediaType: + description: The request is not application/json. + content: + application/json: + schema: {$ref: '#/components/schemas/ErrorResponse'} + InternalError: + description: An internal dependency failed without exposing secrets or backend details. + content: + application/json: + schema: {$ref: '#/components/schemas/ErrorResponse'} diff --git a/services/people-api/src/orgmetra_people_api/__init__.py b/services/people-api/src/orgmetra_people_api/__init__.py index b043bed33..c15437401 100644 --- a/services/people-api/src/orgmetra_people_api/__init__.py +++ b/services/people-api/src/orgmetra_people_api/__init__.py @@ -1,5 +1,12 @@ """Request-edge, governed read, confirmed-hire, and People mutation contracts.""" +from orgmetra_people_api.assignment_correction_http import AssignmentCorrectionAsgiApp +from orgmetra_people_api.assignment_correction_mutations import ( + AssignmentCorrectionMutationCommand, + AssignmentCorrectionMutationPort, + AssignmentCorrectionMutationResult, + correct_assignment_record_category, +) from orgmetra_people_api.auth import ( AuthenticatedPrincipal, AuthenticationFailed, @@ -41,10 +48,15 @@ read_worker_people_record, ) from orgmetra_people_api.postgres import PostgresPeopleReadPort +from orgmetra_people_api.postgres_assignment_corrections import PostgresAssignmentCorrectionMutationPort from orgmetra_people_api.postgres_hire import PostgresHireAcceptancePort from orgmetra_people_api.postgres_mutations import PostgresPeopleMutationPort __all__ = [ + "AssignmentCorrectionAsgiApp", + "AssignmentCorrectionMutationCommand", + "AssignmentCorrectionMutationPort", + "AssignmentCorrectionMutationResult", "AuthenticatedPrincipal", "AuthenticationFailed", "AuthorizedWorkerPeopleView", @@ -64,6 +76,7 @@ "PeopleRecordNotFound", "PositionMutationCommand", "PositionMutationResult", + "PostgresAssignmentCorrectionMutationPort", "PostgresHireAcceptancePort", "PostgresPeopleMutationPort", "PostgresPeopleReadPort", @@ -75,6 +88,7 @@ "WorkerPeopleRecord", "accept_confirmed_hire", "authorize_resource_fields", + "correct_assignment_record_category", "create_assignment_record", "create_employment_record", "create_position_record", diff --git a/services/people-api/src/orgmetra_people_api/assignment_correction_http.py b/services/people-api/src/orgmetra_people_api/assignment_correction_http.py new file mode 100644 index 000000000..db6e70052 --- /dev/null +++ b/services/people-api/src/orgmetra_people_api/assignment_correction_http.py @@ -0,0 +1,290 @@ +"""Purpose-bound ASGI boundary for immutable Assignment category correction.""" + +from __future__ import annotations + +from dataclasses import dataclass +import logging +from secrets import token_urlsafe +from typing import Callable, Mapping +from uuid import UUID, uuid4 + +from orgmetra_hris_kernel import KernelError +from orgmetra_keyverse_adapter import AuthorizationDeniedError, PurposeBoundAccessPolicy + +from orgmetra_people_api.assignment_correction_mutations import ( + AssignmentCorrectionMutationCommand, + AssignmentCorrectionMutationPort, + correct_assignment_record_category, +) +from orgmetra_people_api.auth import ( + AuthenticatedPrincipal, + AuthenticationFailed, + TokenAuthenticator, + extract_bearer_token, +) +from orgmetra_people_api.hire_http import ( + _InvalidHttpRequest, + _PayloadTooLarge, + _UnsupportedMediaType, + _read_json_object, + _require_json_content_type, +) +from orgmetra_people_api.http import AsgiReceive, AsgiSend, _authorization_header, _send_json +from orgmetra_people_api.mutation_http import _parse_command_headers, _send_error +from orgmetra_people_api.mutations import PeopleMutationIntegrityError + +_LOGGER = logging.getLogger(__name__) +_BODY_KEYS = frozenset( + { + "corrected_category_code", + "confirmation_reference", + "evidence_version_code", + } +) +_MAX_UUID_INT = (1 << 128) - 1 +_SUPPORT_REFERENCE_RANDOM_BYTES = 24 + + +def _predecessor_from_path(path: object) -> UUID | None: + """Return the operational predecessor identity for the one owned route.""" + if type(path) is not str: + return None + parts = path.strip("/").split("/") + if len(parts) != 4 or parts[0] != "v1" or parts[1] != "assignment-records" or parts[3] != "category-corrections": + return None + try: + predecessor = UUID(parts[2]) + except (AttributeError, ValueError): + return None + if predecessor.int in (0, _MAX_UUID_INT): + return None + return predecessor + + +def _require_body_string(payload: Mapping[str, object], field_name: str) -> str: + """Require an exact JSON string and leave semantic validation to the command.""" + value = payload.get(field_name) + if type(value) is not str: + raise _InvalidHttpRequest(f"{field_name} must be a string") + return value + + +def _correction_command( + *, + tenant_record_id: UUID, + predecessor_assignment_record_id: UUID, + payload: Mapping[str, object], + idempotency_key: str, + id_factory: Callable[[], UUID], +) -> AssignmentCorrectionMutationCommand: + """Map one exact HTTP body onto the governed application command.""" + if frozenset(payload) != _BODY_KEYS: + raise _InvalidHttpRequest("correction command fields are incomplete or unsupported") + return AssignmentCorrectionMutationCommand( + tenant_record_id=tenant_record_id, + predecessor_assignment_record_id=predecessor_assignment_record_id, + replacement_assignment_record_id=id_factory(), + assignment_supersession_record_id=id_factory(), + audit_event_record_id=id_factory(), + outbox_delivery_record_id=id_factory(), + corrected_category_code=_require_body_string(payload, "corrected_category_code"), + confirmation_reference=_require_body_string(payload, "confirmation_reference"), + evidence_version_code=_require_body_string(payload, "evidence_version_code"), + idempotency_key=idempotency_key, + ) + + +@dataclass(frozen=True, slots=True) +class AssignmentCorrectionAsgiApp: + """Expose one correction command without permitting in-place Assignment mutation. + + The route is ``POST /v1/assignment-records/{assignment_record_id}/category-corrections``. + It authenticates the actor through Keyverse, binds tenant/actor/purpose/idempotency + headers, authorizes only ``assignment_category_code`` for ``correct_record``, and + returns opaque replacement plus supersession identities after the transaction commits. + """ + + authenticator: TokenAuthenticator + correction_policy: PurposeBoundAccessPolicy + mutation_port: AssignmentCorrectionMutationPort + id_factory: Callable[[], UUID] = uuid4 + + def __post_init__(self) -> None: + """Reject incomplete governed dependencies before serving corrections.""" + if not isinstance(self.authenticator, TokenAuthenticator): + raise TypeError("authenticator must implement TokenAuthenticator") + if not isinstance(self.correction_policy, PurposeBoundAccessPolicy): + raise TypeError("correction_policy must be a PurposeBoundAccessPolicy") + if not isinstance(self.mutation_port, AssignmentCorrectionMutationPort): + raise TypeError("mutation_port must implement AssignmentCorrectionMutationPort") + if not callable(self.id_factory): + raise TypeError("id_factory must be callable") + + async def __call__(self, scope: Mapping[str, object], receive: AsgiReceive, send: AsgiSend) -> None: + """Serve one fail-closed correction without exposing credentials or free-text PII.""" + if scope.get("type") != "http": + raise ValueError("AssignmentCorrectionAsgiApp accepts only HTTP ASGI scopes") + if scope.get("method") != "POST": + await _send_error( + send, + status=405, + payload={"error": "method_not_allowed", "message": "Use POST for Assignment category corrections."}, + extra_headers=((b"allow", b"POST"),), + ) + return + + predecessor = _predecessor_from_path(scope.get("path")) + if predecessor is None: + await _send_error( + send, + status=404, + payload={ + "error": "route_not_found", + "message": "Use /v1/assignment-records/{assignment_record_id}/category-corrections.", + }, + ) + return + + try: + headers = _parse_command_headers(scope) + _require_json_content_type(scope) + except _UnsupportedMediaType: + await _send_error( + send, + status=415, + payload={"error": "unsupported_media_type", "message": "Send application/json and retry."}, + ) + return + except (_InvalidHttpRequest, ValueError, TypeError): + await _send_error( + send, + status=400, + payload={"error": "invalid_request", "message": "Correct the governed command headers and retry."}, + ) + return + + try: + bearer_token = extract_bearer_token(_authorization_header(scope)) + principal = await self.authenticator.authenticate(bearer_token) + if not isinstance(principal, AuthenticatedPrincipal): + raise TypeError("authenticator returned an invalid principal") + except AuthenticationFailed: + await _send_error( + send, + status=401, + payload={"error": "authentication_required", "message": "Provide one valid Bearer credential and retry."}, + extra_headers=((b"www-authenticate", b"Bearer"),), + ) + return + except Exception as error: # noqa: BLE001 - identity backend failures stay client-safe. + support_reference = f"err_{token_urlsafe(_SUPPORT_REFERENCE_RANDOM_BYTES)}" + _LOGGER.error( + "Assignment correction authentication failed", + extra={ + "tenant_record_id": str(headers.tenant_record_id), + "predecessor_assignment_record_id": str(predecessor), + "exception_type": type(error).__name__, + "support_reference": support_reference, + }, + ) + await _send_error( + send, + status=500, + payload={ + "error": "internal_error", + "message": "Retry later or contact an Orgmetra operator with the support reference; never include the bearer token.", + }, + support_reference=support_reference, + ) + return + + if principal.tenant_record_id != headers.tenant_record_id or principal.actor_reference != headers.actor_reference: + await _send_error( + send, + status=403, + payload={"error": "access_denied", "message": "Use the tenant and actor bound to the authenticated credential."}, + ) + return + + try: + payload = await _read_json_object(receive) + command = _correction_command( + tenant_record_id=headers.tenant_record_id, + predecessor_assignment_record_id=predecessor, + payload=payload, + idempotency_key=headers.idempotency_key, + id_factory=self.id_factory, + ) + except _PayloadTooLarge: + await _send_error( + send, + status=413, + payload={"error": "payload_too_large", "message": "Send one bounded JSON correction command and retry."}, + ) + return + except (_InvalidHttpRequest, ValueError, TypeError, StopIteration): + await _send_error( + send, + status=400, + payload={"error": "invalid_request", "message": "Correct the category, confirmation, evidence version, and command fields, then retry."}, + ) + return + + try: + result = correct_assignment_record_category( + principal=principal, + command=command, + purpose_code=headers.purpose_code, + policy=self.correction_policy, + mutation_port=self.mutation_port, + ) + except AuthorizationDeniedError: + await _send_error( + send, + status=403, + payload={"error": "access_denied", "message": "Request a purpose and scope authorized to correct only Assignment category."}, + ) + return + except (PeopleMutationIntegrityError, KernelError): + await _send_error( + send, + status=409, + payload={ + "error": "mutation_integrity_conflict", + "message": "The correction cannot be committed safely; refresh the Assignment and retry with current evidence.", + }, + ) + return + except Exception as error: # noqa: BLE001 - persistence details must not cross the HTTP boundary. + support_reference = f"err_{token_urlsafe(_SUPPORT_REFERENCE_RANDOM_BYTES)}" + _LOGGER.error( + "Assignment correction persistence failed", + extra={ + "tenant_record_id": str(headers.tenant_record_id), + "predecessor_assignment_record_id": str(predecessor), + "correlation_reference": f"audit_event_record:{command.audit_event_record_id.hex}", + "exception_type": type(error).__name__, + "support_reference": support_reference, + }, + ) + await _send_error( + send, + status=500, + payload={ + "error": "internal_error", + "message": "Retry later or contact an Orgmetra operator with the support reference; never include the bearer token.", + }, + support_reference=support_reference, + ) + return + + replacement = str(result.replacement_assignment_record_id) + await _send_json( + send, + status=201, + payload={ + "replacement_assignment_record_id": replacement, + "assignment_supersession_record_id": str(result.assignment_supersession_record_id), + }, + extra_headers=((b"location", f"/v1/assignment-records/{replacement}".encode("ascii")),), + ) diff --git a/services/people-api/src/orgmetra_people_api/assignment_correction_mutations.py b/services/people-api/src/orgmetra_people_api/assignment_correction_mutations.py new file mode 100644 index 000000000..e48509f37 --- /dev/null +++ b/services/people-api/src/orgmetra_people_api/assignment_correction_mutations.py @@ -0,0 +1,192 @@ +"""Purpose-bound application contract for Assignment category corrections.""" + +from __future__ import annotations + +from dataclasses import dataclass +from hashlib import sha256 +import json +import re +from typing import Protocol, runtime_checkable +from uuid import UUID + +from orgmetra_keyverse_adapter import AuthorizationDecision, PurposeBoundAccessPolicy + +from orgmetra_people_api.auth import AuthenticatedPrincipal +from orgmetra_people_api.authorization import authorize_resource_fields + +_MAX_UUID_INT = (1 << 128) - 1 +_EXPLICIT_ASSIGNMENT_CATEGORY_CODES = frozenset({"primary", "concurrent_secondary"}) +_CORRECTION_FIELDS = frozenset({"assignment_category_code"}) +_REFERENCE_PATTERN = re.compile(r"^[a-z][a-z0-9_]*:[A-Za-z0-9][A-Za-z0-9._~-]*$") +_VERSION_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$") +_CONFIRMATION_REFERENCE_MAX = 300 +_EVIDENCE_VERSION_MAX = 200 +_IDEMPOTENCY_MIN = 16 +_IDEMPOTENCY_MAX = 200 + + +def _require_operational_uuid(field_name: str, value: object) -> UUID: + """Require an exact operational UUID before any caller-defined behavior runs.""" + if type(value) is not UUID or value.int in (0, _MAX_UUID_INT): + raise ValueError(f"{field_name} must be an operational UUID.") + return value + + +def _require_reference(field_name: str, value: object) -> str: + """Require one exact, bounded namespaced opaque reference.""" + if ( + type(value) is not str + or not 1 <= len(value) <= _CONFIRMATION_REFERENCE_MAX + or _REFERENCE_PATTERN.fullmatch(value) is None + ): + raise ValueError(f"{field_name} must be a namespaced opaque reference of at most 300 characters.") + return value + + +def _require_version(value: object) -> str: + """Require one exact, bounded whitespace-free evidence version token.""" + if ( + type(value) is not str + or not 1 <= len(value) <= _EVIDENCE_VERSION_MAX + or _VERSION_PATTERN.fullmatch(value) is None + ): + raise ValueError("evidence_version_code must be a whitespace-free version token of at most 200 characters.") + return value + + +def _require_idempotency_key(value: object) -> str: + """Require an exact visible-ASCII correction replay key.""" + if type(value) is not str or not (_IDEMPOTENCY_MIN <= len(value) <= _IDEMPOTENCY_MAX): + raise ValueError("idempotency_key must be 16 to 200 visible ASCII characters.") + if any(ord(character) < 0x21 or ord(character) > 0x7E for character in value): + raise ValueError("idempotency_key must be 16 to 200 visible ASCII characters.") + return value + + +@dataclass(frozen=True, slots=True) +class AssignmentCorrectionMutationCommand: + """Human-confirmed command to replace one committed Assignment category fact.""" + + tenant_record_id: UUID + predecessor_assignment_record_id: UUID + replacement_assignment_record_id: UUID + assignment_supersession_record_id: UUID + audit_event_record_id: UUID + outbox_delivery_record_id: UUID + corrected_category_code: str + confirmation_reference: str + evidence_version_code: str + idempotency_key: str + + def __post_init__(self) -> None: + """Fail closed before authorization or persistence on malformed evidence.""" + for field_name in ( + "tenant_record_id", + "predecessor_assignment_record_id", + "replacement_assignment_record_id", + "assignment_supersession_record_id", + "audit_event_record_id", + "outbox_delivery_record_id", + ): + _require_operational_uuid(field_name, getattr(self, field_name)) + if self.predecessor_assignment_record_id == self.replacement_assignment_record_id: + raise ValueError("replacement_assignment_record_id must differ from the predecessor.") + if ( + type(self.corrected_category_code) is not str + or self.corrected_category_code not in _EXPLICIT_ASSIGNMENT_CATEGORY_CODES + ): + raise ValueError("corrected_category_code must be primary or concurrent_secondary.") + _require_reference("confirmation_reference", self.confirmation_reference) + _require_version(self.evidence_version_code) + _require_idempotency_key(self.idempotency_key) + + +@dataclass(frozen=True, slots=True) +class AssignmentCorrectionMutationResult: + """Opaque replacement and provenance identities returned after commit.""" + + replacement_assignment_record_id: UUID + assignment_supersession_record_id: UUID + + def __post_init__(self) -> None: + """Reject malformed adapter results at the service boundary.""" + _require_operational_uuid( + "replacement_assignment_record_id", + self.replacement_assignment_record_id, + ) + _require_operational_uuid( + "assignment_supersession_record_id", + self.assignment_supersession_record_id, + ) + + +def assignment_correction_command_digest( + *, + command: AssignmentCorrectionMutationCommand, + authorization: AuthorizationDecision, +) -> str: + """Hash correction semantics while excluding retry-generated record identities.""" + if type(command) is not AssignmentCorrectionMutationCommand: + raise TypeError("command must be an exact AssignmentCorrectionMutationCommand") + if type(authorization) is not AuthorizationDecision: + raise TypeError("authorization must be an exact AuthorizationDecision") + payload = { + "actor_reference": authorization.actor_reference, + "command_route": "assignment-category-corrections", + "method": "POST", + "purpose_code": authorization.purpose_code, + "semantic_command": { + "confirmation_reference": command.confirmation_reference, + "corrected_category_code": command.corrected_category_code, + "evidence_version_code": command.evidence_version_code, + "predecessor_assignment_record_id": str(command.predecessor_assignment_record_id), + }, + "tenant_record_id": str(command.tenant_record_id), + } + return sha256(json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")).hexdigest() + + +@runtime_checkable +class AssignmentCorrectionMutationPort(Protocol): + """Persist one authorized category correction atomically inside Orgmetra.""" + + def correct_assignment_category( + self, + *, + command: AssignmentCorrectionMutationCommand, + authorization: AuthorizationDecision, + ) -> AssignmentCorrectionMutationResult: + """Commit one linked correction or raise without partial writes.""" + + +def correct_assignment_record_category( + *, + principal: AuthenticatedPrincipal, + command: AssignmentCorrectionMutationCommand, + purpose_code: str, + policy: PurposeBoundAccessPolicy, + mutation_port: AssignmentCorrectionMutationPort, +) -> AssignmentCorrectionMutationResult: + """Authorize exactly one predecessor's category field before correction.""" + if type(command) is not AssignmentCorrectionMutationCommand: + raise TypeError("command must be an exact AssignmentCorrectionMutationCommand") + if not isinstance(mutation_port, AssignmentCorrectionMutationPort): + raise TypeError("mutation_port must implement AssignmentCorrectionMutationPort") + authorization = authorize_resource_fields( + principal=principal, + tenant_record_id=command.tenant_record_id, + resource_tenant_record_id=command.tenant_record_id, + resource_reference=f"assignment_record:{command.predecessor_assignment_record_id.hex}", + purpose_code=purpose_code, + operation_code="correct_record", + resource_kind="assignment_record", + requested_fields=_CORRECTION_FIELDS, + policy=policy, + ) + result = mutation_port.correct_assignment_category( + command=command, + authorization=authorization, + ) + if type(result) is not AssignmentCorrectionMutationResult: + raise TypeError("mutation_port must return an exact AssignmentCorrectionMutationResult") + return result diff --git a/services/people-api/src/orgmetra_people_api/postgres_assignment_corrections.py b/services/people-api/src/orgmetra_people_api/postgres_assignment_corrections.py new file mode 100644 index 000000000..2d766365e --- /dev/null +++ b/services/people-api/src/orgmetra_people_api/postgres_assignment_corrections.py @@ -0,0 +1,496 @@ +"""Atomic PostgreSQL adapter for governed Assignment category corrections.""" + +from __future__ import annotations + +from contextlib import AbstractContextManager +from dataclasses import dataclass +from datetime import datetime +from decimal import Decimal +from typing import Any, Callable +from uuid import UUID + +from orgmetra_hris_kernel import ( + AssignmentFact, + AuditOutboxEvent, + KernelError, + correct_assignment_category as build_assignment_category_correction, + validate_assignment_write, +) +from orgmetra_keyverse_adapter import AuthorizationDecision + +from orgmetra_people_api.assignment_correction_mutations import ( + AssignmentCorrectionMutationCommand, + AssignmentCorrectionMutationResult, + assignment_correction_command_digest, +) +from orgmetra_people_api.mutations import PeopleMutationIntegrityError, idempotency_record_id +from orgmetra_people_api.postgres_mutations import ( + _INSERT_IDEMPOTENCY_SQL, + _LOOKUP_IDEMPOTENCY_SQL, + _POST_LOCK_RECORDED_AT_SQL, + _READ_IDEMPOTENCY_SQL, + _READ_WRITE_SQL, + _TENANT_CONTEXT_SQL, + _assignment_from_row, + _employment_version_from_row, + _position_version_from_row, + _post_lock_recorded_at, + _record_audit, +) + +PostgresConnectionFactory = Callable[[], AbstractContextManager[Any]] + +_CORRECTION_ROUTE = "assignment-category-corrections" +_CORRECTION_FIELDS = frozenset({"assignment_category_code"}) + +_READ_PREDECESSOR_SQL = """ +SELECT + assignment.assignment_record_id, + assignment.employment_record_id, + assignment.person_record_id, + assignment.position_record_id, + assignment.allocation_ratio, + assignment.assignment_category_code, + assignment.effective_from, + assignment.effective_to, + assignment.recorded_from, + assignment.recorded_to +FROM public.assignment_record AS assignment +WHERE assignment.tenant_record_id = %s + AND assignment.assignment_record_id = %s + AND assignment.recorded_to IS NULL +LIMIT 2 +""".strip() + +_LOCK_EMPLOYMENT_VERSIONS_SQL = """ +SELECT + employment.employment_record_id, + version.employment_record_version_id, + employment.person_record_id, + version.employment_status_code, + version.employment_concurrency_code, + version.effective_from, + version.effective_to, + version.recorded_from, + version.recorded_to +FROM public.employment_record AS employment +JOIN public.employment_record_version AS version + ON version.tenant_record_id = employment.tenant_record_id + AND version.employment_record_id = employment.employment_record_id +WHERE employment.tenant_record_id = %s + AND employment.employment_record_id = %s +FOR UPDATE OF employment +""".strip() + +_LOCK_POSITION_VERSIONS_SQL = """ +SELECT + version.position_record_id, + version.position_record_version_id, + version.position_status_code, + version.effective_from, + version.effective_to, + version.recorded_from, + version.recorded_to +FROM public.position_record AS position +JOIN public.position_record_version AS version + ON version.tenant_record_id = position.tenant_record_id + AND version.position_record_id = position.position_record_id +WHERE position.tenant_record_id = %s + AND position.position_record_id = %s +FOR UPDATE OF position +""".strip() + +_LOCK_ASSIGNMENT_PORTFOLIO_SQL = """ +SELECT + assignment.assignment_record_id, + assignment.employment_record_id, + assignment.person_record_id, + assignment.position_record_id, + assignment.allocation_ratio, + assignment.assignment_category_code, + assignment.effective_from, + assignment.effective_to, + assignment.recorded_from, + assignment.recorded_to +FROM public.assignment_record AS assignment +WHERE assignment.tenant_record_id = %s + AND ( + assignment.employment_record_id = %s + OR assignment.position_record_id = %s + ) +ORDER BY assignment.assignment_record_id +FOR UPDATE OF assignment +""".strip() + +_CLOSE_PREDECESSOR_SQL = """ +UPDATE public.assignment_record +SET recorded_to = %s +WHERE tenant_record_id = %s + AND assignment_record_id = %s + AND recorded_to IS NULL +""".strip() + +_INSERT_REPLACEMENT_SQL = """ +INSERT INTO public.assignment_record ( + tenant_record_id, + assignment_record_id, + employment_record_id, + person_record_id, + position_record_id, + allocation_ratio, + assignment_category_code, + effective_from, + effective_to, + recorded_from +) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) +""".strip() + +_INSERT_SUPERSESSION_SQL = """ +INSERT INTO public.assignment_supersession_record ( + tenant_record_id, + assignment_supersession_record_id, + predecessor_assignment_record_id, + replacement_assignment_record_id, + recorded_at +) VALUES (%s, %s, %s, %s, %s) +""".strip() + +_READ_REPLAY_SUPERSESSION_SQL = """ +SELECT + supersession.assignment_supersession_record_id, + supersession.replacement_assignment_record_id +FROM public.assignment_supersession_record AS supersession +WHERE supersession.tenant_record_id = %s + AND supersession.predecessor_assignment_record_id = %s + AND supersession.replacement_assignment_record_id = %s +LIMIT 2 +""".strip() + + +def _is_operational_uuid(value: object) -> bool: + """Return whether a database identity is an exact non-reserved UUID.""" + return type(value) is UUID and value.int not in (0, (1 << 128) - 1) + + +def _is_sha256(value: object) -> bool: + """Return whether a stored command digest is one exact lowercase SHA-256 token.""" + return ( + type(value) is str + and len(value) == 64 + and all(character in "0123456789abcdef" for character in value) + ) + + +def _require_correction_authorization( + *, + authorization: object, + command: AssignmentCorrectionMutationCommand, +) -> AuthorizationDecision: + """Require the exact allow decision for the predecessor category correction.""" + if type(authorization) is not AuthorizationDecision: + raise PeopleMutationIntegrityError("assignment correction requires an exact authorization decision") + if ( + not authorization.allowed + or authorization.tenant_record_id != command.tenant_record_id + or authorization.resource_reference + != f"assignment_record:{command.predecessor_assignment_record_id.hex}" + or authorization.resource_kind != "assignment_record" + or authorization.operation_code != "correct_record" + or authorization.requested_fields != _CORRECTION_FIELDS + or authorization.authorized_fields != _CORRECTION_FIELDS + ): + raise PeopleMutationIntegrityError("assignment correction authorization does not match the predecessor") + return authorization + + +def _assignment_from_locked_row(tenant_record_id: UUID, row: tuple[object, ...]) -> AssignmentFact: + """Reconstruct one locked Assignment without trusting executable Decimal subclasses.""" + if len(row) != 10 or type(row[4]) is not Decimal: + raise PeopleMutationIntegrityError("assignment row is invalid") + return _assignment_from_row(tenant_record_id, row) + + +def _require_one_predecessor( + tenant_record_id: UUID, + rows: list[tuple[object, ...]], +) -> AssignmentFact: + """Require one recorded-open authoritative predecessor inside tenant scope.""" + if len(rows) != 1: + raise PeopleMutationIntegrityError("assignment correction predecessor is missing or ambiguous") + predecessor = _assignment_from_locked_row(tenant_record_id, rows[0]) + if predecessor.recorded.end is not None: + raise PeopleMutationIntegrityError("assignment correction predecessor is already closed") + return predecessor + + +def _require_locked_predecessor( + *, + candidate: AssignmentFact, + portfolio: list[AssignmentFact], +) -> AssignmentFact: + """Re-resolve the predecessor from the deterministically locked Assignment portfolio.""" + matches = [ + assignment + for assignment in portfolio + if assignment.assignment_record_id == candidate.assignment_record_id + ] + if len(matches) != 1: + raise PeopleMutationIntegrityError("assignment correction predecessor is missing or ambiguous") + predecessor = matches[0] + if predecessor.recorded.end is not None: + raise PeopleMutationIntegrityError("assignment correction predecessor is already closed") + if ( + predecessor.employment_record_id != candidate.employment_record_id + or predecessor.person_record_id != candidate.person_record_id + or predecessor.position_record_id != candidate.position_record_id + ): + raise PeopleMutationIntegrityError("assignment correction predecessor identity changed during locking") + return predecessor + + +def _replayed_correction( + cursor: Any, + *, + command: AssignmentCorrectionMutationCommand, + authorization: AuthorizationDecision, +) -> AssignmentCorrectionMutationResult | None: + """Serialize one replay key and return the first committed correction when present.""" + key_parameters = (command.tenant_record_id, _CORRECTION_ROUTE, command.idempotency_key) + cursor.execute(_LOOKUP_IDEMPOTENCY_SQL, key_parameters) + cursor.execute(_READ_IDEMPOTENCY_SQL, key_parameters) + rows = cursor.fetchmany(2) + if not rows: + return None + if len(rows) != 1 or len(rows[0]) != 2: + raise PeopleMutationIntegrityError("assignment correction idempotency row is invalid") + replacement_record_id, stored_digest = rows[0] + if not _is_operational_uuid(replacement_record_id) or not _is_sha256(stored_digest): + raise PeopleMutationIntegrityError("assignment correction idempotency row is invalid") + expected_digest = assignment_correction_command_digest( + command=command, + authorization=authorization, + ) + if stored_digest != expected_digest: + raise PeopleMutationIntegrityError("idempotency key is bound to a different command") + assert isinstance(replacement_record_id, UUID) + cursor.execute( + _READ_REPLAY_SUPERSESSION_SQL, + ( + command.tenant_record_id, + command.predecessor_assignment_record_id, + replacement_record_id, + ), + ) + supersession_rows = cursor.fetchmany(2) + if len(supersession_rows) != 1 or len(supersession_rows[0]) != 2: + raise PeopleMutationIntegrityError("assignment correction replay provenance is invalid") + supersession_record_id, linked_replacement_id = supersession_rows[0] + if ( + not _is_operational_uuid(supersession_record_id) + or type(linked_replacement_id) is not UUID + or linked_replacement_id != replacement_record_id + ): + raise PeopleMutationIntegrityError("assignment correction replay provenance is invalid") + assert isinstance(supersession_record_id, UUID) + return AssignmentCorrectionMutationResult( + replacement_assignment_record_id=replacement_record_id, + assignment_supersession_record_id=supersession_record_id, + ) + + +def _record_correction_idempotency( + cursor: Any, + *, + command: AssignmentCorrectionMutationCommand, + authorization: AuthorizationDecision, + replacement_record_id: UUID, +) -> None: + """Persist semantic replay evidence with the replacement inside the transaction.""" + cursor.execute( + _INSERT_IDEMPOTENCY_SQL, + ( + command.tenant_record_id, + idempotency_record_id( + tenant_record_id=command.tenant_record_id, + command_route_value=_CORRECTION_ROUTE, + idempotency_key=command.idempotency_key, + ), + _CORRECTION_ROUTE, + command.idempotency_key, + assignment_correction_command_digest( + command=command, + authorization=authorization, + ), + replacement_record_id, + ), + ) + + +@dataclass(frozen=True, slots=True) +class PostgresAssignmentCorrectionMutationPort: + """Persist one reviewed Assignment category correction in a tenant transaction.""" + + connection_factory: PostgresConnectionFactory + + def __post_init__(self) -> None: + """Reject an unusable database factory before a protected correction starts.""" + if not callable(self.connection_factory): + raise TypeError("connection_factory must be callable") + + def correct_assignment_category( + self, + *, + command: AssignmentCorrectionMutationCommand, + authorization: AuthorizationDecision, + ) -> AssignmentCorrectionMutationResult: + """Lock, revalidate, replace, link, audit, and bind replay evidence atomically.""" + if type(command) is not AssignmentCorrectionMutationCommand: + raise TypeError("command must be an exact AssignmentCorrectionMutationCommand") + decision = _require_correction_authorization( + authorization=authorization, + command=command, + ) + with self.connection_factory() as connection: + with connection.cursor() as cursor: + cursor.execute(_READ_WRITE_SQL) + cursor.execute(_TENANT_CONTEXT_SQL, (str(command.tenant_record_id),)) + replayed = _replayed_correction( + cursor, + command=command, + authorization=decision, + ) + if replayed is not None: + return replayed + + cursor.execute( + _READ_PREDECESSOR_SQL, + (command.tenant_record_id, command.predecessor_assignment_record_id), + ) + candidate = _require_one_predecessor( + command.tenant_record_id, + cursor.fetchmany(2), + ) + + cursor.execute( + _LOCK_EMPLOYMENT_VERSIONS_SQL, + (command.tenant_record_id, candidate.employment_record_id), + ) + employment_versions = [ + _employment_version_from_row(command.tenant_record_id, row) + for row in cursor.fetchall() + ] + cursor.execute( + _LOCK_POSITION_VERSIONS_SQL, + (command.tenant_record_id, candidate.position_record_id), + ) + position_versions = [ + _position_version_from_row(command.tenant_record_id, row) + for row in cursor.fetchall() + ] + + cursor.execute( + _LOCK_ASSIGNMENT_PORTFOLIO_SQL, + ( + command.tenant_record_id, + candidate.employment_record_id, + candidate.position_record_id, + ), + ) + portfolio = [ + _assignment_from_locked_row(command.tenant_record_id, row) + for row in cursor.fetchall() + ] + predecessor = _require_locked_predecessor( + candidate=candidate, + portfolio=portfolio, + ) + recorded_at = _post_lock_recorded_at(cursor) + try: + closed, replacement, supersession = build_assignment_category_correction( + predecessor, + replacement_assignment_record_id=command.replacement_assignment_record_id, + assignment_supersession_record_id=command.assignment_supersession_record_id, + corrected_category_code=command.corrected_category_code, + recorded_at=recorded_at, + ) + other_assignments = [ + assignment + for assignment in portfolio + if assignment.assignment_record_id != predecessor.assignment_record_id + ] + validate_assignment_write( + replacement, + [*other_assignments, closed, replacement], + employment_versions, + position_versions, + known_at=recorded_at, + ) + except KernelError as error: + raise PeopleMutationIntegrityError(str(error)) from error + + cursor.execute( + _CLOSE_PREDECESSOR_SQL, + ( + recorded_at, + command.tenant_record_id, + predecessor.assignment_record_id, + ), + ) + cursor.execute( + _INSERT_REPLACEMENT_SQL, + ( + replacement.tenant_record_id, + replacement.assignment_record_id, + replacement.employment_record_id, + replacement.person_record_id, + replacement.position_record_id, + replacement.allocation_ratio, + replacement.assignment_category_code, + replacement.effective.start, + replacement.effective.end, + recorded_at, + ), + ) + cursor.execute( + _INSERT_SUPERSESSION_SQL, + ( + supersession.tenant_record_id, + supersession.assignment_supersession_record_id, + supersession.predecessor_assignment_record_id, + supersession.replacement_assignment_record_id, + supersession.recorded_at, + ), + ) + _record_audit( + cursor, + command_tenant=command.tenant_record_id, + event_id=command.audit_event_record_id, + outbox_id=command.outbox_delivery_record_id, + event=AuditOutboxEvent( + event_id=command.audit_event_record_id, + tenant_record_id=command.tenant_record_id, + source_service="people_api", + event_type="orgmetra.people.assignment_category_corrected", + resource_reference=( + f"assignment_record:{command.predecessor_assignment_record_id}" + ), + actor_reference=decision.actor_reference, + purpose_code=decision.purpose_code, + reason_code="assignment_category_corrected", + evidence_version_code=command.evidence_version_code, + result_code="assignment_category_corrected", + occurred_at=recorded_at, + high_impact=True, + confirmation_reference=command.confirmation_reference, + ), + ) + _record_correction_idempotency( + cursor, + command=command, + authorization=decision, + replacement_record_id=replacement.assignment_record_id, + ) + return AssignmentCorrectionMutationResult( + replacement_assignment_record_id=command.replacement_assignment_record_id, + assignment_supersession_record_id=command.assignment_supersession_record_id, + ) diff --git a/services/people-api/tests/test_assignment_correction_adapter_runtime_integrity.py b/services/people-api/tests/test_assignment_correction_adapter_runtime_integrity.py new file mode 100644 index 000000000..780e61e34 --- /dev/null +++ b/services/people-api/tests/test_assignment_correction_adapter_runtime_integrity.py @@ -0,0 +1,118 @@ +"""Regression contract for Assignment correction adapter runtime evidence types.""" + +from __future__ import annotations + +import unittest + +from orgmetra_keyverse_adapter import AuthorizationDecision +from orgmetra_people_api.assignment_correction_mutations import ( + AssignmentCorrectionMutationCommand, + assignment_correction_command_digest, +) +from orgmetra_people_api.mutations import PeopleMutationIntegrityError +from orgmetra_people_api.postgres_assignment_corrections import PostgresAssignmentCorrectionMutationPort +from test_assignment_correction_mutations import ForgedCorrectionCommand, correction_command +from test_postgres_assignment_corrections import correction_authorization + + +class ForgedAuthorizationDecision(AuthorizationDecision): + """Represent caller-defined authorization evidence at the persistence boundary.""" + + +def forged_command() -> AssignmentCorrectionMutationCommand: + """Return a valid-value command whose runtime type is caller-controlled.""" + command = correction_command() + return ForgedCorrectionCommand( + tenant_record_id=command.tenant_record_id, + predecessor_assignment_record_id=command.predecessor_assignment_record_id, + replacement_assignment_record_id=command.replacement_assignment_record_id, + assignment_supersession_record_id=command.assignment_supersession_record_id, + audit_event_record_id=command.audit_event_record_id, + outbox_delivery_record_id=command.outbox_delivery_record_id, + corrected_category_code=command.corrected_category_code, + confirmation_reference=command.confirmation_reference, + evidence_version_code=command.evidence_version_code, + idempotency_key=command.idempotency_key, + ) + + +def forged_authorization() -> AuthorizationDecision: + """Return valid-value allow evidence whose runtime type is caller-controlled.""" + decision = correction_authorization() + return ForgedAuthorizationDecision( + allowed=decision.allowed, + tenant_record_id=decision.tenant_record_id, + actor_reference=decision.actor_reference, + resource_reference=decision.resource_reference, + policy_version_code=decision.policy_version_code, + purpose_code=decision.purpose_code, + operation_code=decision.operation_code, + resource_kind=decision.resource_kind, + requested_fields=decision.requested_fields, + authorized_fields=decision.authorized_fields, + reason_code=decision.reason_code, + next_action=decision.next_action, + ) + + +class ExplodingConnectionFactory: + """Prove malformed runtime evidence is rejected before database access.""" + + def __init__(self) -> None: + """Start with no attempted connection.""" + self.calls = 0 + + def __call__(self) -> object: + """Fail if the persistence boundary reaches the database factory.""" + self.calls += 1 + raise AssertionError("database connection must not be opened") + + +class AssignmentCorrectionAdapterRuntimeIntegrityTests(unittest.TestCase): + """Reject caller-defined command and authorization subtypes before replay or I/O.""" + + def test_digest_rejects_command_subtype(self) -> None: + """Do not hash semantic fields through a caller-defined command runtime type.""" + with self.assertRaisesRegex(TypeError, "exact AssignmentCorrectionMutationCommand"): + assignment_correction_command_digest( + command=forged_command(), + authorization=correction_authorization(), + ) + + def test_digest_rejects_authorization_subtype(self) -> None: + """Do not hash actor or purpose fields through caller-defined authorization evidence.""" + with self.assertRaisesRegex(TypeError, "exact AuthorizationDecision"): + assignment_correction_command_digest( + command=correction_command(), + authorization=forged_authorization(), + ) + + def test_postgres_port_rejects_command_subtype_before_connection(self) -> None: + """Reject a command subtype before transaction or tenant context setup.""" + factory = ExplodingConnectionFactory() + port = PostgresAssignmentCorrectionMutationPort(factory) + + with self.assertRaisesRegex(TypeError, "exact AssignmentCorrectionMutationCommand"): + port.correct_assignment_category( + command=forged_command(), + authorization=correction_authorization(), + ) + + self.assertEqual(factory.calls, 0) + + def test_postgres_port_rejects_authorization_subtype_before_connection(self) -> None: + """Reject forged allow evidence before transaction or tenant context setup.""" + factory = ExplodingConnectionFactory() + port = PostgresAssignmentCorrectionMutationPort(factory) + + with self.assertRaisesRegex(PeopleMutationIntegrityError, "exact authorization decision"): + port.correct_assignment_category( + command=correction_command(), + authorization=forged_authorization(), + ) + + self.assertEqual(factory.calls, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/services/people-api/tests/test_assignment_correction_http.py b/services/people-api/tests/test_assignment_correction_http.py new file mode 100644 index 000000000..ed4c89637 --- /dev/null +++ b/services/people-api/tests/test_assignment_correction_http.py @@ -0,0 +1,383 @@ +"""Executable HTTP and service-OpenAPI contracts for Assignment category correction.""" + +from __future__ import annotations + +import json +from pathlib import Path +import unittest +from uuid import UUID + +from orgmetra_hris_kernel import KernelError +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy +from orgmetra_people_api import AuthenticatedPrincipal, AuthenticationFailed +from orgmetra_people_api.assignment_correction_http import ( + AssignmentCorrectionAsgiApp, + _correction_command, + _predecessor_from_path, + _require_body_string, +) +from orgmetra_people_api.assignment_correction_mutations import ( + AssignmentCorrectionMutationCommand, + AssignmentCorrectionMutationResult, +) +from orgmetra_people_api.hire_http import _InvalidHttpRequest +from orgmetra_people_api.mutations import PeopleMutationIntegrityError + +TENANT = UUID("0198a412-8100-7000-8000-000000000001") +OTHER_TENANT = UUID("0198a412-8100-7000-8000-000000000002") +PREDECESSOR = UUID("0198a412-8100-7000-8000-000000000070") +REPLACEMENT = UUID("0198a412-8100-7000-8000-000000000071") +SUPERSESSION = UUID("0198a412-8100-7000-8000-000000000072") +AUDIT = UUID("0198a412-8100-7000-8000-000000000073") +OUTBOX = UUID("0198a412-8100-7000-8000-000000000074") +IDS = (REPLACEMENT, SUPERSESSION, AUDIT, OUTBOX) + + +class SequentialIdFactory: + """Return deterministic operational UUIDs for one correction request.""" + + def __init__(self, values: tuple[UUID, ...] = IDS) -> None: + """Initialize the request-local UUID sequence.""" + self.values = iter(values) + + def __call__(self) -> UUID: + """Return the next deterministic operational UUID.""" + return next(self.values) + + +class FakeAuthenticator: + """Return one configured principal or error while recording bearer-token use.""" + + def __init__(self, principal: object, *, error: Exception | None = None) -> None: + """Configure the authentication result and optional backend failure.""" + self.principal = principal + self.error = error + self.tokens: list[str] = [] + + async def authenticate(self, bearer_token: str) -> object: + """Record the bearer token and return the configured authentication result.""" + self.tokens.append(bearer_token) + if self.error is not None: + raise self.error + return self.principal + + +class RecordingCorrectionPort: + """Capture authorized corrections or raise a configured persistence error.""" + + def __init__(self, *, error: Exception | None = None) -> None: + """Initialize an empty correction ledger and optional persistence failure.""" + self.error = error + self.calls: list[tuple[AssignmentCorrectionMutationCommand, object]] = [] + + def correct_assignment_category( + self, + *, + command: AssignmentCorrectionMutationCommand, + authorization: object, + ) -> AssignmentCorrectionMutationResult: + """Record one correction call and return its governed identities.""" + self.calls.append((command, authorization)) + if self.error is not None: + raise self.error + return AssignmentCorrectionMutationResult( + replacement_assignment_record_id=command.replacement_assignment_record_id, + assignment_supersession_record_id=command.assignment_supersession_record_id, + ) + + +class AssignmentCorrectionHttpTests(unittest.IsolatedAsyncioTestCase): + """Prove the buyer-facing correction route is narrow, purpose-bound, and fail-closed.""" + + def setUp(self) -> None: + """Build one authorized tenant principal and correction policy per test.""" + self.principal = AuthenticatedPrincipal( + tenant_record_id=TENANT, + actor_reference="keyverse_subject:operator-17", + granted_scope_codes=frozenset({"orgmetra.people.write"}), + ) + self.policy = self._policy() + + def _policy(self, *, purpose: str = "workforce_admin") -> PurposeBoundAccessPolicy: + """Return the exact field-scoped correction policy for the requested purpose.""" + return PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="assignment-correction-v1", + resource_kind="assignment_record", + purpose_code=purpose, + operation_code="correct_record", + required_scope_code="orgmetra.people.write", + permitted_fields=frozenset({"assignment_category_code"}), + ) + + def _headers( + self, + *, + tenant: UUID = TENANT, + actor: str = "keyverse_subject:operator-17", + content_type: bytes = b"application/json", + idempotency: bool = True, + ) -> list[tuple[bytes, bytes]]: + """Build one governed correction request header set.""" + headers = [ + (b"authorization", b"Bearer opaque-token"), + (b"content-type", content_type), + (b"x-tenant-reference", str(tenant).encode("ascii")), + (b"x-actor-reference", actor.encode("ascii")), + (b"x-purpose-code", b"workforce_admin"), + ] + if idempotency: + headers.append((b"idempotency-key", b"assignment-correction-17")) + return headers + + def _app( + self, + *, + authenticator: object | None = None, + policy: object | None = None, + port: object | None = None, + id_factory: object | None = None, + ) -> AssignmentCorrectionAsgiApp: + """Build the correction ASGI app with optional boundary doubles.""" + return AssignmentCorrectionAsgiApp( + authenticator=authenticator if authenticator is not None else FakeAuthenticator(self.principal), + correction_policy=policy if policy is not None else self.policy, + mutation_port=port if port is not None else RecordingCorrectionPort(), + id_factory=id_factory if id_factory is not None else SequentialIdFactory(), + ) + + async def _request( + self, + app: AssignmentCorrectionAsgiApp, + *, + method: str = "POST", + path: object | None = None, + headers: object | None = None, + body: object | None = None, + ) -> tuple[int, dict[bytes, bytes], dict[str, object]]: + """Execute one in-memory ASGI correction request and decode its response.""" + payload = { + "corrected_category_code": "concurrent_secondary", + "confirmation_reference": "human_confirmation:review-42", + "evidence_version_code": "assignment-correction-v1", + } + messages: list[dict[str, object]] = [] + + async def receive() -> dict[str, object]: + """Return one bounded ASGI request body frame.""" + return { + "type": "http.request", + "body": body if body is not None else json.dumps(payload).encode("utf-8"), + "more_body": False, + } + + async def send(message: dict[str, object]) -> None: + """Capture one ASGI response frame for assertions.""" + messages.append(message) + + await app( + { + "type": "http", + "method": method, + "path": path if path is not None else f"/v1/assignment-records/{PREDECESSOR}/category-corrections", + "query_string": b"", + "headers": headers if headers is not None else self._headers(), + }, + receive, + send, + ) + start, response = messages + return int(start["status"]), dict(start["headers"]), json.loads(bytes(response["body"])) + + def test_path_body_and_command_helpers_fail_closed(self) -> None: + """Reject malformed routes and command bodies before governed service execution.""" + for path in ( + object(), + "/v1/assignment-records", + f"/v2/assignment-records/{PREDECESSOR}/category-corrections", + f"/v1/other-records/{PREDECESSOR}/category-corrections", + f"/v1/assignment-records/{PREDECESSOR}/other", + "/v1/assignment-records/not-a-uuid/category-corrections", + f"/v1/assignment-records/{UUID(int=0)}/category-corrections", + f"/v1/assignment-records/{UUID(int=(1 << 128) - 1)}/category-corrections", + ): + self.assertIsNone(_predecessor_from_path(path)) + self.assertEqual( + _predecessor_from_path(f"/v1/assignment-records/{PREDECESSOR}/category-corrections"), + PREDECESSOR, + ) + self.assertEqual(_require_body_string({"field": "value"}, "field"), "value") + with self.assertRaises(_InvalidHttpRequest): + _require_body_string({"field": 1}, "field") + with self.assertRaises(_InvalidHttpRequest): + _correction_command( + tenant_record_id=TENANT, + predecessor_assignment_record_id=PREDECESSOR, + payload={"corrected_category_code": "primary"}, + idempotency_key="assignment-correction-17", + id_factory=SequentialIdFactory(), + ) + command = _correction_command( + tenant_record_id=TENANT, + predecessor_assignment_record_id=PREDECESSOR, + payload={ + "corrected_category_code": "primary", + "confirmation_reference": "human_confirmation:review-42", + "evidence_version_code": "assignment-correction-v1", + }, + idempotency_key="assignment-correction-17", + id_factory=SequentialIdFactory(), + ) + self.assertEqual(command.replacement_assignment_record_id, REPLACEMENT) + + def test_constructor_requires_every_governed_dependency(self) -> None: + """Reject missing or untyped authentication, policy, persistence, and ID dependencies.""" + with self.assertRaisesRegex(TypeError, "authenticator"): + self._app(authenticator=object()) + with self.assertRaisesRegex(TypeError, "correction_policy"): + self._app(policy=object()) + with self.assertRaisesRegex(TypeError, "mutation_port"): + self._app(port=object()) + with self.assertRaisesRegex(TypeError, "id_factory"): + AssignmentCorrectionAsgiApp( + authenticator=FakeAuthenticator(self.principal), + correction_policy=self.policy, + mutation_port=RecordingCorrectionPort(), + id_factory=None, # type: ignore[arg-type] + ) + + async def test_non_http_scope_is_rejected_as_programming_error(self) -> None: + """Reject non-HTTP ASGI scopes instead of interpreting them as correction traffic.""" + async def receive() -> dict[str, object]: + """Return an unused request frame for the non-HTTP scope regression.""" + return {"type": "http.request", "body": b"{}", "more_body": False} + + async def send(message: dict[str, object]) -> None: + """Discard the response because non-HTTP scope handling must raise first.""" + del message + + with self.assertRaisesRegex(ValueError, "only HTTP"): + await self._app()({"type": "websocket"}, receive, send) + + async def test_post_creates_linked_replacement_and_authorizes_only_category(self) -> None: + """Return linked correction identities after category-only authorization succeeds.""" + authenticator = FakeAuthenticator(self.principal) + port = RecordingCorrectionPort() + status, headers, payload = await self._request(self._app(authenticator=authenticator, port=port)) + self.assertEqual(status, 201) + self.assertEqual( + payload, + { + "assignment_supersession_record_id": str(SUPERSESSION), + "replacement_assignment_record_id": str(REPLACEMENT), + }, + ) + self.assertEqual(headers[b"location"], f"/v1/assignment-records/{REPLACEMENT}".encode("ascii")) + self.assertEqual(headers[b"cache-control"], b"no-store") + self.assertEqual(authenticator.tokens, ["opaque-token"]) + command, authorization = port.calls[0] + self.assertEqual(command.predecessor_assignment_record_id, PREDECESSOR) + self.assertEqual(command.corrected_category_code, "concurrent_secondary") + self.assertEqual(command.idempotency_key, "assignment-correction-17") + self.assertEqual(authorization.operation_code, "correct_record") + self.assertEqual(authorization.requested_fields, frozenset({"assignment_category_code"})) + + async def test_request_edge_rejections_stop_before_authentication(self) -> None: + """Reject method, route, media-type, and header errors without invoking identity.""" + authenticator = FakeAuthenticator(self.principal) + app = self._app(authenticator=authenticator) + status, headers, payload = await self._request(app, method="GET") + self.assertEqual((status, headers[b"allow"], payload["error_code"]), (405, b"POST", "method_not_allowed")) + status, _, payload = await self._request(app, path="/v1/assignment-records/not-a-uuid/category-corrections") + self.assertEqual((status, payload["error_code"]), (404, "route_not_found")) + status, _, payload = await self._request(app, headers=self._headers(content_type=b"text/plain")) + self.assertEqual((status, payload["error_code"]), (415, "unsupported_media_type")) + status, _, payload = await self._request(app, headers=self._headers(idempotency=False)) + self.assertEqual((status, payload["error_code"]), (400, "invalid_request")) + self.assertEqual(authenticator.tokens, []) + + async def test_authentication_and_principal_binding_fail_closed(self) -> None: + """Sanitize identity failures and bind actor and tenant to the authenticated principal.""" + denied = self._app(authenticator=FakeAuthenticator(self.principal, error=AuthenticationFailed("denied"))) + status, headers, payload = await self._request(denied) + self.assertEqual((status, headers[b"www-authenticate"], payload["error_code"]), (401, b"Bearer", "authentication_required")) + backend = self._app(authenticator=FakeAuthenticator(self.principal, error=RuntimeError("secret"))) + status, _, payload = await self._request(backend) + self.assertEqual((status, payload["error_code"]), (500, "internal_error")) + self.assertNotIn("secret", json.dumps(payload)) + status, _, payload = await self._request(self._app(authenticator=FakeAuthenticator(object()))) + self.assertEqual((status, payload["error_code"]), (500, "internal_error")) + status, _, payload = await self._request(self._app(), headers=self._headers(tenant=OTHER_TENANT)) + self.assertEqual((status, payload["error_code"]), (403, "access_denied")) + other_actor = AuthenticatedPrincipal( + tenant_record_id=TENANT, + actor_reference="keyverse_subject:other-actor", + granted_scope_codes=frozenset({"orgmetra.people.write"}), + ) + status, _, payload = await self._request(self._app(authenticator=FakeAuthenticator(other_actor))) + self.assertEqual((status, payload["error_code"]), (403, "access_denied")) + + async def test_body_and_identity_failures_return_bounded_client_errors(self) -> None: + """Map oversized, malformed, unsupported, and exhausted-ID requests to bounded 4xx errors.""" + app = self._app() + status, _, payload = await self._request(app, body=b"{" + (b"x" * 65536) + b"}") + self.assertEqual((status, payload["error_code"]), (413, "payload_too_large")) + status, _, payload = await self._request(app, body=b"not-json") + self.assertEqual((status, payload["error_code"]), (400, "invalid_request")) + extra = { + "corrected_category_code": "primary", + "confirmation_reference": "human_confirmation:review-42", + "evidence_version_code": "assignment-correction-v1", + "unexpected": True, + } + status, _, payload = await self._request(app, body=json.dumps(extra).encode()) + self.assertEqual((status, payload["error_code"]), (400, "invalid_request")) + invalid = { + "corrected_category_code": "secondary", + "confirmation_reference": "human_confirmation:review-42", + "evidence_version_code": "assignment-correction-v1", + } + status, _, payload = await self._request(app, body=json.dumps(invalid).encode()) + self.assertEqual((status, payload["error_code"]), (400, "invalid_request")) + status, _, payload = await self._request(self._app(id_factory=SequentialIdFactory(()))) + self.assertEqual((status, payload["error_code"]), (400, "invalid_request")) + + async def test_authorization_integrity_and_backend_failures_are_sanitized(self) -> None: + """Map policy, kernel, integrity, and backend failures without leaking internal details.""" + status, _, payload = await self._request(self._app(policy=self._policy(purpose="different_admin"))) + self.assertEqual((status, payload["error_code"]), (403, "access_denied")) + status, _, payload = await self._request( + self._app(port=RecordingCorrectionPort(error=PeopleMutationIntegrityError("conflict"))) + ) + self.assertEqual((status, payload["error_code"]), (409, "mutation_integrity_conflict")) + status, _, payload = await self._request( + self._app( + port=RecordingCorrectionPort( + error=KernelError("kernel conflict", next_action="refresh the Assignment") + ) + ) + ) + self.assertEqual((status, payload["error_code"]), (409, "mutation_integrity_conflict")) + status, _, payload = await self._request( + self._app(port=RecordingCorrectionPort(error=RuntimeError("database-secret"))) + ) + self.assertEqual((status, payload["error_code"]), (500, "internal_error")) + self.assertNotIn("database-secret", json.dumps(payload)) + + def test_service_openapi_publishes_exact_correction_contract(self) -> None: + """Publish the correction route, scopes, headers, vocabulary, result, and error statuses.""" + schema = (Path(__file__).parents[1] / "assignment-correction.openapi.yaml").read_text(encoding="utf-8") + self.assertIn("/assignment-records/{assignment_record_id}/category-corrections:", schema) + self.assertIn("operationId: correctAssignmentRecordCategory", schema) + self.assertIn("- orgmetra.people.write", schema) + for header in ("Idempotency-Key", "X-Tenant-Reference", "X-Actor-Reference", "X-Purpose-Code"): + self.assertIn(f"name: {header}", schema) + self.assertIn("enum: [primary, concurrent_secondary]", schema) + self.assertIn("replacement_assignment_record_id", schema) + self.assertIn("assignment_supersession_record_id", schema) + for response in ("'400':", "'401':", "'403':", "'404':", "'405':", "'409':", "'413':", "'415':", "'500':"): + self.assertIn(response, schema) + + +if __name__ == "__main__": + unittest.main() diff --git a/services/people-api/tests/test_assignment_correction_mutations.py b/services/people-api/tests/test_assignment_correction_mutations.py new file mode 100644 index 000000000..bccaf7941 --- /dev/null +++ b/services/people-api/tests/test_assignment_correction_mutations.py @@ -0,0 +1,275 @@ +"""Executable contract for purpose-bound Assignment category correction commands.""" + +from __future__ import annotations + +import unittest +from uuid import UUID + +from orgmetra_keyverse_adapter import AuthorizationDeniedError, PurposeBoundAccessPolicy +from orgmetra_people_api.assignment_correction_mutations import ( + AssignmentCorrectionMutationCommand, + AssignmentCorrectionMutationPort, + AssignmentCorrectionMutationResult, + assignment_correction_command_digest, + correct_assignment_record_category, +) +from orgmetra_people_api.auth import AuthenticatedPrincipal + +TENANT = UUID("0198a412-8000-7000-8000-000000000001") +PREDECESSOR = UUID("0198a412-8000-7000-8000-000000000070") +REPLACEMENT = UUID("0198a412-8000-7000-8000-000000000071") +SUPERSESSION = UUID("0198a412-8000-7000-8000-000000000072") +AUDIT_EVENT = UUID("0198a412-8000-7000-8000-000000000080") +OUTBOX = UUID("0198a412-8000-7000-8000-000000000081") +CONFIRMATION = "human_confirmation:assignment-category-review-88" +EVIDENCE = "assignment_category_review:v1" +IDEMPOTENCY = "assignment-correction-17xx" + + +class ForgedUUID(UUID): + """Represent executable UUID behavior at the application trust boundary.""" + + +class ForgedString(str): + """Represent executable string behavior at the application trust boundary.""" + + +class ForgedCorrectionCommand(AssignmentCorrectionMutationCommand): + """Represent caller-defined command subtype behavior at the service boundary.""" + + +def correction_command(**overrides: object) -> AssignmentCorrectionMutationCommand: + """Build one deterministic category-correction command.""" + values: dict[str, object] = { + "tenant_record_id": TENANT, + "predecessor_assignment_record_id": PREDECESSOR, + "replacement_assignment_record_id": REPLACEMENT, + "assignment_supersession_record_id": SUPERSESSION, + "audit_event_record_id": AUDIT_EVENT, + "outbox_delivery_record_id": OUTBOX, + "corrected_category_code": "concurrent_secondary", + "confirmation_reference": CONFIRMATION, + "evidence_version_code": EVIDENCE, + "idempotency_key": IDEMPOTENCY, + } + values.update(overrides) + return AssignmentCorrectionMutationCommand(**values) # type: ignore[arg-type] + + +def correction_policy(*, purpose_code: str = "workforce_admin") -> PurposeBoundAccessPolicy: + """Return the exact purpose-bound correction policy.""" + return PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="assignment-correction-v1", + resource_kind="assignment_record", + purpose_code=purpose_code, + operation_code="correct_record", + required_scope_code="orgmetra.people.write", + permitted_fields=frozenset({"assignment_category_code"}), + ) + + +PRINCIPAL = AuthenticatedPrincipal( + tenant_record_id=TENANT, + actor_reference="keyverse_subject:operator-17", + granted_scope_codes=frozenset({"orgmetra.people.write"}), +) + + +class RecordingCorrectionPort: + """Capture the exact authorized command without persisting HRIS truth.""" + + def __init__(self) -> None: + """Initialize an empty authorized-call ledger.""" + self.calls: list[tuple[AssignmentCorrectionMutationCommand, object]] = [] + + def correct_assignment_category( + self, + *, + command: AssignmentCorrectionMutationCommand, + authorization: object, + ) -> AssignmentCorrectionMutationResult: + """Record one authorized correction call.""" + self.calls.append((command, authorization)) + return AssignmentCorrectionMutationResult( + replacement_assignment_record_id=command.replacement_assignment_record_id, + assignment_supersession_record_id=command.assignment_supersession_record_id, + ) + + +class InvalidResultPort(RecordingCorrectionPort): + """Return an invalid adapter result for service-boundary regression.""" + + def correct_assignment_category( + self, + *, + command: AssignmentCorrectionMutationCommand, + authorization: object, + ) -> object: + """Return a value outside the governed port contract.""" + del command, authorization + return object() + + +class AssignmentCorrectionMutationTests(unittest.TestCase): + """Prove correction authority, evidence, identity, and replay semantics.""" + + def test_authorizes_exact_predecessor_category_field_before_persistence(self) -> None: + """Authorize only the predecessor Assignment category before invoking persistence.""" + port = RecordingCorrectionPort() + result = correct_assignment_record_category( + principal=PRINCIPAL, + command=correction_command(), + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=port, + ) + + self.assertIsInstance(port, AssignmentCorrectionMutationPort) + self.assertEqual(result.replacement_assignment_record_id, REPLACEMENT) + self.assertEqual(result.assignment_supersession_record_id, SUPERSESSION) + authorization = port.calls[0][1] + self.assertEqual(authorization.resource_reference, f"assignment_record:{PREDECESSOR.hex}") + self.assertEqual(authorization.operation_code, "correct_record") + self.assertEqual(authorization.requested_fields, frozenset({"assignment_category_code"})) + self.assertEqual(authorization.authorized_fields, frozenset({"assignment_category_code"})) + + def test_policy_denial_prevents_correction(self) -> None: + """Do not call persistence when purpose-bound authorization denies the correction.""" + port = RecordingCorrectionPort() + with self.assertRaises(AuthorizationDeniedError): + correct_assignment_record_category( + principal=PRINCIPAL, + command=correction_command(), + purpose_code="workforce_admin", + policy=correction_policy(purpose_code="benefits_admin"), + mutation_port=port, + ) + self.assertEqual(port.calls, []) + + def test_command_rejects_malformed_identity_category_and_evidence(self) -> None: + """Reject malformed trust-bearing command values before authorization.""" + cases = ( + lambda: correction_command(tenant_record_id=UUID(int=0)), + lambda: correction_command(predecessor_assignment_record_id=UUID(int=(1 << 128) - 1)), + lambda: correction_command(replacement_assignment_record_id=ForgedUUID(str(REPLACEMENT))), + lambda: correction_command(assignment_supersession_record_id="not-a-uuid"), + lambda: correction_command(replacement_assignment_record_id=PREDECESSOR), + lambda: correction_command(corrected_category_code="legacy_unspecified"), + lambda: correction_command(corrected_category_code=ForgedString("primary")), + lambda: correction_command(confirmation_reference="not-namespaced"), + lambda: correction_command(confirmation_reference=ForgedString(CONFIRMATION)), + lambda: correction_command(confirmation_reference="human_confirmation:" + "a" * 300), + lambda: correction_command(evidence_version_code="has space"), + lambda: correction_command(evidence_version_code=ForgedString(EVIDENCE)), + lambda: correction_command(evidence_version_code="v" * 201), + lambda: correction_command(idempotency_key="short"), + lambda: correction_command(idempotency_key=ForgedString(IDEMPOTENCY)), + lambda: AssignmentCorrectionMutationResult( + replacement_assignment_record_id=UUID(int=0), + assignment_supersession_record_id=SUPERSESSION, + ), + ) + for builder in cases: + with self.subTest(builder=builder), self.assertRaises(ValueError): + builder() + + def test_command_accepts_exact_high_impact_metadata_limits(self) -> None: + """Accept confirmation and evidence metadata exactly at their governed maxima.""" + confirmation_reference = "human_confirmation:" + "a" * 281 + evidence_version_code = "v" * 200 + + command = correction_command( + confirmation_reference=confirmation_reference, + evidence_version_code=evidence_version_code, + ) + + self.assertEqual(len(command.confirmation_reference), 300) + self.assertEqual(len(command.evidence_version_code), 200) + + def test_digest_binds_semantics_but_excludes_generated_correction_ids(self) -> None: + """Keep semantic replay stable across retry-generated correction identities.""" + port = RecordingCorrectionPort() + correct_assignment_record_category( + principal=PRINCIPAL, + command=correction_command(), + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=port, + ) + authorization = port.calls[0][1] + first = assignment_correction_command_digest( + command=correction_command(), + authorization=authorization, + ) + retried = assignment_correction_command_digest( + command=correction_command( + replacement_assignment_record_id=UUID("0198a412-8000-7000-8000-000000000091"), + assignment_supersession_record_id=UUID("0198a412-8000-7000-8000-000000000092"), + audit_event_record_id=UUID("0198a412-8000-7000-8000-000000000093"), + outbox_delivery_record_id=UUID("0198a412-8000-7000-8000-000000000094"), + ), + authorization=authorization, + ) + changed_category = assignment_correction_command_digest( + command=correction_command(corrected_category_code="primary"), + authorization=authorization, + ) + changed_confirmation = assignment_correction_command_digest( + command=correction_command(confirmation_reference="human_confirmation:assignment-category-review-89"), + authorization=authorization, + ) + self.assertEqual(first, retried) + self.assertNotEqual(first, changed_category) + self.assertNotEqual(first, changed_confirmation) + + def test_service_requires_typed_command_port_and_result(self) -> None: + """Reject ungoverned command types, ports, results, and command subtypes.""" + forged_command = ForgedCorrectionCommand( + tenant_record_id=TENANT, + predecessor_assignment_record_id=PREDECESSOR, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + audit_event_record_id=AUDIT_EVENT, + outbox_delivery_record_id=OUTBOX, + corrected_category_code="concurrent_secondary", + confirmation_reference=CONFIRMATION, + evidence_version_code=EVIDENCE, + idempotency_key=IDEMPOTENCY, + ) + with self.assertRaisesRegex(TypeError, "exact AssignmentCorrectionMutationCommand"): + correct_assignment_record_category( + principal=PRINCIPAL, + command=forged_command, + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=RecordingCorrectionPort(), + ) + with self.assertRaisesRegex(TypeError, "AssignmentCorrectionMutationCommand"): + correct_assignment_record_category( + principal=PRINCIPAL, + command=object(), # type: ignore[arg-type] + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=RecordingCorrectionPort(), + ) + with self.assertRaisesRegex(TypeError, "AssignmentCorrectionMutationPort"): + correct_assignment_record_category( + principal=PRINCIPAL, + command=correction_command(), + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=object(), # type: ignore[arg-type] + ) + with self.assertRaisesRegex(TypeError, "AssignmentCorrectionMutationResult"): + correct_assignment_record_category( + principal=PRINCIPAL, + command=correction_command(), + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=InvalidResultPort(), # type: ignore[arg-type] + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/services/people-api/tests/test_assignment_correction_openapi.py b/services/people-api/tests/test_assignment_correction_openapi.py new file mode 100644 index 000000000..21af36af0 --- /dev/null +++ b/services/people-api/tests/test_assignment_correction_openapi.py @@ -0,0 +1,35 @@ +"""Cross-check the Assignment correction OpenAPI with the shared People boundary.""" + +from __future__ import annotations + +from pathlib import Path + + +def _schema_text() -> str: + """Read the published correction schema from the service root.""" + return (Path(__file__).parents[1] / "assignment-correction.openapi.yaml").read_text(encoding="utf-8") + + +def test_correction_openapi_matches_the_shared_mutation_error_envelope() -> None: + """Keep the closed service schema identical to ``mutation_http._send_error`` output.""" + schema = _schema_text() + assert "required: [error_code, message, next_action, support_reference]" in schema + error_block = schema.split(" ErrorResponse:\n", 1)[1].split(" responses:\n", 1)[0] + assert " error_code:\n type: string" in error_block + assert " message:\n type: string" in error_block + assert " next_action:\n type: string" in error_block + assert " support_reference:\n type: string" in error_block + assert " error:\n" not in error_block + + +def test_correction_openapi_bounds_high_impact_evidence_metadata() -> None: + """Published correction metadata limits must match the shared People write contract.""" + schema = _schema_text() + confirmation_block = schema.split(" confirmation_reference:\n", 1)[1].split( + " evidence_version_code:\n", 1 + )[0] + evidence_block = schema.split(" evidence_version_code:\n", 1)[1].split( + " AssignmentCategoryCorrectionResult:\n", 1 + )[0] + assert " maxLength: 300" in confirmation_block + assert " maxLength: 200" in evidence_block diff --git a/services/people-api/tests/test_assignment_correction_openapi_structure.py b/services/people-api/tests/test_assignment_correction_openapi_structure.py new file mode 100644 index 000000000..73759389c --- /dev/null +++ b/services/people-api/tests/test_assignment_correction_openapi_structure.py @@ -0,0 +1,126 @@ +"""Operation-scoped OpenAPI regressions for Assignment category correction.""" + +from __future__ import annotations + +from pathlib import Path +import unittest + + +_ROUTE = "/assignment-records/{assignment_record_id}/category-corrections" +_OPENAPI_PATH = Path(__file__).parents[1] / "assignment-correction.openapi.yaml" +_REQUIRED_HEADERS = ( + "Idempotency-Key", + "X-Tenant-Reference", + "X-Actor-Reference", + "X-Purpose-Code", +) +_ERROR_STATUSES = ("400", "401", "403", "404", "405", "409", "413", "415", "500") + + +def _mapping_block(document: str, *, key: str, indent: int) -> str: + """Return one exact YAML mapping block without accepting a sibling key as evidence.""" + marker = f"{' ' * indent}{key}:" + lines = document.splitlines() + starts = [index for index, line in enumerate(lines) if line == marker] + if len(starts) != 1: + raise AssertionError( + f"expected exactly one YAML key {key!r} at indent {indent}, found {len(starts)}" + ) + + start = starts[0] + end = len(lines) + for index in range(start + 1, len(lines)): + line = lines[index] + if not line.strip(): + continue + current_indent = len(line) - len(line.lstrip(" ")) + if current_indent <= indent: + end = index + break + return "\n".join(lines[start:end]) + + +class AssignmentCorrectionOpenApiStructureTests(unittest.TestCase): + """Keep correction evidence attached to the exact published POST operation.""" + + def setUp(self) -> None: + """Read the dedicated service contract from its repository-owned path.""" + self.schema = _OPENAPI_PATH.read_text(encoding="utf-8") + + def _post_operation(self, document: str | None = None) -> str: + """Resolve only the owned category-correction POST operation.""" + schema = self.schema if document is None else document + paths = _mapping_block(schema, key="paths", indent=0) + route = _mapping_block(paths, key=_ROUTE, indent=2) + return _mapping_block(route, key="post", indent=4) + + def test_service_openapi_binds_the_exact_correction_contract_to_post(self) -> None: + """Bind operation ID, authority, input, output, and statuses to the POST operation.""" + post = self._post_operation() + security = _mapping_block(post, key="security", indent=6) + parameters = _mapping_block(post, key="parameters", indent=6) + request_body = _mapping_block(post, key="requestBody", indent=6) + responses = _mapping_block(post, key="responses", indent=6) + created = _mapping_block(responses, key="'201'", indent=8) + + self.assertIn(" operationId: correctAssignmentRecordCategory", post) + self.assertIn( + " - keyverse_oidc:\n - orgmetra.people.write", + security, + ) + for header in _REQUIRED_HEADERS: + self.assertIn( + f" - name: {header}\n in: header\n required: true", + parameters, + ) + self.assertIn(" required: true", request_body) + self.assertIn( + "$ref: '#/components/schemas/AssignmentCategoryCorrectionCommand'", + request_body, + ) + self.assertIn( + "$ref: '#/components/schemas/AssignmentCategoryCorrectionResult'", + created, + ) + for status in _ERROR_STATUSES: + self.assertIn(f" '{status}':", responses) + + components = _mapping_block(self.schema, key="components", indent=0) + schemas = _mapping_block(components, key="schemas", indent=2) + command = _mapping_block(schemas, key="AssignmentCategoryCorrectionCommand", indent=4) + result = _mapping_block(schemas, key="AssignmentCategoryCorrectionResult", indent=4) + self.assertIn(" enum: [primary, concurrent_secondary]", command) + self.assertIn(" - replacement_assignment_record_id", result) + self.assertIn(" - assignment_supersession_record_id", result) + + def test_sibling_operation_cannot_satisfy_post_operation_identity(self) -> None: + """Prove whole-file token presence cannot substitute for POST-scoped evidence.""" + moved = self.schema.replace( + " operationId: correctAssignmentRecordCategory\n", + " get:\n operationId: correctAssignmentRecordCategory\n", + 1, + ) + self.assertIn("operationId: correctAssignmentRecordCategory", moved) + self.assertNotIn("operationId: correctAssignmentRecordCategory", self._post_operation(moved)) + + def test_same_scope_on_different_security_scheme_cannot_satisfy_post_authority(self) -> None: + """Reject a substituted OIDC scheme even when the People write scope survives.""" + substituted = self.schema.replace( + " - keyverse_oidc:\n - orgmetra.people.write", + " - external_oidc:\n - orgmetra.people.write", + 1, + ) + self.assertNotEqual(substituted, self.schema) + self.assertIn("orgmetra.people.write", self._post_operation(substituted)) + + original_schema = self.schema + self.schema = substituted + try: + with self.assertRaises(AssertionError): + self.test_service_openapi_binds_the_exact_correction_contract_to_post() + finally: + self.schema = original_schema + + +if __name__ == "__main__": + unittest.main() diff --git a/services/people-api/tests/test_assignment_correction_result_runtime_integrity.py b/services/people-api/tests/test_assignment_correction_result_runtime_integrity.py new file mode 100644 index 000000000..b1c309c92 --- /dev/null +++ b/services/people-api/tests/test_assignment_correction_result_runtime_integrity.py @@ -0,0 +1,88 @@ +"""Regression contract for Assignment correction result runtime integrity.""" + +from __future__ import annotations + +import unittest +from uuid import UUID + +from orgmetra_keyverse_adapter import PurposeBoundAccessPolicy +from orgmetra_people_api.assignment_correction_mutations import ( + AssignmentCorrectionMutationCommand, + AssignmentCorrectionMutationResult, + correct_assignment_record_category, +) +from orgmetra_people_api.auth import AuthenticatedPrincipal + +TENANT = UUID("0198a412-8000-7000-8000-000000000001") +PREDECESSOR = UUID("0198a412-8000-7000-8000-000000000070") +REPLACEMENT = UUID("0198a412-8000-7000-8000-000000000071") +SUPERSESSION = UUID("0198a412-8000-7000-8000-000000000072") +AUDIT_EVENT = UUID("0198a412-8000-7000-8000-000000000080") +OUTBOX = UUID("0198a412-8000-7000-8000-000000000081") + + +class ForgedCorrectionResult(AssignmentCorrectionMutationResult): + """Represent caller-defined executable behavior at the result boundary.""" + + +class ForgedResultPort: + """Return a subtype instead of the exact governed correction result.""" + + def correct_assignment_category( + self, + *, + command: AssignmentCorrectionMutationCommand, + authorization: object, + ) -> AssignmentCorrectionMutationResult: + """Return a structurally valid but caller-defined result subtype.""" + del command, authorization + return ForgedCorrectionResult( + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + ) + + +class AssignmentCorrectionResultRuntimeIntegrityTests(unittest.TestCase): + """Keep untrusted result subtypes from crossing the application boundary.""" + + def test_service_rejects_result_subtype_after_port_call(self) -> None: + """Require the exact governed result before HTTP code can consume its fields.""" + principal = AuthenticatedPrincipal( + tenant_record_id=TENANT, + actor_reference="keyverse_subject:operator-17", + granted_scope_codes=frozenset({"orgmetra.people.write"}), + ) + policy = PurposeBoundAccessPolicy( + tenant_record_id=TENANT, + policy_version_code="assignment-correction-v1", + resource_kind="assignment_record", + purpose_code="workforce_admin", + operation_code="correct_record", + required_scope_code="orgmetra.people.write", + permitted_fields=frozenset({"assignment_category_code"}), + ) + command = AssignmentCorrectionMutationCommand( + tenant_record_id=TENANT, + predecessor_assignment_record_id=PREDECESSOR, + replacement_assignment_record_id=REPLACEMENT, + assignment_supersession_record_id=SUPERSESSION, + audit_event_record_id=AUDIT_EVENT, + outbox_delivery_record_id=OUTBOX, + corrected_category_code="concurrent_secondary", + confirmation_reference="human_confirmation:assignment-category-review-88", + evidence_version_code="assignment_category_review:v1", + idempotency_key="assignment-correction-17xx", + ) + + with self.assertRaisesRegex(TypeError, "exact AssignmentCorrectionMutationResult"): + correct_assignment_record_category( + principal=principal, + command=command, + purpose_code="workforce_admin", + policy=policy, + mutation_port=ForgedResultPort(), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/services/people-api/tests/test_postgres_assignment_corrections.py b/services/people-api/tests/test_postgres_assignment_corrections.py new file mode 100644 index 000000000..5a73f6c4e --- /dev/null +++ b/services/people-api/tests/test_postgres_assignment_corrections.py @@ -0,0 +1,270 @@ +"""Executable contract for atomic PostgreSQL Assignment category corrections.""" + +from __future__ import annotations + +from datetime import date, datetime, timezone +from decimal import Decimal +import unittest +from uuid import UUID + +from orgmetra_keyverse_adapter import AuthorizationDecision +from orgmetra_people_api.assignment_correction_mutations import ( + assignment_correction_command_digest, + correct_assignment_record_category, +) +from orgmetra_people_api.mutations import PeopleMutationIntegrityError +from orgmetra_people_api.postgres_assignment_corrections import PostgresAssignmentCorrectionMutationPort +from test_assignment_correction_mutations import ( + IDEMPOTENCY, + PREDECESSOR, + REPLACEMENT, + SUPERSESSION, + TENANT, + PRINCIPAL, + correction_command, + correction_policy, +) +from test_people_mutations import EMPLOYMENT, EMPLOYMENT_VERSION, PERSON, POSITION, POSITION_VERSION +from test_postgres_people_mutations import FakeConnection, ScriptedCursor + +RECORDED_START = datetime(2026, 9, 3, 0, 1, tzinfo=timezone.utc) +CORRECTED_AT = datetime(2026, 9, 3, 0, 2, tzinfo=timezone.utc) +ACTOR = "keyverse_subject:operator-17" + + +def correction_authorization() -> AuthorizationDecision: + """Return the exact allow decision produced by the correction policy.""" + return AuthorizationDecision( + allowed=True, + tenant_record_id=TENANT, + actor_reference=ACTOR, + resource_reference=f"assignment_record:{PREDECESSOR.hex}", + policy_version_code="assignment-correction-v1", + purpose_code="workforce_admin", + operation_code="correct_record", + resource_kind="assignment_record", + requested_fields=frozenset({"assignment_category_code"}), + authorized_fields=frozenset({"assignment_category_code"}), + reason_code="access_permitted", + next_action="continue", + ) + + +def predecessor_row() -> tuple[object, ...]: + """Return one recorded-open primary Assignment eligible for correction.""" + return ( + PREDECESSOR, + EMPLOYMENT, + PERSON, + POSITION, + Decimal("0.5000"), + "primary", + date(2026, 8, 1), + None, + RECORDED_START, + None, + ) + + +def employment_row() -> tuple[object, ...]: + """Return one active Employment version covering the Assignment effective time.""" + return ( + EMPLOYMENT, + EMPLOYMENT_VERSION, + PERSON, + "active", + "exclusive", + date(2026, 8, 1), + None, + RECORDED_START, + None, + ) + + +def position_row() -> tuple[object, ...]: + """Return one open Position version covering the Assignment effective time.""" + return ( + POSITION, + POSITION_VERSION, + "open", + date(2026, 8, 1), + None, + RECORDED_START, + None, + ) + + +class PostgresAssignmentCorrectionMutationTests(unittest.TestCase): + """Prove locking, revalidation, replay, provenance, audit, and rollback boundaries.""" + + def test_correction_locks_revalidates_and_commits_linked_evidence_atomically(self) -> None: + cursor = ScriptedCursor( + [[], [predecessor_row()]], + [[employment_row()], [position_row()], [predecessor_row()]], + clock_timestamp=CORRECTED_AT, + ) + connection = FakeConnection(cursor) + port = PostgresAssignmentCorrectionMutationPort(lambda: connection) + + result = correct_assignment_record_category( + principal=PRINCIPAL, + command=correction_command(), + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=port, + ) + + self.assertEqual(result.replacement_assignment_record_id, REPLACEMENT) + self.assertEqual(result.assignment_supersession_record_id, SUPERSESSION) + sql = [statement for statement, _parameters in cursor.executions] + predecessor_read = next( + i + for i, statement in enumerate(sql) + if "assignment.assignment_record_id = %s" in statement and "LIMIT 2" in statement + ) + employment_lock = next(i for i, statement in enumerate(sql) if "FOR UPDATE OF employment" in statement) + position_lock = next(i for i, statement in enumerate(sql) if "FOR UPDATE OF position" in statement) + portfolio_lock = next( + i + for i, statement in enumerate(sql) + if "OR assignment.position_record_id = %s" in statement + ) + clock_read = sql.index("SELECT pg_catalog.clock_timestamp()") + close_write = next(i for i, statement in enumerate(sql) if statement.startswith("UPDATE public.assignment_record")) + replacement_write = next( + i + for i, statement in enumerate(sql) + if statement.startswith("INSERT INTO public.assignment_record (") + ) + supersession_write = next( + i + for i, statement in enumerate(sql) + if statement.startswith("INSERT INTO public.assignment_supersession_record") + ) + audit_write = next(i for i, statement in enumerate(sql) if "record_audit_outbox_event" in statement) + replay_write = next( + i + for i, statement in enumerate(sql) + if statement.startswith("INSERT INTO public.people_mutation_idempotency_record") + ) + self.assertNotIn("FOR UPDATE", sql[predecessor_read]) + self.assertIn("ORDER BY assignment.assignment_record_id", sql[portfolio_lock]) + self.assertLess(predecessor_read, employment_lock) + self.assertLess(employment_lock, position_lock) + self.assertLess(position_lock, portfolio_lock) + self.assertLess(portfolio_lock, clock_read) + self.assertLess(clock_read, close_write) + self.assertLess(close_write, replacement_write) + self.assertLess(replacement_write, supersession_write) + self.assertLess(supersession_write, audit_write) + self.assertLess(audit_write, replay_write) + self.assertEqual(cursor.fetchall_rows, []) + close_parameters = cursor.executions[close_write][1] + assert close_parameters is not None + self.assertEqual(close_parameters, (CORRECTED_AT, TENANT, PREDECESSOR)) + self.assertIsNone(connection.exit_exception) + + def test_matching_replay_returns_committed_replacement_and_supersession_without_new_writes(self) -> None: + digest = assignment_correction_command_digest( + command=correction_command(), + authorization=correction_authorization(), + ) + cursor = ScriptedCursor( + [[(REPLACEMENT, digest)], [(SUPERSESSION, REPLACEMENT)]], + [], + clock_timestamp=CORRECTED_AT, + ) + port = PostgresAssignmentCorrectionMutationPort(lambda: FakeConnection(cursor)) + + result = correct_assignment_record_category( + principal=PRINCIPAL, + command=correction_command( + replacement_assignment_record_id=UUID("0198a412-8000-7000-8000-000000000091"), + assignment_supersession_record_id=UUID("0198a412-8000-7000-8000-000000000092"), + audit_event_record_id=UUID("0198a412-8000-7000-8000-000000000093"), + outbox_delivery_record_id=UUID("0198a412-8000-7000-8000-000000000094"), + ), + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=port, + ) + + self.assertEqual(result.replacement_assignment_record_id, REPLACEMENT) + self.assertEqual(result.assignment_supersession_record_id, SUPERSESSION) + sql_text = "\n".join(statement for statement, _parameters in cursor.executions) + self.assertNotIn("UPDATE public.assignment_record", sql_text) + self.assertNotIn("INSERT INTO public.assignment_record (", sql_text) + self.assertNotIn("record_audit_outbox_event", sql_text) + + def test_same_key_with_changed_semantics_fails_before_authoritative_locks(self) -> None: + digest = assignment_correction_command_digest( + command=correction_command(), + authorization=correction_authorization(), + ) + cursor = ScriptedCursor([[(REPLACEMENT, digest)]], [], clock_timestamp=CORRECTED_AT) + connection = FakeConnection(cursor) + port = PostgresAssignmentCorrectionMutationPort(lambda: connection) + + with self.assertRaisesRegex(PeopleMutationIntegrityError, "different command"): + correct_assignment_record_category( + principal=PRINCIPAL, + command=correction_command( + corrected_category_code="primary", + idempotency_key=IDEMPOTENCY, + ), + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=port, + ) + + sql_text = "\n".join(statement for statement, _parameters in cursor.executions) + self.assertNotIn("FOR UPDATE OF assignment", sql_text) + self.assertNotIn("UPDATE public.assignment_record", sql_text) + self.assertIs(connection.exit_exception, PeopleMutationIntegrityError) + + def test_missing_predecessor_rolls_back_before_close_or_provenance(self) -> None: + cursor = ScriptedCursor([[], []], [], clock_timestamp=CORRECTED_AT) + connection = FakeConnection(cursor) + port = PostgresAssignmentCorrectionMutationPort(lambda: connection) + + with self.assertRaises(PeopleMutationIntegrityError): + correct_assignment_record_category( + principal=PRINCIPAL, + command=correction_command(), + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=port, + ) + + sql_text = "\n".join(statement for statement, _parameters in cursor.executions) + self.assertNotIn("UPDATE public.assignment_record", sql_text) + self.assertNotIn("INSERT INTO public.assignment_supersession_record", sql_text) + self.assertIs(connection.exit_exception, PeopleMutationIntegrityError) + + def test_failed_portfolio_revalidation_rolls_back_before_any_correction_write(self) -> None: + cursor = ScriptedCursor( + [[], [predecessor_row()]], + [[], [position_row()], [predecessor_row()]], + clock_timestamp=CORRECTED_AT, + ) + connection = FakeConnection(cursor) + port = PostgresAssignmentCorrectionMutationPort(lambda: connection) + + with self.assertRaises(PeopleMutationIntegrityError): + correct_assignment_record_category( + principal=PRINCIPAL, + command=correction_command(), + purpose_code="workforce_admin", + policy=correction_policy(), + mutation_port=port, + ) + + sql_text = "\n".join(statement for statement, _parameters in cursor.executions) + self.assertNotIn("UPDATE public.assignment_record", sql_text) + self.assertNotIn("INSERT INTO public.assignment_record (", sql_text) + self.assertNotIn("INSERT INTO public.assignment_supersession_record", sql_text) + self.assertIs(connection.exit_exception, PeopleMutationIntegrityError) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_assignment_category_correction_postgres.sh b/tests/test_assignment_category_correction_postgres.sh new file mode 100644 index 000000000..512072852 --- /dev/null +++ b/tests/test_assignment_category_correction_postgres.sh @@ -0,0 +1,243 @@ +#!/usr/bin/env bash +set -euo pipefail + +: "${DATABASE_URL:=postgresql://orgmetra:orgmetra@localhost:5432/orgmetra}" + +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0001_foundation_schema.sql +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0002_sealed_evidence_digest.sql +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0017_assignment_category_code.sql + +# A late migration failure must roll back the relation and all dependent DDL. +# Colliding with the trigger function fails after CREATE TABLE has executed, +# proving that BEGIN/COMMIT is the recovery boundary rather than psql autocommit. +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 <<'SQL' +CREATE FUNCTION public.enforce_assignment_supersession_link() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +BEGIN + RETURN NEW; +END; +$$; +SQL + +set +e +atomicity_output="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0018_assignment_category_supersession.sql 2>&1)" +atomicity_status=$? +set -e +if [[ ${atomicity_status} -eq 0 || "${atomicity_output}" != *"enforce_assignment_supersession_link"* ]]; then + echo "assignment supersession migration did not hit the deterministic late conflict: ${atomicity_output}" >&2 + exit 1 +fi + +partial_table="$(psql "${DATABASE_URL}" -Atqc "SELECT pg_catalog.to_regclass('public.assignment_supersession_record') IS NOT NULL;")" +if [[ "${partial_table}" != "f" ]]; then + echo "failed assignment supersession migration left partial schema state" >&2 + exit 1 +fi + +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -c "DROP FUNCTION public.enforce_assignment_supersession_link();" +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0018_assignment_category_supersession.sql + +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 <<'SQL' +INSERT INTO tenant_record (tenant_record_id, tenant_reference) +VALUES + ('10000000-0000-7000-8000-000000000001', 'tenant_alpha'), + ('20000000-0000-7000-8000-000000000001', 'tenant_beta'); +INSERT INTO person_record (tenant_record_id, person_record_id, recorded_from) +VALUES ('10000000-0000-7000-8000-000000000001', '10000000-0000-7000-8000-000000000101', TIMESTAMPTZ '2026-09-03 00:00:00+00'); +INSERT INTO employment_record (tenant_record_id, employment_record_id, person_record_id, recorded_from) +VALUES ('10000000-0000-7000-8000-000000000001', '10000000-0000-7000-8000-000000000111', '10000000-0000-7000-8000-000000000101', TIMESTAMPTZ '2026-09-03 00:00:00+00'); +INSERT INTO organization_unit (tenant_record_id, organization_unit_id, recorded_from) +VALUES ('10000000-0000-7000-8000-000000000001', '10000000-0000-7000-8000-000000000121', TIMESTAMPTZ '2026-09-03 00:00:00+00'); +INSERT INTO job_profile (tenant_record_id, job_profile_id, recorded_from) +VALUES ('10000000-0000-7000-8000-000000000001', '10000000-0000-7000-8000-000000000131', TIMESTAMPTZ '2026-09-03 00:00:00+00'); +INSERT INTO position_record ( + tenant_record_id, position_record_id, organization_unit_id, job_profile_id, recorded_from +) VALUES + ('10000000-0000-7000-8000-000000000001', '10000000-0000-7000-8000-000000000141', '10000000-0000-7000-8000-000000000121', '10000000-0000-7000-8000-000000000131', TIMESTAMPTZ '2026-09-03 00:00:00+00'), + ('10000000-0000-7000-8000-000000000001', '10000000-0000-7000-8000-000000000142', '10000000-0000-7000-8000-000000000121', '10000000-0000-7000-8000-000000000131', TIMESTAMPTZ '2026-09-03 00:00:00+00'); + +INSERT INTO assignment_record ( + tenant_record_id, assignment_record_id, employment_record_id, person_record_id, + position_record_id, allocation_ratio, assignment_category_code, + effective_from, effective_to, recorded_from +) VALUES ( + '10000000-0000-7000-8000-000000000001', '10000000-0000-7000-8000-000000000151', + '10000000-0000-7000-8000-000000000111', '10000000-0000-7000-8000-000000000101', + '10000000-0000-7000-8000-000000000141', 0.5000, 'primary', + DATE '2026-09-03', DATE '2026-10-01', TIMESTAMPTZ '2026-09-03 00:01:00+00' +); +UPDATE assignment_record +SET recorded_to = TIMESTAMPTZ '2026-09-03 00:02:00+00' +WHERE assignment_record_id = '10000000-0000-7000-8000-000000000151'; + +INSERT INTO assignment_record ( + tenant_record_id, assignment_record_id, employment_record_id, person_record_id, + position_record_id, allocation_ratio, assignment_category_code, + effective_from, effective_to, recorded_from +) VALUES ( + '10000000-0000-7000-8000-000000000001', '10000000-0000-7000-8000-000000000152', + '10000000-0000-7000-8000-000000000111', '10000000-0000-7000-8000-000000000101', + '10000000-0000-7000-8000-000000000141', 0.5000, 'concurrent_secondary', + DATE '2026-09-03', DATE '2026-10-01', TIMESTAMPTZ '2026-09-03 00:02:00+00' +); + +INSERT INTO assignment_supersession_record ( + tenant_record_id, + assignment_supersession_record_id, + predecessor_assignment_record_id, + replacement_assignment_record_id, + recorded_at +) VALUES ( + '10000000-0000-7000-8000-000000000001', + '10000000-0000-7000-8000-000000000190', + '10000000-0000-7000-8000-000000000151', + '10000000-0000-7000-8000-000000000152', + TIMESTAMPTZ '2026-09-03 00:02:00+00' +); +SQL + +edge_count="$(psql "${DATABASE_URL}" -Atqc "SELECT count(*) FROM assignment_supersession_record WHERE tenant_record_id='10000000-0000-7000-8000-000000000001' AND predecessor_assignment_record_id='10000000-0000-7000-8000-000000000151' AND replacement_assignment_record_id='10000000-0000-7000-8000-000000000152' AND recorded_at=TIMESTAMPTZ '2026-09-03 00:02:00+00';")" +test "${edge_count}" = "1" + +rls_state="$(psql "${DATABASE_URL}" -Atqc "SELECT relrowsecurity::text || ':' || relforcerowsecurity::text FROM pg_class WHERE oid='public.assignment_supersession_record'::regclass;")" +test "${rls_state}" = "true:true" + +policy_count="$(psql "${DATABASE_URL}" -Atqc "SELECT count(*) FROM pg_policy WHERE polrelid='public.assignment_supersession_record'::regclass AND polname='assignment_supersession_scope_policy';")" +test "${policy_count}" = "1" + +# RLS catalog flags are insufficient evidence. Prove visibility through an +# ordinary NOBYPASSRLS role with absent, matching, and non-matching tenant context. +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 <<'SQL' +CREATE ROLE orgmetra_assignment_correction_reader NOLOGIN NOBYPASSRLS; +GRANT USAGE ON SCHEMA public TO orgmetra_assignment_correction_reader; +GRANT SELECT ON public.assignment_supersession_record TO orgmetra_assignment_correction_reader; +GRANT EXECUTE ON FUNCTION public.current_tenant_record_id() TO orgmetra_assignment_correction_reader; +SET ROLE orgmetra_assignment_correction_reader; + +RESET orgmetra.tenant_record_id; +DO $$ +BEGIN + IF (SELECT count(*) FROM public.assignment_supersession_record) <> 0 THEN + RAISE EXCEPTION 'missing tenant context exposed assignment supersession provenance'; + END IF; +END; +$$; + +SET orgmetra.tenant_record_id = '10000000-0000-7000-8000-000000000001'; +DO $$ +BEGIN + IF (SELECT count(*) FROM public.assignment_supersession_record) <> 1 THEN + RAISE EXCEPTION 'tenant alpha could not read its assignment supersession provenance'; + END IF; +END; +$$; + +SET orgmetra.tenant_record_id = '20000000-0000-7000-8000-000000000001'; +DO $$ +BEGIN + IF (SELECT count(*) FROM public.assignment_supersession_record) <> 0 THEN + RAISE EXCEPTION 'tenant beta observed tenant alpha assignment supersession provenance'; + END IF; +END; +$$; +RESET ROLE; +SQL + +set +e +update_output="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -c "UPDATE assignment_supersession_record SET recorded_at=TIMESTAMPTZ '2026-09-03 00:03:00+00' WHERE assignment_supersession_record_id='10000000-0000-7000-8000-000000000190';" 2>&1)" +update_status=$? +set -e +if [[ ${update_status} -eq 0 || "${update_output}" != *"append-only"* ]]; then + echo "assignment supersession provenance was mutable: ${update_output}" >&2 + exit 1 +fi + +set +e +truncate_output="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -c "TRUNCATE assignment_supersession_record;" 2>&1)" +truncate_status=$? +set -e +if [[ ${truncate_status} -eq 0 || "${truncate_output}" != *"cannot be truncated"* ]]; then + echo "assignment supersession provenance could be truncated: ${truncate_output}" >&2 + exit 1 +fi + +set +e +sentinel_output="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -c "INSERT INTO assignment_supersession_record (tenant_record_id, assignment_supersession_record_id, predecessor_assignment_record_id, replacement_assignment_record_id, recorded_at) VALUES ('10000000-0000-7000-8000-000000000001','00000000-0000-0000-0000-000000000000','10000000-0000-7000-8000-000000000151','10000000-0000-7000-8000-000000000152',TIMESTAMPTZ '2026-09-03 00:02:00+00');" 2>&1)" +sentinel_status=$? +set -e +if [[ ${sentinel_status} -eq 0 || "${sentinel_output}" != *"assignment_supersession_record_id_operational_check"* ]]; then + echo "reserved supersession identity escaped validation: ${sentinel_output}" >&2 + exit 1 +fi + +# Trigger-level linkage validation must fail before uniqueness could otherwise +# hide a malformed second edge from the same predecessor. +set +e +time_output="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -c "INSERT INTO assignment_supersession_record (tenant_record_id, assignment_supersession_record_id, predecessor_assignment_record_id, replacement_assignment_record_id, recorded_at) VALUES ('10000000-0000-7000-8000-000000000001','10000000-0000-7000-8000-000000000191','10000000-0000-7000-8000-000000000151','10000000-0000-7000-8000-000000000152',TIMESTAMPTZ '2026-09-03 00:03:00+00');" 2>&1)" +time_status=$? +set -e +if [[ ${time_status} -eq 0 || "${time_output}" != *"recorded timestamp"* ]]; then + echo "mismatched supersession time escaped linkage validation: ${time_output}" >&2 + exit 1 +fi + +# A replacement with changed business truth is not a category correction. +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 <<'SQL' +INSERT INTO assignment_record ( + tenant_record_id, assignment_record_id, employment_record_id, person_record_id, + position_record_id, allocation_ratio, assignment_category_code, + effective_from, effective_to, recorded_from +) VALUES ( + '10000000-0000-7000-8000-000000000001', '10000000-0000-7000-8000-000000000153', + '10000000-0000-7000-8000-000000000111', '10000000-0000-7000-8000-000000000101', + '10000000-0000-7000-8000-000000000142', 0.5000, 'concurrent_secondary', + DATE '2026-09-03', DATE '2026-10-01', TIMESTAMPTZ '2026-09-03 00:02:00+00' +); +SQL + +set +e +business_output="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -c "INSERT INTO assignment_supersession_record (tenant_record_id, assignment_supersession_record_id, predecessor_assignment_record_id, replacement_assignment_record_id, recorded_at) VALUES ('10000000-0000-7000-8000-000000000001','10000000-0000-7000-8000-000000000192','10000000-0000-7000-8000-000000000151','10000000-0000-7000-8000-000000000153',TIMESTAMPTZ '2026-09-03 00:02:00+00');" 2>&1)" +business_status=$? +set -e +if [[ ${business_status} -eq 0 || "${business_output}" != *"business truth"* ]]; then + echo "non-category replacement escaped supersession validation: ${business_output}" >&2 + exit 1 +fi + +# Same-category replacement is also invalid even when every other fact matches. +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 <<'SQL' +INSERT INTO assignment_record ( + tenant_record_id, assignment_record_id, employment_record_id, person_record_id, + position_record_id, allocation_ratio, assignment_category_code, + effective_from, effective_to, recorded_from +) VALUES ( + '10000000-0000-7000-8000-000000000001', '10000000-0000-7000-8000-000000000154', + '10000000-0000-7000-8000-000000000111', '10000000-0000-7000-8000-000000000101', + '10000000-0000-7000-8000-000000000141', 0.5000, 'primary', + DATE '2026-09-03', DATE '2026-10-01', TIMESTAMPTZ '2026-09-03 00:02:00+00' +); +SQL + +set +e +category_output="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -c "INSERT INTO assignment_supersession_record (tenant_record_id, assignment_supersession_record_id, predecessor_assignment_record_id, replacement_assignment_record_id, recorded_at) VALUES ('10000000-0000-7000-8000-000000000001','10000000-0000-7000-8000-000000000194','10000000-0000-7000-8000-000000000151','10000000-0000-7000-8000-000000000154',TIMESTAMPTZ '2026-09-03 00:02:00+00');" 2>&1)" +category_status=$? +set -e +if [[ ${category_status} -eq 0 || "${category_output}" != *"must change one explicit assignment category"* ]]; then + echo "same-category replacement escaped supersession validation: ${category_output}" >&2 + exit 1 +fi + +# The normalized edge is one-to-one: a predecessor cannot fork and a replacement +# cannot claim multiple predecessors. +set +e +duplicate_output="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -c "INSERT INTO assignment_supersession_record (tenant_record_id, assignment_supersession_record_id, predecessor_assignment_record_id, replacement_assignment_record_id, recorded_at) VALUES ('10000000-0000-7000-8000-000000000001','10000000-0000-7000-8000-000000000193','10000000-0000-7000-8000-000000000151','10000000-0000-7000-8000-000000000152',TIMESTAMPTZ '2026-09-03 00:02:00+00');" 2>&1)" +duplicate_status=$? +set -e +if [[ ${duplicate_status} -eq 0 || "${duplicate_output}" != *"assignment_supersession_predecessor_unique"* ]]; then + echo "predecessor supersession fork escaped uniqueness: ${duplicate_output}" >&2 + exit 1 +fi + +echo "assignment category correction PostgreSQL provenance contract passed" diff --git a/tests/test_assignment_correction_idempotency_postgres.sh b/tests/test_assignment_correction_idempotency_postgres.sh new file mode 100644 index 000000000..667fbf6f0 --- /dev/null +++ b/tests/test_assignment_correction_idempotency_postgres.sh @@ -0,0 +1,44 @@ +#!/usr/bin/env bash +set -euo pipefail + +: "${DATABASE_URL:=postgresql://orgmetra:orgmetra@localhost:5432/orgmetra}" + +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0001_foundation_schema.sql +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0002_sealed_evidence_digest.sql +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0012_people_mutation_idempotency.sql +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0017_assignment_category_code.sql +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0018_assignment_category_supersession.sql +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -f database/migrations/0019_assignment_correction_idempotency_route.sql + +psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 <<'SQL' +INSERT INTO public.tenant_record (tenant_record_id, tenant_reference) +VALUES ('30000000-0000-7000-8000-000000000001', 'correction_idempotency_tenant'); + +INSERT INTO public.people_mutation_idempotency_record ( + tenant_record_id, + people_mutation_idempotency_record_id, + command_route, + idempotency_key, + command_digest, + created_record_id +) VALUES ( + '30000000-0000-7000-8000-000000000001', + '30000000-0000-7000-8000-000000000011', + 'assignment-category-corrections', + 'assignment-correction-17xx', + repeat('a', 64), + '30000000-0000-7000-8000-000000000012' +); +SQL + +route_count="$(psql "${DATABASE_URL}" -Atqc "SELECT count(*) FROM public.people_mutation_idempotency_record WHERE tenant_record_id='30000000-0000-7000-8000-000000000001' AND command_route='assignment-category-corrections';")" +test "${route_count}" = "1" + +set +e +unknown_output="$(psql "${DATABASE_URL}" -v ON_ERROR_STOP=1 -c "INSERT INTO public.people_mutation_idempotency_record (tenant_record_id, people_mutation_idempotency_record_id, command_route, idempotency_key, command_digest, created_record_id) VALUES ('30000000-0000-7000-8000-000000000001','30000000-0000-7000-8000-000000000013','assignment-correction-unknown','assignment-correction-18xx',repeat('b',64),'30000000-0000-7000-8000-000000000014');" 2>&1)" +unknown_status=$? +set -e +if [[ ${unknown_status} -eq 0 || "${unknown_output}" != *"people_mutation_idempotency_route_check"* ]]; then + echo "unknown People mutation route escaped the closed idempotency vocabulary: ${unknown_output}" >&2 + exit 1 +fi